PythonAI算法
Windows 平台零基础部署 Qwen1.5 大模型教程
Windows 平台部署 Qwen1.5 大模型教程。涵盖 NVIDIA 驱动更新、Anaconda 环境配置、CUDA 版本匹配、PyTorch 及 Transformers 库安装。演示了通过 VSCode 运行本地模型，并基于 FastAPI 和 Uvicorn 构建支持流式输出的 RESTful API 接口。提供完整代码示例与依赖管理方案，适用于具备后端基础的开发人员快速上手本地大模型应用开发。
Windows 平台部署 Qwen1.5 大模型教程。涵盖 NVIDIA 驱动更新、Anaconda 环境配置、CUDA 版本匹配、PyTorch 及 Transformers 库安装。演示了通过 VSCode 运行本地模型，并基于 FastAPI 和 Uvicorn 构建支持流式输出的 RESTful API 接口。提供完整代码示例与依赖管理方案，适用于具备后端基础的开发人员快速上手本地大模型应用开发。
[图片：驱动下载页面]
[图片：驱动选择界面]
搞机器学习的，Anaconda 一般绕不开。
Anaconda 是一个用于科学计算的 Python 发行版，支持 Linux, Mac, Windows，包含了众多流行的科学计算、数据分析的 Python 包。Conda 是一个开源的包、环境管理器，可以用于在同一个机器上安装不同版本的软件包及其依赖，并能够在不同的环境之间切换。
既然每个环境可以指定相应的 Python 版本了，那我们就不用安装全局的 Python，先删除一下。
[图片：环境变量设置]
找到原来的 Python 安装的地址的两条记录，删掉。在退出时，记得点确定。
[图片：Path 路径清理]
我们来到官网，点击安装包下载链接。这里有个注意点，记得点一下按时间倒排，才能看到最新的版本，选择符合我们的 Windows 64 位版本下载。
[图片：Anaconda 下载页]
安装过程有很多点击下一步，在这里省略。在这里针对 C 盘比较小的情况，选择安装盘的时候，需要选择 All Users 和改到其它盘。
[图片：安装选项]
我们继续回到刚才删掉 Python 环境变量的 Path 上，把 Anaconda3 的安装地址加进去。
[图片：添加 Anaconda 路径]
配置完环境变量，重启一下电脑。这是环境变量生效最不烧脑的做法。
按 Win + R，输入 cmd 确定进入命令行提示框，输入：
conda --version
[图片：Conda 版本验证]
此时能看到 conda 的版本号。
接入我们输入 python 进入 python 交互环境：
[图片：Python 版本验证]
同样的，我们也能看到 Python 的版本号。
在这里我们可以通过 Ctrl+Z 再加 Enter 或者输入 exit() 来退出环境。
官方的源在国外，下载缓慢且不稳定。我们这里更换成国内的镜像。
[图片：打开终端]
执行以下命令生成 .condarc 文件：
conda config --set show_channel_urls yes
.condarc 文件打开：[图片：配置文件位置]
修改为以下内容：
channels:
- defaults
show_channel_urls: true
default_channels:
- https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main
- https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/r
- https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/msys2
custom_channels:
conda-forge: https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud
msys2: https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud
bioconda: https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud
menpo: https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud
pytorch: https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud
pytorch-lts: https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud
simpleitk: https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud
deepmodeling: https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud/
conda clean -i 清除索引缓存，保证用的是镜像站提供的索引。注：后续如果想切换回官方源，.condarc 文件里面的内容清掉即可。
conda create --name llm python=3.11.5
[图片：创建环境]
在这里我们创建的虚拟环境名称为 llm，选择 Python 的版本选择为 3.11.5。
通过命令：
conda info --envs
[图片：环境列表]
可以看到，我们创建的 llm 已经在里面了。
conda activate llm
[图片：激活环境]
以我们要部署的 Qwen1.5-0.5B-Chat 为例，通过 ModelScope 提供的下载方式，我们选择 Git 方式下载。
git clone https://www.modelscope.cn/qwen/Qwen1.5-0.5B-Chat.git
我把大模型下载到 D:\大模型\Qwen1.5-0.5B-Chat 目录下。
在这之前，我们先查一下我们的 cuda 版本。前面已经装好了相应的 NVIDIA 驱动，我们通过桌面右键，选择'NVIDIA 控制面板'->左下角'系统信息'->切换为'组件'Tab：
[图片：CUDA 版本查询]
可以看到我的 cuda 版本为 12.2。
通过 PyTorch 官网找到相应的安装脚本。
这里选择 PyTorch 2.1.0 版本，之所以不选择更加新的版本，是因为会遇到其它问题。而 cuda 则选择 12.1 版本的，跟电脑的 12.2 比较接近。
conda install pytorch==2.1.0 torchvision==0.16.0 torchaudio==2.1.0 pytorch-cuda=12.1 -c pytorch -c nvidia
我们需要安装 transformers 库：
conda install conda-forge::transformers
在写代码前，先配置一下写代码工具，那是必须的。这里以 VSCode 为例。
通过 VSCode 官网进行下载安装。
[图片：插件市场]
同时推荐安装 autoDocstring - Python Docstring Generator、autopep8。这样 VSCode 在编写 Python 代码时，可以提高代码质量和编写效率：
首先打开引入我们的大模型目录（D:\大模型）
[图片：VSCode 打开目录]
接着按住 Ctrl + Shift + P 打开命令面板，搜索'python 选择解析器'，选择上我们刚刚新建的 llm 环境。这样在 VSCode 中，我们就按照那个环境来跑我们的代码了。
[图片：选择解释器]
qwen.py 文件，写入：from transformers import AutoModelForCausalLM, AutoTokenizer
device = "cuda" # the device to load the model onto
# Now you do not need to add "trust_remote_code=True"
model = AutoModelForCausalLM.from_pretrained(
"Qwen1.5-0.5B-Chat", # 修改大模型位置
torch_dtype="auto",
device_map="auto"
)
tokenizer = AutoTokenizer.from_pretrained("Qwen1.5-0.5B-Chat") # 修改大模型位置
# Instead of using model.chat(), we directly use model.generate()
# But you need to use tokenizer.apply_chat_template() to format your inputs as shown below
# 改成中文提问
prompt = "给我简单介绍一下大型语言模型。"
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)
# Directly use generate() and tokenizer.decode() to get the output.
# Use `max_new_tokens` to control the maximum output length.
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]
# 打印一下助手回复的内容
print(response)
注：这里要修改对大模型的引入路径。
python qwen.py
[图片：运行结果]
发现还是缺了一个依赖，安装：
conda install conda-forge::accelerate
再重新执行，这时候就可以看到模型的回复了。
[图片：模型回复]
如果想在 VSCode 中运行，则需要选择在'在专用终端中运行 Python 文件'。
[图片：VSCode 运行设置]
运行效果如下：
[图片：终端输出]
如果想要流式的，可以改为以下代码：
from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer
from threading import Thread
device = "cuda" # the device to load the model onto
# Now you do not need to add "trust_remote_code=True"
model = AutoModelForCausalLM.from_pretrained(
"Qwen1.5-0.5B-Chat",
torch_dtype="auto",
device_map="auto"
)
tokenizer = AutoTokenizer.from_pretrained("Qwen1.5-0.5B-Chat")
# Instead of using model.chat(), we directly use model.generate()
# But you need to use tokenizer.apply_chat_template() to format your inputs as shown below
# 改成中文提问
prompt = "给我简单介绍一下大型语言模型。"
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)
# Directly use generate() and tokenizer.decode() to get the output.
# Use `max_new_tokens` to control the maximum output length.
streamer = TextIteratorStreamer(
tokenizer, skip_prompt=True, skip_special_tokens=True)
generation_kwargs = dict(model_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(generated_text)
这时候我们借助一个 FastAPI 和 Uvicorn 来实现 API 接口的支持。
conda install fastapi uvicorn
web.py ，写入代码：import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from argparse import ArgumentParser
app = FastAPI()
# 支持 CORS
app.add_middleware(
CORSMiddleware,
allow_origins=['*'],
allow_credentials=True,
allow_methods=['*'],
allow_headers=['*'],
)
@app.get("/")
async def index():
return {"message": "Hello World"}
def _get_args():
parser = ArgumentParser()
parser.add_argument('--server-port',
type=int,
default=8000,
help='Demo server port.')
parser.add_argument('--server-name',
type=str,
default='127.0.0.1',
help='Demo server name. Default: 127.0.0.1, which is only visible from the local computer.'
' If you want other computers to access your server, use 0.0.0.0 instead.',
)
args = parser.parse_args()
return args
if __name__ == '__main__':
args = _get_args()
uvicorn.run(app, host=args.server_name, port=args.server_port, workers=1)
web.pypython web.py
[图片：服务启动]
这时候请求接口，就可以看到返回 hello world ：
[图片：接口测试]
上面已经启动 API 服务了，我们把前面 Qwen 的代码写进去：
from contextlib import asynccontextmanager
import torch
import uvicorn
import time
from transformers import AutoModelForCausalLM, AutoTokenizer
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from argparse import ArgumentParser
from typing import List, Literal, Optional, Union
from pydantic import BaseModel, Field
@asynccontextmanager
async def lifespan(app: FastAPI): # collects GPU memory
yield
if torch.cuda.is_available():
torch.cuda.empty_cache()
torch.cuda.ipc_collect()
app = FastAPI(lifespan=lifespan)
# 支持 CORS
app.add_middleware(
CORSMiddleware,
allow_origins=['*'],
allow_credentials=True,
allow_methods=['*'],
allow_headers=['*'],
)
class ChatMessage(BaseModel):
role: Literal['user', 'assistant', 'system']
content: Optional[str]
class DeltaMessage(BaseModel):
role: Optional[Literal['user', 'assistant', 'system']] = None
content: Optional[str] = None
class ChatCompletionRequest(BaseModel):
model: str
messages: List[ChatMessage]
stream: Optional[bool] = False
class ChatCompletionResponseChoice(BaseModel):
index: int
message: ChatMessage
finish_reason: Literal['stop', 'length']
class ChatCompletionResponseStreamChoice(BaseModel):
index: int
delta: DeltaMessage
finish_reason: Optional[Literal['stop', 'length']]
class ChatCompletionResponse(BaseModel):
model: str
object: Literal['chat.completion', 'chat.completion.chunk']
choices: List[Union[ChatCompletionResponseChoice,
ChatCompletionResponseStreamChoice]]
created: Optional[int] = Field(default_factory=lambda: int(time.time()))
@app.get("/")
async def index():
return {"message": "Hello World"}
@app.post("/v1/chat/completions", response_model=ChatCompletionResponse)
async def create_chat_completion(request: ChatCompletionRequest):
global model, tokenizer
# 简单的错误校验
if request.messages[-1].role != "user":
raise HTTPException(status_code=400, detail="Invalid request")
text = tokenizer.apply_chat_template(
request.messages,
tokenize=False,
add_generation_prompt=True
)
model_inputs = tokenizer([text], return_tensors="pt").to(model.device)
# Directly use generate() and tokenizer.decode() to get the output.
# Use `max_new_tokens` to control the maximum output length.
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]
choice_data = ChatCompletionResponseChoice(
index=0,
message=ChatMessage(role="assistant", content=response),
finish_reason="stop"
)
return ChatCompletionResponse(model=request.model, choices=[choice_data], object="chat.completion")
def _get_args():
parser = ArgumentParser()
parser.add_argument(
'-c',
'--checkpoint-path',
type=str,
default='Qwen1.5-0.5B-Chat',
help='Checkpoint name or path, default to %(default)r',
)
parser.add_argument('--server-port',
type=int,
default=8000,
help='Demo server port.')
parser.add_argument('--server-name',
type=str,
default='127.0.0.1',
help='Demo server name. Default: 127.0.0.1, which is only visible from the local computer.'
' If you want other computers to access your server, use 0.0.0.0 instead.',
)
args = parser.parse_args()
return args
if __name__ == '__main__':
args = _get_args()
# Now you do not need to add "trust_remote_code=True"
model = AutoModelForCausalLM.from_pretrained(
args.checkpoint_path,
torch_dtype="auto",
device_map="auto"
)
tokenizer = AutoTokenizer.from_pretrained(args.checkpoint_path)
uvicorn.run(app, host=args.server_name, port=args.server_port, workers=1)
调用下接口：
[图片：接口调用结果]
可以看到接口已经返回内容了。
这里需要安装 sse_starlette 的库，来支持流式的返回：
pip install sse_starlette
安装完把代码再改一下，通过参数 stream 来判断是否流式返回：
from contextlib import asynccontextmanager
from threading import Thread
import torch
import uvicorn
import time
from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer, BatchEncoding
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from argparse import ArgumentParser
from typing import List, Literal, Optional, Union
from pydantic import BaseModel, Field
from sse_starlette.sse import EventSourceResponse
@asynccontextmanager
async def lifespan(app: FastAPI): # collects GPU memory
yield
if torch.cuda.is_available():
torch.cuda.empty_cache()
torch.cuda.ipc_collect()
app = FastAPI(lifespan=lifespan)
# 支持 CORS
app.add_middleware(
CORSMiddleware,
allow_origins=['*'],
allow_credentials=True,
allow_methods=['*'],
allow_headers=['*'],
)
class ChatMessage(BaseModel):
role: Literal['user', 'assistant', 'system']
content: Optional[str]
class DeltaMessage(BaseModel):
role: Optional[Literal['user', 'assistant', 'system']] = None
content: Optional[str] = None
class ChatCompletionRequest(BaseModel):
model: str
messages: List[ChatMessage]
stream: Optional[bool] = False
class ChatCompletionResponseChoice(BaseModel):
index: int
message: ChatMessage
finish_reason: Literal['stop', 'length']
class ChatCompletionResponseStreamChoice(BaseModel):
index: int
delta: DeltaMessage
finish_reason: Optional[Literal['stop', 'length']]
class ChatCompletionResponse(BaseModel):
model: str
object: Literal['chat.completion', 'chat.completion.chunk']
choices: List[Union[ChatCompletionResponseChoice,
ChatCompletionResponseStreamChoice]]
created: Optional[int] = Field(default_factory=lambda: int(time.time()))
@app.get("/")
async def index():
return {"message": "Hello World"}
@app.post("/v1/chat/completions", response_model=ChatCompletionResponse)
async def create_chat_completion(request: ChatCompletionRequest):
global model, tokenizer
# 简单的错误校验
if request.messages[-1].role != "user":
raise HTTPException(status_code=400, detail="Invalid request")
text = tokenizer.apply_chat_template(
request.messages,
tokenize=False,
add_generation_prompt=True
)
model_inputs = tokenizer([text], return_tensors="pt").to(model.device)
if request.stream:
generate = predict(model_inputs, request.model)
return EventSourceResponse(generate, media_type="text/event-stream")
# Directly use generate() and tokenizer.decode() to get the output.
# Use `max_new_tokens` to control the maximum output length.
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]
choice_data = ChatCompletionResponseChoice(
index=0,
message=ChatMessage(role="assistant", content=response),
finish_reason="stop"
)
return ChatCompletionResponse(model=request.model, choices=[choice_data], object="chat.completion")
async def predict(model_inputs: BatchEncoding, model_id: str):
global model, tokenizer
streamer = TextIteratorStreamer(
tokenizer, skip_prompt=True, skip_special_tokens=True)
generation_kwargs = dict(
model_inputs, streamer=streamer, max_new_tokens=512)
thread = Thread(target=model.generate, kwargs=generation_kwargs)
choice_data = ChatCompletionResponseStreamChoice(
index=0,
delta=DeltaMessage(role="assistant"),
finish_reason=None
)
chunk = ChatCompletionResponse(model=model_id, choices=[
choice_data], object="chat.completion.chunk")
yield "{}".format(chunk.model_dump_json(exclude_unset=True))
thread.start()
for new_text in streamer:
choice_data = ChatCompletionResponseStreamChoice(
index=0,
delta=DeltaMessage(content=new_text),
finish_reason=None
)
chunk = ChatCompletionResponse(model=model_id, choices=[
choice_data], object="chat.completion.chunk")
yield "{}".format(chunk.model_dump_json(exclude_unset=True))
choice_data = ChatCompletionResponseStreamChoice(
index=0,
delta=DeltaMessage(),
finish_reason="stop"
)
chunk = ChatCompletionResponse(model=model_id, choices=[
choice_data], object="chat.completion.chunk")
yield "{}".format(chunk.model_dump_json(exclude_unset=True))
yield '[DONE]'
def _get_args():
parser = ArgumentParser()
parser.add_argument(
'-c',
'--checkpoint-path',
type=str,
default='Qwen1.5-0.5B-Chat',
help='Checkpoint name or path, default to %(default)r',
)
parser.add_argument('--server-port',
type=int,
default=8000,
help='Demo server port.')
parser.add_argument('--server-name',
type=str,
default='127.0.0.1',
help='Demo server name. Default: 127.0.0.1, which is only visible from the local computer.'
' If you want other computers to access your server, use 0.0.0.0 instead.',
)
args = parser.parse_args()
return args
if __name__ == '__main__':
args = _get_args()
# Now you do not need to add "trust_remote_code=True"
model = AutoModelForCausalLM.from_pretrained(
args.checkpoint_path,
torch_dtype="auto",
device_map="auto"
)
tokenizer = AutoTokenizer.from_pretrained(args.checkpoint_path)
uvicorn.run(app, host=args.server_name, port=args.server_port, workers=1)
再调用接口：
[图片：流式接口测试]
流式返回也支持了。
接口更加完善的话可以参考官方文档或社区示例。
至此，Qwen1.5 大模型已在 Windows 环境下完成本地部署及 API 服务化。通过上述步骤，您可以实现从环境搭建、模型加载到接口调用的全流程开发。建议在实际生产环境中进一步考虑安全性、并发处理及资源优化等问题。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online