SSE 不只是流式输出：从协议本质到 LangChain 实战

1. 被误读的 SSE它从来就不只是流式输出很多人第一次接触 SSE是在做 AI 对话界面的时候。前端要一个字一个字往外蹦后端接口不能一次性返回于是搜索流式输出搜到了EventSource抄了一段代码跑通了然后就把 SSE 归类成AI 打字机效果的实现方式。这个认知不能说错但实在太窄了。SSE 的全称是Server-Sent Events它是 HTTP 协议体系里一个正式的、有规范定义的推送机制。浏览器端用EventSource接口去消费服务端按照text/event-stream的格式持续写数据。它解决的问题本质上是在一条长连接上服务端如何主动、持续、有序地把事件推给客户端。流式输出只是它最显眼的一种应用形态而不是它的全部。我见过太多项目把 SSE 用成了伪流式——后端攒一批数据发一次前端收到后整段替换看起来在动实际上延迟很高、体验很糙。也见过有人把 SSE 和 WebSocket 混为一谈觉得能推消息就是一样的。这两者在连接模型、协议开销、重连机制、适用场景上差别很大选错了会在后期付出代价。这篇文章想做的事情很明确把 SSE 从协议层到应用层讲透把EventSource的使用边界讲清楚再落到 LangChain 这类框架的流式输出上看看工业级的实时推送到底是怎么搭起来的。中间会穿插 Node 环境相关的实操细节因为很多坑其实出在环境和工具链上而不是 SSE 本身。适合已经写过一点前端、想认真搞懂实时推送的开发者也适合正在做 AI 应用、被流式输出折磨过的同学。2. SSE 的协议本质一条被低估的长连接2.1 它到底长什么样SSE 的报文格式简单到有点朴素。服务端返回的 Content-Type 是text/event-stream然后按照固定格式往连接里写文本event: message data: {content: 你} data: {content: 好} id: 1 retry: 3000几个关键字段需要记牢data消息体可以多行多行会被拼接成一个字符串行与行之间用换行符连接。event事件类型名前端可以用addEventListener监听自定义事件不写就是默认的message。id事件 ID浏览器会自动记录最后收到的 ID断线重连时会通过Last-Event-ID请求头带回去。retry重连间隔毫秒数告诉浏览器断线后多久重试。以:开头的行是注释常用来做心跳保活。这里有个容易被忽略的点每条消息之间必须用空行分隔。我见过有人写服务端代码时忘了这个空行结果前端一直收不到消息排查半天以为是跨域问题。实际上 SSE 的解析器是靠空行来判定一条消息结束的没有空行数据就一直挂在缓冲区里。2.2 为什么它比轮询和 WebSocket 更适合某些场景先做个对比把三种常见方案摆在一起看维度短轮询WebSocketSSE连接方向客户端反复请求全双工服务端单向推送协议HTTP独立协议握手用 HTTPHTTP自动重连无需自己实现浏览器内置断点续传无需自己实现内置 Last-Event-ID代理/网关兼容好一般需配置升级好就是普通 HTTP数据格式任意二进制/文本文本适用场景低频状态查询双向高频交互服务端单向推送SSE 最大的优势在于它就是普通 HTTP。这意味着它天然穿过大多数代理、网关、负载均衡器不需要额外的协议升级配置。而 WebSocket 在很多企业网络环境里会被中间设备拦截或降级配置起来麻烦得多。另一个被低估的能力是自动重连和断点续传。浏览器发现连接断了会按照retry指定的间隔自动重连并且把最后收到的id通过Last-Event-ID头带回去。服务端只要根据这个 ID 把断线期间的消息补发就实现了消息不丢。这个能力在 WebSocket 里是要自己从零实现的。但 SSE 的短板也很明显单向。客户端不能通过这条连接发消息给服务端要发只能另开一个 HTTP 请求。所以聊天、协作编辑这类需要双向高频通信的场景WebSocket 更合适。而通知推送、日志流、AI 流式输出、进度上报这类服务端说、客户端听的场景SSE 是更省事的选择。2.3 连接数限制这个老问题浏览器对同域名下的 HTTP/1.1 连接数有限制通常是 6 个。SSE 会长期占用其中一个。如果你在一个页面里开了多个EventSource很容易把连接池占满导致其他请求排队。解决办法有两个一是升级到 HTTP/2多路复用让连接数限制基本消失二是多个数据流复用同一个 SSE 连接通过event字段区分不同业务类型。第二种做法在后端稍微复杂一点但前端只需要一个EventSource实例管理起来更清爽。提示如果你的应用还在 HTTP/1.1 上跑并且页面里同时有多个实时数据源优先考虑合并成一条 SSE 连接用事件类型分流而不是开多个 EventSource。3. EventSource 的实战边界能做什么不能做什么3.1 基础用法和那些必须知道的限制EventSource的 API 简单得让人放松警惕const es new EventSource(/api/stream); es.onmessage (e) { console.log(收到:, e.data); }; es.addEventListener(progress, (e) { console.log(进度事件:, e.data); }); es.onerror (err) { console.error(出错了, err); };但它的限制同样明显而且每一条都可能在实际项目里咬你一口不能自定义请求头。EventSource构造函数只接受 URL 和可选的withCredentials你没法加Authorization头。这意味着基于 Token 的鉴权方案要么改用 Cookie要么把 Token 放到 URL 参数里不推荐会进日志要么用 polyfill 库。只能发 GET 请求。想传复杂参数只能拼在 URL 上长度受限语义也差。不能主动关闭后重连。调用es.close()之后浏览器不会再自动重连。如果你希望暂停一下再恢复得自己重新 new 一个。错误处理很粗糙。onerror拿不到 HTTP 状态码你无法区分是 401 鉴权失败还是 500 服务端错误只能靠重连行为间接判断。3.2 用 fetch 替代 EventSource 的取舍因为上面这些限制很多项目干脆放弃EventSource改用fetchReadableStream手动解析 SSE 格式。这样能自定义请求头、能用 POST、能拿到完整的状态码。const resp await fetch(/api/stream, { method: POST, headers: { Content-Type: application/json, Authorization: Bearer ${token} }, body: JSON.stringify({ query: 你好 }) }); const reader resp.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:)) { const data line.slice(5).trim(); if (data) handleMessage(data); } } }代价是你要自己实现重连、自己处理 Last-Event-ID、自己解析事件边界。这段代码看着不长但边界情况很多分片可能把一个多字节字符切断data:后面可能有也可能没有空格空行判定要准确。我踩过的坑是decoder.decode没传{ stream: true }导致中文被截断成乱码排查了很久。所以选型逻辑很清楚如果鉴权简单、只需要 GET、能接受浏览器默认重连行为用 EventSource如果需要自定义头、POST、精细控制重连用 fetch 手动解析。没有绝对优劣看场景。3.3 一个真实的标签返回不完整排查过程热词里有个标签返回未完整怎么处理这个我太有共鸣了。做 AI 流式输出时模型返回的内容里带结构化标签比如answer.../answer前端按标签解析渲染。结果经常出现标签只收到一半的情况ans就断了或者/answer缺了最后的。排查链路是这样的第一步先确认是传输层截断还是解析层截断。在onmessage里把原始e.data打出来如果原始数据本身就是半截标签那是服务端分片的问题如果原始数据完整但渲染出来是半截那是前端解析逻辑的问题。第二步如果是传输层检查服务端的 flush 时机。Node 里用res.write()之后如果没有触发 flush数据会攒在缓冲区里。SSE 场景下通常要配合res.flushHeaders()和确保没有中间件在做压缩缓冲。第三步如果是解析层问题往往出在按 chunk 边界解析。流式数据到达的边界和标签边界没有任何关系一个标签可能横跨三个 chunk。正确做法是维护一个缓冲区每次收到新数据就追加然后尝试匹配完整的标签匹配不到就等下一批。let tagBuffer ; function processChunk(chunk) { tagBuffer chunk; const regex /answer([\s\S]*?)\/answer/g; let match; let lastIndex 0; while ((match regex.exec(tagBuffer)) ! null) { renderAnswer(match[1]); lastIndex regex.lastIndex; } tagBuffer tagBuffer.slice(lastIndex); }这个模式的核心思想是永远不要假设一次收到的数据是一个完整语义单元。流式处理的本质就是和边界不确定性打交道。4. LangChain 流式输出框架帮你做了什么又藏了什么4.1 从 LLM 到前端的完整链路LangChain 的流式输出不是单一环节的事它是一条链路模型 API 的流式响应 → LangChain 的回调/流式接口 → 你的服务端 → SSE → 前端渲染。任何一环没打通前端都看不到逐字效果。先看模型层。主流模型 API 都支持stream: true返回的是一系列 chunk每个 chunk 包含一小段增量文本。LangChain 把这层封装成了stream()方法from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, streamingTrue) for chunk in llm.stream(讲个笑话): print(chunk.content, end, flushTrue)注意streamingTrue这个参数。有些模型封装默认不开流式你不显式打开stream()拿到的可能是一次性返回的完整结果。这个坑我在项目里遇到过前端一直不蹦字查了半天发现是模型初始化时漏了这个参数。再往上一层是 Chain 和 Agent。LangChain 提供了astream和astream_events两个接口。astream返回的是最终输出的流astream_events返回的是全链路事件流包括 LLM 开始、LLM 结束、工具调用、检索等中间步骤。做工业智能体的时候astream_events更有价值因为你能把正在检索知识库正在调用工具这些状态也推给前端。4.2 服务端怎么把 LangChain 的流接到 SSE这是整个链路里最容易出问题的地方。Python 侧用 FastAPI 举例from fastapi import FastAPI from fastapi.responses import StreamingResponse from langchain_openai import ChatOpenAI app FastAPI() llm ChatOpenAI(modelgpt-4o-mini, streamingTrue) async def event_generator(query: str): async for chunk in llm.astream(query): if chunk.content: yield fdata: {chunk.content}\n\n yield data: [DONE]\n\n app.get(/api/stream) async def stream(query: str): return StreamingResponse( event_generator(query), media_typetext/event-stream, headers{ Cache-Control: no-cache, X-Accel-Buffering: no, } )几个关键点值得展开X-Accel-Buffering: no这个头是给 Nginx 看的告诉它不要缓冲这个响应。没有这个头Nginx 默认会攒够一定大小才转发前端就会看到卡一下蹦一大段的效果。这个坑非常隐蔽本地开发直连没问题一上生产经过 Nginx 就出问题。media_typetext/event-stream必须设置否则浏览器不会按 SSE 解析。[DONE]结束标记是 OpenAI 风格的约定前端收到它就主动close()避免连接一直挂着。异步生成器要用async for因为 LangChain 的astream是异步的。如果你用同步的stream配异步框架会阻塞事件循环多个用户同时请求时直接卡死。4.3 前端消费 LangChain 流的完整写法前端这块如果用EventSource因为不能发 POST参数只能拼 URL。更常见的做法是用 fetch 手动解析前面已经给过代码。这里补充一个完整的、带错误处理和结束判定的版本async function chat(query, onChunk, onDone) { const controller new AbortController(); const resp await fetch(/api/stream?query encodeURIComponent(query), { signal: controller.signal }); if (!resp.ok) { throw new Error(HTTP ${resp.status}); } const reader resp.body.getReader(); const decoder new TextDecoder(); let buffer ; try { while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); let idx; while ((idx buffer.indexOf(\n\n)) ! -1) { const rawEvent buffer.slice(0, idx); buffer buffer.slice(idx 2); const dataLines rawEvent .split(\n) .filter(l l.startsWith(data:)) .map(l l.slice(5).trimStart()); const data dataLines.join(\n); if (data [DONE]) { onDone(); return; } onChunk(data); } } } finally { controller.abort(); } }这段代码里buffer.indexOf(\n\n)是核心——按空行切分事件而不是按 chunk 切分。这是 SSE 解析的正确姿势。很多人图省事直接对每个 chunk 做split(data:)遇到跨 chunk 的事件就崩了。5. 环境与工具链那些和 SSE 无关却总在拖后腿的坑5.1 Node 环境安装的常见翻车现场热词里 Node 相关的词占了很大比例说明很多人在环境这一步就卡住了。这些坑和 SSE 本身没关系但会直接影响你能不能跑通 demo。Windows 上npm.ps1无法加载。报错信息是因为在此系统上禁止运行脚本。这是 PowerShell 的执行策略问题不是 Node 的问题。解决办法是以管理员身份运行 PowerShell执行Set-ExecutionPolicy RemoteSigned然后确认。或者改用 CMD 而不是 PowerShell。nvm 切换 Node 版本后全局包丢失。这是 nvm 的设计使然每个 Node 版本有独立的全局包目录。切换版本后需要重新npm install -g你需要的工具。如果嫌麻烦可以用nvm reinstall-packages从旧版本迁移。国内镜像配置。默认源下载慢是常态配置镜像能省很多时间npm config set registry https://registry.npmmirror.comLinux 离线安装。内网环境没法直接下载需要提前在有网机器上下好对应架构的二进制包解压后配置 PATH。注意 glibc 版本兼容性高版本 Node 可能依赖较新的 glibc。node-gyp和 Node 版本对应。编译原生模块时node-gyp需要匹配的 Python 和编译工具链。Node 18 和 Node 20 对 Python 版本要求不同装错了会报一堆看不懂的错。建议用nvm管理版本遇到原生模块编译问题先确认 Node 版本。5.2 流式场景下的分片上传报错热词里有个node 分片上传文件时报错 request aborted。这个和 SSE 是两回事但都涉及流式处理值得说一句。request aborted通常意味着客户端在服务端还没读完流的时候就断开了连接。常见原因客户端超时设置太短、服务端处理太慢、中间代理有超时限制。排查思路是先在服务端监听req.on(aborted)和req.on(close)确认是客户端主动断还是服务端出错。如果是代理超时调整代理的proxy_read_timeout。如果是客户端超时检查 fetch 或 axios 的 timeout 配置。注意SSE 长连接同样会遇到代理超时问题。默认情况下很多代理 60 秒没数据就断连。解决办法是服务端定期发送心跳注释行: ping\n\n保持连接活跃。5.3 一个容易被忽略的心跳机制SSE 连接如果长时间没有数据中间的网络设备可能认为连接空闲而断开。浏览器虽然会自动重连但频繁重连会造成消息重复或丢失。标准做法是服务端每隔 15 到 30 秒发一个心跳const heartbeat setInterval(() { res.write(: heartbeat\n\n); }, 20000); req.on(close, () { clearInterval(heartbeat); });以:开头的行是 SSE 注释浏览器会忽略它但连接保持活跃。这个细节在本地开发时完全看不出问题一上生产就暴露。6. 选型与演进SSE、WebSocket、LangGraph 该怎么摆6.1 SSE 和 WebSocket 不是二选一很多人纠结到底用哪个其实它们可以共存。一个典型的 AI 应用可能是这样对话流式输出用 SSE单向、简单、穿透好协同编辑或实时双向控制用 WebSocket双向、低延迟。按功能模块选而不是全站统一。判断标准可以简化成一句话如果客户端不需要在这条连接上主动发消息优先 SSE。6.2 LangChain 和 LangGraph 的关系热词里反复出现langchain 和 langgraph 的区别这里说清楚。LangChain 是基础框架提供模型封装、Chain、工具、检索等组件。LangGraph 是构建在 LangChain 之上的状态机式编排框架用来处理有循环、有分支、有状态的复杂 Agent 流程。简单说LangChain 适合线性的、简单的调用链LangGraph 适合需要多轮循环、条件分支、人工介入的复杂智能体。做工业智能体时如果流程里有调用工具 → 判断结果 → 决定是否重试这种循环LangGraph 更合适。至于它们是不是过时了我的看法是框架会演进但底层的流式输出、事件驱动、状态管理这些概念不会过时。学框架的同时理解底层机制换框架时迁移成本才低。6.3 流式输出的性能与体验权衡最后说一个实操心得。流式输出不是越快越好。如果模型吐字太快前端每个 chunk 都触发一次 DOM 更新会造成大量重排页面反而卡顿。常见的优化是批量渲染用一个缓冲区攒一小段时间比如 50ms的 chunk然后一次性更新 DOM。这样既保留了逐字效果又减少了渲染压力。let pending ; let timer null; function scheduleRender(text) { pending text; if (timer) return; timer setTimeout(() { renderToDOM(pending); pending ; timer null; }, 50); }这个 50ms 是个经验值可以根据实际体验调整。太小了没效果太大了会感觉卡顿。我在实际项目里踩过的另一个坑是流式输出和 Markdown 渲染结合时未闭合的语法会导致渲染错乱。比如模型正在输出一个代码块三个反引号只到了两个Markdown 解析器会把后面的内容全当成代码。解决办法是渲染前对未闭合的语法做容错处理或者用支持增量渲染的 Markdown 库。这套东西搭下来你会发现 SSE 本身其实很简单难的是把它和框架、环境、前端渲染、网络中间件正确地串起来。每一个环节都有它自己的脾气而经验就是被这些脾气磨出来的。