Python 调用通义千问 Qwen2.5 模型完整流程

3.2 设备部署策略

如需显式指定设备，可结合 PyTorch 进行判断：

import torch
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model.to(device)

推理过程

4.1 构建输入数据

Qwen2.5 采用对话模板格式，需构建包含 system、user 角色的消息列表。

messages = [
    {"role": "system", "content": "请逐步推理，最终答案放入 \boxed{} 中。"},
    {"role": "user", "content": "Find the value of x that satisfies 4x + 5 = 6x + 7."}
]

4.2 生成输出

应用聊天模板后送入模型生成，注意处理 token 长度。

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, 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)

4.3 流式生成

对于实时反馈场景，可使用 TextStreamer 实现逐字输出。

from transformers import TextStreamer
streamer = TextStreamer(tokenizer, skip_special_tokens=True)
model.generate(**model_inputs, max_new_tokens=512, streamer=streamer)

API 调用方案

5.1 DashScope SDK 调用

阿里云 DashScope 提供了官方 SDK，便于集成。

from dashscope import Generation

Generation.api_key = "your_api_key_here"
response = Generation.call(
    model="qwen2.5-7b-instruct",
    messages=[{"role": "user", "content": "你好，通义千问！"}],
    max_tokens=100,
    temperature=0.7
)
print(response.output.text)

5.2 OpenAI 兼容接口

部分本地部署或第三方服务支持 OpenAI 协议，可直接复用相关客户端。

import requests

url = "http://localhost:8000/v1/chat/completions"
headers = {
    "Content-Type": "application/json",
    "Authorization": "Bearer your_api_key_here"
}
data = {
    "model": "Qwen/Qwen2.5-7B-Instruct",
    "messages": [{"role": "user", "content": "请介绍大语言模型。"}],
    "temperature": 0.7,
    "max_tokens": 512
}
response = requests.post(url, headers=headers, json=data)
print(response.json()["choices"][0]["message"]["content"])

5.3 密钥管理

建议将 AccessKey ID 和 Secret 存储在环境变量中，避免硬编码在代码里。

import os
os.environ['ALIYUN_ACCESS_KEY_ID'] = 'your_access_key_id'
os.environ['ALIYUN_ACCESS_KEY_SECRET'] = 'your_access_key_secret'

部署与优化

6.1 vLLM 部署

vLLM 能显著提升推理吞吐量，适合高并发场景。

from vllm import LLM, SamplingParams

llm = LLM(model="path/to/qwen2.5")
sampling_params = SamplingParams(temperature=0.8, top_p=0.9)
prompts = ["Hello, how are you?"]
outputs = llm.generate(prompts, sampling_params)
for output in outputs:
    print(output.text)

6.2 SGLang 与 Ollama

SGLang 支持多种部署模式，Ollama 则适合本地快速运行。

# Ollama 示例
from ollama import Model
model = Model(model_path="qwen2.5")
response = model.generate(prompt="Hello", options={"temperature": 0.8})
print(response["response"])

6.3 多 GPU 分布式

对于超大参数模型，可使用 Distributed PyTorch 进行多卡并行。

import torch.distributed as dist
from torch.nn.parallel import DistributedDataParallel as DDP
# 初始化进程组并包裹模型

函数调用能力

Qwen2.5 支持通过特定模板触发外部函数调用，适用于需要实时数据或执行操作的场景。

7.1 Qwen-Agent 方式

使用 Qwen-Agent 框架可简化函数定义与调用流程。

from qwen_agent.llm import get_chat_model

llm = get_chat_model({
    "model": "Qwen/Qwen2.5-7B-Instruct",
    "model_server": "http://localhost:8000/v1",
    "api_key": "EMPTY"
})

functions = [{"name": "get_weather", "description": "查询天气", "parameters": {...}}]
messages = [{"role": "user", "content": "旧金山现在天气如何？"}]

for responses in llm.chat(messages=messages, functions=functions):
    if fn_call := responses.get("function_call"):
        # 执行实际函数逻辑并返回结果
        pass

7.2 Transformers 原生支持

利用 apply_chat_template 构建包含工具调用的输入。

text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
# 后续生成逻辑同上

常见问题排查

8.1 API 错误处理

网络波动或密钥失效常导致调用失败，建议增加重试机制。

from retry import retry

@retry(tries=3, delay=2)
def call_api(url, headers):
    response = requests.get(url, headers=headers)
    response.raise_for_status()
    return response.json()

8.2 性能优化

• 缓存：对固定查询结果使用 lru_cache。
• 批量处理：合并多个请求以减少网络开销。
• 异步 IO：使用 aiohttp 处理高并发请求。

8.3 参数调优

• temperature：控制随机性，范围 [0, 2)。
• top_p：核采样阈值，影响多样性。
• max_tokens：限制生成长度。

参考示例

以下是一个完整的本地推理脚本，整合了加载、推理与输出。

from transformers import AutoModelForCausalLM, AutoTokenizer

model_name = "Qwen/Qwen2.5-7B-Instruct"
model = AutoModelForCausalLM.from_pretrained(model_name, torch_dtype="auto", device_map="auto")
tokenizer = AutoTokenizer.from_pretrained(model_name)

prompt = "请给我一个关于大型语言模型的简短介绍。"
messages = [{"role": "user", "content": prompt}]
text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)

model_inputs = tokenizer([text], return_tensors="pt").to(model.device)
generated_ids = model.generate(**model_inputs, 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)