1. 为什么流式输出成了 AI Native 应用的生死线做过大模型应用的人都有一个共同体会用户对等待的容忍度在 ChatGPT 出现之后被彻底重置了。以前一个接口转三秒用户觉得正常现在你让他在输入框里干等三秒才看到第一个字他已经在考虑关掉页面了。这不是用户变挑剔了而是流式输出把响应速度这件事从总耗时重新定义成了首字延迟。我最早做 AI 应用的时候用的是最朴素的方式前端发一个请求后端等模型把整段话生成完一次性返回 JSON。功能上没毛病但体验上就是灾难。一段五百字的回答模型生成要八到十二秒用户盯着 loading 转圈中途没有任何反馈很多人直接刷新页面重来。后来改成流式同样的模型、同样的耗时用户感知却完全变了——第一个字一两秒就蹦出来后面的内容像打字机一样往外冒哪怕总时长没变用户也会觉得这玩意儿挺快。这就是AI Native应用和传统 Web 应用在交互范式上的根本差异。传统应用是请求-响应的离散模型一次交互就是一个完整的结果AI 应用是持续生成的连续模型结果本身是一个随时间展开的过程。你的架构如果还停留在攒完再发的思路里那不管模型多强体验都会差一截。而支撑这种连续交互的底层技术就是流式输出架构。从最早的SSEServer-Sent Events到 WebSocket 的双向通信再到这两年逐渐成型的AG-UI协议这条演进路线背后其实是三个问题的不断升级怎么把 token 稳定地推给前端、怎么在推送过程中保持连接可靠、怎么让流式数据不只是文本而是可交互的界面事件。这篇文章我想把这条线完整捋一遍。不是教科书式的协议对比而是我在实际项目里踩过的坑、做过的取舍、以及最后沉淀下来的一套能直接抄作业的方案。如果你正在做 AI 对话产品、Agent 应用或者任何需要边生成边展示的功能这里面的东西应该能帮你少走不少弯路。2. 流式输出的三种技术路线与选型逻辑2.1 SSE最简单也最容易踩坑的方案SSE 本质上就是一个长连接的单向广播。客户端发一个普通的 HTTP 请求服务端把Content-Type设成text/event-stream然后保持连接不关闭持续往里面写数据。每条消息以data:开头以两个换行符结束浏览器端的EventSource会自动帮你解析。它最大的优势是简单。不需要额外的协议握手不需要引入 WebSocket 库服务端就是往一个 HTTP response 里写字符串前端就是监听onmessage。对于服务端单向推送 token这个场景SSE 几乎是天然契合的。但简单的东西往往藏着细节。我列几个实际项目里一定会遇到的问题第一个坑是缓冲。很多人写完 SSE 发现前端要等好几秒才一次性收到一大段而不是逐字冒出来。这十有八九是中间有代理或者框架在缓冲。Nginx 默认会缓冲响应需要显式关掉location /api/stream { proxy_pass http://backend; proxy_buffering off; proxy_cache off; proxy_set_header Connection ; proxy_http_version 1.1; chunked_transfer_encoding off; }后端框架也有类似问题。比如某些 Python 框架默认会等 response 完整才 flush需要手动调用 flush 或者用生成器逐条 yield。Node.js 里如果用 Express记得res.flushHeaders()并且不要用任何会做压缩的中间件——gzip 会把你的流式数据重新攒成块。第二个坑是超时。热搜词里那个stream disconnected before completion: idle timeout waiting for sse我太熟悉了。模型思考时间长一点中间十几秒没吐 token连接就被中间层掐断了。这个问题的根源是各层都有 idle timeout负载均衡器、反向代理、网关、甚至浏览器。解决办法是在应用层加心跳即使没有真实 token也定期发一个注释行: keep-aliveSSE 规范里以冒号开头的行是注释客户端会忽略但能让连接保持活跃。我一般设 15 秒发一次心跳比大多数中间层的 30 秒或 60 秒超时都短稳。第三个坑是重连。EventSource自带重连机制断线后会自动重发请求。这在普通推送场景是好事但在 AI 对话里是灾难——重连意味着重新触发一次模型生成用户会看到内容重复或者计费翻倍。所以生产环境我通常不用原生EventSource而是用fetchReadableStream手动解析这样能完全控制重连逻辑也能带上自定义 headerEventSource不支持自定义 header这是个硬伤。2.2 WebSocket双向能力带来的复杂度WebSocket 和 SSE 的核心区别是双向。SSE 只能服务端推WebSocket 两边都能随时发。那什么时候真的需要双向我的判断标准很简单如果用户在生成过程中需要打断或者追加指令那就需要 WebSocket。比如用户看到模型答到一半发现方向不对想点个停止或者补一句换个角度这种场景 SSE 就很别扭——你只能关掉连接再开一个新的而 WebSocket 可以直接在同一个连接里发一条控制消息。但 WebSocket 的代价是明显的。首先是连接管理你得自己维护连接池、处理断线重连、做心跳保活。热搜词里websocket 心跳机制实现被反复搜就是因为这是绕不过去的坎。一个典型的心跳实现是这样的// 客户端心跳 let heartbeatTimer null; let pongTimeout null; function startHeartbeat(ws) { heartbeatTimer setInterval(() { if (ws.readyState WebSocket.OPEN) { ws.send(JSON.stringify({ type: ping, ts: Date.now() })); pongTimeout setTimeout(() { console.warn(pong 超时主动重连); ws.close(); }, 5000); } }, 25000); } ws.onmessage (e) { const msg JSON.parse(e.data); if (msg.type pong) { clearTimeout(pongTimeout); return; } // 处理正常业务消息 };服务端收到ping要立刻回pong客户端如果 5 秒内没收到pong就认为连接已死主动关闭重连。这个双向确认比单纯发心跳更可靠因为能区分连接还在但服务端卡死和连接正常两种情况。其次是水平扩展。SSE 本质是 HTTP天然能被负载均衡器按普通请求分发WebSocket 是有状态的持久连接一旦连上就绑定了某台服务器。如果你的生成任务可能跨机器比如模型推理在 A 机器业务逻辑在 B 机器就得引入 Redis Pub/Sub 或者消息队列做跨节点转发复杂度直接上一个台阶。所以我的实际选型经验是纯对话展示用 SSE需要中途交互用 WebSocket。不要因为 WebSocket 看起来更高级就无脑上大部分 AI 对话场景 SSE 完全够用而且省掉一大堆连接管理的麻烦。2.3 AG-UI流式数据从文本到事件的跃迁前面两种都是传输层的方案解决的是怎么把字节推过去。但 AI Native 应用发展到今天光推文本已经不够了。想象一个场景模型在回答过程中要调用工具查天气然后根据结果继续回答。如果只推文本前端只能显示正在查询...这种模糊状态用户不知道背后发生了什么。再比如模型要生成一个表格、一个按钮、一个可点击的选项纯文本流根本表达不了。这就是AG-UI这类协议要解决的问题。它把流式输出的内容从文本片段升级成了结构化事件。一个典型的 AG-UI 事件流大概长这样event: run_started data: {runId: abc123, threadId: t1} event: text_message_start data: {messageId: m1, role: assistant} event: text_message_content data: {messageId: m1, delta: 今天} event: text_message_content data: {messageId: m1, delta: 北京} event: tool_call_start data: {toolCallId: tc1, toolName: get_weather} event: tool_call_args data: {toolCallId: tc1, delta: {\city\:} event: tool_call_end data: {toolCallId: tc1} event: text_message_end data: {messageId: m1} event: run_finished data: {runId: abc123}注意这里的关键变化每个事件都有类型和语义。前端不再只是收到一段文字就追加到气泡里而是能根据事件类型做不同的渲染——文本事件追加到消息气泡工具调用事件显示一个可折叠的正在调用 XX 工具卡片状态事件更新顶部的进度指示器。这种设计带来的最大好处是前后端解耦。以前前端要理解这段文本里[TOOL:weather]是什么意思得写一堆正则去解析现在协议层就把语义定义清楚了前端只负责按事件类型渲染后端换模型、换工具、换编排逻辑前端几乎不用改。AG-UI 目前还在演进中不同实现细节有差异但核心思想是一致的把流式输出从字节流抽象成事件流。这也是我认为 AI Native 应用架构接下来一定会走的方向——因为 Agent 的行为越来越复杂纯文本协议承载不了这么多语义。3. 一套能扛住生产的流式架构长什么样3.1 分层设计把生成和传输拆开我见过很多项目把模型调用和 SSE 推送写在一个函数里模型吐一个 token 就直接往 response 里写。这种写法在 demo 阶段没问题但一上生产就各种问题想加个日志得改核心逻辑想换个模型得重写推送想做多路复用根本无从下手。我的做法是强制分层中间用一个事件通道隔开[模型/Agent 层] --事件-- [事件总线] --订阅-- [传输层] --SSE/WS-- [前端]模型层只负责产生事件它不知道外面是 SSE 还是 WebSocket也不知道有几个消费者。传输层只负责把事件序列化后推给客户端它不关心事件是模型生成的还是工具产生的。中间的事件总线可以是一个内存队列也可以接 Redis 做跨进程分发。这个设计的好处我在一个多端项目里体会特别深。同一个对话Web 端用 SSE移动端用 WebSocket还有一个后台任务在消费同样的事件流做审计日志。如果按传统写法我得写三套推送逻辑分层之后三个消费者只是订阅同一个事件通道模型层一行代码都不用改。事件的数据结构我一般定义成这样interface StreamEvent { id: string; // 事件唯一 ID用于断点续传 type: EventType; // 事件类型决定前端如何渲染 timestamp: number; payload: unknown; // 具体内容结构由 type 决定 } type EventType | run.started | message.delta | message.completed | tool.started | tool.delta | tool.completed | run.failed | run.finished;id这个字段很关键。SSE 协议原生支持id:字段客户端断线重连时会带上Last-Event-IDheader服务端可以根据这个 ID 把断线期间的事件补发回去。这就是断点续传的基础。我在做长文档生成的时候一次生成可能几分钟中间网络抖一下很常见有了这个机制用户几乎无感。3.2 背压处理别让快生产者拖垮慢消费者流式系统里一个容易被忽视的问题是背压backpressure。模型生成 token 的速度可能很快比如每秒几十个但客户端网络慢或者前端渲染卡顿消费速度跟不上。如果中间没有缓冲控制事件就会在内存里堆积量大了一个进程就 OOM 了。SSE 场景下这个问题相对好处理因为 HTTP 的 TCP 窗口本身就是天然的背压信号——客户端不读服务端的 write 就会阻塞。但要注意别在服务端做无界缓冲比如先把所有事件塞进一个数组再统一发那就等于放弃了流式。WebSocket 场景要更小心因为ws.send()是异步的如果不管返回值一直发底层缓冲区会涨。我的做法是监控ws.bufferedAmount超过阈值就暂停生产async function pushEvent(ws, event) { const MAX_BUFFER 1024 * 1024; // 1MB while (ws.bufferedAmount MAX_BUFFER) { await new Promise(r setTimeout(r, 10)); } ws.send(JSON.stringify(event)); }这个等缓冲区降下来再发的逻辑本质上就是手动实现背压。虽然简单但能避免很多莫名其妙的崩溃。3.3 错误处理流式场景下的失败比你想的复杂传统请求失败很干脆要么成功返回要么报错。流式请求的失败是部分失败——前面已经推了一千个字推到一半模型报错了这时候怎么办我的处理原则是已经推出去的内容不撤回用事件明确告知失败位置和原因。具体做法是发一个run.failed事件带上已经完成的部分和错误信息前端把错误提示挂在消息末尾而不是清空整个气泡。用户至少能看到已经生成的内容体验上比啪一下全没了好得多。还有一种失败是静默失败连接还在但服务端已经不推数据了。这种最难排查因为前端看起来一切正常就是没内容。我的经验是在协议层加超时——如果 N 秒内没收到任何事件包括心跳客户端主动判定超时并重连。这个 N 一般设 30 到 60 秒比中间层的 idle timeout 短一点确保是客户端先发现异常。4. 实操从零搭一个带断点续传的 SSE 服务4.1 服务端实现要点我用 Node.js 写一个最小可用的例子把前面说的关键点都串起来。核心是一个事件缓冲区加一个推送循环const clients new Map(); // clientId - { res, lastEventId } function handleSSE(req, res) { const clientId req.query.clientId; res.writeHead(200, { Content-Type: text/event-stream, Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, // 关键告诉 Nginx 别缓冲 }); res.flushHeaders(); clients.set(clientId, { res, lastEventId: req.headers[last-event-id] || null }); // 心跳 const heartbeat setInterval(() { res.write(: ping\n\n); }, 15000); req.on(close, () { clearInterval(heartbeat); clients.delete(clientId); }); } function sendEvent(clientId, event) { const client clients.get(clientId); if (!client) return; client.res.write(id: ${event.id}\n); client.res.write(event: ${event.type}\n); client.res.write(data: ${JSON.stringify(event.payload)}\n\n); }几个细节值得展开说。X-Accel-Buffering: no这个 header 是给 Nginx 看的比改 Nginx 配置更灵活因为它是按响应生效的。res.flushHeaders()要显式调用否则某些框架会等 body 才开始发 header前端就一直卡在连接建立阶段。断点续传的实现依赖事件缓冲区。我一般保留最近 5 分钟的事件客户端重连时带上Last-Event-ID服务端从这个 ID 之后开始补发function replayEvents(clientId, lastEventId) { const events eventBuffer.getSince(lastEventId); for (const event of events) { sendEvent(clientId, event); } }这里有个坑补发的事件和实时事件可能重叠。如果补发还没结束新的实时事件就来了顺序会乱。我的做法是补发期间先把实时事件暂存补发完再按序推送。虽然增加了一点复杂度但能保证前端看到的事件流是严格有序的。4.2 前端解析别用 EventSource前面说过EventSource不支持自定义 header而且重连逻辑不可控。我推荐用fetchReadableStream手动解析async function streamChat(prompt, onEvent) { const response await fetch(/api/chat, { method: POST, headers: { Content-Type: application/json, Authorization: Bearer ${token}, }, body: JSON.stringify({ prompt }), }); 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 }); // SSE 以 \n\n 分隔事件 const parts buffer.split(\n\n); buffer parts.pop(); // 最后一段可能不完整留到下次 for (const part of parts) { const event parseSSE(part); if (event) onEvent(event); } } } function parseSSE(raw) { const lines raw.split(\n); const event { type: message, data: , id: null }; for (const line of lines) { if (line.startsWith(event:)) event.type line.slice(6).trim(); else if (line.startsWith(data:)) event.data line.slice(5).trim(); else if (line.startsWith(id:)) event.id line.slice(3).trim(); } if (!event.data) return null; try { event.payload JSON.parse(event.data); } catch { event.payload event.data; } return event; }这里最容易出错的是分块边界。reader.read()返回的 chunk 不保证按事件边界切分可能一个事件被切成两半也可能两个事件挤在一个 chunk 里。所以必须维护一个buffer用\n\n切分最后一段不完整的留到下一轮。这个逻辑我调了很久才稳定新手特别容易在这里翻车。4.3 参数选择超时、心跳、缓冲的取值依据这些参数没有标准答案但有几个经验值可以参考。心跳间隔我一般设15 秒理由是大多数云负载均衡器的 idle timeout 是 60 秒15 秒留了足够余量同时不会太频繁浪费带宽。客户端超时设45 秒比心跳间隔的三倍略短能容忍偶尔丢一两个心跳。事件缓冲区保留时长设5 分钟因为用户断线重连通常发生在几秒到几十秒内5 分钟足够覆盖绝大多数场景同时内存占用可控。如果做的是长文档生成这种可能断线几分钟的场景可以延长到 15 分钟但要配合事件落盘不能全放内存。背压阈值我设1MB这是 WebSocket 缓冲区的经验值。太小会导致频繁暂停影响吞吐太大则失去背压的意义。SSE 场景其实不太需要手动背压TCP 窗口会帮你处理但如果你在服务端做了事件聚合比如攒 10 个 token 发一次那就要注意聚合缓冲区的大小。5. 生产环境那些文档不会告诉你的坑5.1 常见问题速查表我把这些年遇到的高频问题整理成一张表方便对照排查现象可能原因排查方向解决手段前端一次性收到全部内容中间层缓冲检查 Nginx、框架、压缩中间件关 buffering禁用 gzip生成到一半连接断开idle timeout看各层超时配置加心跳缩短心跳间隔断线重连后内容重复重连触发重新生成检查重连逻辑用 Last-Event-ID 续传事件顺序错乱补发与实时事件竞争检查补发逻辑补发期间暂存实时事件内存持续增长事件缓冲区无上限监控缓冲区大小加 TTL 和容量上限部分客户端收不到负载均衡粘性问题检查连接分发用 clientId 做一致性哈希中文乱码分块切断多字节字符检查解码方式用 TextDecoder 的 stream 模式5.2 三个我踩过的真实坑第一个坑是 gzip。有次上线后发现流式效果没了前端要等十几秒才一次性显示。查了半天最后发现是某个中间件默认开了 gzip 压缩。压缩算法需要攒够一定数据才能有效压缩所以它会把流式数据缓冲起来。关掉 gzip 后立刻恢复正常。这个坑的隐蔽性在于本地开发环境没开压缩只有生产环境有所以特别容易漏。第二个坑是 HTTP/2。我们为了性能上了 HTTP/2结果发现 SSE 在某些客户端上表现异常。原因是 HTTP/2 的多路复用会让多个流共享一个 TCP 连接如果其中一个流阻塞可能影响其他流。而且 HTTP/2 的流控机制和 SSE 的长连接配合起来有些微妙。后来我们对流式接口单独走 HTTP/1.1问题就消失了。这不是说 HTTP/2 不能用而是流式场景要特别测试。第三个坑是移动端后台。移动端 App 切到后台后系统会挂起网络连接SSE 直接断掉。用户切回来发现内容停在半路。这个问题的解法是在 App 层监听前后台切换切回前台时用Last-Event-ID主动重连续传。如果没有断点续传机制这个场景基本无解。5.3 监控指标流式系统该看什么传统接口看 QPS、延迟、错误率就够了流式系统还得加几个专属指标。首字延迟TTFTTime To First Token是最重要的用户体验指标它直接决定用户觉得快不快。token 吞吐率反映生成速度突然下降可能是模型服务出问题了。连接存活时长能反映断线频率如果平均存活时间很短说明心跳或超时配置有问题。事件积压量反映背压情况持续增长说明消费跟不上生产。我一般把这几个指标做成看板首字延迟设 P95 告警超过 3 秒就查。这个阈值是根据用户感知定的——超过 3 秒用户就会开始怀疑是不是卡住了。6. 从 SSE 到 AG-UI架构演进背后的思考回头看这条演进路线其实每一步都是被需求推着走的。最早只要能把 token 推过去就行SSE 够了后来要支持打断和交互WebSocket 上场再后来 Agent 行为复杂了纯文本表达不了就有了 AG-UI 这类事件协议。我的判断是未来的 AI Native 应用流式协议一定会往事件化和可组合方向走。因为 Agent 不再是简单的问-答而是规划-调用工具-观察-再规划的循环每一步都需要给用户可见的反馈。文本流承载不了这种复杂度必须有结构化的协议。但我也想说不要为了追新而过度设计。如果你的应用就是简单的对话SSE 加个心跳和断点续传能稳定跑很久。AG-UI 这类协议的价值在复杂 Agent 场景才体现得出来简单场景上它只会增加前后端的对接成本。技术选型永远要看自己的实际需求而不是看哪个词更热。最后分享一个我自己的习惯每次做流式功能我都会先写一个最笨的版本——不用任何框架就是裸的 HTTP 加字符串拼接把整条链路跑通确认每个环节的数据流向。等这个版本稳定了再往上加抽象、加协议、加分层。这样做的原因是流式系统的 bug 往往藏在层与层之间的缝隙里抽象越多排查越难。先把地基打牢上面的楼才盖得稳。