使用 FastAPI 封装 Qwen1.5-0.5B 本地大模型接口（支持流式/非流式）

在本地部署大语言模型后，若想提供 Web 服务供前端或其他系统调用，FastAPI 是一个轻量、高性能且易用的选择。本文将展示如何将已缓存的
Qwen1.5-0.5B 模型（通过 ModelScope 下载）封装为支持 流式（Streaming）与非流式（Batch） 两种模式的 API。

---

1. 安装依赖

使用阿里云镜像加速安装所需库（避免清华源限流）：

pip install -i https://mirrors.aliyun.com/pypi/simple/ fastapi uvicorn sse-starlette

2. 编写 main.py

from fastapi import FastAPI, Query, Body
from fastapi.responses import StreamingResponse
from sse_starlette.sse import EventSourceResponse
from modelscope import AutoTokenizer, AutoModelForCausalLM
import torch

app = FastAPI(
    title="Qwen1.5-0.5B API",
    description="本地 CPU 部署的轻量大模型接口，支持流式与非流式输出"
)

print("正在加载 tokenizer 和模型（CPU 模式）...")
model_id = "qwen/Qwen1.5-0.5B"
tokenizer = AutoTokenizer.from_pretrained(model_id, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(
    model_id,
    device_map="cpu",
    trust_remote_code=True,
    tie_word_embeddings=False
)
model.eval()
print("✅ 模型加载完成！")


def generate_stream(prompt: str, max_new_tokens: int = 128):
    """流式生成器：逐 token 返回"""
    inputs = tokenizer(prompt, return_tensors="pt")
    input_ids = inputs.input_ids
    attention_mask = inputs.attention_mask
    generated = input_ids.clone()

    with torch.no_grad():
        for _ in range(max_new_tokens):
            outputs = model(input_ids=generated, attention_mask=attention_mask)
            next_token_logits = outputs.logits[:, -1, :]
            next_token = torch.argmax(next_token_logits, dim=-1).unsqueeze(0)

            # 遇到结束符则停止
            if next_token.item() == tokenizer.eos_token_id:
                break

            # 拼接新 token
            generated = torch.cat([generated, next_token], dim=1)
            attention_mask = torch.cat([attention_mask, torch.ones((1, 1), dtype=torch.long)], dim=1)

            # 解码新增部分
            new_text = tokenizer.decode([next_token.item()], skip_special_tokens=True)
            if new_text.strip():  # 过滤空字符
                yield new_text


@app.post("/generate")
async def generate(
    prompt: str = Body(..., embed=True, description="输入提示文本"),
    stream: bool = Query(False, description="是否启用流式输出（true/false）"),
    max_new_tokens: int = Query(128, ge=1, le=512, description="最大生成长度（1~512）")
):
    """
    调用 Qwen1.5-0.5B 生成文本
    
    - **prompt**: 用户输入（JSON body）
    - **stream**: 是否流式返回（URL 查询参数）
    - **max_new_tokens**: 控制生成长度（URL 查询参数）
    """
    if stream:
        return EventSourceResponse(generate_stream(prompt, max_new_tokens))
    else:
        inputs = tokenizer(prompt, return_tensors="pt")
        with torch.no_grad():
            outputs = model.generate(**inputs, max_new_tokens=max_new_tokens)
        response = tokenizer.decode(outputs[0], skip_special_tokens=True)
        # 移除重复的 prompt 前缀
        if response.startswith(prompt):
            response = response[len(prompt):].lstrip()
        return {"response": response}

3. 启动服务

在终端执行以下命令启动 API 服务：

uvicorn main:app --host 0.0.0.0 --port 8000 --reload

• –host 0.0.0.0：允许外部访问（如局域网）
• –port 8000：监听端口
• –reload：开发模式，代码修改自动重启

启动后访问 http://localhost:8000/docs 可查看自动生成的交互式 API 文档（可以访问swagger的网络情况下）。

4. 接口测试

✅ 一次性输出（默认）

curl -X POST http://localhost:8000/generate \
  -H "Content-Type: application/json" \
  -d '{"prompt": "你好，请介绍一下你自己。"}'

返回示例：

🌊 流式输出（SSE）

curl -X POST "http://localhost:8000/generate?stream=true" \
  -H "Content-Type: application/json" \
  -d '{"prompt": "你好，请介绍一下你自己。"}'

返回为逐字流式数据（SSE 格式）：

前端可通过 EventSource 或 fetch + ReadableStream 实时接收。