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

一、前言介绍

1.1 背景

后端服务迟早要接大模型：智能问答、简历润色、岗位推荐话术生成，都离不开一次"把用户输入发给模型、把模型回答拿回来"的往返。本文聚焦最朴素也最常用的一条链路——用 OpenAI 官方 SDK 调通一个兼容 OpenAI 协议的大模型接口，并让它在 FastAPI 里以接口形式对外提供

1.2 功能概览

• 一次问答接口：接收问题文本，调用模型，返回完整回答；
• 流式问答接口：same 模型，但以 SSE（text/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 SDK

pipinstallopenai

就这一个包，项目里所有大模型调用都靠它。它不只是调 OpenAI 官方，而是"任何兼容 OpenAI 协议的服务"都能调——这是后面能直连百炼 MaaS 的前提。

2.2 配置 API Key（环境变量）

密钥不放代码里，从环境变量读：

# 项目代码里实际读取的变量名 DASHSCOPE_API_KEY=sk-xxxxxxxx

代码中的位置：

importos raw_key=os.getenv("DASHSCOPE_API_KEY")api_key=raw_key.strip()

• 第 1 行：从环境变量取百炼 API Key；
• 第 2 行：strip()去掉首尾空白，防止复制 Key 时带入换行导致鉴权失败。

2.3 配置兼容端点 base_url

项目代码里写死的端点是阿里云百炼的 MaaS 兼容地址：

base_url="https://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-mode）

OpenAI 把对话接口定义成一套固定的请求/响应形状：messages列表 +model字段，返回choices[0].message.content。只要厂商把自家接口"伪装"成这个形状，OpenAI 官方 SDK 就能原样调用，只需要把base_url指过去。

设计意识：客户端与厂商解耦。今天接这个端点、明天换另一个，只改base_url和model，业务代码一行不动。

3.2 MaaS 端点与 DeepSeek 模型

项目里指向的是阿里云百炼的 MaaS 兼容端点，模型名填deepseek-v4-pro：

base_url="https://ws-xxxx.cn-beijing.maas.aliyuncs.com/compatible-mode/v1"model="deepseek-v4-pro"

模型名必须与端点所在平台提供的清单一致，写错会返回model not found。本文代码里就是deepseek-v4-pro，不另作替换。

3.3 流式 SSE

非流式接口等模型把整段话说完再返回，延迟高、首字时间长。流式接口让模型"边生成边回传"，HTTP 上用SSE（Server-Sent Events）承载：每一片以data: 内容\n\n格式推给前端，结束发data: [DONE]\n\n。FastAPI 用StreamingResponse配合生成器即可实现。

四、代码逻辑拆解（严格对照项目代码）

4.1 请求体模型（schemas）

classLLMCase1(BaseModel):question:str=Field(...,description="问题")

• 第 1 行：BaseModel继承，Pydantic v2 的请求体；
• 第 2 行：question用Field(...)必填，缺字段 FastAPI 自动返回 422，省去手写校验。

另一个预留的会话模型：

classLLMCase2(BaseModel):user_id:str=Field(...,description="用户ID")session_id:str=Field(...,description="会话ID")message:str=Field(...,description="消息")

• 三个字段全必填，为后续"多轮对话 + 会话隔离"预留结构（本篇先不展开多轮记忆）。

4.2 密钥读取与客户端初始化

importosfromopenaiimportOpenAI raw_key=os.getenv("DASHSCOPE_API_KEY")api_key=raw_key.strip()client=OpenAI(api_key=api_key,base_url="https://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 一次问答接口（case1）

@llm1_router.post("/case1",summary="LLM1-case1")asyncdefcase1_api(llm1:LLMCase1):completion=client.chat.completions.create(model="deepseek-v4-pro",messages=[{"role":"system","content":"You are a helpful assistant."},{"role":"user","content":llm1.question},],)ai_reply=completion.choices[0].message.contentreturn{"code":1,"message":"请求成功","data":{"ai_reply":ai_reply}}

• 第 1 行：prefix="/llm1"的路由组下挂/case1，summary会显示在 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 流式问答接口（case2）

defstream_chunk(user_querstr:str):client=OpenAI(api_key=api_key,base_url=BASE_URL)completion=client.chat.completions.create(model="deepseek-v4-pro",messages=[{"role":"system","content":"You are a helpful assistant."},{"role":"user","content":user_querstr},],stream=True,stream_options={"include_usage":True},)foriincompletion:ifi.choices:choise=i.choices[0]ifchoise.delta:deita=choise.deltaifdeita.content:yieldf"data:{deita.content}\n\n"yield"data: [DONE]\n\n"

• 第 5 行：stream=True打开流式，SDK 不再等完整结果，而是返回一个可迭代对象，每轮给一片增量；
• 第 6 行：stream_options={"include_usage": True}让最后一片带上 token 用量统计（计费/监控用）；
• 第 8–13 行：遍历增量，i.choices[0].delta.content是"这一片增量文字"；用if层层判空，是因为心跳包、首片、结束片可能choices或delta为空；
• 第 14 行：yield f"data:{内容}\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_type="text/event-stream")

• media_type="text/event-stream"告诉浏览器这是 SSE 流，否则会被当成普通文本一次性缓冲。

4.5 路由注册到 FastAPI

fromapp.apis.llm.case1importllm1_router app.include_router(llm1_router)

• 一行把大模型路由组挂进应用，/llm1/case1、/llm1/case2即生效，Swagger 里归到"文本处理"标签下。

4.6 最小可运行验证脚本（case.py）

脱离 Web 框架，单独验证连通性：

importosfromopenaiimportOpenAI raw_key=os.getenv("DASHSCOPE_API_KEY")api_key=raw_key.strip()client=OpenAI(api_key=api_key,base_url="https://ws-xxxx.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",)defget_response():completion=client.chat.completions.create(model="deepseek-v4-pro",messages=[{"role":"system","content":"You are a helpful assistant."},{"role":"user","content":"国内大模型哪个最好？"},],)returncompletion.choices[0].message.contentprint(get_response())

• 与接口代码共用同一套客户端初始化逻辑，只是把问题写死、直接print；
• 用来在不起 FastAPI 的情况下先确认 Key、端点、模型名三件套是否配通，是排障第一招。