当你的业务需要把本地部署的大模型（如 Ollama、vLLM）暴露给前端或第三方调用时，直接甩一个裸推理地址既不安全也不利于扩展。FastAPI 能帮我们快速封装一层标准、可鉴权、支持流式大模型对话 API 的服务，本文给出可直接上线的完整实现。
一、为什么不直接调模型，而要包一层 API
本地推理框架（Ollama、vLLM、Xinference）都自带 OpenAI 兼容接口，看似可以直接给前端用。但在真实项目里，裸接口至少有三个问题：第一，没有统一的鉴权，任何人拿到地址就能无限调用，GPU 资源很快被打满；第二，缺少请求审计、限流与日志，出问题无法排查；第三，前端拿到的字段格式五花八门，业务方每次都要适配不同的后端。用 FastAPI 包一层，既能复用各家都支持的 OpenAI 协议，又能把鉴权、限流、流式转发、灰度路由等能力集中落地，后端换模型对调用方完全透明。
本文目标很明确：对外暴露一个与 OpenAI 兼容、支持 token 级流式返回的 /v1/chat/completions 接口，内部把请求转发给本地的 Ollama 或 vLLM，并补齐生产环境必需的鉴权与部署细节。
这种”自有网关 + 本地推理”的架构在三类场景里特别常见：其一是企业内部知识库问答，模型跑在隔离网段，只允许通过网关按部门鉴权后访问；其二是面向用户的智能客服或写作助手，需要把多家模型统一成同一套接口，方便做 A/B 与灰度；其三是降本需求，把公有云 API 调用逐步迁移到自托管模型时，网关层零改动就能切换后端。无论哪种，网关都是承上启下的那一层，值得认真写。
二、项目骨架与环境准备
新建一个最小项目，依赖只有 FastAPI、Uvicorn 以及官方 OpenAI Python SDK（我们用它的异步客户端来对接后端，避免自己拼 HTTP）。把依赖写进 requirements.txt 方便用我们此前介绍的 uv 包管理器一键装好。
# requirements.txt
fastapi==0.115.0
uvicorn[standard]==0.30.6
openai==1.50.0
python-dotenv==1.0.1
# 安装（使用 uv 可显著提速）
# uv pip install -r requirements.txt
项目的目录结构保持扁平即可：main.py 放路由与启动逻辑，config.py 放密钥与后端地址，.env 存放敏感配置。这样后续要接 CI/CD（例如我们讲过的 GitHub Actions 流水线）也很容易。
除了鉴权，还建议在网关层顺手加上请求审计与简易限流：用中间件记录每个 key 的调用次数与时间戳，超过阈值就返回 429。这么做有两个好处，一是防止单个调用方把 GPU 打满影响他人，二是为后续按量计费或配额管理预留数据。这些逻辑和路由解耦，放在依赖函数或中间件里即可，不影响核心转发代码。
三、定义请求模型与鉴权依赖
为了与 OpenAI 协议对齐，请求体采用同样的字段命名。我们用 Pydantic 做校验，并用 APIKeyHeader 实现一个最简鉴权依赖：只要请求头带了合法的 X-API-Key 就放行，否则返回 403。生产环境可把它换成 JWT 或对接网关的统一鉴权。
from fastapi import FastAPI, Depends, HTTPException, Request
from fastapi.security import APIKeyHeader
from pydantic import BaseModel, Field
from typing import List, Optional
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
VALID_KEYS = {"sk-demo-2026"} # 实际请从环境变量或配置中心读取
async def verify_key(key: Optional[str] = Depends(api_key_header)):
if not key or key not in VALID_KEYS:
raise HTTPException(status_code=403, detail="Invalid API Key")
return key
class Message(BaseModel):
role: str
content: str
class ChatRequest(BaseModel):
model: str = "qwen2.5:7b"
messages: List[Message]
stream: bool = True
temperature: float = 0.7
max_tokens: Optional[int] = 1024
四、对接 OpenAI 兼容的本地推理后端
Ollama 与 vLLM 都实现了 OpenAI 兼容协议，区别只是 base_url 和 api_key。用官方 SDK 的异步客户端 AsyncOpenAI 统一对接，业务代码完全不用关心背后跑的是哪家。下面把后端地址抽成配置，方便在 Ollama 与 vLLM 之间切换。
from openai import AsyncOpenAI
# Ollama 本地默认地址
OLLAMA_CLIENT = AsyncOpenAI(
base_url="http://localhost:11434/v1",
api_key="ollama", # Ollama 不校验 key，占位即可
)
# vLLM 启动后默认暴露的 OpenAI 兼容地址
VLLM_CLIENT = AsyncOpenAI(
base_url="http://localhost:8000/v1",
api_key="EMPTY", # vLLM 默认不校验 key
)
def pick_client(model: str) -> AsyncOpenAI:
# 简单的模型路由：名字含 vllm 走 vLLM，否则走 Ollama
return VLLM_CLIENT if "vllm" in model else OLLAMA_CLIENT
如果你已经在用 Dify 配合 Ollama 搭过低代码工作流，这里可以把它看作同一套协议下的”自己手写的服务端版本”——只不过路由、鉴权、限流全部掌握在自己手里。
五、用 SSE 实现 token 流式返回
流式对话的核心是 SSE（Server-Sent Events）。后端以 text/event-stream 持续向前端推送 data: 行，每行是一条 JSON，前端逐块渲染即可实现”打字机”效果。FastAPI 的 StreamingResponse 天然支持这种边生成边返回的模式。注意：必须用异步客户端 + 异步生成器，否则一个请求会阻塞整个 worker。
为什么选 SSE 而不是 WebSocket？因为对话接口本质是”一问一答”的请求响应模型，SSE 复用普通 HTTP 语义，天然兼容 CDN、负载均衡与现有监控，调试时直接用 curl 就能看到流；WebSocket 则要建立长连接状态，跨层运维更复杂。另外，若链路里隔着会断连的代理，可每隔十几秒发一行 : ping 注释保活，避免空闲连接被回收。实践中 SSE 对这类场景是最省心的方案。
import json
from fastapi.responses import StreamingResponse
def sse_payload(delta: str) -> str:
chunk = {"choices": [{"delta": {"content": delta}, "finish_reason": None}]}
return f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"
async def event_generator(req: ChatRequest, client: AsyncOpenAI):
try:
stream = await client.chat.completions.create(
model=req.model,
messages=[m.model_dump() for m in req.messages],
stream=True,
temperature=req.temperature,
max_tokens=req.max_tokens,
)
async for part in stream:
delta = part.choices[0].delta.content or ""
if delta:
yield sse_payload(delta)
yield "data: [DONE]\n\n"
except Exception as e:
yield sse_payload(f"[backend error] {e}")
yield "data: [DONE]\n\n"
六、完整路由与启动
把上面的零件拼起来：一个兼容 OpenAI 的 /v1/chat/completions 路由，非流式请求直接返回完整 JSON，流式请求返回 StreamingResponse。同时加上 CORS 中间件，方便前端（比如我们在 VS Code 里起的前端调试页）跨域调用。
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI(title="LLM Chat Gateway")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], allow_methods=["*"], allow_headers=["*"],
)
@app.post("/v1/chat/completions")
async def chat(req: ChatRequest, _: str = Depends(verify_key)):
client = pick_client(req.model)
if not req.stream:
resp = await client.chat.completions.create(
model=req.model,
messages=[m.model_dump() for m in req.messages],
stream=False,
)
return resp.model_dump()
return StreamingResponse(
event_generator(req, client),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
)
# 启动：uvicorn main:app --host 0.0.0.0 --port 8080 --workers 2
七、前端如何消费流式接口
前端不需要任何额外库，用浏览器原生的 fetch + ReadableStream 即可按行解析 SSE。关键是拿到 response.body.getReader() 后持续读取，按 \n\n 切分事件，遇到 [DONE] 结束。这种写法在所有现代浏览器都可用。
async function streamChat(messages) {
const res = await fetch("/v1/chat/completions", {
method: "POST",
headers: { "Content-Type": "application/json", "X-API-Key": "sk-demo-2026" },
body: JSON.stringify({ model: "qwen2.5:7b", messages, stream: true }),
});
const reader = res.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
while (true) {
const { value, done } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split("\n\n");
buffer = lines.pop() || "";
for (const line of lines) {
const payload = line.trim();
if (!payload.startsWith("data:")) continue;
const data = payload.slice(5).trim();
if (data === "[DONE]") return;
const json = JSON.parse(data);
const token = json.choices?.[0]?.delta?.content || "";
process.stdout.write(token); // 实际项目中替换为追加到 UI
}
}
}
八、生产部署与并发调优
本地 uvicorn 单进程跑 demo 没问题，上线要关注并发、超时与反代缓冲三个点。下表给出关键参数与取值建议，整体思路和我们之前部署 Nginx 反向代理时的优化一致。
当单实例撑不住流量时，横向扩展也很简单：把这份 FastAPI 服务用 Docker 多副本起来，前面挂一层负载均衡（Nginx 或云 LB），由于每个请求都是无状态的，扩容不需要改任何业务代码。如果后端推理本身是瓶颈，再在网关层加一个按模型的请求队列，把并发收敛到后端能承受的窗口内，既能保护 GPU 又能保证先来先服务的公平感。整套链路做到了”前端稳定、网关可控、后端可换”。
| 参数 | 建议值 | 原因 |
|---|---|---|
| uvicorn workers | CPU 核数 / 2，且 ≤ 后端并发上限 | 流式长连接占 worker，过多会撑爆文件描述符 |
| proxy_buffering | off（Nginx 必须关） | 否则响应被缓冲，前端收不到实时流 |
| proxy_read_timeout | 300s 以上 | 长对话生成慢，避免中途断流 |
| client_max_body_size | 1m 足够 | 对话请求体很小，限制可挡滥发 |
| CORS | 生产收紧到具体域名 | demo 用 *，上线务必收敛 |
九、常见坑与排查清单
流式接口上线后，90% 的”收不到流”问题都出在中间层缓冲或异步写法上。下面这张表覆盖了排障优先级最高的情况。
| 现象 | 根因 | 修复 |
|---|---|---|
| 前端一次性收到全部内容，没有打字机效果 | Nginx 开了 proxy_buffering | 加 proxy_buffering off; 与 chunked_transfer_encoding on; |
| 请求卡住直到后端超时 | 用了同步 OpenAI 客户端 / 同步 def | 改用 AsyncOpenAI + async def 路由 |
| 并发一高就 503 | worker 数超过后端承载 | 下调 workers，或加队列限流 |
| 浏览器报 CORS | 中间件未放行对应域名/方法 | 在 CORSMiddleware 显式配置 allow_origins |
| 客户端断开后后端仍在生成 | 未感知断开 | 注入 Request，捕获断开异常提前 return |
十、小结
用 FastAPI 封装一层流式大模型对话 API，本质是把”本地推理能力”变成”可控、可观测、可扩展的 HTTP 服务”。本文从鉴权依赖、OpenAI 兼容客户端、SSE 流式生成、前端消费到生产部署与排障，给出了一条能直接上线的完整路径。把它和 Nginx 反代、CI/CD 流水线组合起来，你就能以很低成本拥有一个专属的对话后端，随时切换 Ollama、vLLM 等不同模型而不动前端一行代码。
