SSE 流式输出实战：从协议原理到 Node 与 LangChain 实现

1. 为什么 SSE 总被当成普通流式用错1.1 一个被叫烂了的名字SSE 到底是什么SSE全称 Server-Sent Events中文一般叫服务器推送事件。名字里带推送两个字很多人第一反应就是这不就是个长连接吗然后随手拿它当 WebSocket 的廉价替代品或者干脆当成HTTP 流式响应的别名。这两种理解都不算错但都不够准确也正是这种模糊认知导致后面一堆坑连接莫名其妙断了、标签返回不完整、idle timeout 报错、代理层把缓冲打开导致消息延迟几十秒才到。先把定义钉死SSE 是建立在 HTTP/1.1 之上的单向、文本、基于事件的推送协议。客户端发起一个普通的 GET 请求服务端返回Content-Type: text/event-stream然后这条 HTTP 连接就不关闭服务端可以持续往这条连接里写数据客户端通过EventSource这个浏览器原生 API 逐条接收。它和 WebSocket 的核心区别在于三点单向只能服务端推客户端、纯文本二进制要自己编码、走标准 HTTP不需要协议升级握手。这三点决定了它的适用边界——AI 流式输出、日志实时推送、进度条更新、通知广播这类服务端说、客户端听的场景SSE 是最省事的选择而需要双向通信的聊天室、协同编辑还是老老实实上 WebSocket。1.2 为什么大家会把它和流式输出混为一谈因为在大模型应用爆发之后SSE 几乎成了流式输出的默认载体。你调 OpenAI 兼容接口返回的就是text/event-streamLangChain 的astream事件流底层也是 SSEVue 前端做 AI 对话打字机效果用的还是EventSource或者fetch ReadableStream。于是很多人产生了一个错觉流式输出 SSE。其实流式输出是一种数据交付模式边生成边发送而不是攒完再发SSE 只是实现它的一种传输协议。你完全可以用 chunked transfer encoding 做流式输出而不带 SSE 的事件格式也可以用 WebSocket 做流式输出。反过来SSE 也不一定用于流式输出比如定时推送一条系统通知它也是 SSE。把这两个概念分开是理解后面所有问题的前提。我见过太多人排查流式输出卡住的问题结果发现根本不是模型的问题而是 Nginx 把text/event-stream当普通响应做了缓冲数据全堵在代理层。这种问题你不把协议层和数据层分开看是永远找不到根因的。1.3 这篇文章适合谁看如果你正在做下面任何一件事这篇内容应该能帮你省下不少调试时间用 Node 写 SSE 服务端前端用EventSource接收但消息总是延迟或者断连用 LangChain / LangGraph 做流式输出想搞清楚stream、astream、astream_events到底该用哪个前端做 AI 对话标签比如 Markdown 代码块返回不完整渲染出来是半截的遇到stream disconnected before completion: idle timeout waiting for SSE这种报错不知道怎么定位想搞清楚 SSE 和 WebSocket 到底怎么选以及 LangChain 和 LangGraph 在流式这块的区别。内容会从协议原理讲到 Node 实现再到 LangChain 的流式抽象最后落到前端渲染和排错。全程按为什么这么设计来讲而不是丢一段代码让你抄。2. SSE 协议层EventSource 到底在干什么2.1 一条 SSE 消息长什么样SSE 的报文格式简单到有点朴素。服务端往连接里写的内容是一行一行的纯文本每行以字段名开头冒号分隔值空行表示一条消息结束。四个标准字段data:消息正文可以多行多行会被拼成一个字符串中间用换行连接event:自定义事件名客户端可以用addEventListener(事件名, ...)监听不写就是默认的message事件id:消息 ID客户端断线重连时会带上Last-Event-ID请求头服务端可以据此续传retry:重连间隔毫秒数告诉浏览器断线后等多久再重连。一个典型的消息块长这样id: 42 event: token data: {content:你好}注意最后那个空行它是消息结束的标志少了它浏览器就不会触发事件。这是新手最容易踩的坑之一——用res.write()拼字符串时忘了补\n\n结果前端一直收不到消息还以为是网络问题。还有一点data:后面如果内容本身包含换行必须拆成多个data:行不能直接塞一个带\n的字符串。比如要发一段 Markdowndata: 第一行 data: 第二行 data: 第三行浏览器收到后event.data会是第一行\n第二行\n第三行。这个规则看起来啰嗦但它保证了 SSE 的解析器足够简单任何语言都能几十行实现一个客户端。2.2 EventSource 的自动重连好用但会咬人EventSource最贴心的设计就是自动重连。连接断了浏览器会默认等 3 秒或者服务端retry:指定的时间自动重连并且带上Last-Event-ID。这个特性让 SSE 在弱网环境下表现相当稳但也带来两个隐蔽问题。第一个问题是重连风暴。如果服务端因为某个 bug 一直返回 500浏览器会不停地重连几秒钟一次日志瞬间被刷爆。解决办法是在服务端对 SSE 端点做限流或者在客户端监听onerror连续失败 N 次后主动close()并降级到轮询。第二个问题是重复消费。如果服务端没有正确实现Last-Event-ID的续传逻辑重连后会把已经发过的消息再发一遍。前端如果直接 append 到消息列表就会出现重复内容。我的做法是每条消息带一个单调递增的id前端维护一个lastId收到id小于等于lastId的消息直接丢弃。这个去重逻辑只有几行但能省掉大量为什么消息重复了的困惑。提示EventSource只能发 GET 请求不能自定义请求头也不能带 body。如果你的接口需要鉴权要么把 token 放 query string注意日志脱敏要么改用fetch ReadableStream 手动解析 SSE 格式。这是EventSource最大的能力边界。2.3 用 curl 直接看 SSE 流最朴素的调试手段调试 SSE 最有效的工具不是浏览器 DevTools而是curl。因为 DevTools 的 Network 面板对text/event-stream的支持时好时坏有时候数据到了它也不刷新。直接上命令行curl -N -H Accept: text/event-stream http://localhost:3000/stream-N是关键参数意思是禁用 curl 的输出缓冲让数据一到就打印。不加-N你会看到 curl 攒了一大坨才吐出来误以为服务端没在流式发送。这个参数我强烈建议背下来排查 SSE 问题的第一步永远是先curl -N看原始字节流。如果curl -N能看到数据一条条出来但浏览器收不到那问题一定在中间层代理、网关、CDN或者前端解析如果curl -N也是攒一堆才出来那问题在服务端或者服务端前面的代理。这个二分法能帮你快速缩小范围。3. Node 服务端实现从裸写到生产级3.1 最小可用的 SSE 端点Node 里写一个 SSE 端点核心就是设置正确的响应头然后保持连接不关闭。用原生http模块const http require(http); http.createServer((req, res) { if (req.url /stream) { res.writeHead(200, { Content-Type: text/event-stream, Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no }); let id 0; const timer setInterval(() { id; res.write(id: ${id}\n); res.write(data: ${JSON.stringify({ time: Date.now() })}\n\n); }, 1000); req.on(close, () { clearInterval(timer); res.end(); }); } }).listen(3000);这段代码有几个细节值得说。Cache-Control: no-cache是必须的否则中间层可能缓存响应Connection: keep-alive明确告诉对方别关连接X-Accel-Buffering: no是给 Nginx 看的禁用它的响应缓冲这个头不加Nginx 默认会把 SSE 数据攒起来你就等着消息延迟吧。req.on(close)里的清理逻辑是生产环境必须有的。客户端断开关标签页、断网时如果不清理定时器和相关资源服务端会内存泄漏。我见过一个项目因为忘了这个跑了两天内存涨到 4G最后 OOM 挂掉。3.2 心跳机制解决 idle timeout 的正解stream disconnected before completion: idle timeout waiting for SSE这个报错几乎每个做 SSE 的人都遇到过。它的根因是连接长时间没有数据流动中间层负载均衡、网关、云厂商的 LB认为连接空闲主动掐断。解决办法就是心跳。服务端每隔 15 到 30 秒发一个注释行以冒号开头的行客户端会忽略const heartbeat setInterval(() { res.write(: heartbeat\n\n); }, 15000);注释行不触发任何事件纯粹是为了让连接上有字节流动骗过空闲检测。为什么是 15 秒因为大多数云厂商的 LB 默认空闲超时是 60 秒取 15 秒留足余量。如果你的环境超时更短就相应调小。心跳还有一个副作用它能帮你检测死连接。如果res.write()返回 false 或者抛错说明连接已经不可写这时候就该清理资源了。所以心跳不只是保活也是健康检查。3.3 背压处理别让慢客户端拖垮服务端SSE 是单向推送服务端只管写但如果客户端消费慢比如网络差、页面卡顿res.write()会返回false表示内核缓冲区满了。这时候如果你继续无脑写数据会堆在 Node 的内存里最终把进程撑爆。正确的做法是监听drain事件function safeWrite(res, chunk) { const ok res.write(chunk); if (!ok) { // 暂停生产等 drain return new Promise(resolve res.once(drain, resolve)); } return Promise.resolve(); }对于 AI 流式输出这种场景如果客户端消费不过来其实更好的策略是丢弃中间 token只保留最新的完整状态。因为用户看的是最终结果中间过程丢几帧无所谓。这个取舍要根据业务定但一定要有背压意识不能假设客户端永远跟得上。3.4 多路复用与连接数管理一个页面开多个 SSE 连接是很常见的比如同时有通知流和对话流。浏览器对同域名下的 HTTP/1.1 连接数有限制通常 6 个SSE 连接会占用这个配额。如果开太多其他请求会被阻塞。HTTP/2 下这个问题缓解很多因为多路复用让连接数不再是瓶颈。服务端这边每个 SSE 连接都是一个常驻的 socket要关注文件描述符上限。Linux 默认ulimit -n是 1024单机跑几千个 SSE 连接就会撞墙。生产环境记得调大并且用ss -s或者netstat监控连接数。注意SSE 连接是有状态的这意味着它不能像普通 HTTP 请求那样随便水平扩展。如果你有多台服务端客户端连到哪台就只能收到哪台推的消息。要做集群推送得引入 Redis Pub/Sub 或者消息队列做广播让每台服务端都订阅同一份消息源。4. LangChain 流式输出stream、astream 与事件流4.1 LangChain 的流式抽象到底解决什么问题裸写 SSE 能跑通但一旦接入大模型问题就复杂了模型返回的是 token 流工具调用Agent会产生中间步骤链式调用Chain有多个环节你希望把这些都统一成一条流推给前端。LangChain 的流式 API 就是干这个的。它提供了几个层次的接口很多人搞不清该用哪个接口粒度适用场景stream/astream最终输出 token简单对话只要最终文本astream_events所有内部事件需要展示工具调用、中间步骤astream_log状态变更日志调试、可观测性astream返回的是最终输出的增量适合用户只关心答案的场景。astream_events返回的是细粒度事件流每个 token、每次工具调用、每个链的起止都会作为一个事件冒出来适合做那种思考过程可见的 Agent 界面。选哪个的判断标准很简单前端要不要展示中间过程。不要就用astream要就用astream_events。别一上来就用astream_events它的数据量和复杂度都高一个量级简单场景用它是杀鸡用牛刀。4.2 astream_events 的事件结构拆解astream_events吐出的事件每个都是一个字典关键字段有event事件类型比如on_chat_model_stream、on_tool_start、on_tool_end、on_chain_startname产生事件的组件名data事件负载on_chat_model_stream的data.chunk就是那个 tokenrun_id/parent_ids用于还原调用树知道这个事件属于哪个环节。一个典型的消费循环async for event in chain.astream_events(input, versionv2): kind event[event] if kind on_chat_model_stream: chunk event[data][chunk] if chunk.content: yield chunk.content elif kind on_tool_start: yield f\n[调用工具: {event[name]}]\nversionv2这个参数别漏v1 和 v2 的事件结构不一样网上很多老教程还是 v1 的写法照抄会报错。这里有个实战经验on_chat_model_stream事件的chunk.content有时候是空字符串尤其是模型返回工具调用的时候。前端如果无脑 append会出现一堆空消息。所以一定要判断if chunk.content再发。4.3 LangChain 和 LangGraph 在流式上的区别这是被问得最多的问题之一。简单说LangChain 是链式编排LangGraph 是图式编排。链是线性的图可以有环、有分支、有状态。流式方面LangGraph 的astream支持stream_mode参数可以选values每步的完整状态、updates每步的增量、messages消息 token 流、custom自定义。这个设计比 LangChain 更灵活因为图执行过程中状态是显式的你可以精确控制推什么。LangChain 的astream_events在 LangGraph 里也能用但 LangGraph 更推荐用stream_modemessages来拿 token 流语义更清晰。如果你在做多轮对话 工具调用的 AgentLangGraph 的状态管理会让流式输出好写很多因为每一步的状态变化都是可追踪的。至于LangChain 和 LangGraph 是不是过时了这种问题我的看法是工具没有过时只有合不合适。LangChain 的生态和集成度依然是最大的LangGraph 在复杂 Agent 编排上更顺手。两者不是替代关系很多项目是混用的。5. 前端接收与渲染标签不完整的坑5.1 EventSource vs fetch 流式读取前端接 SSE 有两条路。第一条是EventSource简单自动重连但只能 GET、不能带 header。第二条是fetchReadableStream手动解析 SSE 格式灵活但要把重连、解析都自己写。const res await fetch(/api/chat, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ question }) }); 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\n); buffer lines.pop(); for (const line of lines) { if (line.startsWith(data:)) { const data line.slice(5).trim(); // 处理 data } } }这里的关键是buffer的处理网络分片不保证按消息边界到达一次read()可能拿到半条消息也可能拿到一条半。所以必须维护一个缓冲区用\n\n切分最后一段不完整的留在 buffer 里等下次。这个逻辑写错就会出现消息偶尔丢失或者JSON 解析失败的诡异 bug。5.2 标签返回不完整怎么处理AI 流式输出 Markdown 时最典型的问题是模型正在输出一个代码块 还没闭合前端就渲染了结果整个页面格式乱掉。或者输出到一半的**加粗**只有前半截。处理思路有两个层次。第一层是渲染层容错用 Markdown 渲染器时对未闭合的代码块做特殊处理比如自动补全结尾的 或者把未闭合部分当纯文本渲染。很多 Markdown 库如 marked有breaks和容错选项但默认行为不一定符合预期要实测。第二层是数据层缓冲不要每个 token 都触发渲染而是用requestAnimationFrame或者节流比如 50ms 一次批量更新。这样既减少渲染压力也能让不完整的标签有更大概率在下一次渲染前闭合。实测下来50ms 的节流对打字机效果几乎无感但渲染性能提升明显。还有一个技巧维护一个已确认完整的文本和一个待确认的尾巴。用正则找出最后一个可能未闭合的代码块或标签把它放到待确认区只渲染前面的完整部分。这个逻辑稍微复杂但对代码类输出效果很好。5.3 Vue 里的流式对话实现要点Vue 做流式对话核心是把接收到的 token 增量地拼到响应式变量上。用ref存消息内容每收到一个 token 就content.value token。Vue 的响应式系统会自动触发视图更新。但要注意两点。第一高频更新会触发大量重渲染如果消息很长性能会掉。解决办法是用shallowRef配合手动触发或者把内容按段落拆分只更新变化的段落。第二滚动到底部的逻辑要处理好用户手动往上翻的时候不要强制拉回底部否则体验很差。判断方法是监听滚动事件记录用户是否在底部附近。const content ref(); const isAtBottom ref(true); function onToken(token) { content.value token; if (isAtBottom.value) { nextTick(() scrollToBottom()); } }这个是否在底部的判断是流式对话体验的关键细节很多 demo 都忽略了导致用户想回看历史时被不断打断。6. 常见问题排查速查表6.1 连接类问题现象可能原因排查方法消息延迟几十秒才到代理层缓冲加X-Accel-Buffering: no检查 Nginxproxy_buffering连接几分钟后断开空闲超时加心跳检查 LB 的 idle timeout 配置浏览器收不到但 curl 正常前端解析错误检查\n\n分隔和data:前缀重连后消息重复未处理 Last-Event-ID前端按 id 去重连接数上不去文件描述符限制调大ulimit -n检查连接泄漏6.2 流式输出类问题stream disconnected before completion这个报错八成是心跳没做或者间隔太长。先确认服务端有没有定时发注释行再看中间层的超时配置。如果心跳正常还报那就是模型侧响应太慢超过了某个环节的超时需要检查模型调用的超时设置。标签不完整的问题本质是渲染时机和数据完整性的错配。记住一个原则渲染可以滞后但不能超前。宁可让用户多等 50ms 看到完整内容也不要让他看到半截乱码。6.3 环境类问题热词里出现了一堆 Node 安装相关的问题比如npm : 无法加载文件 npm.ps1因为在此系统上禁止运行脚本。这是 Windows PowerShell 的执行策略问题解决办法是以管理员身份运行Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned还有nvm切换版本后全局包丢失的问题这是因为每个 Node 版本有独立的全局目录。切换版本后需要重新npm install -g安装全局工具。用nvm的话建议把常用的全局工具写进一个脚本切版本后一键重装。node-gyp和 Node 版本对应的问题通常出现在装原生模块时。node-gyp需要 Python 和 C 编译工具链Windows 上还要装 Visual Studio Build Tools。如果不想折腾优先找有没有预编译的替代包。7. 我踩过的几个真实坑第一个坑是在 Nginx 后面忘了关缓冲。本地开发一切正常一上测试环境消息就延迟 30 秒。查了半天代码最后发现是 Nginx 默认proxy_buffering on把 SSE 数据全缓存了。加上proxy_buffering off;和X-Accel-Buffering: no之后立刻正常。这个坑的教训是SSE 的问题先怀疑中间层。第二个坑是心跳间隔设太长。一开始设了 60 秒结果云厂商的 LB 是 60 秒超时卡在边界上偶尔断偶尔不断特别难排查。后来改成 15 秒再没出过问题。心跳间隔要明显小于超时时间留足余量别卡边界。第三个坑是前端没做去重。重连之后消息重复用户看到两遍同样的回答。加上基于 id 的去重逻辑后解决。这个逻辑虽然简单但一定要在项目初期就加上后期补会很麻烦因为要改数据结构。第四个坑是LangChain 事件流里空 content 没过滤。工具调用的时候chunk.content是空字符串前端 append 了一堆空消息看起来像卡住了。加上if chunk.content判断后正常。这种细节官方文档不会强调只能自己踩。8. 选型建议SSE、WebSocket 还是轮询最后说说选型。判断标准就一条通信方向。只需要服务端推、客户端听SSE。省事走标准 HTTP自动重连代理友好。需要双向实时通信WebSocket。SSE 做不了客户端主动推。实时性要求不高、消息量小轮询。实现最简单但延迟和资源消耗都高。AI 流式输出这个场景SSE 几乎是标准答案因为模型生成就是单向的。除非你要做用户中途打断生成这种交互那可能需要 WebSocket 或者用fetch的AbortController来中断请求——后者其实更简单fetch流式读取配合AbortController就能实现打断不一定非要上 WebSocket。我个人在实际项目里的组合是对话流用 fetch ReadableStream因为要 POST 带 body还要能中断通知流用 EventSource因为要自动重连且是纯推送。两套并存各取所长。这个组合用了大半年稳定性没问题。如果你还在纠结 LangChain 和 LangGraph 用哪个我的建议是先看你的编排复杂度。线性链用 LangChain 够了有分支、有循环、有显式状态就用 LangGraph。流式这块两者都能满足LangGraph 的stream_mode在复杂场景下更可控。工具是死的场景是活的别为了用新工具而用新工具。