Server-Sent Events (SSE) 实现流式输出

2025-01-14

字数：875字 | 预计阅读时长：3分钟

在 ChatGPT 出现之前，大部分 Web 接口都是“请求-响应”模式：用户发一个请求，服务器处理完（可能需要几秒），然后一次性把结果扔回来。但在大模型时代，生成一段长文本可能需要 10 秒甚至更久。如果让用户盯着空白屏幕等 10 秒，体验会非常糟糕。于是，SSE (Server-Sent Events) 再次回到了聚光灯下。它允许服务器一边生成内容，一边通过长连接把数据“推”给前端，也就是我们看到的“打字机效果”。

1.什么是 SSE？

SSE（Server-Sent Events）是一种基于 HTTP 的单向通信机制。

HTTP 标准：它不需要像 WebSocket 那样复杂的握手协议，它本质上就是一个长连接的 HTTP 请求。
单向通信：服务端 -> 客户端。这非常适合 AI 生成内容的场景（用户问一次，AI 持续吐字）。
简单易用：浏览器原生支持 EventSource API，后端实现也只是多了一个 yield。

2.代码展示

之前做的一个AI知识库项目需要用到SSE实现流式输出，这里展示一下SSE流式输出部分代码。

后端实现：FastAPI + sse_starlette

from sse_starlette.sse import EventSourceResponse
from langchain_core.messages import HumanMessage, AIMessage

# 定义生成器函数
def stream_generator(session_id: str, text: str):
    # 1. 获取上下文（实现记忆功能）
    history = get_history(session_id)
    current_messages = history + [HumanMessage(content=text)]
    
    # 2. 调用大模型流式接口
    for chunk in model.stream(current_messages):
        content = chunk.content
        if content:
            # 3. 按照 SSE 格式推送数据
            # 格式通常是: data: <内容>\n\n
            yield {
                "event": "message",
                "data": content
            }
            
    # 4. 结束标记
    yield {"data": "[[END]]"}

# API 接口
@app.get("/chat_stream")
async def chat_stream(text: str, session_id: str):
    return EventSourceResponse(stream_generator(session_id, text))

关键点解析：

yield: 它的作用是“产出”数据但不结束函数，函数会暂停在这里，直到下一次循环。这就实现了“生成一点，发送一点”。
EventSourceResponse: 自动处理 HTTP Header（如 Content-Type: text/event-stream），告诉浏览器这是一个流。

前端实现：原生 EventSource + 气泡 UI

前端部分，使用 EventSource。为了支持 Header 传参（如 Token），推荐使用 event-source-polyfill。

建立连接

// 引入 polyfill
// <script src="eventsource.min.js"></script>

const url = new URL('http://127.0.0.1:9000/chat_stream');
url.searchParams.set('text', "你好");
url.searchParams.set('session_id', "user_123");

// 建立连接
const es = new EventSourcePolyfill(url, {
    headers: { Authorization: 'Bearer my-token' }
});

接收数据与渲染

const aiMessageDiv = document.createElement('div'); // 创建气泡
let currentText = "";

es.onmessage = (e) => {
    // 1. 判断是否结束
    if (e.data === '[[END]]') {
        es.close(); // 关闭连接
        return;
    }

    // 2. 拼接数据（增量渲染）
    currentText += e.data;
    aiMessageDiv.innerText = currentText; 
    
    // 3. 自动滚动到底部
    scrollToBottom();
};

es.onerror = () => {
    console.log("连接异常");
    es.close();
};

3.遇到的坑与解决方案

我在项目开发过程中遇到了以下问题：

CORS 跨域问题：
- SSE 对跨域非常敏感。需要在 FastAPI 中配置 CORSMiddleware，允许前端域名访问。
连接断开重连：
- 原生 EventSource 会自动重连，这在聊天场景有时不需要（比如回答已经结束了）。一定要在后端发送明确的结束标记（如 [[END]]），前端收到后手动 .close()。
Nginx 缓存缓冲：
- 如果你部署在 Nginx 后面，Nginx 默认会缓存一部分响应再发送，这会导致流式效果失效（变成一卡一卡的）。需要关闭 Nginx 的 proxy_buffering。