AI应用流式响应实现：SSE + LLM打字机效果完整代码
用AI API做产品的人迟早会遇到一个问题：模型生成一段长回答要好几秒，用户盯着空白屏幕等，体验很差。ChatGPT那种逐字输出的效果，技术上怎么实现？答案就是Server-Sent Events，简称SSE。
这篇文章给出从前端到后端的完整实现代码。不是讲理论，是直接能跑的代码。拿来改改就能用进你的AI产品里。
SSE比WebSocket更适合LLM流式输出
很多人第一反应是用WebSocket做双向通信。但对LLM流式输出这个场景来说，SSE更合适。
原因是LLM流式输出的数据流是单向的：服务端往客户端推token，客户端不需要在生成过程中往回发消息。SSE正是为这种单向推送设计的。WebSocket的优势在于双向通信，但它的协议更重，需要握手升级、维护连接状态、处理心跳包。
根据MDN Web Docs，SSE使用text/event-stream MIME类型，每个事件以双换行符结束。浏览器原生支持EventSource API，不用装任何库。OpenAI官方文档也明确说明，他们的streaming基于HTTP streaming（stream=true）over SSE实现。
一个要注意的限制：MDN文档指出，HTTP/1.1下SSE有最大6个并发连接的限制（每个浏览器+域名）。如果你开多个标签页同时用SSE，可能遇到连接打不开的问题。HTTP/2下默认最大100个并发流，基本够用。现在大部分部署平台都支持HTTP/2了。
后端：Python FastAPI流式接口
后端用Python的FastAPI实现。选FastAPI是因为它原生支持异步和StreamingResponse，不需要额外装东西。
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import OpenAI
import json
app = FastAPI()
client = OpenAI(api_key="your-api-key")
async def generate_stream(prompt: str):
stream = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
stream=True,
)
for chunk in stream:
if chunk.choices[0].delta.content is not None:
data = json.dumps({"text": chunk.choices[0].delta.content})
yield f"data: {data}\n\n"
yield "data: [DONE]\n\n"
@app.get("/chat")
async def chat(prompt: str):
return StreamingResponse(
generate_stream(prompt),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no",
}
)几个关键点要说明。
media_type必须设为text/event-stream，浏览器才会用EventSource处理这个响应。不设的话浏览器当普通HTTP响应处理，不会流式渲染。
X-Accel-Buffering: no这个header是给Nginx看的。默认Nginx会缓冲后端响应，攒够一定量才往客户端发。加上这个header告诉Nginx别缓冲，收到就转发。如果你用Cloudflare，默认不缓冲SSE响应，不用管这个。
Cache-Control: no-cache防止浏览器或CDN缓存SSE响应。每个请求都是新的流，不能复用缓存。
前端：EventSource逐字渲染
前端用浏览器原生的EventSource API。不用装任何npm包。
const source = new EventSource("/chat?prompt=" + encodeURIComponent(userInput));
let fullText = "";
const outputEl = document.getElementById("output");
source.onmessage = (event) => {
if (event.data === "[DONE]") {
source.close();
return;
}
const data = JSON.parse(event.data);
fullText += data.text;
outputEl.textContent = fullText;
};
source.onerror = () => {
source.close();
};这段代码做三件事：建立连接、接收每个chunk拼接到fullText、渲染到页面上。
一个常见问题：EventSource的onmessage只在服务端发送不带event:字段的消息时触发。如果服务端指定了事件类型（比如event: token），需要用addEventListener("token", handler)监听。上面的后端代码没指定事件类型，所以onmessage就能收到。
另一个注意点：EventSource默认会自动重连。连接断开后，浏览器3秒后自动重发请求。对LLM流式输出来说这通常不是你要的行为。生成到一半断了，重连后从头开始生成没有意义。在onerror里手动调用source.close()阻止自动重连。
进阶：支持中断生成
用户可能想在生成过程中点停止按钮。EventSource本身不支持客户端主动关闭连接，但source.close()可以断开。服务端需要检测客户端断开。
from starlette.requests import Request
async def generate_stream(prompt: str, request: Request):
stream = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
stream=True,
)
try:
for chunk in stream:
if await request.is_disconnected():
stream.close()
break
if chunk.choices[0].delta.content is not None:
data = json.dumps({"text": chunk.choices[0].delta.content})
yield f"data: {data}\n\n"
except Exception as e:
yield f"data: {json.dumps({'error': str(e)})}\n\n"
yield "data: [DONE]\n\n"
@app.get("/chat")
async def chat(request: Request, prompt: str):
return StreamingResponse(
generate_stream(prompt, request),
media_type="text/event-stream",
)request.is_disconnected()是FastAPI/Starlette提供的异步方法，检测客户端是否断开连接。客户端调用source.close()后，服务端下次循环到这个检查时返回True，退出循环并关闭OpenAI的stream。
不调用stream.close()的话，OpenAI那边还会继续生成token，你的API额度继续扣。虽然客户端看不到了，但费用在跑。这一点很多人踩坑。
部署踩坑清单
第一，反向代理缓冲。Nginx默认缓冲upstream响应。除了加X-Accel-Buffering: no header，也可以在nginx.conf里针对location加proxy_buffering off。Cloudflare Pages和Workers默认不缓冲，但如果用了CDN缓存规则，要确保SSE路由不被缓存。
第二，超时设置。LLM生成长回答可能需要30秒以上。Nginx的proxy_read_timeout默认60秒，生成超时60秒的响应会被截断。设成300秒比较安全。
第三，CORS问题。前端和API不在同一个域名下时，EventSource需要服务端返回正确的CORS header。跨域SSE不支持withCredentials发送Cookie，认证用Authorization header或URL参数传递token。
第四，连接数限制。HTTP/1.1下每个域名最多6个SSE连接。页面同时开多个聊天窗口可能触发限制。用HTTP/2部署，或者用单个SSE连接多路复用。
成本优化建议
流式响应不影响API调用费用。OpenAI官方文档说明，streaming和非streaming的token计费相同。但流式模式有个隐性成本：如果用户中途停止生成，已经生成的token仍然计费。所以做好中断逻辑不只是UX问题，也直接影响成本。
一个可操作的优化：在前端限制用户输入长度。长prompt不只增加输入token费用，生成时间也更长，用户更可能中途停止。用maxlength属性或前端校验控制输入在2000字符以内，大部分场景够用了。
