做 Agent 的人早晚会卡在同一个地方流式输出。模型侧吐 token 是流式的前端 UI 却需要一个一个字段地更新中间任何一环处理不好用户看到的就是半个字挂在那里转圈或者整段对话卡到超时。DeepSeek-Harness 这一期把整条链路单独拎出来讲核心就一句话StreamChunk 是管道里的最小传输单位UI 是管道的终点。从模型返回的字节流到前端渲染出来的增量文本中间需要一套严谨的协议、缓冲和状态同步机制。这篇基于我实际调试 Harness 流式管道的经验把从 StreamChunk 到 UI 的完整路径拆开揉碎讲清楚适合正在做 LLM Agent 前后端联调、或者想把输出体验做好的同学参考。1. 为什么 Agent 场景必须有一套独立的流式输出管道1.1 一次性返回的“黑盒”体验有多糟糕先说一个反例。早期做 Agent 原型的时候很多团队图省事让后端一次性把完整回答拼好再返回。模型生成时间短则几秒长则几十秒尤其涉及多轮工具调用时这段时间前端只有一个 loading 动画。用户在等一个不确定长度的结果第一反应就是刷新页面或者怀疑服务挂了。这个体验放在聊天窗口里尤其致命——聊天窗口的用户预期是“我说一句你应一句”而且是边说边出字。流式输出的价值不只是“看起来快”它直接改变用户对系统状态的判断。首字延迟从几秒甚至十几秒压缩到几百毫秒用户立刻知道系统在响应、在生成、没死。对于 Agent 这种经常要串多个步骤的复杂系统来说还能把“当前正在执行工具调用”“已经拿到中间结果”“正在生成最终回复”这些状态逐步透出而不是全程一个 spinner。1.2 Agent 场景比普通 Chat 复杂在哪普通 Chat 的流式输出本质上是一条文本线模型出 token前端追加文本。但 Agent 场景的流式输出要复杂得多原因是流里混着多种类型的信息增量文本模型生成的回复片段直接展示给用户。工具调用事件Agent 决定调某个工具附带参数 JSON 片段。状态变更从 planning 切到 execution再切到 final answer。元数据耗时、token 用量、中间结果摘要。这一堆东西如果只用一个字符串管道往下推前端根本没法区分“这段话是给我看的”还是“这个字段是内部状态”——所以必须要有StreamChunk这样的结构化单元来承载。DeepSeek-Harness 的做法是对模型输出做流式解析把不同类型的增量封装成带类型的 Chunk通过统一的管道推给 UI 层。前端只需要按 type 分派处理逻辑文本追加到对话区工具事件更新状态面板状态变更切换指示器互不干扰。1.3 管道设计决定 Agent 系统的扩展边界另外一点很重要流式输出管道的设计直接决定 Agent 系统的扩展边界。你今天只做一个聊天框文本流就够了明天想加一个工具调用可视化面板就得在管道里新增事件类型和后端解析逻辑后天想支持用户中途打断又得在管道的取消语义上下功夫。DeepSeek-Harness 把 StreamChunk 设计成一套自包含的消息协议而不是把解析逻辑散落在模型层和 UI 层就是为了让后续加功能不推倒重来。说白了管道是 Agent 系统的“血管”血管长什么样决定了你能长多大。2. StreamChunk流式传输的最小结构化单元2.1 Chunk 的设计思路StreamChunk 不是简单地把文本切成小块。它需要同时满足三个角色后端解析器的产出、管道的传输载具、前端渲染的数据源。所以在设计上每个 Chunk 应当是自描述的——前端拿到一个 Chunk 不用回看历史就知道这是什么、怎么处理。一个典型的 StreamChunk 结构长这样TypeScript 表述export type StreamChunkType | text_delta | tool_call_start | tool_call_delta | tool_call_end | status_change | meta; export interface StreamChunk { type: StreamChunkType; chunkId: string; // 全局递增ID用于乱序检测与调试 sessionId: string; // 会话标识多开场景下路由用 sequence: number; // 序号前端按此排序 timestamp: number; // 产物时间利于延迟统计 payload: { content?: string; // 文本增量、参数增量 toolName?: string; // 工具名 toolCallId?: string; // 本次调用唯一ID status?: running | done; // 状态事件 metrics?: Recordstring, any; // 元数据 }; }字段不多但每个字段都在解决具体问题type让前端能够用 switch 或映射表分派逻辑不用写一堆 if 判断字符串特征。sequence解决乱序问题。虽然 HTTP 长连接下乱序很少见但在代理重连、多路复用场景下仍有可能前端拿到之后就做一次简单的比较缓存。chunkId主要是给排查问题用的。线上定位“某个字丢了”的时候没有全局唯一 ID 只能靠猜。sessionId是给多会话场景留的口子。你一旦在同一个页面做多 Agent 并行没有 sessionId 的管道就废了。2.2 为什么 payload 要设计成多态而不是大而全的字段集很多团队一开始图方便把可能用到的字段全部塞进 Chunk{ type: text_delta, content: ..., toolName: null, toolCallId: null, status: null, metrics: null, }这种设计的坏处是知道足够但语义模糊。前端处理 text_delta 时还得判断toolName有没有值没法对单个 Chunk 做紧凑校验也浪费网络流量。更好的做法是让 payload 跟随 type 变化形成可辨识联合type StreamChunk | { type: text_delta; sequence: number; payload: { content: string } } | { type: tool_call_start; sequence: number; payload: { toolName: string; toolCallId: string } } | { type: tool_call_delta; sequence: number; payload: { argDelta: string } } | { type: status_change; sequence: number; payload: { status: Phase } };TypeScript 的类型收窄可以在编译期就排除“处理 text_delta 却发现 payload 里没有 content”这类错误。后端语言的序列化框架比如 Pydantic discriminated union也能做同样的事。这是把管道当成“协议”而不是“数据结构”来设计的关键差异。2.3 Chunk 的合并规则与边界条件流式解析的另一个核心问题是合并规则。模型输出天然是一串字节你要在合适的位置切分出一个个有意义的 Chunk而不是平均切块。比如 DeepSeek-Harness 解析工具调用时需要识别出参数 JSON 的起始和终止边界在边界不完整的时候把字节缓存住等数据攒够了再生成 Chunk。实际操作中我总结的合并原则是三条文本类增量按语义片段划分比如按句子、按换行、按 Markdown 块边界。完全不切只等一次性 flush 的话UI 端就会长时间无响应切得太碎每几个字符一个 Chunk则前端渲染压力大、网络包数量爆炸。我的经验是把单次 Chunk 文本控制在 20~100 字之间具体结合模型吐字速度和 UI 刷新帧率来调。工具参数的 JSON 增量必须等边界闭合才发。否则前端拿到半截 JSON 去解析直接抛异常。后端解析器要做括号配对检测欠括号就继续缓存。状态变更事件必须单独成块优先于文本增量发送。这样前端才能先切状态再渲染文本避免出现“状态栏写着跑批对话框里却在回答用户”的错位。注意Chunk 的切分粒度直接影响前端渲染帧率。做流式输出时最高效的方式是按 UI 的刷新节奏做一个小的 buffer window一般 50~100ms攒一窗数据再 flush 给前端既能保证平滑又不至于每吐一个字就发一次 HTTP 响应。3. 从 StreamChunk 到 UI 的链路构建3.1 传输层选型SSE 为什么比 WebSocket 更适合默认场景现在流式传输的技术选型基本就两个SSEServer-Sent Events和 WebSocket。很多人默认 WebSocket 更强但 Agent 场景下绝大部分时间 SSE 是更合理的默认选择。SSE 是单向的服务端持续把事件推给客户端客户端只用普通的 HTTP 连接接收。这正好匹配 LLM 流式输出的模型——模型在生成客户端在看。而 WebSocket 是双向的功能虽强但实现复杂度、连接管理成本、断线重连的语义都更重。Agent 场景里用户确实需要“取消生成”或“发送新消息”这些交互可以走独立的普通 HTTP POST 接口不需要为此把整个传输层升级成 WebSocket避免两端状态机复杂化。DeepSeek-Harness 的做法我记得也是 SSE 为主链路、用户指令走独立请求这套组合在工程上最简单可靠。SSE 协议本身不复杂一个简单的 event stream 响应体长这样data: {type:status_change,sequence:1,payload:{status:planning}} data: {type:text_delta,sequence:2,payload:{content:我正在}} data: {type:text_delta,sequence:3,payload:{content:准备调用工具}}每行data:就是一条消息用两个换行符分隔。客户端用EventSource或者fetch ReadableStream 就能消费。3.2 服务端流式管道的实现要点在 Harness 这一侧流式管道大致分为四层模型适配层—— 对接各家模型的 stream 接口统一转成内部的 Chunk 流。DeepSeek-Harness 用的应该是 OpenAI 风格的流式协议stream_options: {include_usage: true}这样可以在流尾拿到完整 token 统计。解析与编排层—— 做两部分工作识别工具调用边界、构造状态事件。模型输出里如果触发了函数调用要以已输出的 function name 未闭合的 JSON 参数作为一个连续流直到}闭合才发出tool_call_end。背压与缓冲层—— 模型吐字速度可能比前端消费速度快也可能慢。服务端要在中间做缓冲和削峰快时攒着按窗口发慢时至少维持心跳。这一步处理不好会出现两类线上事故缓冲区无限增长导致内存被拖垮或者前端长时间收不到数据以为断了。协议封装层—— 把 Chunk 序列化成 SSE 帧格式处理编码强制 UTF-8、心跳帧每 15 秒发一个: ping注释行防止中间代理超时掐断连接。核心代码Node.js 风格伪代码大致长这样async function handleLLMStream(req, res) { res.writeHead(200, { Content-Type: text/event-stream; charsetutf-8, Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, // 关闭Nginx缓冲否则SSE会被吞 }); const buffer []; let lastFlush Date.now(); for await (const rawChunk of modelStream) { const chunks parser.parse(rawChunk); // 解析成StreamChunk数组 for (const chunk of chunks) { buffer.push(chunk); } // 每 80ms 或攒满 32 个 chunk 就 flush 一次 if (chunks.length 0 || Date.now() - lastFlush 80) { flushBuffer(); lastFlush Date.now(); } } flushBuffer(); // 发送结束标记 res.write(data: ${JSON.stringify({ type: meta, payload: { done: true } })}\n\n); res.end(); // 记得给前端一个干净的结束标记否则EventSource会一直挂着 }真正做的时候有两个细节务必注意X-Accel-Buffering: no这行头很重要。很多团队在本地联调时好好的一上 Nginx 就发现前端半天刷不出字多数是因为 Nginx 开了缓冲把 SSE 帧按块吞了。反向代理后面做流式输出这个地方必须设置。结束标记要显式发送。SSE 连接如果不主动关闭EventSource会按要求自动重连前端可能拿到一堆重复内容。要么发 done 标记让前端主动 close要么在服务端直接 end 连接。3.3 前端消费与增量渲染的基本模式前端拿到 StreamChunk 并不等于 UI 直接更新。中间还有一层状态管理和渲染协调我通常会用一个 reducer 模式来组织interface StreamState { text: string; toolCalls: Recordstring, ToolCallState; phase: idle | planning | executing | finalizing; metrics?: Recordstring, any; } const streamReducer (state: StreamState, chunk: StreamChunk): StreamState { switch (chunk.type) { case text_delta: return { ...state, text: state.text chunk.payload.content }; case tool_call_start: { const { toolCallId, toolName } chunk.payload; return { ...state, toolCalls: { ...state.toolCalls, [toolCallId]: { toolName, status: running, args: }, }, }; } case tool_call_delta: { const { toolCallId, argDelta } chunk.payload; return { ...state, toolCalls: { ...state.toolCalls, [toolCallId]: { ...state.toolCalls[toolCallId], args: (state.toolCalls[toolCallId]?.args || ) argDelta, }, }, }; } case status_change: return { ...state, phase: chunk.payload.status }; default: return state; } };每次收到一个 StreamChunk就通过dispatch更新 React/Vue 的状态UI 因为绑定了状态而自动更新。增量文本不需要整段重渲染框架的 diff 机制天然支持字符串插到末尾。但这里有一个容易踩的坑不要在 render 函数里直接做 chunk 拼接。如果把状态存在 ref 里每次渲染时手动 append 文本React 18 的并发渲染特性会导致文本顺序错乱。一定要用不可变更新的方式如上面的 reducer 写法保证每次 render 都基于同一份快照。4. UI 层交互与状态同步的工程细节4.1 工具调用状态的可视化Agent 和普通 Chat 在 UI 上的最大区别在于工具调用可视化。如果只是在文本流里输出一行“调用工具search”用户看不懂对故障排查也没帮助。我的建议是把工具调用做成一卡片样式跟随流式状态变化而发生状态迁移收到tool_call_start渲染一张卡片标题“正在调用 xxxx 工具”显示 spinner。收到一段段tool_call_delta卡片展开一个等宽字体区域显示正在累积的参数 JSON。收到tool_call_endspinner 变成对勾或绿色状态条展示工具返回的结果摘要。这套 UI 的更新节奏要和 Chunk 对齐。参数 delta 可能以每秒多次的频率到来卡片里的 JSON 内容刷新频繁要保证滚动条不跳、文字不闪。我的做法是工具参数区域只在累积了新的完整行时才刷新而不是每个 delta 都触发重渲染。可以在组件里做个节流50ms 的窗口内只渲染一次最新值。const ToolCallCard ({ toolCallId }) { const toolCall useAppSelector((s) s.stream.toolCalls[toolCallId]); const [displayArgs, setDisplayArgs] useState(); useEffect(() { const timer setInterval(() { setDisplayArgs(toolCall?.args || ); }, 60); return () clearInterval(timer); }, [toolCall?.args]); // 60ms 节流后渲染避免高频 delta 打爆 DOM 更新 return ( div classNametool-card div classNametool-name{toolCall?.toolName}/div pre classNametool-args{displayArgs}/pre /div ); };4.2 取消生成的三层协作用户点了“停止生成”如果只是断掉前端的 EventSource服务端模型还在继续跑浪费算力还会留下一个半截状态。正确做法是三层协作UI 层调用后端的 cancel 接口同时本地立即关闭 SSE 连接、把状态切到 idle给一个“已取消”的占位提示。服务端收到 cancel 请求后在运行时层给模型 stream 发 cancellation token。OpenAI 系 SDK 一般有AbortController或类似机制把它挂在流式请求的上下文里。同时服务端销毁该会话的缓存和缓冲区。Agent 循环层取消不只意味着停掉输出流还意味着当前工具调用链要中断、后续状态机要复位。如果不把 cancel 传播到 Agent 执行器可能出现“前端已取消、后台还在调工具”的资源泄漏。async function cancelGeneration(sessionId) { const session sessions.get(sessionId); if (session?.abortController) { session.abortController.abort(); // 模型流停止 session.state cancelled; // Agent状态复位 session.buffer []; } }4.3 竞态与过期 Chunk 的处理前端状态管理还会遇到一类隐蔽问题用户连续发送两条消息前一条的流还没结束后一条已经开始了。此时两条流的 StreamChunk 会在前端打架。我的处理方式是在发送新消息时执行一次“管道重置”给当前流加一个generationId每次新请求生成新 ID。reducer 里存currentGenerationId收到 Chunk 时先比较 ID不一致就丢弃。新消息发出时旧流即使还在回调也直接忽略。这样既不会出现文本互相穿插也不会把旧工具调用状态渲染到新会话里。很多团队把这个逻辑漏掉导致用户连续提问后 UI 出现“上一个搜索卡片混在新回答里”的灵异现象多半就是这里的问题。if (chunk.generationId ! state.currentGenerationId) { return state; // 丢弃过期流的chunk }5. 流式管道运维实战与问题速查5.1 本地联调与线上实测的差异清单流式管道本地跑通不难难在线上环境拓扑变了之后就出各种怪问题。我把踩过的坑整理成一张速查表方便排查时对照现象根因排查方式与修复前端长时间无输出反代缓冲Nginx等吞了流后端加X-Accel-Buffering: no或反代层关闭 buffering 与压缩输出不完整结尾被截断框架或代理层有 body 大小限制上调proxy_buffers/client_max_body_size确认是传输层截断还是模型结束出现乱序文本倒序文本前端在渲染层做了直接 append用 reducer 快照更新禁止在渲染函数里改 ref停止按钮点了还在出字Cancel 信号未传到模型层检查 AbortController 是否真正挂在流式请求上服务端是否还持有旧会话页面内存缓慢上涨工具参数 delta 高频重渲染用节流/防抖控制高频刷新及时清理结束状态的 ToolCalls 记录偶发性连接中断自动重连没有配心跳帧客户端对重连自动做幂等恢复服务端每 15 秒发心跳JSON 参数解析失败半截 JSON 被提前封装成 chunk服务端解析器做括号配对边界检测参数闭合后才 emit首字延迟过高flush 窗口太大把缓冲区 flush 时间调低或改成“有新字节立即发、兼做窗口削峰”双模式5.2 三个排查工具与技巧流式管道排查有个特点报错往往没有堆栈只有现象描述——“用户看到半行字停了”。这种问题靠加日志不好定位更需要主动的观测手段技巧一给每个 Chunk 打时间戳算端到端延迟分布。服务端生成 Chunk 到前端 render 之间的耗时拆成三段分别统计模型产出耗时、网络传输耗时、前端渲染耗时。哪段异常晚就查哪段。前端可以直接用performance.now()减去 chunk.timestamp 得到近似传输耗时。技巧二录音回放式排查。在服务端把原始 SSE 帧按原样存一份扇出一份到日志文件出问题时用同一份日志在本地重新跑前端看哪一步渲染出错。这个技巧能大幅缩短“模拟复现”的时间因为线上流式时序很难模拟。技巧三双端日志 ID 对齐。前端每次渲染异常把对应 chunk 的 chunkId 记下来后端日志里按 chunkId 搜同一时刻的前 20 条 chunk确认是否是解析层丢块。没有 chunkId 这套排查根本做不了这也是为什么我强调 Chunk 必须有全局唯一 ID。5.3 生产环境做好降级预案流式管道和普通接口最大的不同是链路长、中间态多、故障表现往往不是报错而是“卡住”。所以在生产环境我强烈建议做两个降级开关降级为完整返回。如果服务端检测到模型流式接口报错超时自动降级为等整个结果出来一次性返回。虽然体验差点但至少用户能拿到结果。降级为同步刷新。前端如果检测到 SSE 长时间无帧且无心跳自动退出流式模式改轮询一个“当前完整回答”的只读接口保证用户不会永远卡在半个字上。降级逻辑平时不会触发但真有并发尖峰或者模型不稳定时这套预案能挡住网上绝大部分“服务挂了”的负面反馈。6. 个人踩坑记录与扩展想法6.1 我实际踩过的三个流式坑第一个坑把 SSE 的 heartbeat 发成了data: ping。这会直接污染前端的 EventSource 事件流前端会把 “ping” 当作一条文本 delta 拼到回答里。正确做法是发一个不带 data 的注释行: ping或者用自定义事件名隔离。第二个坑多个 Agent 并行时共用同一个 reducer。原本只支持单 Agent 的应用把管道从单路改成多路后我没有给每个 session 建独立的 reducer 切片导致两个 Agent 的文本互相穿插。后面改成按 sessionId 建 map 结构才干净。第三个坑前端做了超时自动断连超时阈值设置太短。原本设了 60 秒但一次慢速工具调用跑了 40 多秒加上生成时间就超过 60 秒前端自动断开重连用户看到进度条莫名回退。强烈建议超时检测只看“心跳是否存活”不要看“是否有数据”因为工具调用期间模型可能确实长时间不输出。6.2 后续可以继续扩展的方向流式管道一旦稳定运行我计划在三个方向扩展多模态输出管道现在的 StreamChunk 只覆盖文本和工具参数后续要支持图片生成、音频片段时可以在 type 上扩展media_deltapayload 里带 base64 分片和 MIME 类型。浅层状态可视化把 token 级别的消耗和阶段耗时做成微型图表撑进 UI 侧边栏。回放与评测驱动把线上真实流式数据保存后离线回放给新的 Agent 版本做对比评测看流式管道改动是否引入时序问题。流式输出管道做扎实了Agent 系统才谈得上“可感知、可打断、可观测”。上面这套从 StreamChunk 设计到 UI 落地的链路是我在 DeepSeek-Harness 项目里反复打磨后的通用做法配置细节可能因为版本而异但协议分层、背压控制、前端不可变更新这几个核心思路是可以直接照搬的。做的时候宁可多花点功夫把 Chunk 结构和状态机理清楚也不要等上线后用户来帮你测卡顿。