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

pip install openai

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

2.2 配置 API Key（环境变量）

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

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

代码中的位置：

import os
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）

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

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

另一个预留的会话模型：

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

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

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

import os
from openai import OpenAI

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")
async def case1_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.content
    return {"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）

def stream_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},
    )
    for i in completion:
        if i.choices:
            choise = i.choices[0]
            if choise.delta:
                deita = choise.delta
                if deita.content:
                    yield f"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="流式回答")
async def case2_api(llm1: LLMCase1):
    return StreamingResponse(stream_chunk(llm1.question), media_type="text/event-stream")

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

4.5 路由注册到 FastAPI

from app.apis.llm.case1 import llm1_router
app.include_router(llm1_router)

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

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

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

import os
from openai import OpenAI

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",
)

def get_response():
    completion = client.chat.completions.create(
        model="deepseek-v4-pro",
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": "国内大模型哪个最好？"},
        ],
    )
    return completion.choices[0].message.content

print(get_response())

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