企业官网建设流程全解析

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

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

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

一、背景与技术选型

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

选型：

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

环境依赖：

pipinstallopenai fastapi uvicorn

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

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

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

2.1 非流式调用llm/case1.py

importosfromopenaiimportOpenAI 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/modelsmodel="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

importosfromopenaiimportOpenAI 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=[]forchunkincompletion:ifchunk.choices:choice=chunk.choices[0]ifchoice.delta:delta=choice.deltaifdelta.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

frompydanticimportBaseModel,FieldclassLLMCase1(BaseModel):question:str=Field(...,title="问题",description="问题")

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

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

case1：非流式（一次性返回）

importosfromfastapiimportAPIRouterfromopenaiimportOpenAIfromapp.schemas.llm_case1importLLMCase1 llm_day01_router=APIRouter(prefix="/llm-day01",tags=["LLM-DAY01"])@llm_day01_router.post("/case1",summary="LLM-DAY01-CASE1")asyncdefcase1_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.contentreturn{"code":1,"message":"请求成功","data":{"ai_reply":ai_reply},}

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

case2：SSE 流式输出

fromstarlette.responsesimportStreamingResponsedefstream_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})forchunkincompletion:ifchunk.choices:choice=chunk.choices[0]ifchoice.delta:delta=choice.deltaifdelta.content:yieldf"data:{delta.content}\n\n"# SSE 标准格式：data: xxx\n\nyield"data: [done]\n\n"# 结束标记@llm_day01_router.post("/case2",summary="LLM-DAY01-CASE2")asyncdefcase2_api(llmCase1Request:LLMCase1):returnStreamingResponse(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.pyfromapp.apis.llm.case1_apiimportllm_day01_router app.include_router(llm_day01_router)

启动后接口路径：

• POST /llm-day01/case1非流式
• POST /llm-day01/case2SSE 流式

四、踩坑清单（重点）

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 走环境变量）。