server代码：fastapi_qwen3.py

import torch
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from transformers import AutoModelForCausalLM, AutoTokenizer
import uvicorn

app = FastAPI()

# 1. 设置模型路径和设备
# 请确保这里的路径正确指向你的 Qwen3.5-4B 文件夹
model_path = "/data2/wanghq/models/Qwen/Qwen3.5-4B"
# model_path = "/data2/wanghq/models/Qwen/Qwen3-4B"
# model_path = "/data2/wanghq/models/Qwen/Qwen3-1.7B"
device = "cuda:0" # 明确指定使用第一块 4090

print(f"正在从 {model_path} 加载模型到 {device}...")

# 2. 加载分词器和模型
tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(
    model_path,
    torch_dtype=torch.bfloat16, # 4090 必用 bf16，性能最好
    device_map={"": device},
    trust_remote_code=True,
    attn_implementation="sdpa" # 开启闪速注意力机制
)

# 定义请求数据结构
class GroundingRequest(BaseModel):
    prompt: str
    system_prompt: str = "你是一个由阿里巴巴开发的人工智能助手，请根据用户的指令提供准确、简洁的帮助。"

@app.post("/parse_instruction")
async def parse_instruction(request: GroundingRequest):
    try:
        # 构造符合 Qwen 3.5 格式的对话
        messages = [
            {"role": "system", "content": request.system_prompt},
            {"role": "user", "content": request.prompt}
        ]
        
        # 应用模板（enable_thinking=False 关闭 Qwen3 默认的思考模式）
        text = tokenizer.apply_chat_template(
            messages,
            tokenize=False,
            add_generation_prompt=True,
            enable_thinking=False
        )
        
        model_inputs = tokenizer([text], return_tensors="pt").to(device)

        # 生成回复
        generated_ids = model.generate(
            **model_inputs,
            max_new_tokens=256,
            do_sample=False, # 设为 False 保证输出高度一致稳定
            repetition_penalty=1.1
        )
        
        # 裁剪掉输入部分，只保留回复内容
        response_ids = [
            output_ids[len(input_ids):] for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)
        ]
        
        response_text = tokenizer.batch_decode(response_ids, skip_special_tokens=True)[0]
        
        return {
            "status": "success",
            "result": response_text.strip()
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    # 运行在 8000 端口
    uvicorn.run(app, host="0.0.0.0", port=8000)

测试代码：test_qwen3_api.py

import requests

API_URL = "http://192.168.112.159:8000/parse_instruction"

def test_parse_instruction():
    payload = {
        "prompt": "从后面语句中获得用户想要图书名称，直接输出书名，不要输出其他信息。语句:请帮我拿桌上的 图像处理这本书"
    }

    response = requests.post(API_URL, json=payload)
    assert response.status_code == 200, f"请求失败，状态码: {response.status_code}"

    data = response.json()
    assert data["status"] == "success", f"返回状态异常: {data}"

    result = data["result"]
    print(f"模型返回: {result}")
    assert "图像处理" in result, f"期望包含 '图像处理'，实际返回: {result}"
    print("✅ 测试通过！")

if __name__ == "__main__":
    test_parse_instruction()

测试结果

测试命令

wanghq@wanghqdeMacBook-Pro ~ % curl -X POST "http://192.168.112.159:8000/parse_instruction"      -H "Content-Type: application/json"      -d '{"prompt": "从后面语句中获得用户想要图书名称，直接输出书名，不要输出其他信息。语句:请帮我拿桌上的 图像处理这本书"}'

测试结果：