1. 流式输出为什么是 Agent 体验的分水岭做 Agent 应用的人迟早会撞上同一堵墙模型明明已经在吐字了前端却要等整段生成结束才一次性渲染用户盯着转圈图标干等十几秒体感直接崩掉。这个问题的根子不在模型而在流式输出管道——从模型侧产出的StreamChunk到最终落到 UI 上的一个个字符中间那条链路没打通。我最早做 DeepSeek 接入的时候也踩过这个坑。当时图省事直接调非流式接口等response完整返回再setState。本地测试没问题一上真实网络就露馅一个稍长的回答要等 8 到 15 秒用户以为卡死了反复点发送结果并发请求堆了一堆。后来改成流式首字延迟压到 1 秒以内同样的模型、同样的网络体验完全是两个产品。这篇要聊的就是 DeepSeek-Harness 里那条流式输出管道。核心链路可以概括成一句话模型侧 SSE 吐出StreamChunkHarness 层做解析、聚合、状态管理最后通过事件或回调推给 UI 逐帧渲染。听起来简单但每一层都有坑chunk 边界怎么切、reasoning_content和content怎么分流、工具调用参数是分片到达的怎么拼、UI 侧怎么避免每个 token 都触发一次重渲染。适合谁看如果你正在做 Agent 开发、接过 DeepSeek API、或者自己写 LLM 框架的流式层这篇能帮你少走弯路。如果你只是好奇流式输出到底难在哪我也会用生活化的类比把原理讲清楚。下面按管道的实际数据流向一层一层拆。2. StreamChunk 的真实结构别被文档的简化示例骗了2.1 一个 chunk 里到底装了什么很多人对StreamChunk的认知停留在一小段文本实际远不止。DeepSeek 的流式响应遵循 OpenAI 兼容格式每个 SSEdata:行反序列化后大致长这样{ id: chatcmpl-xxx, object: chat.completion.chunk, created: 1710000000, model: deepseek-chat, choices: [ { index: 0, delta: { role: assistant, content: 流, reasoning_content: null }, finish_reason: null } ], usage: null }关键在delta字段。它不是完整消息而是增量。第一个 chunk 通常只带rolecontent为空中间 chunk 带content的一两个字最后一个 chunk 的finish_reason变成stop或tool_callscontent为空。如果你把每个 chunk 的content直接当完整句子处理就会得到一堆碎片。这里有个反直觉的点chunk 的切分粒度不由你控制。模型侧按 token 生成但网络传输和 SSE 缓冲会重新分片。你可能收到一个 chunk 里含三个字也可能一个字被拆到两个 chunk。所以解析层必须做字符串拼接而不是假设一个 chunk 一个语义单元。2.2 reasoning_content 与 content 的双通道DeepSeek 的推理模型如deepseek-reasoner会额外输出reasoning_content也就是思维链。它和最终content是两条独立的流在同一个 delta 里交替出现。我见过有人把两者拼到一个字符串里结果 UI 上思维链和答案混在一起用户一脸懵。正确的做法是在 Harness 层维护两个缓冲区class StreamAccumulator: def __init__(self): self.content [] self.reasoning [] self.tool_calls {} def feed(self, chunk): delta chunk[choices][0][delta] if delta.get(reasoning_content): self.reasoning.append(delta[reasoning_content]) if delta.get(content): self.content.append(delta[content]) if delta.get(tool_calls): self._merge_tool_calls(delta[tool_calls])这样 UI 侧可以决定思维链折叠展示、答案实时渲染两者互不干扰。实测下来把思维链单独放一个可折叠区域用户对模型在思考的感知会强很多等待焦虑明显下降。2.3 finish_reason 的三种收尾信号流式结束不是没有更多 chunk 了这么简单。finish_reason有三个常见值处理逻辑完全不同finish_reason含义后续动作stop正常生成完毕关闭流落库完整消息length达到 max_tokens 截断提示用户可继续或自动续写tool_calls模型要调工具暂停渲染进入工具执行分支我踩过的坑是只判断stop遇到tool_calls时流已经关了但工具没执行Agent 直接卡死。后来在 Harness 里加了状态机收到tool_calls就切到PENDING_TOOL状态等工具结果回填后再发起下一轮流式请求。这个状态流转是 Agent 循环的核心后面第 4 节会展开。3. 从 SSE 字节流到结构化 chunk 的解析链路3.1 SSE 协议本身的坑半包与粘包SSEServer-Sent Events走的是 HTTP 长连接服务端按data: {...}\n\n的格式推送。但 TCP 是字节流不保证消息边界。你read()一次拿到的可能是半个 JSONdata: {choices:[{delta:{con一个半事件data: {...}\n\ndata: {...}\n多个事件粘在一起直接json.loads必炸。正确做法是维护一个行缓冲按\n切分遇到空行才认为一个事件结束class SSEParser: def __init__(self): self.buffer def feed(self, raw: bytes): self.buffer raw.decode(utf-8) events [] while \n\n in self.buffer: raw_event, self.buffer self.buffer.split(\n\n, 1) for line in raw_event.split(\n): if line.startswith(data: ): payload line[6:] if payload.strip() [DONE]: events.append({done: True}) else: events.append(json.loads(payload)) return events注意[DONE]这个哨兵值它不是 JSON不能直接 parse。我第一次写的时候忘了处理程序在流结束时抛异常排查了半天。3.2 用 httpx 还是 requests流式场景的选择requests的iter_lines()能用但它是同步阻塞的在 Agent 这种需要同时处理多路流的场景下很别扭。我现在的默认选择是httpx因为它原生支持 async配合async for处理流非常自然import httpx async def stream_chat(prompt: str): async with httpx.AsyncClient(timeoutNone) as client: async with client.stream( POST, https://api.deepseek.com/chat/completions, headers{Authorization: fBearer {API_KEY}}, json{ model: deepseek-chat, messages: [{role: user, content: prompt}], stream: True, }, ) as resp: parser SSEParser() async for raw in resp.aiter_bytes(): for event in parser.feed(raw): if event.get(done): return yield eventtimeoutNone很关键。流式请求可能持续几十秒默认超时会在生成中途掐断连接。但也不能完全不管我一般设一个read timeout比如 60 秒无数据才断避免连接假死。3.3 解析层的性能别在热路径上做重活每个 chunk 都要过一遍解析这段代码在热路径上性能敏感。几个实测有效的优化避免重复 decodeaiter_bytes拿到的是 bytes只在拼接时 decode 一次不要每个 chunk 都 encode/decode 来回转。JSON 解析用 orjson比标准库json快 2 到 3 倍chunk 量大时差距明显。缓冲区用 list 而非字符串拼接Python 里str 是 O(n²)用list.append最后.join才是 O(n)。这些优化单看都是小钱但流式场景下每秒可能几十上百个 chunk累积起来就是 UI 卡不卡的区别。4. Harness 层的状态管理流式与工具调用的交织4.1 工具调用参数是分片到达的这是流式管道里最容易翻车的地方。当模型决定调用工具时tool_calls不是一次性给全的而是这样分片// chunk 1 {delta: {tool_calls: [{index: 0, id: call_abc, function: {name: get_weather, arguments: }}]}} // chunk 2 {delta: {tool_calls: [{index: 0, function: {arguments: {\ci}}]}} // chunk 3 {delta: {tool_calls: [{index: 0, function: {arguments: ty\: \北}}]}} // chunk 4 {delta: {tool_calls: [{index: 0, function: {arguments: 京\}}}]}}arguments是一段一段拼起来的 JSON 字符串必须按index聚合等finish_reason tool_calls时才能json.loads。我见过有人每个 chunk 都尝试 parse arguments结果全是JSONDecodeError。聚合逻辑def _merge_tool_calls(self, delta_tool_calls): for tc in delta_tool_calls: idx tc[index] if idx not in self.tool_calls: self.tool_calls[idx] {id: , name: , arguments: } slot self.tool_calls[idx] if tc.get(id): slot[id] tc[id] fn tc.get(function, {}) if fn.get(name): slot[name] fn[name] if fn.get(arguments): slot[arguments] fn[arguments]4.2 Agent 循环的状态机设计流式输出不是孤立的它嵌在 Agent 的思考-行动-观察循环里。我用一个显式状态机来管IDLE - STREAMING - (finish_reason) ├─ stop - COMPLETED ├─ length - TRUNCATED - 可续写 └─ tool_calls - PENDING_TOOL - EXECUTING_TOOL - STREAMING (下一轮)关键点流式渲染和工具执行不能并行。工具参数没拼完就执行必然拿到残缺 JSON。所以收到tool_calls后UI 要先把已渲染的文本定格显示正在调用工具的中间态等工具结果回来再开新流。这个中间态设计很影响体验。我一开始直接静默等待用户以为卡了。后来加了个正在查询天气...的提示配合工具名动态生成等待感立刻不一样。4.3 多轮流式的上下文拼接Agent 跑多轮时每轮的流式结果都要追加到messages里作为下一轮的输入。这里有个细节assistant 消息要包含完整的tool_callstool 消息要带tool_call_id否则下一轮请求会被服务端拒绝。messages.append({ role: assistant, content: accumulator.content_text or None, tool_calls: [ { id: slot[id], type: function, function: {name: slot[name], arguments: slot[arguments]}, } for slot in accumulator.tool_calls.values() ], }) messages.append({ role: tool, tool_call_id: slot[id], content: json.dumps(tool_result), })漏掉tool_call_id是最常见的 400 错误来源报错信息还不直观得对着文档一行行核。5. 把 chunk 推到 UI渲染策略与性能取舍5.1 每个 token 都 setState 是性能杀手最直觉的做法是每收到一个 chunk 就更新一次 UI 状态。React 里就是每个 chunk 一次setState。实测下来长回答场景下每秒几十次重渲染主线程直接卡住输入框都打不出字。我的解法是批量刷新用一个 ref 累积文本通过requestAnimationFrame或固定间隔比如 50ms统一提交一次。const bufferRef useRef(); const rafRef useRef(null); function onChunk(text) { bufferRef.current text; if (rafRef.current) return; rafRef.current requestAnimationFrame(() { setDisplayText(bufferRef.current); rafRef.current null; }); }这样渲染频率被压到 60fps 上限视觉上依然是逐字蹦出但 CPU 占用降了一个数量级。50ms 这个间隔是体感和性能的平衡点再长会有明显顿挫感。5.2 打字机效果CSS 还是 JS有人喜欢用 CSS 动画做打字机但流式场景下不适用——文本长度是动态增长的CSS 动画没法跟着变。老老实实用 JS 控制文本切片配合光标闪烁的伪元素效果最稳。另一个细节是自动滚动。新内容不断追加容器要自动滚到底部。但用户手动往上翻看历史时不能强制拉回底部。判断逻辑const el containerRef.current; const isAtBottom el.scrollHeight - el.scrollTop - el.clientHeight 40; if (isAtBottom) { el.scrollTop el.scrollHeight; }40px 的容差是给几乎到底的情况留的余量实测比严格等于更符合直觉。5.3 Markdown 增量渲染的难题流式文本里经常含 Markdown比如代码块。问题是代码块的可能这个 chunk 才出现下一个 chunk 才有内容。如果每个 chunk 都整体重新 parse Markdown未闭合的代码块会渲染错乱。我的处理是延迟闭合检测到未闭合的代码块标记时先按纯文本渲染等闭合标记到达再切换成代码块样式。或者更简单粗暴——流式过程中只做轻量格式化换行、加粗流结束后再做一次完整 Markdown 渲染。后者实现成本低视觉上也就最后闪一下大多数场景能接受。6. 实测中的异常与排查清单6.1 流中断了怎么办网络抖动、服务端限流都会导致流中途断掉。表现是aiter_bytes抛异常或者长时间没有新 chunk。我的处理是带重试的续传记录已接收的content重连时把已有内容作为assistant前缀塞回去让模型接着写。虽然会重复消耗一点 token但比让用户重新提问体验好得多。判断卡住用超时设一个chunk_timeout比如 30 秒超过没新数据就主动断开重连。6.2 常见问题速查表现象可能原因排查方向首字迟迟不来未开 stream / 网络慢检查stream: true测首包延迟文本重复重试时未去重前缀对比已接收内容与重连返回JSON 解析报错半包未缓冲检查 SSEParser 行缓冲逻辑工具调用失败arguments 未拼完确认在 finish_reason 后才 parseUI 卡顿每 chunk 一次 setState改批量刷新400 错误缺 tool_call_id核对 messages 结构6.3 几个我踩过的具体坑坑一[DONE]没处理。前面提过流结束哨兵不是 JSON忘了判断就抛异常。坑二reasoning_content和content顺序错乱。推理模型有时先吐一大段 reasoning再吐 content有时交替。UI 侧如果假设reasoning 一定在前会渲染错位。正确做法是各自独立缓冲按到达顺序分别追加。坑三并发流共享状态。多个会话同时流式时如果 accumulator 是全局单例内容会串。每个请求必须持有独立的 accumulator 实例用请求 id 做 key。坑四代理层缓冲。如果中间隔了反向代理某些配置会缓冲整个响应再转发流式直接失效。排查时先直连 API 确认再逐层加代理定位。7. 关于流式管道的一点个人体会做流式输出这两年最大的感受是它不是一个功能而是一条贯穿模型、网络、框架、UI 的完整链路。任何一层偷懒用户都能感知到。模型侧再快UI 每 token 重渲染照样卡解析层再稳代理缓冲一开全白搭。我现在搭新 Agent 项目流式管道是第一批要跑通的东西比业务逻辑还优先。因为它是体验的地基地基不稳上面盖什么都是歪的。具体到 DeepSeek-Harness 这套我的经验是解析层用行缓冲加 orjson状态层用显式状态机管工具调用UI 层用 rAF 批量刷新这三板斧下去基本能覆盖 90% 的流式场景。剩下 10% 是各种边界超长回答的截断续写、多工具并行的参数聚合、断线重连的去重。这些没有银弹只能一个个 case 攒。但每解决一个管道的鲁棒性就上一个台阶。等你把这些坑都趟过一遍再回头看流式输出这四个字会发现它背后是一整套工程取舍而不是调个 API 那么简单。