1. 流式输出为什么值得单独拎出来讲做 Agent 应用的人迟早会撞上同一个问题模型明明已经在吐字了前端却像死机一样白屏好几秒等整段回答一次性蹦出来。用户体感就是“卡”哪怕后端推理速度再快这个体验也是不及格的。流式输出要解决的就是这件事——把模型逐 token 生成的内容实时地、按顺序地推到用户眼前。我这次在 DeepSeek-Harness 这套 Agent 框架里专门把流式输出管道从底层到 UI 完整梳理了一遍。核心链路其实就一句话模型侧产出StreamChunk经过一层层转换和封装最终变成浏览器里一个字一个字往外冒的效果。但真动手做你会发现中间藏着不少坑SSE 连接莫名其妙断掉、chunk 边界把多字节字符劈成两半、前端渲染顺序错乱、空闲超时把长回答掐死。这篇内容适合两类人看。一类是正在给 Agent 接流式接口的后端同学你需要搞清楚StreamChunk的数据结构怎么设计、SSE 怎么封装才稳另一类是前端同学你要处理的是怎么把一串异步到达的片段平滑地渲染成对话气泡。哪怕你用的是 Vue 还是 React底层逻辑是通的。我会把 DeepSeek-Harness 里这套管道的设计思路、关键代码结构、以及我踩过的坑都摊开讲尽量让你看完就能照着搭一套。2. 整体管道设计与数据流转思路2.1 从模型输出到 UI 的完整链路先把整条链路画清楚不然后面聊细节容易迷路。DeepSeek-Harness 里一次流式对话数据大致经过这么几站模型推理层逐 token 产出原始文本片段适配层把原始片段包装成统一的StreamChunk对象传输层通过 SSE 把StreamChunk序列化后推给客户端客户端 SDK 接收并解析 SSE 事件还原成StreamChunk状态管理层把 chunk 累积、合并、维护对话状态UI 层订阅状态变化增量渲染到界面这条链路里StreamChunk是贯穿始终的“通用货币”。不管底层是 DeepSeek 的 API、还是本地 vLLM 部署的模型甚至是别的兼容接口到了适配层统统转成StreamChunk。这样上层逻辑就不用关心模型来源换模型只需要换适配器。为什么要在中间加一层统一结构而不是直接把模型返回的 JSON 透传给前端我试过透传的方案结果是前端代码里到处是if (provider deepseek)这种判断模型一换就得改前端。加了StreamChunk这层抽象之后前端只认一种数据结构适配的脏活全压在后端维护成本低太多了。2.2 为什么选 SSE 而不是 WebSocket流式输出常见的传输方案有几种轮询、WebSocket、SSE。DeepSeek-Harness 选的是 SSE这个选择值得说道说道。轮询最简单但延迟高、请求量大模型吐字是连续的轮询的节奏对不上体验很差。WebSocket 是双向的能力最强但对于“服务端单向推、客户端只接收”这种场景它有点重了——要维护连接状态、要处理心跳、要自己定义消息协议而且很多网关和负载均衡对 WebSocket 的支持不如普通 HTTP 友好。SSE 的本质是“长连接的 HTTP 响应”服务端保持连接不关闭持续往响应体里写data: xxx\n\n格式的文本。它的优势很明确基于标准 HTTP穿透性好浏览器原生EventSource就能用断线还能自动重连。对于 Agent 对话这种“一问一答、服务端持续推送”的场景SSE 是最贴合的选择。注意SSE 是单向的客户端要发消息得另开一个普通 POST 请求。所以典型模式是“POST 发起对话 SSE 接收流式回复”两个请求配合。2.3 StreamChunk 的结构设计考量StreamChunk到底该包含哪些字段这个设计直接决定了后续好不好用。我见过一些实现只塞一个text字段结果遇到工具调用、思考过程、结束标记就抓瞎。DeepSeek-Harness 里的StreamChunk大致长这样interface StreamChunk { id: string; // 本次流式响应的唯一标识 type: ChunkType; // 片段类型text / tool_call / reasoning / done / error content: string; // 文本内容 index: number; // 片段序号用于排序和去重 finishReason?: string; // 结束原因stop / length / tool_calls metadata?: Recordstring, unknown; // 扩展字段 }type字段是关键。Agent 场景下模型输出不只有正文还可能有思考链、工具调用参数、工具返回结果。如果全混在一个content里前端没法区分该把哪部分渲染成正文、哪部分折叠成“思考中”。用type分开UI 就能按类型走不同的渲染分支。index字段用来兜底。SSE 理论上保序但网络抖动、重连之后可能出现乱序或重复带上序号前端就能做去重和排序保证最终拼接出来的文本是对的。3. 核心细节解析与实操要点3.1 SSE 事件格式与封装逻辑SSE 的协议格式看着简单真写起来细节不少。一个标准的事件长这样event: message id: 42 data: {id:abc,type:text,content:你,index:0}注意几个点。data:后面跟的是内容多个data:行会拼接事件之间用空行分隔event:和id:是可选的。DeepSeek-Harness 里我封装了一个SSEWriter专门负责把StreamChunk序列化成这个格式class SSEWriter: def __init__(self, response): self.response response self.counter 0 def write_chunk(self, chunk: StreamChunk): self.counter 1 payload json.dumps(chunk.to_dict(), ensure_asciiFalse) self.response.write(fid: {self.counter}\n) self.response.write(fevent: {chunk.type}\n) self.response.write(fdata: {payload}\n\n) self.response.flush() # 关键必须 flushflush()这行是命门。很多框架默认会缓冲响应你不主动 flush数据就攒在缓冲区里等攒够一批才发出去流式效果直接没了。我一开始就栽在这调试半天以为 SSE 没生效其实是没 flush。ensure_asciiFalse也别漏。默认json.dumps会把中文转成\uXXXX转义虽然功能上没问题但传输体积变大而且调试时看着难受。关掉之后中文原样输出清爽很多。3.2 多字节字符的边界处理这是流式输出里最阴险的坑之一。模型吐的是 token一个中文字符可能被拆成多个字节分几次到达。如果你在字节层面直接切分很可能把一个 UTF-8 字符劈成两半前端解码出来就是乱码。举个例子“深”这个字的 UTF-8 编码是E6 B7 B1三个字节。如果第一个 chunk 只到了E6 B7你直接decode(utf-8)就会抛异常或者出乱码。解决办法是维护一个字节缓冲区每次收到新数据先追加到缓冲区然后尝试解码遇到不完整的字符就留着等下一批class StreamDecoder: def __init__(self): self.buffer b def feed(self, data: bytes) - str: self.buffer data try: text self.buffer.decode(utf-8) self.buffer b return text except UnicodeDecodeError as e: # 保留不完整的尾部字节 valid self.buffer[:e.start] self.buffer self.buffer[e.start:] return valid.decode(utf-8, errorsignore)这个逻辑看着简单但少了它中文场景下随机出现乱码而且复现困难非常折磨人。我建议所有做流式的地方都加上这层解码缓冲别偷懒。3.3 空闲超时与心跳保活热词里有个很典型的报错stream disconnected before completion: idle timeout waiting for sse。这个问题的根源是模型在生成过程中可能有一段“思考期”比如调用工具、等待外部结果这期间没有新 token 产出SSE 连接上长时间没有数据流动中间的反向代理或网关就判定连接空闲直接掐断。解决办法是加心跳。在服务端定时往流里写注释行以:开头的行会被 SSE 客户端忽略保持连接活跃async def keepalive(response, interval15): while True: await asyncio.sleep(interval) response.write(: keepalive\n\n) response.flush()15 秒是个比较稳妥的间隔大多数网关的空闲超时在 30 到 60 秒15 秒能留出足够余量。同时客户端也要配置合理的超时别用默认值。EventSource默认没有超时但如果你用的是fetch手动解析流记得给读取操作设一个较长的超时或者干脆不设靠心跳维持。提示心跳间隔别设太短比如 1 秒一次会白白增加流量和 CPU 开销。15 到 30 秒是甜区。4. 实操过程与核心环节实现4.1 服务端流式接口的完整实现把前面几块拼起来一个完整的服务端流式接口大概是这样。我用 FastAPI 举例因为它对异步流式响应的支持比较顺手from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() async def stream_generator(prompt: str): decoder StreamDecoder() async for raw in call_model_api(prompt): text decoder.feed(raw) if not text: continue chunk StreamChunk( idcurrent_request_id, typetext, contenttext, indexnext_index(), ) yield format_sse(chunk) # 结束标记 yield format_sse(StreamChunk( idcurrent_request_id, typedone, content, indexnext_index(), finishReasonstop, )) app.post(/chat/stream) async def chat_stream(prompt: str): return StreamingResponse( stream_generator(prompt), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, # 关键禁用 Nginx 缓冲 }, )X-Accel-Buffering: no这个头是给 Nginx 看的。Nginx 默认会缓冲上游响应导致流式数据被攒起来前端看到的还是“一次性输出”。加上这个头Nginx 就知道不要缓冲直接透传。如果你前面还有别的代理也得确认它们的缓冲策略。media_type必须是text/event-stream这是 SSE 的标准 MIME 类型浏览器和客户端 SDK 靠它识别。4.2 客户端解析与状态累积客户端这边我用fetch加ReadableStream手动解析比EventSource灵活因为EventSource只支持 GET而发起对话通常需要 POST 带 body。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(utf-8); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); // 按空行切分事件 const events buffer.split(\n\n); buffer events.pop(); // 最后一段可能不完整留着 for (const event of events) { const chunk parseSSEEvent(event); if (chunk) onChunk(chunk); } } }decoder.decode(value, { stream: true })里的stream: true很重要它告诉解码器“后面还有数据”遇到不完整的多字节字符会先缓存不会直接报错。这跟服务端的解码缓冲是同一个道理两端都要处理。buffer.split(\n\n)之后pop()出来的最后一段是可能被截断的半个事件必须留到下一轮拼接。如果直接处理它会解析失败。4.3 前端增量渲染与性能取舍拿到StreamChunk之后怎么渲染到界面也有讲究。最朴素的做法是每来一个 chunk 就setState触发一次重渲染。token 生成快的时候一秒可能来几十个 chunk每次都重渲染React 或 Vue 的 diff 压力会很大页面会卡。我的做法是加一个小的批量窗口。用一个缓冲区攒 chunk每 50 毫秒统一 flush 一次到状态里let pending ; let timer null; function onChunk(chunk) { if (chunk.type text) { pending chunk.content; if (!timer) { timer setTimeout(() { appendToMessage(pending); pending ; timer null; }, 50); } } }50 毫秒是体感和性能的平衡点。人眼对 50 毫秒的延迟基本无感但渲染次数能降一个数量级。如果追求极致流畅可以用requestAnimationFrame对齐浏览器刷新节奏效果更好。另外长对话里消息列表会越来越长每次重渲染整个列表也是浪费。记得给消息组件加memo或者用虚拟列表只渲染可视区域。5. 常见问题与排查技巧实录5.1 流式输出问题速查表我把实际调试中遇到的高频问题整理成一张表方便对照排查现象可能原因排查方向解决手段前端白屏几秒后一次性输出响应被缓冲检查 flush、代理缓冲头主动 flush加 X-Accel-Buffering: no中文随机乱码多字节字符被截断检查解码逻辑加字节缓冲用 stream 模式解码长回答中途断开空闲超时看报错是否含 idle timeout加心跳调大网关超时chunk 顺序错乱重连或并发检查 index 是否连续用 index 排序去重连接建立但收不到数据事件格式错误抓包看原始字节确认 data: 后有空格、事件间有空行结束标记丢失生成器异常退出检查异常处理用 try/finally 保证发 done5.2 几个容易忽略的实操心得第一个心得是关于错误处理的。流式过程中如果模型侧抛异常连接已经建立了你不能直接返回 HTTP 500因为响应头早就发出去了。正确做法是把错误也包装成一个StreamChunktype设为error推给前端让前端优雅地展示错误提示。我见过不少实现遇到异常直接断流前端只能显示“连接中断”用户一脸懵。第二个心得是关于done标记的。一定要在流正常结束时发一个明确的结束 chunk。前端靠它来判断“回答结束了”进而停止 loading 动画、启用输入框。如果只靠连接关闭来判断遇到异常断流时前端无法区分“正常结束”和“出错中断”。第三个心得是关于日志的。流式场景下日志特别难打因为数据是碎片化的。我的做法是给每个请求分配一个request_id所有 chunk 的日志都带上这个 id事后按 id 聚合就能还原完整的流。调试时这个太有用了不然一堆碎片日志根本对不上。注意生产环境别把每个 chunk 的完整内容都打进日志量太大了。只记 id、type、index 和长度需要详细内容时再开 debug 级别。5.3 并发场景下的额外考量Agent 应用往往要扛并发多个用户同时对话每个对话一条 SSE 连接。这里有几个点要注意。连接数是有上限的。浏览器对同一域名的 HTTP/1.1 连接数限制在 6 个左右SSE 长连接会占满这个配额导致其他请求排队。解决办法是上 HTTP/2多路复用不受这个限制。如果暂时上不了 HTTP/2就得控制单页面的 SSE 连接数或者用共享连接的方式。服务端每个 SSE 连接都占一个协程或线程并发高了资源消耗不小。要设一个合理的并发上限超出的请求排队或拒绝别让服务被拖垮。同时给每个连接设一个最长存活时间防止僵尸连接堆积。还有一点是背压。如果客户端消费速度跟不上服务端生产速度数据会在缓冲区堆积。SSE 本身没有背压机制得靠应用层控制。简单做法是给缓冲区设上限超了就断开或降速。6. 从 StreamChunk 到 UI 的工程化收尾把这条管道跑通之后我最大的感受是流式输出的难点不在“流”本身而在边界处理。模型输出的边界、字节的边界、事件的边界、渲染的边界每一处边界都是 bug 的温床。DeepSeek-Harness 这套设计把StreamChunk作为统一契约把适配的复杂度收敛到后端一层这个思路我觉得是对的值得借鉴。如果你正准备给自己的 Agent 接流式输出我的建议是先别急着上复杂框架用最朴素的fetch加手动解析把链路跑通把解码缓冲、心跳、结束标记这几个基础件做扎实。等链路稳了再考虑抽成 SDK、加批量渲染优化。顺序反了的话底层不稳上层优化都是空中楼阁。最后分享一个我调试时的小技巧写一个 mock 的流式接口故意按字节切分、故意加延迟、故意中途断流用它来压测你的客户端解析逻辑。真实模型的行为不好复现但 mock 接口可以把各种边界情况都造出来。我靠这个 mock 接口提前发现了三个隐藏 bug比在真实环境里碰运气高效多了。