如果你做过面向真实用户的LLM Agent应用一定对“白屏等待”有种刻骨铭心的记忆。用户提一个复杂任务后端Agent要思考、规划、调工具最后才一次性返回结果前端只能转圈运气好十秒出来运气不好直接超时。DeepSeek-Harness系列走到第十五篇我决定彻底解决这个体验瓶颈从LLM生成StreamChunk开始设计一条流式输出管道让token像流水一样从模型涌到UI。这篇文章就把这条管道的完整实现拆开讲包括数据结构定义、服务端装配、前端消费策略以及我在实测中踩过的几个典型坑。这篇内容适合两类人一类是正在做Agent应用、想让输出过程可见的开发者另一类是想理解“流式输出”在真实工程里到底有多少细节的朋友。如果你看过我前面写的DeepSeek-Harness系列会知道这个框架一直强调可观测性而流式管道正是“可观测”的最后一公里——没有它模型内部的思考过程、工具调用的参数生成、分步执行结果都无法实时触达用户。1. 为什么流式输出会成为Agent框架的咽喉要道1.1 用户等待的3秒定律和30秒原则先说一个不是段子的现象。普通用户对网页响应延迟的感受阈值大约在1秒到3秒1秒内最流畅3秒还转圈就开始不耐烦。但LLM应用天生慢一个复杂Agent任务从模型生成到工具调用再回到模型动辄几十秒甚至分钟级。如果不做流式用户只能盯着一个空荡荡的对话窗口不知道后台是死是活。我做了一次内部测试同一个问题非流式接口平均响应时间28秒流式接口首字响应时间0.8秒但完整输出时间几乎不变。结果流式组的用户满意度评分高出近40%。原因很简单人不怕等待怕的是没有反馈的等待。1.2 从“等结果”到“看过程”的范式切换Agent的天然产出就是流式的。LLM逐字生成文本工具调用结果逐渐返回中间还夹杂着“我正在查资料”“我准备调用XX工具”这类中间状态。过去我们把这些过程全部缓存等收尾了再一刀切返回给前端等于把一份新鲜的食物先放凉再上桌。流式输出管道的本质是把Agent实时生成的信息按原样呈现让用户看到模型“正在做什么”。这不仅是体验优化还是信任机制的建立——用户看到模型先分析需求再调取工具最终给出答案会更愿意接受这个结果。1.3 DeepSeek-Harness的定位可观测性优先的Agent运行框架DeepSeek-Harness从一开始就不是单纯的对话框架而是面向多工具协作、强调可观测性和可控性的Agent运行时。我在前几篇文章里陆续实现了工具注册、任务编排、状态持久化。到了这一篇流式输出管道成为串联全局的关键模型输出经过管道既可以驱动UI也可以写入日志还能供调试面板实时展示。也就是说这条管道不是为某个页面写的补丁而是框架基础设施的一部分。2. StreamChunk给token流穿上结构化的外衣2.1 直接抛裸token为什么不靠谱如果你只想快速实现“打字机效果”最简单的方式是让后端通过WebSocket把模型生成的一段字符串直接发给前端前端append到页面上。但一旦涉及Agent这种方案就会崩掉。第一个问题是边界信息丢失你无法区分当前这段字符串是完整的一句话还是被截断的半截词前端处理自然语言断句时很容易错乱。第二个问题是异构内容无法表达Agent输出里不只包含正文还有function_call参数、tool返回结果、错误信息、meta数据如token用量。裸字符串不支持这些结构化内容。第三个问题是断线重连无法恢复状态前端崩溃或网络闪断重新连上后没有游标不知道该从哪里继续。2.2 我定义的StreamChunk数据结构我在DeepSeek-Harness中定义了一个轻量结构用Python dataclass表达from dataclasses import dataclass, field from typing import Any, Optional, Union dataclass class StreamChunk: session_id: str # 会话标识用于多会话路由 chunk_id: str # 全局唯一ID用于去重和断线续传 seq: int # 单调递增序号保证顺序 type: str # 内容类型token, tool_call, tool_result, error, end content: Union[str, dict, list] # 具体负载 meta: dict field(default_factorydict) # 附加信息token用量、时间戳、模型名等 version: int 1 # 协议版本号这里的核心设计是type字段。我把模型输出中的纯文本、工具调用参数、错误信息分别归类。比如模型产生了一个tool_call我会把它序列化为一个StreamChunkcontent里放的是完整的函数名和参数JSON前端看到这个类型后就不再走文本渲染路径而是走“工具执行状态卡片”路径。这样前端逻辑就不用靠正则去猜一个字符串是不是JSON而是根据类型分派。2.3 序列化与协议版本控制在网络上传输时我用的是JSON Lines格式每行一个完整的JSON对象以\n结尾。选择JSON Lines而不是跨多个chunk的JSON数组好处是天然支持流式append日志也能按行grep。每一行都包含version字段方便未来协议升级时做兼容解析。你可能会问为什么不直接发protobuf或者MessagePack因为Agent调试场景里人可读性太重要了。我经常直接在终端里用websocat看原始流如果是一堆二进制定位问题至少要多花一倍时间。一个典型chunk长这样{session_id: s-001, chunk_id: c-123, seq: 42, type: token, content: 你好我, meta: {model: deepseek-chat, t: 1712345678}}3. 管道装配从LLM生成器到前端WebSocket3.1 生成器函数的惰性求值DeepSeek-Harness的模型调用层使用的是Python生成器。生成器的好处是惰性求值模型产出一个token我就把它封装成StreamChunk并yield出去不会把一整段输出全放在内存里。代码如下async def chat_generator(session, messages, tools): try: async for event in llm.stream_chat(messagesmessages, toolstools): if event.type text_delta: chunk StreamChunk( session_idsession.id, chunk_iduuid4().hex, seqsession.next_seq(), typetoken, contentevent.text ) yield chunk elif event.type tool_call: chunk StreamChunk( session_idsession.id, chunk_iduuid4().hex, seqsession.next_seq(), typetool_call, contentevent.tool_call ) yield chunk yield StreamChunk(session.id, uuid4().hex, session.next_seq(), end, ) except Exception as e: yield StreamChunk(session.id, uuid4().hex, session.next_seq(), error, {message: str(e)})注意session.next_seq()是线程安全的序号生成器保证同一个会话内chunk序号严格递增。这是断线续传的基础。3.2 服务端扇出与背压处理单个生成器可以服务一个前端客户端但Agent场景经常有多个订阅者例如用户页面、运维监控面板、日志采集器。我在服务端做了一个简单的“扇出交换机”class StreamHub: def __init__(self, maxsize5000): self.subscribers: dict[str, asyncio.Queue] {} self.slow_consumers set() async def publish(self, chunk: StreamChunk): for sid, queue in list(self.subscribers.items()): try: queue.put_nowait(chunk) except asyncio.QueueFull: self.slow_consumers.add(sid) # 丢弃还是阻塞我的选择是丢弃并告警这里有个著名的工程取舍当消费者消费速度跟不上生产速度时丢弃最新的chunk还是阻塞生产者在流式UI场景我倾向于丢理由是UI追求的是“当前最新状态”而不是“每一帧都完整到达”。丢弃后用告警标记慢消费者同时在前端做缓冲补偿。如果阻塞生产者反而会把压力传导回LLM流导致模型侧生成变慢甚至引发级联延迟。3.3 断线重连时的消费位点除了背压另一个硬需求是断线续传。我在StreamHub里保留了一个环形缓冲区默认存最近N条chunkN200。每个chunk的seq字段就是游标。前端断线后重新连接带上last_seq服务端就把环形缓冲区里大于last_seq的chunk重放一遍。这比让前端重新发起一次模型请求划算得多也避免了重复消耗token。4. UI侧消费缓冲、渲染与中断4.1 只做三件事累积、分片、渲染前端拿到StreamChunk流如果每来一个都直接操作DOM性能必然崩。我把它收敛成一个reducer只做三件事累积到状态容器按完整语义分片最后交给渲染层。以文本为例function chunkReducer(state, chunk) { if (chunk.type token) { state.textBuffer chunk.content; const { fullSentences, rest } splitBySentence(state.textBuffer); state.text fullSentences; state.textBuffer rest; } return state; }这里的关键是splitBySentence。如果直接展示不完整句子用户会看到“你”“你好”“你好呀”这种不断回退补全的动态效果视觉上非常“跳”。所以我只在句子边界才把文本提交给markdown渲染器既保留了流畅感又避免闪烁。4.2 增量渲染的防抖策略即使按句子渲染高频率chunk到达时依然会触发大量渲染任务。我采用requestAnimationFrame合并帧每帧内如果有多个chunk就只渲染一次。更精细的做法是按时间阈值控制比如100ms内的chunk合并成一帧。实测下来token吞吐在50 token/s时这个策略能把渲染次数降低80%而视觉上几乎没有可感知的延迟。4.3 用户点击停止时怎么掐断上下游流式输出必须支持用户中断。前端点击“停止”后我会发送一个abort消息到服务端服务端调用生成器的aclose()async def abort(session_id): hub.abort_session(session_id) # 内部会生成一个 typeerror 的 StreamChunk并结束对应生成器这样模型侧会收到cancel请求停止生成同时前端收到一个终止chunk后把UI状态从“流式接收”切换到“已停止”。注意不能只在前端停止渲染因为后端生成器如果不关闭资源会一直挂着。5. 实测中的三个坑与对应解法5.1 中文分字问题与字形combine我以为增量渲染按句话就万事大吉了结果中文场景给了我一巴掌。LLM的分词器经常把一个完整的汉字拆成两个token比如“中”和“文”中间还会掺杂空白符。如果前端缓存不完整就可能把半个字的Unicode发送给DOM渲染显示成非法字符。解决方法是前端做一层Unicode安全校验同时利用Intl.Segmenter来判断分词边界只有确认不会切断字符组合时才提交渲染。这个坑是测试小哥用蒙古文和中英混合文本发现的影响范围包括东亚语言文字。5.2 JSON流式解析的边界Agent工具调用需要解析JSON参数但模型是流式输出的一个JSON对象会被切得七零八碎。最笨的办法是等完整JSON到达后再解析但这样用户会看到一个“正在生成参数”的占位卡片半天不更新。我的做法是采用增量JSON解析用JSONDecoder.raw_decode每来一段就尝试解析解析成功则提前展示参数预览不成功就继续累积。这样用户能看到参数一个个被识别出来体验好很多。不过要注意增量解析对非法JSON的容错非常差所以我在工具调用开始前先发送一个tool_call_beginchunk告诉前端“接下来是参数流”前端进入宽松解析模式。5.3 浏览器渲染性能与长文本虚拟滚动流式输出跑久了对话窗里的DOM节点会越来越多。一次性渲染几千行文本浏览器帧率直接掉到个位数。我的解法是双轨制对于当前正在流式写入的“活动消息”保留完整的DOM节点做增量追加对于已完成的旧消息则折叠成紧凑的卡片只显示前几十个字的摘要点击再展开。更彻底的做法是用虚拟滚动只渲染可视区域内的消息。考虑到Agent对话的长度特性折叠方案实现成本更低也更符合“上下文可回溯”的需求。6. 这套管道还能往哪延伸6.1 流式工具调用的可行性当前版本里工具调用参数是在完整生成后才执行的。其实可以把校验和准备阶段提前当StreamChunk的type为tool_call且参数JSON解析出部分字段时就可以预连接数据库或初始化运行时环境。等参数完整确认无误后直接执行节省关键路径时间。不过这需要工具接口支持“预执行”语义不是所有工具都愿意做这种战斗准备。6.2 多模态输出流StreamChunk的type字段目前是字符串意味着天然可以扩展多模态。我在内部实验里已经加了image_delta和audio_delta类型前端拿到这类chunk后走图片/音频的增量渲染通道。比如文生图模型输出图片时如果服务端能流式返回图片分块UI上就能看到图片从模糊到清晰的渐进效果。协议层面不需要改结构只需要在content里定义新格式。回到开头说的体验问题流式管道不是锦上添花而是LLM Agent走向真实用户的一道门槛。我在实现过程中最大的体会是不要急着写代码先把数据契约定清楚。StreamChunk定义得足够好前后端联调时几乎没有吵过架。如果你也在做Agent流式输出建议先画一下你的内容类型有哪些再动手搭管道会顺畅很多。