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

FastAPI 集成通义千问Qwen实战 Day1非流式调用与 SSE 流式输出关键词FastAPI、通义千问、Qwen、OpenAI 兼容协议、SSE 流式、Python本文记录我在招聘系统项目里接入大模型能力的第一天用阿里云百炼DashScope提供的兼容 OpenAI 协议接口先跑通原生脚本再封装成 FastAPI 接口覆盖「一次性返回」和「打字机式流式返回」两种最常见姿势。一、背景与技术选型项目里想给简历 / JD 模块加智能能力JD 自动生成、简历要点提取等第一步是把大模型调通。选型模型通义千问qwen-plus阿里云百炼平台提供协议百炼提供兼容 OpenAI的接口所以直接用官方openaiSDK只需改base_url和api_key即可不用学新 SDKWeb 框架FastAPI项目本身后端就是 FastAPI Tortoise ORM MySQL两种链路非流式请求完等模型生成完毕一次性返回完整回答流式SSE边生成边推前端打字机效果环境依赖pipinstallopenai fastapi uvicornAPI Key 在阿里云百炼控制台获取用环境变量DASHSCOPE_API_KEY注入不要硬编码进代码。二、先跑通原生脚本脱离 Web 框架在写接口前先用最小脚本验证链路通不通。2.1 非流式调用llm/case1.pyimportosfromopenaiimportOpenAI clientOpenAI(# 若没有配置环境变量请用百炼API Key将下行替换为api_keysk-xxxapi_keyos.getenv(DASHSCOPE_API_KEY),base_urlhttps://ws-ulkao56twirebft4.cn-beijing.maas.aliyuncs.com/compatible-mode/v1,)completionclient.chat.completions.create(# 模型列表https://help.aliyun.com/zh/model-studio/getting-started/modelsmodelqwen-plus,messages[{role:system,content:You are a helpful assistant.},{role:user,content:人为什么要睡觉?},],temperature0.75,)print(completion.choices[0].message.content)要点OpenAI(api_key..., base_url...)初始化客户端base_url 指向百炼的兼容模式 /v1client.chat.completions.create(...)发起对话参数model模型名messages对话历史每项{role: ..., content: ...}。system设定人设user是用户输入temperature采样温度越大越发散取结果completion.choices[0].message.content2.2 流式调用llm/case2.pyimportosfromopenaiimportOpenAI clientOpenAI(api_keyos.environ[DASHSCOPE_API_KEY],base_urlhttps://ws-ulkao56twirebft4.cn-beijing.maas.aliyuncs.com/compatible-mode/v1,)completionclient.chat.completions.create(modelqwen-plus,messages[{role:system,content:You are a helpful assistant.},{role:user,content:请介绍一下自己}],streamTrue,# 开启流式stream_options{include_usage:True}# 拿 token 用量)chunks[]forchunkincompletion:ifchunk.choices:choicechunk.choices[0]ifchoice.delta:deltachoice.deltaifdelta.content:print(delta.content)chunks.append(delta.content)# 暂存片段res.join(chunks)# 最后 join比 高效print(res)要点streamTrue开启流式stream_options{include_usage: True}才能拿到 token 用量逐块迭代chunk.choices[0].delta.content是这一小段增量文本用list暂存片段最后.join(chunks)拼接——比字符串逐次高效三、封装成 FastAPI 接口3.1 请求体校验app/schemas/llm_case1.pyfrompydanticimportBaseModel,FieldclassLLMCase1(BaseModel):question:strField(...,title问题,description问题)只收一个question字段用 Pydantic 做参数校验。3.2 接口app/apis/llm/case1_api.pycase1非流式一次性返回importosfromfastapiimportAPIRouterfromopenaiimportOpenAIfromapp.schemas.llm_case1importLLMCase1 llm_day01_routerAPIRouter(prefix/llm-day01,tags[LLM-DAY01])llm_day01_router.post(/case1,summaryLLM-DAY01-CASE1)asyncdefcase1_api(llmCase1Request:LLMCase1):clientOpenAI(api_keyos.getenv(DASHSCOPE_API_KEY),base_urlhttps://ws-ulkao56twirebft4.cn-beijing.maas.aliyuncs.com/compatible-mode/v1,)completionclient.chat.completions.create(modelqwen-plus,messages[{role:system,content:你是一个智能助手},{role:user,content:llmCase1Request.question},],temperature0.75,)ai_replycompletion.choices[0].message.contentreturn{code:1,message:请求成功,data:{ai_reply:ai_reply},}返回沿用项目统一的响应信封{code, message, data}。case2SSE 流式输出fromstarlette.responsesimportStreamingResponsedefstream_chunk(user_question:str):clientOpenAI(api_keyos.environ[DASHSCOPE_API_KEY],base_urlhttps://ws-ulkao56twirebft4.cn-beijing.maas.aliyuncs.com/compatible-mode/v1,)completionclient.chat.completions.create(modelqwen-plus,messages[{role:system,content:You are a helpful assistant.},{role:user,content:user_question}],streamTrue,stream_options{include_usage:True})forchunkincompletion:ifchunk.choices:choicechunk.choices[0]ifchoice.delta:deltachoice.deltaifdelta.content:yieldfdata:{delta.content}\n\n# SSE 标准格式data: xxx\n\nyielddata: [done]\n\n# 结束标记llm_day01_router.post(/case2,summaryLLM-DAY01-CASE2)asyncdefcase2_api(llmCase1Request:LLMCase1):returnStreamingResponse(contentstream_chunk(llmCase1Request.question),media_typetext/event-stream)要点stream_chunk是一个生成器函数含yield逐片产出 SSE 文本SSE 格式固定为data: 内容\n\n必须是双换行前端EventSource才能正确切分事件最后yield data: [done]\n\n作为结束信号通知前端回答完毕用StreamingResponse(content生成器, media_typetext/event-stream)把生成器包成流式响应FastAPI 普通函数不能yield流式必须返回StreamingResponse3.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 流式四、踩坑清单重点API Key 别硬编码用环境变量。但注意代码里两处不一致——非流式用os.getenv缺省返回None静默失败生成器里用os.environ[...]缺省会抛KeyError。建议统一用os.getenv并显式判断为空时返回友好错误。base_url 与 Key 地域强绑定百炼的 base_url 必须是兼容模式/v1且要和 API Key 所属地域一致否则鉴权直接失败。SSE 必须双换行data: xxx\n\n少一个\n前端EventSource解析不出事件。流式结束要发标记最后yield data: [done]\n\n让前端知道回答结束、关闭连接。流式拿 token 用量stream_options{include_usage: True}才有chunk.usage。字符串拼接用 list join流式片段别用用chunks.append后.join(chunks)性能更好。FastAPI 流式不能用普通函数 yield必须返回StreamingResponse并传入生成器。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 走环境变量。