FastAPI 集成通义千问（Qwen）实战 Day1：非流式调用与 SSE 流式输出

关键词：FastAPI、通义千问、Qwen、OpenAI 兼容协议、SSE 流式、Python

本文记录我在招聘系统项目里接入大模型能力的第一天：用阿里云百炼（DashScope）提供的兼容 OpenAI 协议接口，先跑通原生脚本，再封装成 FastAPI 接口，覆盖「一次性返回」和「打字机式流式返回」两种最常见姿势。

一、背景与技术选型

项目里想给简历 / JD 模块加智能能力（JD 自动生成、简历要点提取等），第一步是把大模型调通。

选型：

  • 模型：通义千问 qwen-plus（阿里云百炼平台提供）
  • 协议：百炼提供 兼容 OpenAI 的接口，所以直接用官方 openai SDK，只需改 base_url 和 api_key 即可，不用学新 SDK
  • Web 框架：FastAPI（项目本身后端就是 FastAPI + Tortoise ORM + MySQL）
  • 两种链路：
    1. 非流式：请求完等模型生成完毕，一次性返回完整回答
    2. 流式（SSE）：边生成边推，前端打字机效果

环境依赖：

pip install openai fastapi uvicorn

API Key 在阿里云百炼控制台获取，用环境变量 DASHSCOPE_API_KEY 注入，不要硬编码进代码。

二、先跑通原生脚本（脱离 Web 框架）

在写接口前，先用最小脚本验证链路通不通。

2.1 非流式调用 llm/case1.py

import os
from openai import OpenAI

client = OpenAI(
    # 若没有配置环境变量，请用百炼API Key将下行替换为：api_key="sk-xxx"
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://ws-ulkao56twirebft4.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",
)

completion = client.chat.completions.create(
    # 模型列表：https://help.aliyun.com/zh/model-studio/getting-started/models
    model="qwen-plus",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "人为什么要睡觉?"},
    ],
    temperature=0.75,
)

print(completion.choices[0].message.content)

要点：

  • OpenAI(api_key=..., base_url=...) 初始化客户端，base_url 指向百炼的兼容模式 /v1
  • client.chat.completions.create(...) 发起对话，参数：
    • model：模型名
    • messages：对话历史，每项 {"role": ..., "content": ...}。system 设定人设，user 是用户输入
    • temperature：采样温度，越大越发散
  • 取结果：completion.choices[0].message.content

2.2 流式调用 llm/case2.py

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["DASHSCOPE_API_KEY"],
    base_url="https://ws-ulkao56twirebft4.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",
)

completion = client.chat.completions.create(
    model="qwen-plus",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "请介绍一下自己"}
    ],
    stream=True,                                  # 开启流式
    stream_options={"include_usage": True}        # 拿 token 用量
)

chunks = []
for chunk in completion:
    if chunk.choices:
        choice = chunk.choices[0]
        if choice.delta:
            delta = choice.delta
            if delta.content:
                print(delta.content)
                chunks.append(delta.content)      # 暂存片段

res = ''.join(chunks)                             # 最后 join，比 += 高效
print(res)

要点：

  • stream=True 开启流式，stream_options={"include_usage": True} 才能拿到 token 用量
  • 逐块迭代：chunk.choices[0].delta.content 是这一小段增量文本
  • 用 list 暂存片段，最后 ''.join(chunks) 拼接——比字符串逐次 += 高效

三、封装成 FastAPI 接口

3.1 请求体校验 app/schemas/llm_case1.py

from pydantic import BaseModel, Field


class LLMCase1(BaseModel):
    question: str = Field(..., title="问题", description="问题")

只收一个 question 字段，用 Pydantic 做参数校验。

3.2 接口 app/apis/llm/case1_api.py

case1：非流式（一次性返回）
import os
from fastapi import APIRouter
from openai import OpenAI

from app.schemas.llm_case1 import LLMCase1

llm_day01_router = APIRouter(prefix="/llm-day01", tags=["LLM-DAY01"])


@llm_day01_router.post("/case1", summary="LLM-DAY01-CASE1")
async def case1_api(llmCase1Request: LLMCase1):
    client = OpenAI(
        api_key=os.getenv("DASHSCOPE_API_KEY"),
        base_url="https://ws-ulkao56twirebft4.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",
    )
    completion = client.chat.completions.create(
        model="qwen-plus",
        messages=[
            {"role": "system", "content": "你是一个智能助手"},
            {"role": "user", "content": llmCase1Request.question},
        ],
        temperature=0.75,
    )
    ai_reply = completion.choices[0].message.content
    return {
        "code": 1,
        "message": "请求成功",
        "data": {"ai_reply": ai_reply},
    }

返回沿用项目统一的响应信封 {code, message, data}。

case2：SSE 流式输出
from starlette.responses import StreamingResponse


def stream_chunk(user_question: str):
    client = OpenAI(
        api_key=os.environ["DASHSCOPE_API_KEY"],
        base_url="https://ws-ulkao56twirebft4.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",
    )
    completion = client.chat.completions.create(
        model="qwen-plus",
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": user_question}
        ],
        stream=True,
        stream_options={"include_usage": True}
    )
    for chunk in completion:
        if chunk.choices:
            choice = chunk.choices[0]
            if choice.delta:
                delta = choice.delta
                if delta.content:
                    yield f"data: {delta.content}\n\n"   # SSE 标准格式：data: xxx\n\n
    yield "data: [done]\n\n"                            # 结束标记


@llm_day01_router.post("/case2", summary="LLM-DAY01-CASE2")
async def case2_api(llmCase1Request: LLMCase1):
    return StreamingResponse(
        content=stream_chunk(llmCase1Request.question),
        media_type="text/event-stream"
    )

要点：

  • stream_chunk 是一个 生成器函数（含 yield），逐片产出 SSE 文本
  • SSE 格式固定为 data: 内容\n\n，必须是双换行，前端 EventSource 才能正确切分事件
  • 最后 yield "data: [done]\n\n" 作为结束信号，通知前端回答完毕
  • 用 StreamingResponse(content=生成器, media_type="text/event-stream") 把生成器包成流式响应
  • FastAPI 普通函数不能 yield，流式必须返回 StreamingResponse

3.3 路由注册

app/apis/llm/__init__.py 预留空文件做包入口，main.py 里注册：

# main.py
from app.apis.llm.case1_api import llm_day01_router

app.include_router(llm_day01_router)

启动后接口路径：

  • POST /llm-day01/case1 非流式
  • POST /llm-day01/case2 SSE 流式

四、踩坑清单（重点）

  1. API Key 别硬编码：用环境变量。但注意代码里两处不一致——非流式用 os.getenv（缺省返回 None 静默失败），生成器里用 os.environ["..."]（缺省会抛 KeyError）。建议统一用 os.getenv 并显式判断为空时返回友好错误。
  2. base_url 与 Key 地域强绑定：百炼的 base_url 必须是兼容模式 /v1，且要和 API Key 所属地域一致，否则鉴权直接失败。
  3. SSE 必须双换行：data: xxx\n\n，少一个 \n 前端 EventSource 解析不出事件。
  4. 流式结束要发标记：最后 yield "data: [done]\n\n" 让前端知道回答结束、关闭连接。
  5. 流式拿 token 用量：stream_options={"include_usage": True} 才有 chunk.usage。
  6. 字符串拼接用 list + join：流式片段别用 +=，用 chunks.append 后 ''.join(chunks) 性能更好。
  7. FastAPI 流式不能用普通函数 yield：必须返回 StreamingResponse 并传入生成器。
  8. temperature 按场景调：demo 用 0.75 偏发散；生产场景（如 JD 生成）建议调低拿稳定输出。

五、小结

Day1 跑通了通义千问在 FastAPI 下的两种调用姿势：

  • 非流式：适合后台任务、批量处理
  • SSE 流式：适合对话界面、打字机效果

后续可继续做：

  • 多轮对话（维护 messages 历史）
  • 工具调用（function calling，接招聘业务 API）
  • 落地招聘场景：JD 自动生成、简历要点提取、候选人匹配

注：本文所有代码片段均来自当天真实提交的后端文件（llm/case1.py、llm/case2.py、app/apis/llm/case1_api.py、app/schemas/llm_case1.py、main.py），仅做脱敏（API Key 走环境变量）。

# fastapi
Logo
亚马逊云科技技术品牌专区

更多推荐

cover

百万Token不是数字游戏：Kimi K3完整吞下10万行代码库后，我测出了它的真实理解力

avatar 亚马逊云科技技术品牌专区
cover

【2026最新版】Codex换肤助手下载和使用教程：一键美化 Codex 客户端，支持自定义主题

avatar 亚马逊云科技技术品牌专区

FastAPI + OpenAI 兼容协议 + DeepSeek 实战：大模型 Function Calling 工具调用全拆解

模型本身不能直接查天气、查数据库。我们告诉模型"你有这些函数可用"，模型在觉得需要时返回"我想调用某个函数、参数是这样"，真正的函数由我们本地执行，再把结果交还给模型整理成话。这就是"模型决策 + 本地执行"的混合模式。

avatar 亚马逊云科技技术品牌专区