一 背景
在之前我们通过使用千问的公网在线API，实现了对大模型的调用。但出于对数据安全与隐私保护、网络稳定性、定制化需求、监管合规、知识产权保护、自主可控性、业务连续性以及成本效益等多方面的考虑，在有些场景下，调用在线API的方法就不可行了，此时需要使用一些已经训练好的基模进行本地化部署。通过本地化部署，可以更好地满足自身特定需求，确保业务的合法、稳定、连续运行，并提高对模型的掌控能力。
这篇文章中，我们通过将模搭社区开源的大模型部署到本地，并实现简单的对话和RAG。
二 开发框架介绍
ModelScope Library
ModelScope Library是魔搭社区提供的一个能够快速、方便的使用社区提供的各类模型的Python library，其中包含了ModelScope官方模型的实现，以及使用这些模型进行推理，finetune等任务所需的数据预处理，后处理，效果评估等功能相关的代码，同时也提供了简单易用的API，以及丰富的使用样例。通过调用library，用户可以只写短短的几行代码，就可以完成模型的推理、训练和评估等任务，也可以在此基础上快速进行二次开发，实现自己的创新想法。本文中我们使用这个库进行模型的加载。
ModelScope Library支持的模型不光局限于huggingface的transformers架构类的模型，并且社区提供大量的中文大语言模型，更适合我们在国内下载，也方便学习及使用。
三 环境准备
1. 环境检查
本地实验环境：
系统：Win11
显卡：1070（8G显存）
首先更新显卡驱动到最新版本，可以去官网下载或者直接在NVIDIA Geforce Experience中直接更新驱动到最新版本，新版本的驱动向下兼容更多版本的CUDA。
查看显卡驱动支持的CUDA的最高版本，小于等于此版本的CUDA均可以使用。CMD或powershell中执行如下命令：
nvidia-smi
在https://pytorch.org/查看当前最新版PyTorch支持最低Python版本为3.8，支持CUDA的11.8和12.1版本，后面我们选择安装12.1版本。
最终生成的命令可以拷贝出来，下文需要使用。
2. 安装CUDA 12.1（可选）
此步骤可选，不安装的话后面Torch会自动安装
下载地址：
https://developer.nvidia.com/cuda-12-1-1-download-archive
下载完成后直接安装即可，如果已经安装需要先卸载后再装。
3. 安装conda
conda可以用来管理Python环境，后面我们会使用conda创建一个Python3.10的运行环境。
下载地址：https://www.anaconda.com/download
安装完成后，为了能在命令行中使用，需要将conda的相关目录加入环境变量，例如安装在D:\developer\anaconda，则需要将以下目录添加到PATH中：
D:\developer\anaconda D:\developer\anaconda\Scripts D:\developer\anaconda\Library\bin D:\developer\anaconda\Library\mingw-w64\bin
打开powershell，执行conda init初始化conda的powershell和cmd环境，linux下会初始化bash环境，初始化后方便进入conda创建的Python环境。
4. 使用conda创建PyTorch环境
我们使用conda创建一个Python版本为3.10的Python运行环境，在命令行中执行如下命令：
conda create -n pytorch python=3.10 conda activate pytorch
使用上文中安装PyTorch的命令安装PyTorch
conda install pytorch torchvision torchaudio pytorch-cuda=12.1 -c pytorch -c nvidia
5. 下载模型
我们可以去模搭社区获取模型，国内的地址，下载速度快，不需要魔法可以直接访问。
模型库地址：https://modelscope.cn/models
这里使用Qwen1.5-0.5B-Chat这个对话模型进行体验，模型较小，占用内存少，生成速度快。
模型地址：https://modelscope.cn/models/qwen/Qwen1.5-0.5B-Chat/summary
点击模型文件 -> 下载模型，可支持两种下载方式：Sdk和Git
我们通过git的方式将模型文件下载到本地
mkdir Qwen && cd Qwen git clone https://www.modelscope.cn/qwen/Qwen1.5-0.5B-Chat.git cd ..
四 加载模型
1. 模型功能验证
可以使用modelscope Library加载模型，使用方法与transformers相同，使用AutoModelForCausalLM.from_pretrained方法和AutoTokenizer.from_pretrained从本地文件中加载，如果路径不存在，这两个方法会自动到modelscope下载模型文件。
需要先安装modelscope库：
pip install modelscope transformers
使用量化模型的话需要安装以下库：
pip install optimum auto-gptq
创建一个Python文件，放到与上文Qwen文件夹同级的目录中，内容如下：
from threading import Thread from modelscope import (AutoModelForCausalLM, AutoTokenizer) from transformers import TextIteratorStreamer device = "cuda" # 将模型加载到哪个硬件，此处为GPU model = AutoModelForCausalLM.from_pretrained( "Qwen/Qwen1.5-0.5B-Chat", # 模型文件夹路径 device_map="auto" ) tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen1.5-0.5B-Chat") while True: user_input = input("请输入问题（q退出）：") if user_input.lower() == "q": print("exit") break try: messages = [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": user_input} ] text = tokenizer.apply_chat_template( messages, tokenize=False, add_generation_prompt=False ) inputs = tokenizer([text], return_tensors="pt").to(device) streamer = TextIteratorStreamer(tokenizer) generation_kwargs = dict(inputs, streamer=streamer, max_new_tokens=512) thread = Thread(target=model.generate, kwargs=generation_kwargs) thread.start() generated_text = "" count = 0 for new_text in streamer: generated_text += new_text print(new_text, end="", flush=True) print() except Exception as e: print(f"出错了：{str(e)}")
上面的代码首先从本地模型文件夹中加载了模型和分词器，然后我们在一个循环中接收用户输入，并将输入处理后通过大模型进行内容生成。我们可以通过python运行上面的文件，运行后，就可以测试了，就测试运行效果如下：
2. LangChain加载本地模型
到目前为止，我们已经在本地跑起来了一个千问0.5B大语言模型，接下来需要让langchain能够加载这个本地模型。
如果要用langchain加载模型，我们需要继承langchain.llms.base.LLM 类，并且重写_llm_type, _call方法，因为我们需要支持流式输出，就需要重写_stream方法。可参考langchain的官方文档：Custom LLM | 🦜️🔗 LangChain
下面是这个类的代码：
from abc import ABC from threading import Thread from typing import Any, List, Mapping, Optional, Iterator from langchain.callbacks.manager import CallbackManagerForLLMRun from langchain.llms.base import LLM from langchain_core.outputs import GenerationChunk from modelscope import AutoModelForCausalLM, AutoTokenizer from transformers import TextIteratorStreamer device = "cuda" # the device to load the model onto model = AutoModelForCausalLM.from_pretrained( "Qwen/Qwen1.5-0.5B-Chat", device_map="auto" ) tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen1.5-0.5B-Chat") class QwenLocalLLM(LLM, ABC): max_token: int = 10000 temperature: float = 0.01 top_p = 0.9 def __init__(self): super().__init__() @property def _llm_type(self) -> str: return "Qwen" def _call( self, prompt: str, stop: Optional[List[str]] = None, run_manager: Optional[CallbackManagerForLLMRun] = None, **kwargs: Any ) -> str: messages = [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": prompt} ] text = tokenizer.apply_chat_template( messages, tokenize=False, add_generation_prompt=True ) model_inputs = tokenizer([text], return_tensors="pt").to(device) generated_ids = model.generate( model_inputs.input_ids, max_new_tokens=512 ) generated_ids = [ output_ids[len(input_ids):] for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids) ] response = tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0] return response @property def _identifying_params(self) -> Mapping[str, Any]: """Get the identifying parameters.""" return {"max_token": self.max_token, "temperature": self.temperature, "top_p": self.top_p, "history_len": self.history_len} def _stream( self, prompt: str, stop: Optional[List[str]] = None, run_manager: Optional[CallbackManagerForLLMRun] = None, **kwargs: Any, ) -> Iterator[GenerationChunk]: try: messages = [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": prompt} ] text = tokenizer.apply_chat_template( messages, tokenize=False, add_generation_prompt=False ) inputs = tokenizer([text], return_tensors="pt").to(device) streamer = TextIteratorStreamer(tokenizer) generation_kwargs = dict(inputs, streamer=streamer, max_new_tokens=512) thread = Thread(target=model.generate, kwargs=generation_kwargs) thread.start() generated_text = "" for new_text in streamer: generated_text += new_text print(new_text, end="", flush=True) yield GenerationChunk( text=new_text ) print() except Exception as e: print(f"出错了：{str(e)}") yield GenerationChunk( text=f"生成失败: {str(e)}" )
五 后续方向
虽然私有化部署大模型花费的精力很多，但是大模型私有化部署的好处很多，比如：
数据安全：能够更好地保护企业的数据隐私和安全。
定制化：可以根据企业的具体需求进行定制和优化。
掌控权：企业对模型具有更大的掌控权，可以进行灵活的管理和调整。
效率提升：可针对企业特定业务流程进行优化，提高工作效率。
稳定性：减少对外部网络和服务的依赖，提高系统的稳定性。
合规性：有助于满足企业在数据隐私和安全方面的合规要求。
品牌建设：打造具有企业自身特色的人工智能解决方案，提升品牌形象。
资源优化：根据企业的实际情况进行资源分配和优化，避免不必要的浪费。
后续我们会继续学习探索。
