聊天流式输出是怎么实现的？SSE 从原理到代码

陈陈老师2026/03/20649 阅读

用 ChatGPT 的时候你会发现回复是流式的，一个字一个字往外蹦。这个背后是 SSE（Server-Sent Events），比 WebSocket 简单得多，一个 HTTP 连接就能搞定。这篇讲讲怎么手写。

SSE 的原理：服务端响应头设置 Content-Type: text/event-stream，然后持续往连接里写数据，格式是 `data: {json}

`。浏览器端的 EventSource API 会自动解析。

服务端（FastAPI 为例）：

Python

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio, json

app = FastAPI()

@app.post("/chat")
async def chat(prompt: str):
    async def gen():
        for token in ["你", "好", "，", "世", "界"]:
            yield f"data: {json.dumps({'token': token}, ensure_ascii=False)}

"
            await asyncio.sleep(0.1)
    return StreamingResponse(gen(), media_type="text/event-stream")

前端：

JavaScript

const es = new EventSource('/chat-stream');
es.onmessage = (e) => {
  const data = JSON.parse(e.data);
  appendText(data.token);
};

几个坑：

第一，EventSource 只支持 GET。如果你想用 POST 传参数，EventSource 干不了，要么把参数放 query string，要么用 fetch + ReadableStream 自己解析：

JavaScript

const res = await fetch('/chat', { method: 'POST', body: JSON.stringify({prompt}) });
const reader = res.body.getReader();
const decoder = new TextDecoder();
while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  processChunk(decoder.decode(value));
}

第二，要设置合适的 flush。如果服务端有代理层（nginx），记得关掉缓冲，不然流式变成一次性输出。nginx 配置 proxy_buffering off。

第三，断线重连。EventSource 自带重连，但如果用 fetch 方案，要自己处理断线逻辑。

流式输出的体验差距真的很大，同样的模型，流式感觉快了三倍。

还没有评论，来抢沙发～

评论0