SSE 协议与 AI 流式接口：从 HTTP 长连接到逐字输出

2022 年 11 月，ChatGPT 发布。很多人第一次注意到一个交互细节：屏幕上的文字不是整段跳出来的，而是一个字一个字、像打字机一样"敲"出来的。这个效果的背后不是前端在逐字拆分已经收到的完整响应，而是后端在通过一条长连接，把生成的文字实时地、分块地推送给浏览器。

这条长连接用的不是 WebSocket，而是一个在 HTML5 规范里沉寂了很多年的协议——SSE（Server-Sent Events）。它在 AI 对话场景里重新成了行业首选，但很多人对它底层的 HTTP 机制仍然一知半解。这篇笔记从 HTTP 协议层面开始，把 SSE 的工作原理、和普通 JSON 接口的差异、与 WebSocket 的取舍、以及工程实践中的各种细节彻底梳理清楚。

---

一、SSE 到底是什么：不是"新技术"，是"老规范的新场景"

1.1 规范出处

SSE 的全称是 Server-Sent Events，由 WHATWG 在 HTML5 规范中定义，最早出现在 2012 年左右。它对应的协议规范文档是 WHATWG 的 HTML Living Standard § 9.2（原 W3C 的 EventSource 规范，W3C 版本后来被废弃，以 WHATWG 的为准）。

SSE 的核心设计目标非常简单：让服务器能够通过一条普通的 HTTP 连接，持续向客户端推送文本数据。它不是一个独立的传输协议，而是建立在普通 HTTP/1.1（或 HTTP/2）之上的应用层消息格式。

1.2 SSE 的 HTTP 握手：一条普通的 GET 请求

SSE 连接的建立过程就是一条普通的 HTTP GET 请求：

GET /chat/stream HTTP/1.1
Host: api.example.com
Accept: text/event-stream
Cache-Control: no-cache

服务器如果支持 SSE，返回的响应头是这样的：

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

注意三个关键响应头：

• Content-Type: text/event-stream：告诉客户端，这个响应的内容是 SSE 格式的数据流，不是一次性的 HTML 或 JSON
• Cache-Control: no-cache：告诉浏览器和中间代理不要缓存这个响应，因为数据是实时生成的
• Connection: keep-alive：保持 TCP 连接不关闭，服务器会持续在这条连接上推送数据

另外，如果前面有 Nginx 等反向代理，通常建议在响应头中加上 X-Accel-Buffering: no——它告诉 Nginx 立即关闭这个响应的缓冲，不要等到攒够一整块才转发。很多 SSE 的"卡顿"问题不是服务器生成慢，而是被代理缓冲了。

和普通 JSON 接口最大的区别就在这里：普通接口的响应体是一次性发完的，客户端收到完整的响应后连接就关闭了；SSE 的响应体是流式的——服务器可以随时在连接上追加新的数据块，客户端逐块读取。

1.3 SSE 的数据格式：不是 JSON，是 text/event-stream

SSE 的消息格式是纯文本，不是 JSON。一条 SSE 消息由以下字段组成：

data: {"token": "Hello", "index": 0}

每个字段以 字段名: 值 的形式出现，以 \n\n（两个换行）结束一条消息。支持的字段有：

字段	含义	是否必填
data	消息的数据内容	否（但通常有）
id	消息的事件 ID，用于断线重连	否
event	事件类型名称，前端用 addEventListener 监听	否
retry	断线后重新连接的时间间隔（毫秒）	否

一个更复杂的例子：

event: token
data: {"text": "Hello", "index": 0}
id: 1
retry: 3000

event: token
data: {"text": " world", "index": 1}
id: 2

event: done
data: {"finish_reason": "stop"}
id: 3

这里服务器发了三条消息：

• 第一条是 token 事件，携带了 "Hello" 这个 token，事件 ID 是 1，如果连接断了 3 秒后重连
• 第二条是 token 事件，携带了 " world"
• 第三条是 done 事件，表示流结束

前端用 EventSource 监听：

const source = new EventSource('/chat/stream');

source.addEventListener('token', (event) => {
    const data = JSON.parse(event.data);
    appendToChatWindow(data.text);  // 把文字追加到聊天窗口
});

source.addEventListener('done', (event) => {
    source.close();  // 流结束，关闭连接
});

注意 SSE 的 data 字段是纯文本。如果你要传结构化数据，需要自己把 JSON 字符串化后放到 data 里，前端再 JSON.parse 解析。

---

二、SSE 和普通 JSON 接口的核心差异

2.1 连接模型：短连接 vs 长连接

普通 JSON 接口（REST API）的连接模型是请求-响应-关闭：

客户端                    服务器
  │                        │
  │─── POST /api/chat ─────→│
  │   {"prompt": "你好"}     │
  │                        │
  │←── 200 OK ─────────────│
  │   {"response": "你好！有什么可以帮你的？"}
  │                        │
  │   [连接关闭]            │

整个过程可能耗时几秒到几十秒（如果 LLM 生成时间长），但客户端在这几秒里什么都看不到，只能干等。等连接关闭后，一次性拿到完整的响应。

SSE 的连接模型是请求-持续响应：

客户端                    服务器
  │                        │
  │─── GET /chat/stream ───→│
  │                        │
  │←── data: {"text": "你"} ─│
  │←── data: {"text": "好"} ─│
  │←── data: {"text": "！"} ─│
  │←── data: {"text": "有"} ─│
  │←── data: {"text": "什"} ─│
  │   ...                  │
  │←── event: done ────────│
  │                        │
  │   [连接关闭]            │

服务器收到请求后立即开始响应，每生成一个字（或一个 token）就推送一条 data 消息。前端收到消息后立刻渲染到屏幕上，用户看到文字在"流动"。

这个差异的底层是 HTTP 的流式传输（streaming）。在 HTTP/1.1 中，服务器可以用 Transfer-Encoding: chunked 来分块发送响应；在 HTTP/2 中，服务器可以在同一个 Stream 上持续发送 DATA 帧。SSE 不依赖特定的 HTTP 版本，但它天然受益于 HTTP/2 的多路复用——多个 SSE 连接可以共享同一个 TCP 连接。

2.2 数据格式：JSON 对象 vs event-stream 文本

普通 JSON 接口的响应体是一个完整的 JSON 对象：

{
  "choices": [{
    "message": {
      "role": "assistant",
      "content": "你好！有什么可以帮你的？"
    },
    "finish_reason": "stop"
  }]
}

SSE 的响应体是流式的文本事件：

data: {"choices": [{"delta": {"content": "你"}}]}

data: {"choices": [{"delta": {"content": "好"}}]}

data: {"choices": [{"delta": {"content": "！"}}]}

data: [DONE]

OpenAI 的 Chat Completions API（stream=true 模式）就是用的这种格式。每条 data 消息里是一个 JSON 字符串，表示当前生成的增量内容（delta）。

2.3 推送方向：双向 vs 单向

这是 SSE 和 WebSocket 最关键的区别。SSE 是服务端单向推送——服务器可以持续向客户端发消息，但客户端不能在同一条连接上向服务器发消息。如果客户端需要发消息（比如用户输入了新问题），需要另开一条普通的 HTTP POST 请求。

WebSocket 是全双工——客户端和服务器可以在同一条连接上随时互相发消息。

对于 AI 对话场景，这个单向特性恰好够用。用户输入问题 → 发 POST 请求 → 服务器开始生成回答 → 通过 SSE 逐字推送 → 用户看到打字机效果。整个过程中客户端不需要在 SSE 连接上发任何消息。

2.4 一张完整的对比表

维度	普通 JSON 接口	SSE	WebSocket
连接模型	短连接，请求-响应-关闭	长连接，服务端持续推送	长连接，全双工
数据格式	一次性 JSON 对象	流式 text/event-stream	二进制或文本帧
推送方向	服务端 → 客户端（一次性）	服务端 → 客户端（持续）	双向
协议依赖	普通 HTTP/1.1	普通 HTTP/1.1+	WebSocket 握手（HTTP Upgrade）
浏览器支持	universally	universally（EventSource）	universally
自动断线重连	无	内置（EventSource）	需手动实现
跨域（CORS）	标准 CORS	标准 CORS	需额外处理
代理/防火墙穿透	无问题	可能受代理缓冲影响	可能受防火墙拦截（非 80/443 端口）
典型场景	REST API、CRUD	AI 流式输出、实时通知	聊天室、多人协作、游戏

---

三、SSE 的底层 HTTP 机制

3.1 HTTP/1.1 下的 SSE：Chunked Transfer Encoding

在 HTTP/1.1 下，SSE 依赖 Transfer-Encoding: chunked（或 Content-Length 不设置时的隐式流式传输）。服务器不需要在响应开始时就知道总内容长度，而是分块发送：

HTTP/1.1 200 OK
Content-Type: text/event-stream
Transfer-Encoding: chunked

a\r\n                          ← 第一个 chunk，长度 10 字节
data: Hello\n\n\r\n
a\r\n                          ← 第二个 chunk，长度 10 字节
data: World\n\n\r\n
0\r\n                          ← 结束标记
\r\n

浏览器（或 EventSource）持续从 TCP 连接上读取数据，每遇到 \n\n 就解析出一条 SSE 消息，触发对应的 JavaScript 事件。

3.2 HTTP/2 下的 SSE：Stream 多路复用

HTTP/2 把多个请求/响应复用在同一个 TCP 连接上，每个请求是一个独立的 Stream。SSE 在 HTTP/2 下不再需要 Transfer-Encoding: chunked——HTTP/2 的 DATA 帧本身就支持分片传输。

HTTP/2 Stream 5
  HEADERS 帧: :status=200, content-type=text/event-stream
  DATA 帧: "data: Hello\n\n"
  DATA 帧: "data: World\n\n"
  DATA 帧（END_STREAM）: "data: [DONE]\n\n"

HTTP/2 的多路复用对 SSE 有两个好处：

1. 一个 TCP 连接可以承载多个 SSE 流：在 HTTP/1.1 中，浏览器对同一个域名通常只开 6 个并发 TCP 连接。如果页面同时和多个 AI 模型对话，连接数可能耗尽。HTTP/2 下所有 SSE 流共享一个 TCP 连接
2. HTTP/2 的流优先级：可以为 SSE 流设置更高的优先级，确保实时推送不被其他大文件下载阻塞

3.3 HTTP/3 下的 SSE：QUIC 流

HTTP/3 基于 QUIC（UDP），每个 Stream 是独立的 QUIC 流。QUIC 为每个 Stream 维护独立的拥塞控制和丢包恢复，这意味着：

• 一个 SSE 流的丢包不会阻塞同一个连接上的其他请求
• 0-RTT 连接建立可以加速 SSE 连接的首包时间

但 HTTP/3 目前（2026 年）的部署仍然不如 HTTP/2 普及，SSE 在 HTTP/3 上的优势更多是理论层面的。

---

四、SSE 在 AI 对话中的具体实现

4.1 OpenAI API 的 SSE 格式

OpenAI 的 Chat Completions API 在 stream=true 模式下返回 SSE：

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","created":1234567890,"model":"gpt-4","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]}

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","created":1234567890,"model":"gpt-4","choices":[{"index":0,"delta":{"content":"你"},"finish_reason":null}]}

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","created":1234567890,"model":"gpt-4","choices":[{"index":0,"delta":{"content":"好"},"finish_reason":null}]}

data: [DONE]

每条 data 消息是一个 JSON 对象，包含 choices[0].delta.content 字段，表示当前生成的增量文本。[DONE] 是一个特殊标记，表示流结束。

注意 OpenAI 的 SSE 没有使用 event 字段——所有消息都是默认的 message 事件类型。理论上前端可以用 source.onmessage 监听，但这里有一个关键的限制：原生 EventSource API 有两个致命缺陷，导致它无法直接调用 OpenAI 等 AI 厂商的流式接口。

4.1.1 原生 EventSource 的两个致命缺陷

1. 只支持 GET 请求：EventSource 的构造函数不接受配置请求方法——它永远是 GET。但 OpenAI 的 /v1/chat/completions 接口强制要求 POST 请求来提交 messages 上下文，GET 方式根本传不了复杂的对话历史
2. 不支持自定义 Header：EventSource 的规范中没有暴露 headers 选项，无法传入 Authorization: Bearer sk-xxx

这意味着原生 EventSource 只能用于简单的 SSE 端点（比如你自己写的、不需要认证的 GET 接口），无法直接对接 OpenAI、Claude、Gemini 等厂商的 API。

4.1.2 业界真实方案：Fetch API + 流式读取

目前前端对接 AI 流式接口的主流做法是用 fetch 发送 POST 请求，然后通过 response.body.getReader() 手动按数据块读取和解析：

async function streamChat(prompt) {
    const response = await fetch('https://api.openai.com/v1/chat/completions', {
        method: 'POST',
        headers: {
            'Content-Type': 'application/json',
            'Authorization': 'Bearer sk-xxx'
        },
        body: JSON.stringify({
            model: 'gpt-4',
            messages: [{ role: 'user', content: prompt }],
            stream: true
        })
    });

    const reader = response.body.getReader();
    const decoder = new TextDecoder('utf-8');
    let buffer = '';

    while (true) {
        const { done, value } = await reader.read();
        if (done) break;

        buffer += decoder.decode(value, { stream: true });

        // SSE 消息以 \n\n 分隔，按行解析
        const lines = buffer.split('\n');
        buffer = lines.pop();  // 最后一个可能是不完整的行，保留在 buffer 中

        for (const line of lines) {
            if (line.startsWith('data: ')) {
                const data = line.slice(6);
                if (data === '[DONE]') return;
                const chunk = JSON.parse(data);
                const text = chunk.choices?.[0]?.delta?.content || '';
                appendToChatWindow(text);
            }
        }
    }
}

这里的关键点是 buffer 的处理——reader.read() 返回的数据块不一定恰好按 SSE 消息的边界切分，可能一条消息被拆成两个 chunk，也可能一个 chunk 里包含多条消息。所以需要用 buffer 累积数据，按 \n 分行后逐行解析。

4.1.3 更优雅的方案：@microsoft/fetch-event-source

微软开源的 @microsoft/fetch-event-source 库完全模拟了 EventSource 的事件处理逻辑（onmessage、addEventListener），但底层使用 fetch，因此支持 POST 请求、自定义 Header 和 Request Body：

import { fetchEventSource } from '@microsoft/fetch-event-source';

fetchEventSource('https://api.openai.com/v1/chat/completions', {
    method: 'POST',
    headers: {
        'Content-Type': 'application/json',
        'Authorization': 'Bearer sk-xxx'
    },
    body: JSON.stringify({
        model: 'gpt-4',
        messages: [{ role: 'user', content: '你好' }],
        stream: true
    }),
    onmessage(event) {
        if (event.data === '[DONE]') {
            // 返回 Promise.reject 来关闭流
            throw new Error('Stream finished');
        }
        const chunk = JSON.parse(event.data);
        const text = chunk.choices?.[0]?.delta?.content || '';
        appendToChatWindow(text);
    }
});

这是目前前端对接 AI 流式接口的标准件——保留了 EventSource 的简洁 API 风格，同时突破了 GET 和 Header 的限制。

4.1.4 后端自行封装的 SSE 代理

另一种做法是后端提供一个 GET 接口作为 SSE 代理——前端用原生 EventSource 连接这个代理接口，后端再向 OpenAI 发 POST 请求转发流式响应：

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import httpx

app = FastAPI()

@app.get('/chat/stream')  # GET 接口，EventSource 可以直接连
async def chat_stream(prompt: str):
    async def generate():
        async with httpx.AsyncClient() as client:
            async with client.stream(
                'POST', 'https://api.openai.com/v1/chat/completions',
                headers={'Authorization': 'Bearer sk-xxx'},
                json={
                    'model': 'gpt-4',
                    'messages': [{'role': 'user', 'content': prompt}],
                    'stream': True
                }
            ) as response:
                async for line in response.aiter_lines():
                    if line.startswith('data: '):
                        yield line + '\n'

    return StreamingResponse(
        generate(),
        media_type='text/event-stream',
        headers={'Cache-Control': 'no-cache'}
    )

这种方案的优点是前端代码最简洁（原生 EventSource 直接连 GET 接口），缺点是后端多了一层转发，增加了延迟和服务器负载，且认证信息（API Key）需要从前端传到后端再传给 OpenAI，增加了暴露面。

4.2 后端生成 SSE 的原理

后端生成 SSE 的核心是不缓存响应、不等待完整内容、逐块写入。以 Python FastAPI 为例：

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

app = FastAPI()

async def generate_tokens():
    tokens = ["你", "好", "！", "有", "什", "么", "可", "以", "帮", "你", "的", "？"]
    for token in tokens:
        await asyncio.sleep(0.1)  # 模拟 LLM 生成延迟
        yield f"data: {{\"text\": \"{token}\"}}\n\n"
    yield "data: [DONE]\n\n"

@app.get('/chat/stream')
async def chat_stream():
    return StreamingResponse(
        generate_tokens(),
        media_type='text/event-stream',
        headers={'Cache-Control': 'no-cache'}
    )

关键点：

• StreamingResponse 接受一个异步生成器，每 yield 一次就向客户端推送一块数据
• media_type='text/event-stream' 设置正确的 Content-Type
• 生成器里的 \n\n 是 SSE 消息的结束标记

如果是和真实的 LLM（如 OpenAI API）对接，后端需要把 LLM 的流式响应转发给客户端：

import httpx

async def proxy_openai_stream(prompt: str):
    async with httpx.AsyncClient() as client:
        async with client.stream(
            'POST', 'https://api.openai.com/v1/chat/completions',
            headers={'Authorization': 'Bearer sk-xxx'},
            json={
                'model': 'gpt-4',
                'messages': [{'role': 'user', 'content': prompt}],
                'stream': True
            }
        ) as response:
            async for line in response.aiter_lines():
                if line.startswith('data: '):
                    yield line + '\n'  # 直接转发 SSE 行

4.3 前端渲染打字机效果

前端收到 SSE 消息后，需要把文字逐字追加到聊天窗口：

let currentMessage = '';

source.onmessage = (event) => {
    if (event.data === '[DONE]') {
        source.close();
        return;
    }
    const chunk = JSON.parse(event.data);
    const text = chunk.choices?.[0]?.delta?.content || '';
    currentMessage += text;
    updateMessageBubble(currentMessage);
};

这里的 updateMessageBubble 负责把当前累积的文本渲染到 UI 上。如果每次收到 token 都重新渲染整个消息内容，对于长消息会有性能问题。更优化的做法是用 DocumentFragment 或虚拟 DOM 的增量更新。

---

五、断线重连与错误处理

5.1 EventSource 的自动重连机制

EventSource 内置了断线重连机制。如果连接意外断开（网络波动、服务器重启、代理超时），浏览器会自动尝试重新连接。重连的行为由以下因素决定：

1. 服务器发送的 retry 字段：服务器可以在某条消息中设置 retry: 5000，表示如果连接断了，客户端应该等 5000 毫秒后再重连
2. EventSource 的默认行为：如果没有收到 retry 字段，浏览器默认使用大约 3 秒的重试间隔，并且这个间隔会指数退避（第一次 3 秒，第二次 6 秒，第三次 12 秒……）
3. 服务器返回的 Last-Event-ID 头：重连时，浏览器会在请求头中带上 Last-Event-ID，值是断线前最后收到的事件 ID。服务器可以据此恢复断点，只发送后续内容

const source = new EventSource('/chat/stream');

source.onerror = (error) => {
    console.log('连接断开，EventSource 会自动重连');
    // 不需要手动调用 source.connect()，浏览器会自动处理
};

5.2 服务器端的断点续传

服务器需要处理 Last-Event-ID 头来支持断点续传：

@app.get('/chat/stream')
async def chat_stream(request: Request):
    last_event_id = request.headers.get('Last-Event-ID')
    
    async def generate():
        # 如果有 last_event_id，只发送后续内容
        start_index = int(last_event_id) + 1 if last_event_id else 0
        for i, token in enumerate(tokens[start_index:], start=start_index):
            yield f"id: {i}\ndata: {{\"text\": \"{token}\"}}\n\n"
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(generate(), media_type='text/event-stream')

注意：AI 对话场景中，断点续传的实现难度比普通通知类 SSE 高得多——因为 LLM 的生成过程是有状态的，服务器需要在内存或缓存中保存生成到一半的上下文。对于短回答（几秒钟生成完），断点续传的意义不大；对于长文档生成（几十秒到几分钟），这个功能就很重要了。

更现实的情况是：由于 LLM 推理的不可预测性和高昂的 GPU 成本，目前绝大多数大模型厂商（OpenAI、Anthropic、Google 等）的流式 API 都不支持基于 Last-Event-ID 的断点续传。一旦连接断开，客户端只能带上完整的对话历史重新发起新的请求。SSE 的断线重连机制在 AI 对话场景中更多是"重新拉取"而非"从中断处继续"。

5.3 代理和防火墙的问题

SSE 的一个潜在问题是中间代理的缓冲。某些代理服务器（尤其是企业级防火墙、CDN 边缘节点）会缓冲响应内容，等收到完整的响应体后再转发给客户端。这会让 SSE 的实时推送效果大打折扣——所有消息被缓冲在一起，客户端在最后才一次性收到。

解决方案：

• 在响应头中明确设置 Cache-Control: no-cache, no-store 和 X-Accel-Buffering: no（Nginx 专用）
• 如果使用 Nginx 反向代理，在配置中关闭缓冲：proxy_buffering off;
• 在 SSE 消息中定期发送心跳消息（空白行或注释行），防止代理认为连接空闲而断开

async def generate_with_heartbeat():
    for token in tokens:
        yield f"data: {{\"text\": \"{token}\"}}\n\n"
        await asyncio.sleep(0.1)
    # 定期发送心跳（每 15 秒）
    yield ": heartbeat\n\n"  # 以冒号开头的行是 SSE 注释，会被 EventSource 忽略

---

六、SSE vs WebSocket：什么时候该选哪个

6.1 技术差异

维度	SSE	WebSocket
协议	普通 HTTP，无需升级	HTTP Upgrade → WebSocket 协议
推送方向	服务端 → 客户端（单向）	双向
数据格式	纯文本（text/event-stream）	二进制或文本帧
自动重连	内置	需手动实现
浏览器 API	EventSource	WebSocket
跨域	标准 CORS	需单独处理 Origin 检查
代理穿透	通常无问题（HTTP 标准端口）	可能被防火墙拦截非 80/443 端口
消息大小	无限制（流式）	单帧最大 2^63 字节（实际受实现限制）
心跳机制	需手动发送注释行	内置 Ping/Pong 帧

6.2 选型决策

选 SSE 的场景：

• AI 流式对话（ChatGPT 式打字机效果）
• 实时股票行情、比分推送
• 服务器日志实时推送
• 社交媒体实时通知（点赞、评论）

这些场景的共同点是：数据从服务器单向流向客户端，客户端不需要在同一条连接上发消息。

选 WebSocket 的场景：

• 实时聊天室（双方都需要随时发消息）
• 多人协作编辑（Google Docs 式实时同步）
• 在线游戏（低延迟双向通信）
• 需要频繁双向交互的实时应用

6.3 为什么 AI 对话选 SSE 而不是 WebSocket

AI 对话场景有几个特点，让 SSE 成为更自然的选择：

1. 单向通信足够：用户输入问题 → POST 请求 → 服务器生成回答 → SSE 推送。整个过程客户端不需要在推送连接上发消息
2. HTTP 基础设施兼容：SSE 走标准的 HTTP，不需要额外的 WebSocket 服务器配置，Nginx、CDN、负载均衡器都原生支持
3. 自动重连：EventSource 的自动重连机制减少了大量样板代码
4. CORS 天然支持：SSE 的跨域和 REST API 一样，用标准 CORS 头即可。WebSocket 的跨域需要单独处理
5. 调试更简单：SSE 的响应可以用 curl 直接查看，WebSocket 需要专门的客户端工具

ChatGPT、Claude、Gemini 的流式 API 全部使用 SSE，这不是偶然——而是经过工程验证后的行业共识。

---

七、工程实践中的常见问题

7.1 Nginx 反向代理配置

如果使用 Nginx 反向代理 SSE 服务，必须关闭缓冲，否则 SSE 的实时性会被破坏：

location /chat/ {
    proxy_pass http://backend;
    proxy_http_version 1.1;
    proxy_set_header Connection '';
    proxy_buffering off;           # ← 关键：关闭缓冲
    proxy_cache off;               # ← 关键：关闭缓存
    proxy_read_timeout 86400s;     # ← SSE 连接可能很长
}

proxy_read_timeout 默认是 60 秒，SSE 连接如果超过这个时间没有数据，Nginx 会主动断开。对于 LLM 生成时间可能超过 60 秒的场景，需要调大这个值，或者定期发送心跳消息。

7.2 多 worker 环境下的 SSE

如果你用 Gunicorn + Uvicorn 部署 FastAPI，并且开了多个 worker（--workers 4），需要注意：SSE 连接会被负载均衡到某个 worker 上，后续的 POST 请求可能被路由到另一个 worker。如果需要在 POST 请求和 SSE 连接之间共享状态（比如会话上下文），需要使用外部的状态存储（Redis、数据库），而不是进程内存。

7.3 客户端取消请求的处理

当用户关闭聊天窗口或切换页面时，浏览器会自动断开 SSE 连接（TCP FIN）。服务器端的生成器会收到一个异常（asyncio.CancelledError 或 GeneratorExit），应该优雅地处理：

async def generate_tokens():
    try:
        for token in tokens:
            yield f"data: {{\"text\": \"{token}\"}}\n\n"
            await asyncio.sleep(0.1)
    except asyncio.CancelledError:
        # 客户端断开连接，清理资源
        logger.info("客户端断开 SSE 连接")
        raise

如果不处理这个异常，可能会导致资源泄漏（比如 LLM 推理任务在后台继续运行，消耗 GPU 资源）。

---

八、总结

SSE 不是新技术，但它在 AI 流式对话场景里找到了最合适的应用场景。它的核心优势可以用一句话概括：用标准的 HTTP 协议实现了服务端到客户端的实时推送，不需要额外的协议升级，不需要复杂的双向通信管理。

和普通 JSON 接口相比，SSE 的区别在于连接模型（长连接流式传输 vs 短连接一次性响应）和数据格式（text/event-stream vs JSON）。和 WebSocket 相比，SSE 的区别在于推送方向（单向 vs 双向）和协议复杂度（标准 HTTP vs 协议升级）。

对于 AI 对话这种"客户端发一个问题，服务器持续推送回答"的场景，SSE 的单向推送能力恰好够用，同时避免了 WebSocket 的部署复杂性和协议开销。这也是为什么从 OpenAI 到国内各大模型厂商，流式 API 的行业首选都是 SSE。