用 FastAPI + SSE 把大模型的「打字机」效果搬进你自己的后端:完整可运行实战
引言
如果你用过官方对话页面,一定见过那种「一个字一个字往外蹦」的打字机效果。它不只是好看——首 token 延迟被摊薄到每一个字上,用户感知到的响应速度会明显变快,也更容易在长回答时不至于盯着一片空白发呆。但当你把模型接到自己的业务后端,默认拿到的往往是一整段文本挤在最后一次响应里。本文用 FastAPI + Server-Sent Events(SSE),手把手实现一个可独立运行的流式输出接口,并配上一个能跑的极简前端,所有代码直接复制即可用。
一、为什么选 SSE 而不是 WebSocket
流式推送有两个常见方案:WebSocket 和 SSE。WebSocket 是双向全双工通道,适合游戏、聊天室这种两端高频互发的场景;而大模型回答是「服务端单向持续推送、客户端基本只收不发」,SSE 刚好匹配——它基于普通 HTTP,浏览器原生 EventSource 即可接收,不需要额外握手库,nginx 也能直接透传。对大多数 AI 应用来说,SSE 是更轻的选择。
二、后端:FastAPI 流式接口
核心思路是让接口返回一个 StreamingResponse,内容类型为 text/event-stream,内部用生成器逐段 yield。下面这段代码对接 OpenAI 兼容接口,把模型返回的增量 token 包成 SSE 格式推给前端:
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import httpx, os, json
app = FastAPI()
API_KEY = os.getenv("OPENAI_API_KEY")
BASE_URL = "https://api.openai.com/v1"
def event_generator(prompt: str):
payload = {
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": prompt}],
"stream": True,
}
with httpx.Client(timeout=60) as client:
with client.stream("POST", f"{BASE_URL}/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}"},
json=payload) as r:
for line in r.iter_lines():
if not line or not line.startswith("data:"):
continue
data = line[len("data:"):].strip()
if data == "[DONE]":
yield "data: [DONE]\\n\\n"
break
delta = json.loads(data)["choices"][0]["delta"].get("content", "")
if delta:
yield f"data: {delta}\\n\\n"
@app.get("/chat/stream")
def chat_stream(q: str):
return StreamingResponse(event_generator(q),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache",
"X-Accel-Buffering": "no"})
关键点只有三个:stream=True 让上游增量返回;iter_lines 逐行读 SSE 分片;X-Accel-Buffering: no 是给 nginx 的反向代理关缓冲用的,漏了它前端会等全部结束才一次性收到。
三、前端:EventSource 收流
前端用浏览器原生 EventSource 即可,不需要任何依赖:
<div id="box"></div>
<script>
const es = new EventSource("/chat/stream?q=" + encodeURIComponent("用一句话解释什么是 SSE"));
es.onmessage = (e) => {
if (e.data === "[DONE]") { es.close(); return; }
document.getElementById("box").innerText += e.data;
};
</script>
innerText += 就是打字机效果本身——每收到一段就追加到页面。
四、生产化三道避坑
总结
流式输出不是炫技,它是改善大模型产品体感的最低成本改动之一。FastAPI 的 StreamingResponse 配合 SSE,几十行就能把「挤牙膏」变成「打字机」。先把最小闭环跑通,再补超时、背压、鉴权三件套,你的接口就能稳稳地「边想边说」了。
网硕互联帮助中心



评论前必须登录!
注册