LLM 代码实现：Qwen 模型下载与调用指南

基础调用

from transformers import AutoModelForCausalLM, AutoTokenizer
from transformers.generation import GenerationConfig

# 加载模型和 tokenizer
path = "./model/Qwen/Qwen-1_8B-Chat"
tokenizer = AutoTokenizer.from_pretrained(path, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(
    path,
    device_map="auto",
    trust_remote_code=True
).eval()

# 配置生成参数
model.generation_config = GenerationConfig.from_pretrained(path, trust_remote_code=True)

# 对话测试
response, history = model.chat(tokenizer, "你好", history=None)
print(response)

设备映射（Device Map）

device_map 用于控制模型在计算设备上的分布。

• "cpu"：仅使用 CPU，速度慢但无需 GPU。
• "auto"：自动平衡所有可用 GPU。
• "balanced"：在所有 GPU 上均衡切分模型。
• "sequential"：按顺序分配模型分片。

自定义显存分配：

from accelerate import infer_auto_device_map

device_map = infer_auto_device_map(model, max_memory={0: "10GiB", 1: "10GiB", "cpu": "30GiB"})
print(model.hf_device_map)

性能优化

官方推荐支持 fp16 或 bf16 精度的显卡用户安装 flash-attention 以提升效率。

git clone https://github.com/Dao-AILab/flash-attention
cd flash-attention && pip install .

生成参数配置

GenerationConfig 允许精细控制输出质量。

model.generation_config = GenerationConfig.from_pretrained(
    path,
    temperature=0.8,      # 多样性：越高越随机
    top_k=5,              # 候选词数量
    repetition_penalty=1.2, # 抑制重复
    do_sample=True,       # 是否采样
    return_unused_kwargs=True
)

关键参数说明：

• temperature：控制文本随机性。0.2 更确定，1.0 更发散。
• top_k：限制考虑的最高概率词的数量。
• top_p：累积概率阈值，动态调整候选词范围。
• repetition_penalty：惩罚重复词语，值越大重复越少。
• max_new_tokens：限制最大生成长度。

多轮对话脚本实现

以下实现了一个完整的交互类，支持清空历史、设置系统提示词。

from transformers import AutoModelForCausalLM, AutoTokenizer
import os
import sys

class QwenChat:
    def __init__(self, model_path="./model/Qwen/Qwen-1_8B-Chat"):
        self.path = model_path
        print(f"正在加载模型：{model_path}")
        self.tokenizer = AutoTokenizer.from_pretrained(self.path, trust_remote_code=True)
        self.model = AutoModelForCausalLM.from_pretrained(
            self.path,
            device_map="auto",
            trust_remote_code=True
        ).eval()
        print("模型加载完成")

    def clear_screen(self):
        os.system('clear' if os.name != 'nt' else 'cls')
        return [], ""

    def chat_qwen(self, system_prompt="你是一位有用的助手"):
        history = []
        while True:
            try:
                prompt = input("user：")
                if prompt.lower() in ["clc", "exit", "quit"]:
                    if prompt.lower() == "clc":
                        history = []
                        print("历史记录已清空")
                        continue
                    else:
                        break
                
                response, history = self.model.chat(
                    self.tokenizer, 
                    prompt, 
                    history=history, 
                    system=system_prompt
                )
                print(f"assistant：\n{response}\n")
            except KeyboardInterrupt:
                print("\n程序中断")
                break
            except Exception as e:
                print(f"发生错误：{e}")

if __name__ == '__main__':
    # 示例：设定外婆的语气
    qwen = QwenChat()
    qwen.chat_qwen(system_prompt="请用外婆的语气和我说话，语气温柔亲切")

流式输出（Streaming）

为了提升用户体验，支持流式输出可以实时显示生成的文字。

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

model_path = "./model/Qwen/Qwen-1_8B-Chat"
tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(model_path, device_map="auto", trust_remote_code=True).eval()

input_text = "请介绍一下人工智能"
inputs = tokenizer(input_text, return_tensors="pt").to(model.device)

for output in model.generate(**inputs, streamer=tokenizer.get_streamer(), max_new_tokens=100):
    print(tokenizer.decode(output, skip_special_tokens=True), end="", flush=True)

常见问题排查

1. 显存溢出（OOM）：

   • 尝试减小 batch_size。
   • 使用量化版本（如 INT8/INT4）。
   • 增加 device_map 的分片策略。

2. 导入错误：

   • 确保安装了 transformers>=4.30.0。
   • 检查 trust_remote_code=True 是否开启。

3. 生成乱码：

   • 检查 tokenizer 是否匹配模型版本。
   • 确认 bos_token_id 和 eos_token_id 配置正确。

部署为 API 服务

使用 FastAPI 将模型封装为 RESTful 接口。

from fastapi import FastAPI
from pydantic import BaseModel
import uvicorn

class ChatRequest(BaseModel):
    message: str
    system_prompt: str = "你是一个智能助手"

app = FastAPI()

@app.post("/chat")
async def chat(req: ChatRequest):
    # 此处复用上述 QwenChat 类的实例
    response, _ = model.chat(tokenizer, req.message, history=[], system=req.system_prompt)
    return {"reply": response}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

通过上述步骤，您可以完成 Qwen 模型的本地部署、调试及基础应用开发。根据实际业务需求，可进一步集成 LangChain 框架或微调模型以适应特定领域。