SSE 流式输出详解 —— 从原理到 AI 实战

前言：为什么 AI 回复要”一个字一个字”往外蹦？

用过 ChatGPT 或 Claude 的人都注意过这个现象：当你发出一条消息，AI 的回复不是一次性弹出来的，而是一个字一个字、一行一行地流出来，就像有人在打字。

这不是为了”显得在思考”的视觉效果，而是有实实在在的技术原因：

• LLM 生成一个完整回复可能需要 10-30 秒，如果等全部生成完再发给用户，体验极差
• 流式输出让用户立即看到响应开始，大幅降低感知延迟
• 某些场景（如实时翻译、对话、代码补全）里用户甚至不需要等完整答案——看到一半就知道对不对

这套技术背后的协议就叫 SSE（Server-Sent Events，服务器推送事件）。本文从底层原理讲到 AI API 实战，帮你彻底搞懂它。

---

第一章：SSE 是什么？

1.1 几种实时通信方式对比

方式	方向	协议	复杂度	适用场景
短轮询	客户端→服务器	HTTP	⭐ 最简单	不需要实时、定时查一下就行
长轮询	客户端→服务器	HTTP	⭐⭐	模拟实时、兼容性好
SSE	服务器→客户端	HTTP	⭐⭐	单向推送（AI 回复、通知、日志）
WebSocket	双向	WS	⭐⭐⭐	聊天、游戏、协作编辑

1.2 SSE 的核心特点

┌──────────┐                        ┌──────────┐
│  浏览器   │  ── HTTP GET ──────→  │  服务器   │
│ (客户端)  │  ←── text/event-stream │ (AI API) │
└──────────┘   持续推送，不断开连接   └──────────┘

• 单向：服务器 → 客户端。客户端只需发一次 HTTP 请求，服务器持续推送
• 基于 HTTP：不需要特殊协议，普通 HTTP 就行，天然支持代理、负载均衡、鉴权
• 自动重连：浏览器原生 EventSource API 在连接断开时会自动重连
• 纯文本：数据格式是纯文本，易于调试和理解
• 轻量：比 WebSocket 简单得多，不需要握手升级协议

1.3 什么时候用 SSE，什么时候用 WebSocket？

场景	推荐	原因
AI 对话流式输出	SSE	单向推送，不需要客户端发消息
服务器日志实时查看	SSE	单向，简单
股票行情推送	SSE	单向数据流
通知推送	SSE	单向，自动重连
在线聊天	WebSocket	需要双向实时通信
多人协作编辑	WebSocket	双向高频通信
游戏	WebSocket	低延迟双向

---

第二章：SSE 协议细节

2.1 HTTP 层面的约定

客户端发起一个普通的 HTTP GET 请求：

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

服务器返回一个特殊的响应：

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

关键在 Content-Type: text/event-stream——浏览器看到这个就知道这是一个 SSE 流，不会关闭连接。

2.2 数据格式

SSE 的数据体由若干字段组成，每条消息以空行分隔。核心字段：

event: 事件类型（可选，默认 "message"）
id: 消息 ID（可选，用于断线重连的 Last-Event-ID）
data: 数据内容（可以有多个 data 行，会拼接）
retry: 重连间隔（毫秒，可选）

简单示例：

data: 你好

data: 这是第二条消息
data: 这是第二条的续行

event: error
data: 出错了

id: 42
data: {"token": "Hello"}

2.3 浏览器端：EventSource API

浏览器原生支持 SSE，几行代码就能用：

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

// 默认事件（event 字段为空或为 "message"）
source.onmessage = (event) => {
  console.log('收到数据:', event.data);
  console.log('消息ID:', event.lastEventId);
};

// 自定义事件
source.addEventListener('error', (event) => {
  console.error('流错误:', event.data);
});

// 连接打开
source.addEventListener('open', () => {
  console.log('连接已建立');
});

// 连接错误（会自动重连）
source.onerror = (event) => {
  if (source.readyState === EventSource.CLOSED) {
    console.log('连接已关闭（不会重连）');
  } else {
    console.log('连接出错，正在重连...');
  }
};

// 手动关闭
source.close();

2.4 Node.js 服务端实现

const http = require('http');

http.createServer((req, res) => {
  if (req.url === '/stream') {
    // 设置 SSE 响应头
    res.writeHead(200, {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive',
      'Access-Control-Allow-Origin': '*',
    });

    // 发送注释（保持连接活跃，防止代理超时）
    const keepAlive = setInterval(() => {
      res.write(': keepalive\n\n');
    }, 15000);

    let count = 0;
    const sendInterval = setInterval(() => {
      count++;
      // 标准格式：data + 空行
      res.write(`data: 这是第 ${count} 条消息\n`);
      res.write(`id: ${count}\n\n`);

      if (count >= 10) {
        clearInterval(sendInterval);
        clearInterval(keepAlive);
        res.end();
      }
    }, 1000);

    // 客户端断开时清理
    req.on('close', () => {
      clearInterval(sendInterval);
      clearInterval(keepAlive);
    });
  }
}).listen(3000);

💡 : keepalive 是 SSE 的注释语法（冒号开头），浏览器会忽略，但能防止代理服务器因长时间无数据而断开连接。

---

第三章：AI 中的 SSE 流式输出

3.1 为什么 LLM 天生适合 SSE？

LLM 是逐 token 生成的（自回归），天然就是流式的：

用户输入："解释一下量子计算"
        │
        ▼
┌─────────────────────────────┐
│ LLM 逐 token 生成：          │
│ 量 → 子 → 计 → 算 → 是      │
│ → 一 → 种 → 利 → 用         │
│ → 量 → 子 → 力 → 学         │
│ → ...                        │
└─────────────────────────────┘
        │ 每个 token 生成后立即发送
        ▼
┌─────────────────────────────┐
│ SSE:                         │
│ data: 量                     │
│ data: 子                     │
│ data: 计                     │
│ data: 算                     │
│ ...                          │
└─────────────────────────────┘

相比”等全部生成完再一次性返回”：

• ⏱️ 首字延迟（TTFB）从 30 秒降到 0.5 秒
• 🧠 用户体验从”在等”变成”在看”
• 💰 可以在生成到一半时取消（省钱）

3.2 OpenAI API 的流式调用

OpenAI 的 Chat Completions API 设置 stream: true 即可启用 SSE：

import requests
import json

url = "https://api.openai.com/v1/chat/completions"
headers = {
    "Authorization": "Bearer YOUR_API_KEY",
    "Content-Type": "application/json",
}

data = {
    "model": "gpt-4o",
    "messages": [{"role": "user", "content": "用三句话介绍量子计算"}],
    "stream": True,  # 关键：开启流式
}

# stream=True 让 requests 不立即读取全部响应体
response = requests.post(url, json=data, headers=headers, stream=True)

for line in response.iter_lines():
    if not line:
        continue
    
    line = line.decode("utf-8")
    
    # OpenAI SSE 格式：data: {"choices": [{"delta": {"content": "量"}}]}
    if line.startswith("data: "):
        json_str = line[6:]  # 去掉 "data: " 前缀
        
        if json_str == "[DONE]":
            print("\n--- 完成 ---")
            break
        
        chunk = json.loads(json_str)
        content = chunk["choices"][0]["delta"].get("content", "")
        if content:
            print(content, end="", flush=True)

输出效果：

量子计算是一种利用量子力学原理进行信息处理的计算范式，
它使用量子比特代替经典比特，通过叠加和纠缠实现并行计算。
在特定问题上，量子计算机可以指数级超越经典计算机。
--- 完成 ---

3.3 OpenAI SSE 的事件结构

OpenAI 的每个 SSE 事件是一个 JSON 块：

{
  "id": "chatcmpl-xxx",
  "object": "chat.completion.chunk",
  "created": 1714567890,
  "model": "gpt-4o",
  "choices": [
    {
      "index": 0,
      "delta": {
        "content": "量"  // 这是新增的文本内容
      },
      "finish_reason": null  // 最后一条会变成 "stop"
    }
  ]
}

流结束时发送：

data: [DONE]

3.4 Anthropic Claude 的流式调用

Claude API 也支持 SSE，语法类似：

import anthropic

client = anthropic.Anthropic(api_key="YOUR_KEY")

with client.messages.stream(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "介绍一下机器学习"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

Claude 的流式事件类型更丰富：

事件类型	含义
message_start	消息开始
content_block_start	内容块开始（包括 thinking 块）
content_block_delta	内容增量（实际的文本）
content_block_stop	内容块结束
message_delta	消息级增量（如 stop_reason）
message_stop	消息结束

---

第四章：前端实战 —— 打造 AI 聊天界面

4.1 用 EventSource 对接 OpenAI

// ⚠️ EventSource 不支持 POST 请求和自定义 Header！
// 所以直接用 fetch + ReadableStream 手动解析 SSE

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

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

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

    buffer += decoder.decode(value, { stream: true });
    
    // 按行分割
    const lines = buffer.split('\n');
    buffer = lines.pop() || ''; // 未完成的行放回 buffer

    for (const line of lines) {
      if (!line.startsWith('data: ')) continue;
      const jsonStr = line.slice(6);
      if (jsonStr === '[DONE]') return;

      try {
        const chunk = JSON.parse(jsonStr);
        const content = chunk.choices?.[0]?.delta?.content;
        if (content) {
          // 实时渲染到界面
          appendToChat(content);
        }
      } catch (e) {
        // 忽略解析错误
      }
    }
  }
}

// 使用
await chatWithAI('你好，请介绍一下你自己');

4.2 完整的聊天 UI 实现

<!DOCTYPE html>
<html>
<head>
  <title>AI Chat</title>
  <style>
    body { font-family: system-ui; max-width: 700px; margin: 40px auto; }
    #chat { border: 1px solid #ddd; border-radius: 8px; padding: 20px; min-height: 400px; margin-bottom: 20px; }
    .msg { margin: 8px 0; padding: 10px 14px; border-radius: 12px; max-width: 80%; }
    .user { background: #e3f2fd; margin-left: auto; }
    .ai { background: #f5f5f5; }
    .cursor::after { content: '▊'; animation: blink 1s step-end infinite; }
    @keyframes blink { 50% { opacity: 0; } }
    #input-area { display: flex; gap: 10px; }
    #input { flex: 1; padding: 12px; border: 1px solid #ddd; border-radius: 8px; font-size: 16px; }
    button { padding: 12px 24px; border: none; border-radius: 8px; background: #1976d2; color: white; cursor: pointer; font-size: 16px; }
    button:hover { background: #1565c0; }
    button:disabled { opacity: 0.5; cursor: not-allowed; }
  </style>
</head>
<body>
  <div id="chat"></div>
  <div id="input-area">
    <input id="input" type="text" placeholder="输入消息...">
    <button id="send">发送</button>
  </div>

  <script>
    const chat = document.getElementById('chat');
    const input = document.getElementById('input');
    const sendBtn = document.getElementById('send');
    const API_KEY = 'YOUR_KEY';
    
    function addMsg(role, text) {
      const div = document.createElement('div');
      div.className = `msg ${role}`;
      div.textContent = text;
      chat.appendChild(div);
      chat.scrollTop = chat.scrollHeight;
      return div;
    }

    async function send() {
      const prompt = input.value.trim();
      if (!prompt) return;
      
      addMsg('user', prompt);
      input.value = '';
      sendBtn.disabled = true;

      const aiDiv = addMsg('ai', '');
      aiDiv.classList.add('cursor');

      try {
        const res = await fetch('https://api.openai.com/v1/chat/completions', {
          method: 'POST',
          headers: {
            'Content-Type': 'application/json',
            'Authorization': `Bearer ${API_KEY}`,
          },
          body: JSON.stringify({
            model: 'gpt-4o',
            messages: [{ role: 'user', content: prompt }],
            stream: true,
          }),
        });

        const reader = res.body.getReader();
        const decoder = new TextDecoder();
        let buffer = '';

        while (true) {
          const { done, value } = await reader.read();
          if (done) break;
          
          buffer += decoder.decode(value, { stream: true });
          const lines = buffer.split('\n');
          buffer = lines.pop() || '';
          
          for (const line of lines) {
            if (!line.startsWith('data: ') || line.includes('[DONE]')) continue;
            try {
              const chunk = JSON.parse(line.slice(6));
              const content = chunk.choices?.[0]?.delta?.content;
              if (content) aiDiv.textContent += content;
            } catch(e) {}
          }
          chat.scrollTop = chat.scrollHeight;
        }
      } catch (e) {
        aiDiv.textContent = '请求失败：' + e.message;
      }
      
      aiDiv.classList.remove('cursor');
      sendBtn.disabled = false;
    }

    sendBtn.addEventListener('click', send);
    input.addEventListener('keydown', (e) => {
      if (e.key === 'Enter' && !e.shiftKey) { e.preventDefault(); send(); }
    });
  </script>
</body>
</html>

效果：实时流式打字、闪烁光标、自动滚动。保存为 .html 就可以直接用。

---

第五章：服务端中转 —— 保护 API Key

前端直接调 OpenAI API 会暴露 API Key，正确做法是加一个后端代理：

// server.js —— Node.js 中转服务
const express = require('express');
const app = express();

app.use(express.json());
app.use(express.static('public'));  // 前端页面

app.post('/api/chat', async (req, res) => {
  const { prompt } = req.body;

  // 设置 SSE 响应头
  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    'Connection': 'keep-alive',
  });

  try {
    // 转发给 OpenAI
    const aiResponse = await fetch('https://api.openai.com/v1/chat/completions', {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Authorization': `Bearer ${process.env.OPENAI_API_KEY}`,  // Key 在服务端
      },
      body: JSON.stringify({
        model: 'gpt-4o',
        messages: [{ role: 'user', content: prompt }],
        stream: true,
      }),
    });

    const reader = aiResponse.body.getReader();
    const decoder = new TextDecoder();

    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      
      const text = decoder.decode(value, { stream: true });
      // 直接透传原始 SSE 数据
      res.write(text);
    }

    res.end();
  } catch (error) {
    res.write(`event: error\ndata: ${JSON.stringify({ error: error.message })}\n\n`);
    res.end();
  }
});

app.listen(3000, () => console.log('Server on http://localhost:3000'));

这样前端只需调 /api/chat，API Key 安全存放在服务端环境变量中。

---

第六章：高级话题

6.1 SSE vs WebSocket 的技术细节

维度	SSE	WebSocket
协议	HTTP/1.1 或 HTTP/2	独立协议（ws:// wss://）
浏览器 API	EventSource	WebSocket
二进制数据	不支持（仅文本）	支持
自动重连	原生支持	需手动实现
连接数限制	HTTP/1.1 每域名 6 个	无限制
HTTP/2 多路复用	完美支持	不适用
穿透代理	天然支持	可能被某些代理阻断
消息方向	单向（S→C）	双向

关键优势：在 HTTP/2 下，多个 SSE 连接可以复用同一条 TCP 连接，不再有 6 连接限制。

6.2 流式思考（Streaming Thinking）

Claude 等模型支持”思考过程”流式输出——在正式回答之前，先把推理过程实时展示给用户：

// content_block_start: thinking
{
  "type": "content_block_start",
  "content_block": {
    "type": "thinking",
    "thinking": "让我分析这个问题..."
  }
}

// content_block_delta: thinking
{
  "type": "content_block_delta",
  "delta": {
    "type": "thinking_delta",
    "thinking": "首先考虑时间复杂度..."
  }
}

这给用户透明度和信任感——知道 AI 不是瞎猜的。

6.3 断线续传

EventSource 的 lastEventId 机制支持断线续传：

// 客户端自动在重连时发送 Last-Event-ID
// 服务端读取并恢复
const lastId = req.headers['last-event-id'];
if (lastId) {
  // 从这条消息之后继续发送
  resumeFrom(lastId);
}

在 AI 场景中，这意味着如果网络断了，可以恢复连接继续输出而不是从头开始。

6.4 多模态流式输出

最新的多模态模型支持流式输出混合内容（文本 + 图片 + 音频）：

data: {"type": "text", "content": "这是生成的文字..."}

data: {"type": "image", "content": "base64...", "mime": "image/png"}

data: {"type": "audio", "content": "base64...", "mime": "audio/wav"}

前端需要解析不同的 content type 并分别渲染。

---

第七章：调试与排错

常见问题

Q：EventSource 为什么连不上？

检查：

1. 响应头 Content-Type: text/event-stream 是否正确
2. 跨域问题 → 服务端加 Access-Control-Allow-Origin
3. EventSource 不支持自定义 Header → 用 fetch + ReadableStream 代替

Q：为什么中间会断？

可能原因：

• 代理服务器超时（nginx 默认 60s） → 加 : keepalive 注释或调大超时
• 服务端没处理 req.on('close') → 客户端断开后还在写数据

Q：数据乱了怎么办？

SSE 以空行分隔消息。如果数据中有空行会乱。处理时积累 buffer，只在遇到完整事件后才解析。

Q：浏览器标签页切到后台，SSE 会暂停吗？

不会停止接收，但 setTimeout/setInterval 会被降频。UI 更新尽量用 requestAnimationFrame。

---

总结

知识点	一句话
SSE 是什么	HTTP 长连接，服务器单向推送文本事件流
和 WebSocket 区别	SSE 单向（S→C）、基于 HTTP、更简单；WS 双向、独立协议
AI 为什么用 SSE	LLM 逐 token 生成，天然流式，首字延迟极低
EventSource	浏览器原生 API，但只支持 GET + 标准 Header
fetch + ReadableStream	手动解析 SSE，支持 POST + 自定义 Header
生产环境注意	服务端代理保护 Key、防超时、断线续传

SSE 不是什么新技术（2009 年就标准化了），但伴随着 LLM 的爆发，它成为了 AI 应用中最核心的通信协议之一。掌握它，你就能理解市面上几乎所有 AI 产品的”打字效果”是怎么实现的。