★ FEATURED ARTICLE 1. 流式输出为什么是 Agent 体验的分水岭做过 Agent 项目的人大概都有这个体会模型能力再强如果用户发完消息后盯着空白屏幕等上十几秒才一次性蹦出全部回答产品的“智能感”会瞬间打对折。流式输出解决的正是这个心理落差问题——它让 token 一个接一个地出现在界面上用户能实时看到模型在“思考”和“组织语言”哪怕总耗时没变主观等待感也会大幅下降。在 DeepSeek-Harness 这套 Agent 框架里流式输出管道是一条贯穿始终的数据链路从模型侧产生StreamChunk经过中间层的解析、缓冲、事件分发最终抵达 UI 层完成渲染。这条链路上任何一环处理不当都会出现大家熟悉的那几种症状——文字一顿一顿地跳、SSE 连接莫名断开、工具调用结果和文本混在一起乱序显示、或者干脆卡在“正在输入”状态再也不动。这篇内容面向的是正在做 Agent 应用开发、需要自己封装流式接口的工程师。我会把 DeepSeek-Harness 里从StreamChunk到 UI 的完整管道拆开讲清楚包括数据结构设计、SSE 封装逻辑、事件分发机制、UI 渲染策略以及我在实际调试中踩过的坑。读完你应该能自己搭一条稳定可用的流式管道而不是只会调一个现成的 SDK。2. StreamChunk 的数据结构设计与拆解2.1 为什么不能直接用原始字符串流很多人第一次做流式输出图省事直接把模型返回的文本片段往前端推前端收到什么就追加什么。这种做法在纯聊天场景下勉强能用但放到 Agent 场景里很快就会崩。原因在于 Agent 的输出不是单一文本流它同时包含好几类信息普通回复文本、工具调用的参数增量、工具执行结果、思考过程、结束标记、错误信息。这些内容如果混在一个字符串流里前端根本没法区分该把哪段渲染成正文、哪段渲染成工具卡片。StreamChunk的核心价值就是给流式数据加上类型标签和结构化字段。它不是一个简单的字符串包装而是一个带有明确语义的事件对象。在 DeepSeek-Harness 的设计里每个 chunk 至少需要携带这几个维度的信息事件类型、内容载荷、序号、以及可选的元数据。2.2 StreamChunk 的字段设计我实际用下来比较顺手的一套字段结构是这样的from dataclasses import dataclass, field from typing import Optional, Literal, Dict, Any dataclass class StreamChunk: event_type: Literal[ text_delta, # 正文文本增量 tool_call_start, # 工具调用开始 tool_call_delta, # 工具参数增量 tool_result, # 工具执行结果 thinking_delta, # 思考过程增量 usage, # token 用量 error, # 错误事件 done # 流结束 ] content: str index: int 0 tool_call_id: Optional[str] None tool_name: Optional[str] None metadata: Dict[str, Any] field(default_factorydict)这里有几个设计决策值得展开说。event_type用字面量类型而不是枚举是为了序列化方便JSON 传输时直接就是字符串前端判断也直观。index字段是全局递增序号它的作用后面讲断线重连时会重点说。tool_call_id和tool_name只在工具相关事件里出现用 Optional 处理。content字段对文本类事件是文本片段对工具参数增量是 JSON 字符串的片段。这里有个容易踩的坑工具调用的参数是分多次增量传过来的你不能每收到一个 delta 就尝试json.loads必须等参数完整后再解析。我见过有同学在tool_call_delta阶段就急着解析 JSON结果每次都报解析错误还以为是模型输出有问题。2.3 序号与时间戳的隐藏作用index这个字段看起来不起眼但它是保证流式数据不乱序的关键。SSE 在传输层理论上是有序的但一旦中间经过了消息队列、多路复用或者前端做了异步处理顺序就可能被打乱。有了全局递增的 index前端可以做两件事一是检测丢包发现 index 跳号就知道中间丢了 chunk二是做重排序缓冲。我在 metadata 里通常还会塞一个ts时间戳单位毫秒。这个时间戳不是为了显示而是用来做性能分析的。把每个 chunk 的到达时间打出来你就能算出首 token 延迟、平均 token 间隔、以及有没有出现异常的停顿。调试流式卡顿时这张时间线图比任何日志都直观。注意content字段在工具参数增量场景下是 JSON 片段不要在这个阶段做任何解析或校验老老实实做字符串拼接等tool_call_start对应的完整参数收齐后再统一解析。3. SSE 封装把 StreamChunk 变成可传输的事件流3.1 SSE 协议格式回顾SSEServer-Sent Events本质上是基于 HTTP 长连接的单向文本推送协议。它的报文格式非常朴素每条消息由若干字段行组成字段行以字段名: 值的形式书写消息之间用空行分隔。最常用的字段是event事件名、data数据载荷、id消息 ID、retry重连间隔。一个典型的 SSE 消息长这样event: text_delta id: 42 data: {content:你好,index:42}注意每条消息末尾必须有一个空行这是分隔符。很多自己手写 SSE 服务端的同学会漏掉这个空行导致客户端一直等不到消息结束表现为“数据发了但前端不显示”。这个坑我在早期项目里踩过排查了大半天才发现是少了个\n\n。3.2 服务端封装逻辑把StreamChunk转成 SSE 消息核心就是一个序列化函数。但这里有几个细节决定了管道的健壮性。import json def chunk_to_sse(chunk: StreamChunk) - str: payload { content: chunk.content, index: chunk.index, tool_call_id: chunk.tool_call_id, tool_name: chunk.tool_name, metadata: chunk.metadata, } data json.dumps(payload, ensure_asciiFalse) lines [ fevent: {chunk.event_type}, fid: {chunk.index}, fdata: {data}, , ] return \n.join(lines)ensure_asciiFalse是必须的否则中文会被转义成\uXXXX虽然功能上没问题但传输体积会膨胀而且调试时看日志很难受。id字段直接用 chunk 的 index这样前端在断线重连时可以通过Last-Event-ID请求头告诉服务端“我收到哪了”服务端就能从断点继续推。3.3 心跳与超时控制SSE 连接最怕的就是中间有代理或网关它们对空闲连接有超时限制。如果模型思考时间较长中间几十秒没有数据推送连接就可能被掐断前端收到的是stream disconnected before completion: idle timeout waiting for sse这类错误。解决办法是定期发送心跳。心跳可以是一条注释消息以冒号开头的行也可以是自定义的ping事件async def heartbeat(interval: float 15.0): while True: await asyncio.sleep(interval) yield : keep-alive\n\n心跳间隔的选择有讲究。太短会浪费带宽太长又起不到保活作用。我的经验值是 15 秒这个数字能覆盖绝大多数网关的 30 秒或 60 秒空闲超时。如果你的部署环境中间层特别多可以压到 10 秒。提示心跳消息用注释行冒号开头而不是自定义事件是因为注释行会被客户端自动忽略不会触发任何业务逻辑最干净。如果你需要前端感知心跳来做连接健康检测那就用自定义事件。3.4 响应头设置SSE 响应必须设置正确的 Content-Type否则浏览器可能不会按事件流处理headers { Content-Type: text/event-stream; charsetutf-8, Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, }X-Accel-Buffering: no这个头是给 Nginx 看的告诉它不要缓冲这个响应。如果不加Nginx 默认会缓冲响应体导致前端收到的数据是一批一批来的而不是实时的。这个坑非常隐蔽因为本地开发直连服务时一切正常一上生产经过 Nginx 就变成“伪流式”了。4. 事件分发与前端解析机制4.1 前端如何消费 SSE浏览器原生提供了EventSourceAPI 来消费 SSE但它有个硬伤只支持 GET 请求没法带请求体。而 Agent 对话通常需要把消息历史、工具配置等一大坨数据发给服务端用 GET 拼 query string 不现实。所以实际项目里更常见的是用fetch配合ReadableStream手动解析。async function consumeStream(url, body, onChunk) { const response await fetch(url, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify(body), }); 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 }); const messages buffer.split(\n\n); buffer messages.pop(); for (const msg of messages) { parseAndDispatch(msg, onChunk); } } }这里的关键点是buffer的处理。网络传输是字节流一个 SSE 消息可能被拆成多个 TCP 包到达也可能多个消息挤在一个包里。所以必须维护一个缓冲区按\n\n切分最后一段不完整的留在 buffer 里等下次。decoder.decode的{ stream: true }参数也很重要它保证多字节字符比如中文被拆包时不会解码成乱码。4.2 事件解析与类型分发拿到单条 SSE 消息后要把它解析回结构化的数据再根据事件类型分发到不同的处理函数function parseAndDispatch(raw, onChunk) { const lines raw.split(\n); let eventType message; let data ; for (const line of lines) { if (line.startsWith(event:)) { eventType line.slice(6).trim(); } else if (line.startsWith(data:)) { data line.slice(5).trim(); } } if (!data) return; const payload JSON.parse(data); onChunk({ eventType, ...payload }); }分发逻辑本身不复杂但事件类型和 UI 状态的映射关系需要仔细设计。文本增量追加到当前消息气泡工具调用开始要新建一个工具卡片工具结果要回填到对应卡片思考过程可能折叠显示错误事件要弹出提示。这套映射如果设计得乱UI 就会出现内容错位。4.3 工具调用的增量拼接工具调用是流式解析里最容易出错的部分。模型返回工具调用时参数是分多个 delta 传过来的而且不同工具的调用可能交错出现。你需要按tool_call_id维护一个待拼接的缓冲区const pendingToolCalls new Map(); function handleToolDelta(chunk) { const id chunk.tool_call_id; if (!pendingToolCalls.has(id)) { pendingToolCalls.set(id, { name: chunk.tool_name, args: }); } const call pendingToolCalls.get(id); call.args chunk.content; }等收到该工具调用的结束标记通常是tool_call_start之后参数收齐或者专门的结束事件再对args做一次JSON.parse然后触发实际执行。这里要特别注意不要假设工具调用是串行的模型可能一次返回多个工具调用它们的 delta 会交错到达必须用 Map 按 id 隔离。注意工具参数拼接完成后解析 JSON 时一定要做 try-catch。模型偶尔会生成格式不规范的 JSON直接 parse 会抛异常导致整个流处理中断。捕获异常后可以把原始字符串作为参数传给工具让工具自己容错或者返回一个友好的错误提示。5. UI 渲染策略与性能优化5.1 增量渲染 vs 批量渲染最直觉的做法是每收到一个 text_delta 就更新一次 DOM。但 token 的到达频率可能很高一秒几十个 chunk每次都触发 DOM 更新和重排页面会明显卡顿尤其是在消息很长的时候。我的做法是做一个渲染节流用一个缓冲区累积文本增量用requestAnimationFrame或者固定时间间隔比如 50ms批量刷新到 DOM。这样既保证了视觉上的流畅感又大幅降低了渲染压力。let renderBuffer ; let rafPending false; function scheduleRender(text) { renderBuffer text; if (rafPending) return; rafPending true; requestAnimationFrame(() { appendToMessage(renderBuffer); renderBuffer ; rafPending false; }); }50ms 这个间隔是视觉上的一个甜点值。人眼对 20ms 以内的更新基本感知不到差异而 50ms 约等于 20fps已经足够流畅同时把渲染次数压到了原来的几分之一。5.2 自动滚动与用户意图的冲突流式输出时消息不断变长通常需要自动滚动到底部让用户看到最新内容。但如果用户主动往上滚去看历史消息你还强制滚动到底部体验就很糟糕。解决办法是维护一个“是否贴底”的状态监听滚动事件当用户滚动位置距离底部小于某个阈值比如 50px时标记为贴底状态此时自动滚动一旦用户往上滚超过阈值取消贴底状态停止自动滚动并显示一个“回到底部”的按钮。function isNearBottom(el, threshold 50) { return el.scrollHeight - el.scrollTop - el.clientHeight threshold; }这个细节看起来小但它是区分“能用”和“好用”的关键。很多开源聊天界面就栽在这一点上用户想回看内容时被不断拽到底部非常烦躁。5.3 长消息的虚拟化当对话轮次很多、单条消息很长时DOM 节点数量会爆炸滚动会变得卡顿。这时候需要考虑虚拟列表只渲染视口内的消息。不过虚拟列表和流式渲染结合时有个矛盾正在流式输出的那条消息高度在不断变化虚拟列表的高度估算会失准。我的处理方式是正在流式输出的消息不参与虚拟化始终真实渲染已经完成的历史消息才纳入虚拟列表。这样既保证了流式体验又控制了 DOM 规模。等流结束后再把这条消息“冻结”进虚拟列表。6. 常见故障排查与避坑实录6.1 流式卡顿与断连问题速查流式管道出问题时症状往往相似但根因各异。我整理了一张排查表按现象倒推可能的原因现象可能原因排查方向首 token 迟迟不来模型侧排队、prompt 过长看服务端日志的首 chunk 时间文字一顿一顿跳中间层缓冲、渲染未节流检查 Nginx 缓冲头、前端渲染频率连接中途断开网关空闲超时、心跳缺失抓包看断开时间点、加心跳中文乱码解码未处理多字节拆包检查 TextDecoder 的 stream 参数工具结果错位tool_call_id 未隔离检查增量拼接的 Map 逻辑内容重复显示断线重连未去重用 index 做去重判断6.2 断线重连与幂等处理SSE 断线后浏览器原生EventSource会自动重连并带上Last-Event-ID请求头。但如果你用的是 fetch 手动解析就得自己实现重连逻辑。重连时最关键的是幂等服务端要根据客户端上报的最后一个 index从下一个 chunk 开始推避免重复。我在服务端会维护一个短期的 chunk 缓存比如最近 1000 个重连请求带上last_index后先检查缓存里有没有有就直接补发缺失部分没有就让客户端重新发起完整请求。这个缓存不需要持久化进程内存里放个环形缓冲区就够了。6.3 并发场景下的资源控制Agent 服务通常要扛并发每个活跃的流式连接都占着一个 HTTP 长连接和一部分内存。如果并发量上来连接数可能成为瓶颈。几个实用的控制手段一是设置单连接的最长存活时间超时强制关闭防止僵尸连接堆积二是限制单用户的并发流数量避免一个用户开一堆标签页把资源占满三是在网关层做连接数限流超过阈值直接拒绝新连接而不是拖垮整个服务。提示流式接口的并发能力和普通接口不是一个量级压测时一定要用真实的流式客户端而不是用普通 HTTP 请求打。普通请求测出来的 QPS 对流式场景没有参考意义。6.4 我踩过的几个具体坑第一个坑是data字段里的换行。JSON 序列化后如果内容里本身有换行符虽然json.dumps会转义成\n但如果你手动拼字符串忘了转义SSE 解析时就会把一行数据拆成两行导致 JSON 解析失败。所以永远用json.dumps而不是手动拼接。第二个坑是代理层的缓冲。除了 Nginx有些云厂商的负载均衡也会缓冲响应。排查这类问题的通用方法是在服务端每发一个 chunk 就 flush 一次同时在客户端打印每个 chunk 的到达时间。如果时间间隔很规律地成批出现基本就是中间层在缓冲。第三个坑是前端状态更新和 React 渲染的批处理。在 React 里连续 setState 会被批处理合并如果你依赖每次 setState 都触发一次副作用可能会漏掉中间状态。流式场景下更推荐用 ref 存缓冲区用状态只控制最终渲染。7. 从 StreamChunk 到 UI 的完整链路回顾把整条链路串起来看一个 chunk 的旅程是这样的模型侧生成 token被包装成带类型和序号的StreamChunk服务端把 chunk 序列化成 SSE 消息加上心跳保活通过设置了正确响应头的 HTTP 长连接推出去前端用 fetch 读取字节流按分隔符切分消息解析出事件类型和数据再根据事件类型分发到不同的处理器文本增量进渲染缓冲区工具增量进拼接 Map最后经过节流批量刷新到 DOM配合贴底检测和虚拟化保证滚动流畅。这条链路上每一环都有它的存在理由也都有对应的坑。我个人的体会是流式输出最难的不是把数据推出去而是处理各种边界情况——断线、乱序、缓冲、并发、渲染性能。把这些边界都处理好了用户感受到的就是那种“丝滑”的实时对话体验而这恰恰是 Agent 产品最直观的品质体现。后续如果要做多模态的流式输出比如图片生成过程的逐步呈现、或者语音的边合成边播放这套StreamChunk加事件分发的架构可以直接扩展只需要增加新的事件类型和对应的 UI 处理器就行。管道本身的设计是通用的这也是当初把它抽象成独立一层而不是塞在业务逻辑里的原因。
阅读完成 · 觉得有帮助？
