AI 配方：使用 FastAPI 流式输出 OpenAI 响应

一份即用配方：FastAPI 端点通过 SSE 将 LLM token 流式传输到浏览器，生产细节（刷新行为、断连、代理缓冲）已处理完毕。十分钟搞定。

服务端

python
main.py
import json
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI
from pydantic import BaseModel
app = FastAPI()
client = AsyncOpenAI()  # 读取 OPENAI_API_KEY
class ChatIn(BaseModel):
    prompt: str
@app.post('/chat/stream')
async def chat_stream(body: ChatIn, request: Request):
    async def gen():
        stream = await client.chat.completions.create(
            model='gpt-4o-mini',
            messages=[{'role': 'user', 'content': body.prompt}],
            stream=True,
        )
        async for chunk in stream:
            # 客户端断开？停止生成，避免浪费 token。
            if await request.is_disconnected():
                break
            token = chunk.choices[0].delta.content or ''
            if token:
                yield f'data: {json.dumps({"token": token})}\n\n'
        yield 'data: [DONE]\n\n'
    return StreamingResponse(
        gen(),
        media_type='text/event-stream',
        headers={
            'Cache-Control': 'no-cache',
            'X-Accel-Buffering': 'no',   # 告知 nginx 不要缓冲
        },
    )

运行：uvicorn main:app --reload。关键点：

async def + AsyncOpenAI — 如果使用同步客户端，会阻塞事件循环，导致服务器上其他请求停滞（经典错误；参见 同步 vs 异步 LLM 调用）。

request.is_disconnected() — 没有它，用户关闭标签页后，生成过程仍会继续运行，浪费你的费用。

data: ...\n\n 格式 — 空行终止每个 SSE 事件；忘记它浏览器将收不到任何内容。

每个 token 用 JSON 包裹 — 原始 token 中的换行符会破坏 SSE 格式；json.dumps 确保安全。

客户端

POST 请求体无法与浏览器原生 EventSource 配合，因此使用 fetch 读取流：

javascript
async function chat(prompt, onToken) {
  const resp = await fetch('/chat/stream', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ prompt }),
  });
  const reader = resp.body.getReader();
  const decoder = new TextDecoder();
  let buf = '';
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buf += decoder.decode(value, { stream: true });
    const events = buf.split('\n\n');
    buf = events.pop();                       // 保留不完整的尾部
    for (const ev of events) {
      const data = ev.replace(/^data: /, '');
      if (data === '[DONE]') return;
      onToken(JSON.parse(data).token);
    }
  }
}
chat('用两句话解释 SSE', t => { output.textContent += t; });

缓冲区拆分操作很重要：网络数据块与 SSE 事件边界不对齐，因此始终在 \n\n 处拆分并保留剩余部分。

生产环境检查清单

nginx：X-Accel-Buffering: no 头部按响应处理；否则为该路由设置 proxy_buffering off;。

Serverless/PaaS：确认你的平台支持流式响应——有些默认会缓冲整个响应体。

流中错误：HTTP 状态码已发送，因此改为发送错误事件——在循环的 try/except 中执行 yield f'data: {json.dumps({"error": str(e)})}\n\n'，并在客户端处理。

认证：这是一个普通的 POST 请求——你现有的 FastAPI 认证依赖项无需修改即可使用。

变体

任意提供商：Anthropic 的 SDK 以相同方式流式传输（async with client.messages.stream(...)），任何兼容 OpenAI 的端点——包括本地 Ollama——只需更改 base_url 即可。

使用 Next.js 而非 FastAPI：Vercel AI SDK 封装了整个配方——参见 Vercel AI SDK vs LangChain.js。

为什么用 SSE 而不是 WebSocket：单向 token 流不需要 socket；更深入的解释见 SSE 指南 和 流式 vs 轮询。

---

*最后更新：2026 年 6 月。*