FastAPI + OpenAI SDK 实战：接入 DeepSeek 大模型与流式问答全流程拆解

项目实践FastAPI 接入大模型与 LangChain 配置FastAPI OpenAI SDK 实战接入 DeepSeek 大模型与流式问答全流程拆解一、前言介绍1.1 背景1.2 功能概览1.3 调用模型总览二、环境准备OpenAI 依赖下载与配置2.1 下载安装 OpenAI SDK2.2 配置 API Key环境变量2.3 配置兼容端点 base_url2.4 目录结构三、知识点讲解3.1 OpenAI 兼容模式compatible-mode3.2 MaaS 端点与 DeepSeek 模型3.3 流式 SSE四、代码逻辑拆解严格对照项目代码4.1 请求体模型schemas4.2 密钥读取与客户端初始化4.3 一次问答接口case14.4 流式问答接口case24.5 路由注册到 FastAPI4.6 最小可运行验证脚本case.pyFastAPI OpenAI SDK 实战接入 DeepSeek 大模型与流式问答全流程拆解一、前言介绍1.1 背景后端服务迟早要接大模型智能问答、简历润色、岗位推荐话术生成都离不开一次把用户输入发给模型、把模型回答拿回来的往返。本文聚焦最朴素也最常用的一条链路——用 OpenAI 官方 SDK 调通一个兼容 OpenAI 协议的大模型接口并让它在 FastAPI 里以接口形式对外提供1.2 功能概览一次问答接口接收问题文本调用模型返回完整回答流式问答接口same 模型但以 SSEtext/event-stream逐字吐字前端体验接近打字机入参校验用 Pydantic 模型约束请求体LangChain 配置用ChatDeepSeek封装同一模型便于后续接链Chain、记忆Memory、检索Retriever。1.3 调用模型总览客户端 → FastAPI 路由async def → Pydantic 校验入参 → OpenAI 客户端 / LangChain ChatModel → 大模型兼容端点base_url → 模型DeepSeek → 同步返回 or SSE 流式返回二、环境准备OpenAI 依赖下载与配置这一节把OpenAI 这套东西怎么装、怎么配单独拎出来讲清楚和业务代码拆解分开方便照抄。2.1 下载安装 OpenAI SDKpipinstallopenai就这一个包项目里所有大模型调用都靠它。它不只是调 OpenAI 官方而是任何兼容 OpenAI 协议的服务都能调——这是后面能直连百炼 MaaS 的前提。2.2 配置 API Key环境变量密钥不放代码里从环境变量读# 项目代码里实际读取的变量名 DASHSCOPE_API_KEYsk-xxxxxxxx代码中的位置importos raw_keyos.getenv(DASHSCOPE_API_KEY)api_keyraw_key.strip()第 1 行从环境变量取百炼 API Key第 2 行strip()去掉首尾空白防止复制 Key 时带入换行导致鉴权失败。2.3 配置兼容端点 base_url项目代码里写死的端点是阿里云百炼的 MaaS 兼容地址base_urlhttps://ws-xxxx.cn-beijing.maas.aliyuncs.com/compatible-mode/v1/compatible-mode/v1是兼容开关缺了 SDK 会按官方域名去请求必然 404。模型名跟着这个端点走项目里填的是deepseek-v4-pro。2.4 目录结构app/ ├── apis/ │ └── llm/ │ └── case1.py # 大模型接口一次问答 流式问答 ├── schemas/ │ └── llm_case1.py # 请求体模型 main.py # 路由注册 case.py # 最小可运行验证脚本脱离 Web 框架三、知识点讲解3.1 OpenAI 兼容模式compatible-modeOpenAI 把对话接口定义成一套固定的请求/响应形状messages列表 model字段返回choices[0].message.content。只要厂商把自家接口伪装成这个形状OpenAI 官方 SDK 就能原样调用只需要把base_url指过去。设计意识客户端与厂商解耦。今天接这个端点、明天换另一个只改base_url和model业务代码一行不动。3.2 MaaS 端点与 DeepSeek 模型项目里指向的是阿里云百炼的 MaaS 兼容端点模型名填deepseek-v4-probase_urlhttps://ws-xxxx.cn-beijing.maas.aliyuncs.com/compatible-mode/v1modeldeepseek-v4-pro模型名必须与端点所在平台提供的清单一致写错会返回model not found。本文代码里就是deepseek-v4-pro不另作替换。3.3 流式 SSE非流式接口等模型把整段话说完再返回延迟高、首字时间长。流式接口让模型边生成边回传HTTP 上用SSEServer-Sent Events承载每一片以data: 内容\n\n格式推给前端结束发data: [DONE]\n\n。FastAPI 用StreamingResponse配合生成器即可实现。四、代码逻辑拆解严格对照项目代码4.1 请求体模型schemasclassLLMCase1(BaseModel):question:strField(...,description问题)第 1 行BaseModel继承Pydantic v2 的请求体第 2 行question用Field(...)必填缺字段 FastAPI 自动返回 422省去手写校验。另一个预留的会话模型classLLMCase2(BaseModel):user_id:strField(...,description用户ID)session_id:strField(...,description会话ID)message:strField(...,description消息)三个字段全必填为后续多轮对话 会话隔离预留结构本篇先不展开多轮记忆。4.2 密钥读取与客户端初始化importosfromopenaiimportOpenAI raw_keyos.getenv(DASHSCOPE_API_KEY)api_keyraw_key.strip()clientOpenAI(api_keyapi_key,base_urlhttps://ws-xxxx.cn-beijing.maas.aliyuncs.com/compatible-mode/v1,)第 3 行从环境变量取密钥不落代码第 4 行strip()去掉首尾空白防止复制 Key 时带入换行导致鉴权失败第 6–9 行构造 OpenAI 客户端base_url指向 MaaS 兼容端点api_key作为 Bearer 令牌随请求发出。设计意识客户端构造成本低但每次请求都 new 一个没必要高并发下建议做成模块级单例或连接池避免重复握手。4.3 一次问答接口case1llm1_router.post(/case1,summaryLLM1-case1)asyncdefcase1_api(llm1:LLMCase1):completionclient.chat.completions.create(modeldeepseek-v4-pro,messages[{role:system,content:You are a helpful assistant.},{role:user,content:llm1.question},],)ai_replycompletion.choices[0].message.contentreturn{code:1,message:请求成功,data:{ai_reply:ai_reply}}第 1 行prefix/llm1的路由组下挂/case1summary会显示在 Swagger第 2 行用 Pydantic 模型收参自动校验第 4 行create发起一次对话model指定deepseek-v4-pro第 5–9 行messages是角色数组system设定助手人设user放用户问题——这是 OpenAI 协议的标准对话结构第 10 行choices[0].message.content取模型文本回答第 11–13 行包成{code, message, data}统一返回体前端按data.ai_reply取答案。4.4 流式问答接口case2defstream_chunk(user_querstr:str):clientOpenAI(api_keyapi_key,base_urlBASE_URL)completionclient.chat.completions.create(modeldeepseek-v4-pro,messages[{role:system,content:You are a helpful assistant.},{role:user,content:user_querstr},],streamTrue,stream_options{include_usage:True},)foriincompletion:ifi.choices:choisei.choices[0]ifchoise.delta:deitachoise.deltaifdeita.content:yieldfdata:{deita.content}\n\nyielddata: [DONE]\n\n第 5 行streamTrue打开流式SDK 不再等完整结果而是返回一个可迭代对象每轮给一片增量第 6 行stream_options{include_usage: True}让最后一片带上 token 用量统计计费/监控用第 8–13 行遍历增量i.choices[0].delta.content是这一片增量文字用if层层判空是因为心跳包、首片、结束片可能choices或delta为空第 14 行yield fdata:{内容}\n\n按 SSE 格式吐字\n\n是 SSE 的分片分隔符缺了前端收不到第 15 行结束标志data: [DONE]前端据此关闭连接。路由侧用StreamingResponse包裹生成器llm1_router.post(/case2,summary流式回答)asyncdefcase2_api(llm1:LLMCase1):returnStreamingResponse(stream_chunk(llm1.question),media_typetext/event-stream)media_typetext/event-stream告诉浏览器这是 SSE 流否则会被当成普通文本一次性缓冲。4.5 路由注册到 FastAPIfromapp.apis.llm.case1importllm1_router app.include_router(llm1_router)一行把大模型路由组挂进应用/llm1/case1、/llm1/case2即生效Swagger 里归到文本处理标签下。4.6 最小可运行验证脚本case.py脱离 Web 框架单独验证连通性importosfromopenaiimportOpenAI raw_keyos.getenv(DASHSCOPE_API_KEY)api_keyraw_key.strip()clientOpenAI(api_keyapi_key,base_urlhttps://ws-xxxx.cn-beijing.maas.aliyuncs.com/compatible-mode/v1,)defget_response():completionclient.chat.completions.create(modeldeepseek-v4-pro,messages[{role:system,content:You are a helpful assistant.},{role:user,content:国内大模型哪个最好},],)returncompletion.choices[0].message.contentprint(get_response())与接口代码共用同一套客户端初始化逻辑只是把问题写死、直接print用来在不起 FastAPI 的情况下先确认 Key、端点、模型名三件套是否配通是排障第一招。