做Agent应用的朋友估计都碰到过这个场景模型推理速度一般任务链又长——一轮规划、两次工具调用、再来一轮生成中间还夹着参数校验和上下文整理。如果还是“攒完整个响应再吐给前端”的同步方式用户看到的永远是一个转了三四十秒的loading圈体验基本等于劝退。我在DeepSeek-Harness这个Agent框架里正好完整趟了一遍这条链路今天把流式输出管道的设计和落地细节拆开聊底层推理引擎吐出的token流如何一步步变成Service层的StreamChunk最终驱动UI按增量渲染指标、实时展示工具调用过程。这篇文章不适合上来就要跑通demo的人读更适合那些正要给Agent系统做流式改造、或者已经在做但被各种边界问题折磨的开发者。我会把数据结构设计、SSE协议选择、服务端背压处理、前端渲染优化、常见坑点都过一遍同时给出可直接参考的代码示例。无论你用的是DeepSeek还是其他模型这套管道设计思路都能搬走。1. 整体设计与思路拆解Agent场景的流式管道和聊天场景根本不是一回事1.1 从“一问一答”到“任务执行”流式输出的需求升级了普通聊天场景的流式输出其实很简单模型生成什么前端就显示什么一行messages.forEach就能完事。但到了Agent场景就不一样了。一个Agent任务最少也要经历“LLM规划 → 工具调用 → 工具结果回填 → 再次推理”这好几个循环。每个环节都需要向用户实时反馈进展否则用户根本不知道系统是在干活还是卡死了。我见过不少团队上来就照搬聊天场景的流式方案结果前端只能显示最终答案中间的工具调用过程全是空白。用户看到光标转了两分钟最后突然冒出一段话根本没法判断中间发生了什么。所以DeepSeek-Harness在设计流式输出管道时一开始就没有把“流式”单纯理解成“文本打字机效果”而是把整条输出链路当成了一个带结构化事件的任务执行流水线。这里的关键差异在于聊天场景下流的是“内容”Agent场景下流的是“状态内容工具调用指令”的复合流。一个完整的事件流里既要有文本delta也要有工具调用的起止标记、参数片段、执行结果还有当前阶段的状态变更。这些事件用同一个数据模型封装才能让前端按统一的逻辑消费。1.2 四层管道架构Token流、协议层、StreamChunk、UI层我们在DeepSeek-Harness里把流式输出链路拆成了四层每一层职责单一边界清楚第一层是推理引擎的token流。这一层由模型服务提供通常是一段异步生成器按token粒度往外吐数据。要注意的是不同引擎的token流格式不统一有的带usage统计有的带工具调用的增量参数有的只是纯文本。第二层是传输协议层。我们最终选用SSEServer-Sent Events作为服务端到前端的主协议理由后面单独说。这一层负责把上层的结构化事件编码成标准SSE格式交给HTTP长连接。第三层是StreamChunk数据模型。这是整个管道的地基所有业务事件不管是文本、工具调用还是状态变更都统一成一种chunk类型再带上一批元字段。前端不需要关心底层协议细节只需要消费标准chunk。第四层是UI状态层。前端拿到chunk流之后经过合并、去重、渲染节流最终驱动界面增量更新。这四层拆开之后每层都能独立测试和替换。比如今天想从SSE换成WebSocket只需要重写第二层想换一种前端框架只需要动第四层。我在实际项目中把StreamChunk的数据结构稳定下来之后后续换模型服务、换前端框架都很顺基本没动过核心逻辑。1.3 为什么选择SSE而不是WebSocket这是个绕不开的问题。WebSocket能做双向通信Agent场景里确实有“前端主动取消任务”的诉求看起来WS更合适。但我在真实项目中权衡之后还是选了SSE为主通道原因有三点。第一SSE基于HTTP天然兼容现有的负载均衡、网关、日志方案。我们团队所有基础设施都是围绕HTTP建的Nginx、网关鉴权、日志中间件全都现成。引入WebSocket意味着要额外管理连接状态、心跳、断线重连、横向扩展时的连接路由这些成本对于Agent系统的输出管道来说太重了。第二Agent系统的输出方向本质上是单向的。虽然需要支持“取消任务”但这个控制指令完全可以用一个独立的POST接口发到服务端不需要为此单独建立一条双向长连接。反而单一方向的数据流让协议层的实现简化了很多。第三SSE自带断线重连和last-event-id机制这在弱网环境里对前端开发极其友好。我之前调研过如果自己用WebSocket实现重连、续传、心跳这些要写一堆代码。后来确认了SSE在浏览器端的兼容性足够好就定了。当然SSE也有短板比如最大并发连接数限制HTTP/1.1下浏览器对同一域名限制6个连接。在Agent后台系统内部这不是问题即使将来做面向大量用户的公开服务也可以通过HTTP/2或域名分片解决。这里可以给一个结论自用和内部工具场景无脑选SSE如果是面向几十万C端用户的高频实时应用再认真评估WebSocket方案。2. StreamChunk的数据模型设计先把“流”拆成标准积木2.1 事件类型设计从text_delta到tool_callStreamChunk是整个管道里最重要的契约它的设计直接决定了前后端的开发体验。我在DeepSeek-Harness里把它定义成一个带type字段的联合类型目前固定的事件类型有七种事件类型含义关键字段session_start一次任务流开始session_id, agent_config, timestampstatusAgent阶段状态变更status(planning/tool_executing/thinking/final), message, detailtext_delta文本增量片段delta, seq, reasoningtool_call_start工具调用开始tool_name, call_id, arguments(初始参数)tool_call_args_delta工具参数的增量片段call_id, args_deltatool_call_end工具调用结束call_id, result_summary, duration_mssession_end任务流正常结束usage, total_duration_ms, error?你可能会问为什么status、text_delta、tool_call相关的事件都要塞进同一个chunk结构直接前端搞几个独立事件不是更清晰吗实际操作下来统一模型有几个明显好处前端解析逻辑只用写一套中间层做日志采集、监控、限流也只需要处理一种格式。如果每个事件类型都走独立协议那客户端代码光switch分支就要写一大堆后续加一个新事件类型还得改解析器。2.2 必填字段与顺序保障在设计StreamChunk时有几个字段我是踩过坑后才坚持留下的。第一个是session_id。这个字段在哪一层都要传服务端用它绑定日志客户端用它区分多轮任务。如果漏了这个字段前端同时开两个会话时就会串流。第二个是seq序列号。这可能是整个管道里最容易被忽略但又最重要的字段。实际场景中Agent可能同时发起多个工具调用多个工具的执行结果会异步回来导致chunk并不是严格按照发送顺序到达前端。如果没有seq做排序基准前端拿到的工具调用结果可能是乱的。我在早期的版本里就没加seq结果工具较多时文本和工具状态经常对不上排查起来极其痛苦。第三个是timestamp。日志追踪和性能观测都需要它。特别是定位“首字延迟”和“两个chunk之间的间隔”时没有时间戳就只能靠猜。StreamChunk的TypeScript类型定义大致长这样type StreamChunk | { type: session_start; session_id: string; seq: number; timestamp: number } | { type: status; session_id: string; seq: number; timestamp: number; status: AgentStatus; message: string } | { type: text_delta; session_id: string; seq: number; timestamp: number; delta: string; reasoning?: string } | { type: tool_call_start; session_id: string; seq: number; timestamp: number; call_id: string; tool_name: string; arguments: Recordstring, unknown } | { type: tool_call_args_delta; session_id: string; seq: number; timestamp: number; call_id: string; args_delta: string } | { type: tool_call_end; session_id: string; seq: number; timestamp: number; call_id: string; result_summary: string; duration_ms: number } | { type: session_end; session_id: string; seq: number; timestamp: number; usage: TokenUsage; error?: string };实际传输时每个chunk整体JSON序列化后作为SSE的一行data字段发出。前端解析后又还原成这个联合类型整个开发体验很顺滑。2.3 分段策略按Token、按句子还是按JSON边界这一节是想聊服务端在生成chunk时究竟该多频繁地往外吐数据。如果每次模型输出一个token就发出去网络开销会爆炸如果攒到整个响应结束再发那就不叫流式了。我在项目里总结出的策略是“按语义边界分段加兜底长度限制”。文本内容按句子或标点切分例如遇到句号、感叹号、分号就攒够一个chunk发出去。这样前端能获得相对完整的语义单元打字机效果也足够自然。如果模型连续输出了很长一段没有标点的内容就是到了30~50个token的兜底长度也要强制发一次避免前端长时间等不到更新。工具调用参数则按JSON片段边界切分。模型在生成工具参数时会逐步吐出JSON如果每次只发几个字符的args_delta前端做实时展示时还要自行拼接JSON麻烦不说还容易出问题。实际操作中我一般让推理引擎攒够一个完整的JSON key-value片段后再作为args_delta发出。这样前端可以直接展示“当前已解析出的参数”不需要自己维护半截JSON的解析状态。对了有一点必须提醒不要试图在后端提前解析工具参数。模型吐出的参数流可能是半截的硬要解析JSON很容易报错。我们在管道里只负责“搬运”不负责“解析”参数解析的工作留给工具执行器去干职责分开后系统稳定多了。3. 服务端实现从推理引擎到SSE服务3.1 异步生成器与有界队列服务端的核心是一个异步生成器。DeepSeek-Harness的架构里推理引擎单独跑在Worker进程中API服务通过队列接收Worker产出的事件。这里最关键的工程点是队列必须带大小上限即所谓的有界队列bounded queue。用一个生活化的类比如果推理引擎是水龙头API服务是水杯那你必须在水杯和水龙头之间放一个有限容量的水桶。如果水桶无限大当客户端消费变慢时服务端内存就会被不断堆积的事件撑爆。我们在上线前压测时遇到过这个问题并发20个任务时内存直接飙升到几个GB后来给队列加上上限内存立刻稳定下来。Python服务端用asyncio实现大概长这样import asyncio from collections import deque from typing import AsyncGenerator, Deque, Optional class StreamQueue: def __init__(self, maxsize: int 200): self._queue: Deque[dict] deque() self._maxsize maxsize self._waiters: Deque[asyncio.Future] deque() async def put(self, chunk: dict) - None: while len(self._queue) self._maxsize: # 队列满了就阻塞形成背压 waiter asyncio.get_event_loop().create_future() self._waiters.append(waiter) await waiter self._queue.append(chunk) def get_nowait(self) - Optional[dict]: if not self._queue: return None chunk self._queue.popleft() if self._waiters: waiter self._waiters.popleft() waiter.set_result(None) return chunk async def stream(self) - AsyncGenerator[dict, None]: while True: chunk self.get_nowait() if chunk is None: await asyncio.sleep(0.001) continue yield chunk这里队列满时put会阻塞推理引擎就会被拖慢整个链路自动形成背压。前端消费慢后端就慢不会出现内存无界增长。这是整个服务端实现里最重要的一件事。3.2 背压、取消与超时处理背压是核心但只有背压还不够。Agent任务的执行时间可能很长用户中途可能关闭页面或者点“停止”这时候服务端必须及时取消任务释放推理引擎的连接和计算资源。我们的做法是每个流式请求都绑定一个task_id前端取消时调用一个POST接口比如/agent/{task_id}/cancel服务端在收到取消请求后向推理Worker发送取消信号然后从StreamQueue中丢弃属于该task的所有尚未发送的chunk最后关闭SSE连接。超时处理也要分两个层面一是整体超时一个Agent任务如果跑了太久比如超过10分钟必须强制终止防止资源泄漏二是单次网络发送超时SSE连接在长时间无数据时会通过心跳维持如果多次心跳无响应服务端主动断开连接释放文件描述符。3.3 SSE编码细节事件、心跳与错误码SSE的协议很简单本质上就是一个HTTP长连接后端点开响应头的Content-Type: text/event-stream然后把事件按格式写出去。每条事件之间用空行分隔。完整格式长这样event: message data: {type:text_delta,delta:你好,seq:12,session_id:s_001,timestamp:1699999999} event: message data: {type:status,status:tool_executing,message:正在调用搜索工具,seq:13,session_id:s_001,timestamp:1699999999}有一个容易被忽略的细节是“心跳”。SSE规范里规定可以用注释行作为keepalive信号因为如果长连接超过一定时间没有任何数据代理服务器可能会掐断连接。我们的做法是如果队列里连续3秒没有chunk产出就发一行注释data: ping作为心跳确保连接长时间空闲时也不会被中间设备断开。错误处理也要走协议。不要在HTTP层面直接返回错误码否则会破坏已建立的SSE连接。标准的做法是在SSE连接内部发一个error类型事件让前端能捕获并展示。我们约定三种错误类型错误码含义前端处理retryable_error临时性错误例如推理引擎超时重试提示稍后重试不关闭会话fatal_error不可恢复错误例如上下文超限关闭会话展示错误信息cancelled用户主动取消静默关闭不提示错误4. 前端消费与UI渲染打字机、工具状态与增量渲染4.1 用fetch ReadableStream解析SSE而不是EventSource前端实现细节方面我发现很多人第一反应是用EventSource。但EventSource有两点限制一是它只能做GET请求无法携带POST body、自定义headers比如鉴权token第二个是不能自定义取消逻辑。所以在DeepSeek-Harness的前端里我最终选了fetch ReadableStream手动解析SSE。核心思路是把响应体当成一个文本流按SSE协议的空行边界切开然后逐条解析出data字段。这段代码是后续所有渲染逻辑的基础async function streamAgentOutput(url: string, body: object, onChunk: (chunk: StreamChunk) void) { const response await fetch(url, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify(body), }); if (!response.ok || !response.body) { throw new Error(HTTP ${response.status}); } 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事件以空行分隔 const events buffer.split(\n\n); buffer events.pop() ?? ; for (const rawEvent of events) { const dataLines rawEvent .split(\n) .filter((line) line.startsWith(data: )) .map((line) line.slice(6)); try { const chunk JSON.parse(dataLines.join(\n)); if (chunk.type) onChunk(chunk as StreamChunk); } catch { // 忽略半截JSON等下一个事件 } } } }这里有一个经验自己写SSE解析器时一定要把buffer的剩余部分留到下一轮再拼接否则事件恰好被TCP分包切开时即所谓的“粘包/断包”现象数据就会丢。上面代码里events.pop()那行就是这个作用。4.2 React状态更新优化别把每个chunk都setState前端最容易犯的第二个性能错误是每收到一个chunk就调用一次React的setState。LLM流式输出时chunk频率可以高达每秒钟几十个如果每个都触发React组件的重新渲染页面会在打字机效果还没出现时先卡死。我们的做法是引入一个简单的渲染调度层。chunk到达后先放进一个内存缓存数组然后用requestAnimationFrame把当前缓存“合并快照”写入React state。这样的效果是React的渲染频率被限制在每帧最多一次约60fps每秒的setState次数从几十次降到最多六十次实际渲染开销大幅下降。简化后的代码如下const pendingChunks: StreamChunk[] []; let scheduled false; function enqueueChunk(chunk: StreamChunk) { pendingChunks.push(chunk); if (scheduled) return; scheduled true; requestAnimationFrame(() { scheduled false; const snapshot pendingChunks.splice(0, pendingChunks.length); appendChunksToState(snapshot); }); }这里本质上是一个“合并写”的思路。把模型的输出看作连续的数据流而不是离散的状态快照UI只需要按帧读取最新状态即可。实测下来这一层调度代码加上后聊天页面的滚动和输入框操作再也没有卡顿。4.3 工具调用状态如何驱动UIAgent场景里工具调用期间的UI是一个高频需求。模型正在调用搜索工具时界面上应该出现“正在搜索”的提示框搜索完成后应该展示结果摘要如果调用的是一个耗时较长的工具甚至还要展示进度条。这块逻辑完全由tool_call_*事件驱动。tool_call_start事件到来时UI创建一张工具卡片显示工具名和初始参数tool_call_args_delta事件到来时更新卡片上的参数展示区tool_call_end事件到来时替换为结果摘要和执行耗时。工具卡片与文本输出分别渲染不要混在一起。这样做的好处是视觉上“对话流”和“执行过程”天然分区用户既能快速定位最终答案也能回看中间过程。我在实际验收测试中发现这个工具卡片的展示对用户信任感提升非常明显——至少每个运行中/失败步骤都有明确的视觉证据用户不会再觉得系统是“突然就出结果了”。4.4 流式Markdown与代码块的展示策略最后一个前端难题是流式Markdown的渲染。如果每收到一个text_delta就重新解析一遍完整的Markdown遇到半截代码块时高亮和渲染都会闪烁。这个问题我在早期版本里被折磨过很久。后来总结出一套方案对最终的纯文本展示区维护一个不断累积的markdown缓冲渲染时每次只在前端对“已经完整结束的行”做markdown解析未完成的部分以纯文本显示。具体做法是先把markdown源文本按行拆分只对已满足条件比如代码块中出现了闭合的三个反引号的段落做解析渲染其他行暂时用等宽字体原样展示。这样即使模型当前正在输出一段代码用户看到的也是“逐渐成长”的代码块而不是一个不停闪烁乱跳的渲染结果。一个更简单的兜底方案是所有流式渲染都走“最终一致”策略渲染结果允许短暂延迟但必须稳定。数据先可读展示纯文本然后等完整段落后再升级为markdown富文本。这个策略在我的实际项目中非常管用。5. 常见问题与排查技巧实录5.1 半截JSON、半截Markdown一切“半截问题”的根源做流式管道必然要面对半截数据。模型在吐数据时不会考虑你前端好不好处理它只会按自己的节奏输出。半截JSON会导致工具参数解析失败半截Markdown会导致渲染闪烁半截SSE事件会导致前端误判流结束。解决半截问题的总原则是不在错误的时机做多余的解析。工具参数的完整解析推迟到tool_call_end之后再做Markdown富文本渲染推迟到段落完整后再做SSE事件必须等遇到空行分隔符才算完整。所有解析器只能处理“完整数据”半截数据一律缓存等待补充。5.2 SSE连接频繁断开、服务端资源泄漏我在上线初期遇到过一个典型的线上问题用户同时开启多个Agent任务后服务端的文件描述符数量持续上涨最后触发too many open files。排查了很久才发现是前端在页面切换时关闭了SSE连接但服务端的StreamQueue还有数据在产出连接关闭事件没有及时清理队列导致任务资源一直悬空。解决方案是加“连接关闭兜底清理”SSE连接的close事件触发时立刻取消与该会话绑定的Agent任务并清空对应队列。不要等Agent任务自然结束因为Agent任务的时长可能长达几分钟。这个清理钩子一定要写在服务端不能依赖前端的主动通知。5.3 多路工具调用并发时内容乱序Agent框架中多个工具并行调用时前端偶尔拿到乱序的chunk。文本流和工具状态交错在一起工具卡片可能比文本晚出现也可能早出现。这个问题在加了seq序列号之后基本能解决但还有更细的一层前端要做轻量的“按会话分组按序号排序”的缓冲队列。有三个字段缺一不可session_id用于分组seq用于排序timestamp用于兜底判断。前端攒够了连续序号的chunk后就批量渲染遇到跳号的情况等待后续数据补齐超过一定等待时间比如3秒则记录告警并强制渲染已到达内容避免整条输出被一个缺失chunk卡死。5.4 高并发场景下服务端背压失效很多团队问“AI Agent怎么扛并发”。坦白讲Agent场景扛并发远比普通HTTP API复杂因为每个请求都会占用一个长连接和一个推理Worker。就算你API网关层并发做得再好推理引擎的并发上限就摆在那里。我们最终的方案是三层并发控制API网关层做常规限流Queue层设置了最多同时执行的Agent任务数信号量控制推理引擎层单独控制并发度。如果Queue积压超过阈值直接拒绝新任务不要无限排队。无限排队在Agent场景里等于慢性自杀因为用户等不起任务状态也没法清晰反馈。说句实话这个方案并没有惊艳的优化就是把该限制的地方都限制住系统才稳定下来。5.5 首字延迟过高流式变“假流式”最后聊一个容易被忽视的指标首字延迟Time To First TokenTTFT。流式改造后如果从请求发出到第一个chunk到达的TTFT时间还是好几秒那用户感知跟没做流式根本没区别。我们在项目里对TTFT做了强监控任何任务从发起到产出第一个chunk的耗时超过1500ms都要告警。优化TTFT的思路核心是不要让Agent的规划阶段阻塞在流式管道前。让状态事件session_start、status在规划一开始就立即发给前端而不要等到第一个文本token就绪。前端收到status事件后立刻渲染“正在思考/正在规划”的界面给用户一种“系统已经在动了”的感觉。这是加“假装在干活”吗不是规划阶段本来就在工作只是没有文本输出而已把这段过程可视化对体验是实打实的提升。写在最后的一点体会给DeepSeek-Harness做了这轮流式输出管道改造之后我的感受是流式输出的难点从来不在“流”本身而在于你把哪些信息放进了流里。只流文本你得到的是普通聊天框流了结构化事件你得到的是一个可观测、可中断、可追踪的Agent执行面板。StreamChunk这个模型我建议从项目一开始就固定下来它决定了前后端并行开发的效率也决定了后期功能扩展的难度。我早期吃过没带seq字段、没做背压控制、图省事直接用EventSource的亏哪一个都让人在深夜加班。希望这篇文章能帮你把这些坑提前绕开。