SSE协议实现LLM流式输出：从原理到实战的Token推送指南

1. 从“蹦”字说起：为什么SSE的Token流式输出如此直观？

最近在折腾大语言模型（LLM）的应用集成，特别是那种需要实时看到模型“思考”过程的场景，比如聊天机器人或者代码补全。大家可能都见过，在类似ChatGPT的界面上，文字是一个个词（Token）蹦出来的，而不是等整段话生成完再一次性显示。这种体验非常流畅，能让人感知到模型正在“工作”。为了实现这个效果，我绕开了更复杂的WebSocket，选择了一个更轻量、更直接的协议：Server-Sent Events。

SSE，全称Server-Sent Events，本质上是一个基于HTTP的长连接。它允许服务器主动向客户端推送数据，但客户端只能接收，不能主动向这个连接发送数据（当然，你可以另开请求）。这个特性让它特别适合像LLM流式输出、实时日志、股票报价这类“服务器说，客户端听”的场景。当你看到Token“一个个蹦出来”时，背后很可能就是SSE在默默工作。与WebSocket的全双工相比，SSE更简单，天然支持断线重连和事件ID，对于不需要双向高频交互的实时推送来说，是性价比极高的选择。

那么，这个“蹦出来”的过程到底是怎么发生的？它不仅仅是前端的一个动画效果。从服务器生成第一个Token，到它出现在你的浏览器里，中间涉及了模型推理、文本编码、网络传输、前端渲染等多个环节的紧密协作。理解这个过程，不仅能帮你更好地调试和优化应用，还能让你明白为什么有时候它会“卡顿”或者“中断”。接下来，我们就深入这个数据流的内部，看看一个Token的“旅程”。

2. SSE协议核心：它是如何让数据“流”起来的？

要理解Token怎么流，得先理解SSE这个管道是怎么建的。SSE的通信模型非常简洁，完全基于标准的HTTP/HTTPS。

2.1 连接建立与数据格式

客户端（通常是浏览器）通过创建一个EventSource对象来发起连接。这个对象会向指定的服务器URL发送一个GET请求，但有一个关键的请求头：Accept: text/event-stream。服务器识别到这个头，就知道客户端期望的是一个SSE流，而不是普通的HTTP响应。

一旦连接建立，服务器会保持这个HTTP连接处于打开状态，并开始以特定的格式发送数据。SSE的数据格式有严格的规范，它不是随便的JSON流，而是由几种类型的行组成：

• 数据行：以data:开头，后面跟着实际要发送的数据。一行数据可以分多次发送，最终会拼接起来。

  data: {"token": "Hello"} data: {"token": " world"}

  上面的两行会被客户端解析为两个独立的事件数据。

• 事件行：以event:开头，用于定义事件类型。客户端可以监听特定类型的事件。

  event: message data: {"token": "Hello"} event: token data: {"token": " world"}

  这样，前端就可以用addEventListener('token', ...)来专门处理Token流。

• ID行：以id:开头，用于设置事件的ID。在连接意外中断后重连时，客户端可以通过Last-Event-ID头告诉服务器“我从哪个ID之后的数据开始要”，从而实现断点续传。

• 重试时间行：以retry:开头，后面跟毫秒数，用于建议客户端在连接断开后多久重试。

一个标准的、完整的SSE响应流看起来是这样的：

HTTP/1.1 200 OK Content-Type: text/event-stream Cache-Control: no-cache Connection: keep-alive event: start data: {"status": "streaming started"} id: 1 event: token data: {"token": "The", "finished": false} id: 2 event: token data: {"token": " quick", "finished": false} id: 3 event: token data: {"token": " brown", "finished": false} event: done data: {"status": "streaming finished"}

注意，每个事件之间用两个换行符\n\n分隔。这是SSE协议规定的分隔符，服务器在发送时必须确保格式正确。

2.2 与WebSocket和长轮询的对比

为什么选SSE而不是别的？这里有个简单的对比：

对于LLM流式输出这种典型的“服务器推送生成结果”的场景，SSE的优势非常明显：实现简单、开销小、功能恰好够用。你不需要处理WebSocket复杂的握手和帧解析，也不用忍受长轮询带来的延迟和服务器压力。

实操心得：在服务端，确保你的响应头正确设置是第一步。Content-Type: text/event-stream和Cache-Control: no-cache是必须的。另外，很多框架（如Flask、Spring Boot）的默认输出缓冲区可能会为了效率而缓存数据，导致Token无法立即发出。你需要手动刷新输出流。在Python Flask中，这通常意味着设置response.headers后，使用yield生成器来流式返回，并确保每个事件后刷新。

3. 从模型到网络：Token的生成与推送链路拆解

现在，我们把镜头对准服务器内部。一个Token从LLM模型里“诞生”到被SSE推送出去，要走完一段精细的流水线。

3.1 模型端的流式生成

现代LLM的推理接口（如OpenAI API、各类开源模型的generate接口）大多支持流式输出。其核心原理是自回归生成。模型在生成下一个Token时，是基于之前所有已生成的Token来计算的。流式接口允许我们在模型每生成一个Token后，就立即将其返回，而不是等待整个序列生成完毕。

例如，使用transformers库调用一个本地模型：

from transformers import AutoModelForCausalLM, AutoTokenizer import torch model = AutoModelForCausalLM.from_pretrained("your-model") tokenizer = AutoTokenizer.from_pretrained("your-model") inputs = tokenizer("Hello, how are you?", return_tensors="pt") # 非流式生成（一次性返回） outputs = model.generate(**inputs, max_new_tokens=50) full_text = tokenizer.decode(outputs[0], skip_special_tokens=True) print(full_text) # 一次性打印全部结果 # 流式生成（逐步返回） for output in model.generate(**inputs, max_new_tokens=50, streamer=streamer): # 注意：这里output可能是整个序列，需要配合streamer # 更常见的做法是使用库提供的专用streamer pass

实际上，更常见的做法是使用像TextStreamer这样的工具，它会在模型生成每个新Token时回调一个函数。

关键点在于：模型生成Token的速度（Tokens per second, TPS）直接决定了推送的频率。如果模型推理很慢，那么前端看到Token“蹦”出来的间隔就会很长，体验就是“卡顿”。

3.2 服务端的流式响应封装

模型返回Token流之后，我们需要用SSE格式把它包装起来，并通过HTTP响应发送出去。这里以Python的FastAPI框架为例，展示一个典型的后端处理流程：

from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse import asyncio import json app = FastAPI() async def fake_llm_streamer(prompt: str): """模拟一个慢速的LLM流式生成器""" simulated_tokens = ["思考", "中", "，", "请", "稍", "候", "。"] for token in simulated_tokens: # 模拟模型推理耗时 await asyncio.sleep(0.5) # 以SSE格式生成数据 # 注意：SSE要求数据行以`data:`开头，并以两个换行符结束 yield f"data: {json.dumps({'token': token, 'finished': False})}\n\n" # 发送结束信号 yield f"data: {json.dumps({'token': '', 'finished': True})}\n\n" @app.post("/stream") async def stream_completion(request: Request): data = await request.json() prompt = data.get("prompt", "") # 关键：返回StreamingResponse，并设置正确的媒体类型 return StreamingResponse( fake_llm_streamer(prompt), media_type="text/event-stream", headers={ 'Cache-Control': 'no-cache', 'Connection': 'keep-alive', 'X-Accel-Buffering': 'no' # 针对Nginx代理的重要设置 } )

这段代码揭示了几个重要细节：

1. 异步生成器：我们使用async函数和yield来创建一个数据流。这允许我们在等待模型生成下一个Token时，不会阻塞整个服务器。
2. 格式严格：每个yield返回的字符串必须是一个完整的SSE事件，以两个换行符\n\n结尾。
3. 响应头：StreamingResponse会自动设置Content-Type: text/event-stream。我们额外添加的头部是为了防止代理服务器（如Nginx）进行缓冲。X-Accel-Buffering: no这个头对于经过Nginx的反向代理至关重要，否则Nginx可能会缓存数据直到达到一定大小，导致Token无法实时推送。

3.3 网络传输与代理的挑战

即使服务器正确发送了流式数据，它仍然需要穿越复杂的网络环境才能到达浏览器。这里有几个常见的“拦路虎”：

• 反向代理缓冲：如前所述，Nginx、Apache等反向代理默认会缓冲上游（你的应用服务器）的响应，以提高传输效率。这对于SSE是致命的。你必须在代理配置中为SSE路径禁用缓冲。

  location /stream { proxy_pass http://your_backend; proxy_set_header Connection ''; proxy_http_version 1.1; proxy_buffering off; # 关键：关闭代理缓冲 proxy_cache off; chunked_transfer_encoding off; proxy_read_timeout 3600s; # 设置长超时 }

• 负载均衡器：云服务商的负载均衡器（如ALB, ELB）也可能有缓冲行为，需要检查其配置。
• 浏览器连接数限制：同一个域名下，浏览器对并发HTTP连接数有限制（通常是6个）。如果你的页面同时打开了多个SSE连接，可能会受到限制。可以考虑使用HTTP/2，它支持多路复用，能更好地处理多个并发流。

踩坑记录：我曾遇到一个诡异的问题，前端接收Token时总是成批到达，而不是一个个来。排查了很久，最后发现是云平台的负载均衡器层面有一个默认的“响应缓冲”策略被开启了。关闭后，流立即变得顺畅。所以，当流不“流”时，检查链路上的每一环（应用服务器、反向代理、负载均衡器、CDN）的缓冲配置是必须的。

4. 前端接收与渲染：让Token在屏幕上“蹦”出来

服务器端的流水线打通了，数据像小溪一样流到了浏览器。前端的工作就是接住这些数据，并平滑地展示给用户。

4.1 使用EventSource API建立连接

现代浏览器原生支持EventSourceAPI，用它来连接SSE端点非常简单：

// 建立SSE连接 const eventSource = new EventSource('/api/stream?prompt=你好世界'); // 监听未指定类型的消息（默认事件类型） eventSource.onmessage = (event) => { const data = JSON.parse(event.data); console.log('收到Token:', data.token); // 将Token渲染到DOM document.getElementById('output').innerText += data.token; }; // 监听特定类型的事件（如果服务器发送了`event: token`） eventSource.addEventListener('token', (event) => { const data = JSON.parse(event.data); handleToken(data); }); // 监听错误 eventSource.onerror = (error) => { console.error('EventSource failed:', error); // EventSource在错误时会自动尝试重连，可根据需要关闭 // eventSource.close(); };

EventSource会自动处理连接管理、断线重连（根据服务器返回的retry时间）和事件分发。但它有一个局限性：它只能发送GET请求，且不能自定义请求头（如Authorization头）。对于需要认证的API，这就成了问题。

4.2 更灵活的选择：使用Fetch API读取流

为了解决EventSource的限制，我们可以使用更底层的Fetch API来读取SSE流。这给了我们完全的灵活性：

async function streamWithFetch(prompt) { const response = await fetch('/api/stream', { method: 'POST', headers: { 'Content-Type': 'application/json', // 可以自定义任何头，比如认证头 'Authorization': 'Bearer your-token' }, body: JSON.stringify({ prompt: prompt }) }); const reader = response.body.getReader(); const decoder = new TextDecoder(); let buffer = ''; try { while (true) { const { done, value } = await reader.read(); if (done) break; // 将接收到的Uint8Array块解码为文本 buffer += decoder.decode(value, { stream: true }); // 处理缓冲区，按SSE格式（\n\n）分割事件 const lines = buffer.split('\n\n'); // 最后一个可能是不完整的事件，放回缓冲区 buffer = lines.pop() || ''; for (const line of lines) { if (line.startsWith('data: ')) { const dataStr = line.slice(6).trim(); // 去掉'data: ' if (dataStr) { try { const data = JSON.parse(dataStr); // 渲染Token appendTokenToUI(data.token); } catch (e) { console.error('解析JSON失败:', e, dataStr); } } } // 可以同样处理event:, id: 等行 } } } finally { reader.releaseLock(); } } function appendTokenToUI(token) { // 这里是渲染逻辑 const outputEl = document.getElementById('output'); // 简单的追加 outputEl.textContent += token; // 更复杂的做法：可以创建动画效果，滚动到底部等 }

使用Fetch API虽然代码量多了，但优势明显：支持POST、可自定义请求头、能更精细地控制连接和错误处理。你需要手动解析SSE格式，但这并不复杂。

4.3 优化渲染体验与性能

仅仅把Token追加到DOM上，可能会遇到性能问题（如果响应很长）或者体验问题（滚动、光标闪烁）。

• 防抖动渲染：不要每收到一个Token就更新一次DOM。可以使用requestAnimationFrame或者简单的计时器来批量更新。

  let tokenQueue = []; let isRendering = false; function scheduleRender() { if (!isRendering) { isRendering = true; requestAnimationFrame(() => { const outputEl = document.getElementById('output'); outputEl.textContent += tokenQueue.join(''); tokenQueue = []; isRendering = false; // 滚动到底部 outputEl.scrollTop = outputEl.scrollHeight; }); } } function appendToken(token) { tokenQueue.push(token); scheduleRender(); }

• 处理中文等多字节字符：LLM返回的Token可能不是完整的字符（特别是在使用BPE等分词器时）。一个Token可能只是一个字的一部分（如“中”字的编码）。直接渲染可能会导致乱码。更稳健的做法是，前端维护一个缓冲区，将Token拼接成完整的字符串后再用TextDecoder解码，或者依赖后端返回已经解码好的文本片段。
• 光标与输入框：如果是在类似聊天输入框的场景中流式输出，需要小心处理光标位置和防止用户输入干扰。一种常见做法是在一个只读的<div>中渲染流式输出，而不是可编辑的<textarea>。

5. 实战中的疑难杂症与排查指南

理论很美好，但现实总会遇到各种问题。下面是一些在实现SSE流式Token时常见的问题和排查思路。

5.1 连接秒断或无法建立

• 症状：前端一建立连接，立即触发onerror，状态码可能是0、404或500。
• 排查：
  1. 检查URL和CORS：确保SSE端点URL正确。由于EventSource和Fetch可能受到CORS策略限制，确保后端设置了正确的CORS头（Access-Control-Allow-Origin等）。
  2. 检查响应头：用浏览器开发者工具的“网络”标签查看SSE请求的响应头。必须包含Content-Type: text/event-stream。如果看到的是application/json，说明后端路由处理错误，返回了普通JSON响应。
  3. 检查代理缓冲：如前所述，这是最常见的原因。在Nginx等代理的访问日志和错误日志中查找线索，或尝试直接连接后端服务（绕过代理）以确认问题。

5.2 Token成批到达，不“实时”

• 症状：前端不是一个个收到Token，而是停顿几秒后，一次性收到一大段。
• 排查：
  1. 服务端刷新缓冲：确认你的后端代码在yield每个Token后，是否立即刷新了输出缓冲区。在Python的Flask中，可能需要response.flush()；在Java Servlet中，需要response.getWriter().flush()。
  2. 禁用各级缓冲：这是重中之重。从上到下检查：
     • 应用框架缓冲：查阅框架文档，是否有针对流式响应的特殊配置。
     • Web服务器缓冲：例如Gunicorn的--log-level debug可能有助于观察。
     • 反向代理缓冲：Nginx的proxy_buffering off。
     • 负载均衡器/云服务缓冲：在云平台控制台检查负载均衡器或API网关的配置。
  3. 网络延迟与拥塞：虽然不常见，但极高的网络延迟或丢包也可能导致TCP层的数据累积。

5.3 连接意外断开与重连机制

• 症状：流传输到一半突然停止，前端触发onerror。
• 原因与处理：
  • 服务器超时：HTTP服务器、代理或负载均衡器可能有连接超时设置（例如30秒、60秒）。对于长时间的LLM生成，需要调高这些超时时间。
  • 防火墙/中间件：某些网络中间件会杀死长时间空闲的连接。服务器可以定期发送SSE注释行（以:开头的行，如: keepalive\n\n）作为心跳包来保活。
  • 实现健壮的重连：EventSource有内置重连，但Fetch需要自己实现。一个简单的策略是：在断开后，等待一个指数退避的时间（如1秒、2秒、4秒…），然后重新连接，并尝试携带最后收到的事件ID（如果服务器支持）以恢复。

5.4 内存泄漏与资源管理

• 问题：用户频繁开始新的流式请求而不关闭旧的连接，或者在单页应用（SPA）中切换页面时未清理连接。
• 解决方案：
  • 前端主动关闭：在组件卸载或开始新请求前，务必调用eventSource.close()或中止Fetch的ReadableStream。
  • 后端连接管理：对于已断开但服务器端未感知的“僵尸连接”，服务器应设置读超时，并定期清理无效连接。在Go或Node.js中，可以通过监听request.close或connection.on(‘close’)事件来释放相关资源（如中止LLM生成进程）。

个人经验：调试SSE流，最强大的工具就是浏览器开发者工具的“网络”标签。选中你的SSE请求，查看“响应”选项卡。如果它是流式的，你会看到数据在实时地一行行增加。如果它一直处于“等待”状态，然后突然出现完整响应，那缓冲的嫌疑就非常大。另一个有用的技巧是在服务器端每个Token推送后打印带时间的日志，这样你就能清晰地看到数据是什么时候离开服务器进程的，有助于定位缓冲发生在哪一层。