1. 从一次流式输出卡顿说起为什么要啃 runLoop 这条流水线如果你用过 Pi-Agent 做过稍微复杂一点的智能体任务大概率遇到过这种场景用户发一句话模型开始一个字一个字往外蹦前端看着挺流畅但中间某个工具调用一插进来整个输出就卡住了等工具返回之后才继续吐字。更诡异的是有时候工具还没执行完模型那边已经把后续内容全生成完了结果前端拿到的是一个先卡后爆的体验。这个问题我第一次遇到的时候直觉是网络抖动抓了半天包才发现问题根本不在网络上而在 Pi-Agent 内部的runLoop 消息流水线和LLM 流式响应的衔接方式上。Pi-Agent 作为一个把 LLM 调用、工具执行、消息回写串起来的智能体框架它的核心其实就是一个循环拿消息 → 调模型 → 解析流式输出 → 判断要不要调工具 → 执行工具 → 把结果塞回消息列表 → 再调模型。这个循环就是 runLoop而流式响应是这个循环里唯一一个持续产出的数据源。所以这篇东西不是那种泛泛而谈的框架介绍而是我实际读 Pi-Agent 源码、改过它的流式处理逻辑之后把 runLoop 这条流水线拆开讲清楚。核心想解决三个问题流式响应在 runLoop 里到底是怎么被消费的、工具调用和流式输出为什么会互相阻塞、以及怎么改才能让两者真正并行起来。适合已经用过 Pi-Agent、想深入定制它行为的人也适合正在自己写 Agent 框架、想参考一套成熟流式处理设计的同学。读源码这件事最怕的就是从入口一路顺着调用栈往下读读到最后忘了自己在哪。我的习惯是先找到那条主干数据流Pi-Agent 里这条主干就是消息在 runLoop 里的流转路径。抓住它其他分支都是挂在上面的枝叶。2. runLoop 的主干结构消息是怎么一圈一圈转起来的2.1 一次完整的循环到底经历了什么Pi-Agent 的 runLoop 本质上是一个while循环每一轮做四件事组装当前消息上下文、发起 LLM 请求、消费流式响应、根据响应决定是否继续循环。听起来简单但魔鬼在消费流式响应这一步因为它不是一个同步返回而是一个异步事件流。我用一个简化后的伪代码把主干画出来注意这不是源码原文是我按源码逻辑重写的结构方便你理解async def run_loop(self, messages): while True: # 1. 组装上下文注入系统提示、历史消息、工具定义 context self.build_context(messages) # 2. 发起流式请求拿到一个异步迭代器 stream await self.llm.stream_chat(context) # 3. 消费流式响应 assistant_message await self.consume_stream(stream) # 4. 把助手消息追加进历史 messages.append(assistant_message) # 5. 判断是否有工具调用 if not assistant_message.tool_calls: break # 没有工具调用循环结束 # 6. 执行工具把结果作为新消息塞回去 for tool_call in assistant_message.tool_calls: result await self.execute_tool(tool_call) messages.append(build_tool_message(tool_call, result)) # 回到 while 顶部带着工具结果再问一次模型这个结构里第 3 步consume_stream是整条流水线的心脏。它决定了流式响应怎么被拆解、怎么被累积、什么时候触发工具调用。很多人读源码卡住就是卡在这一步——因为流式响应的数据格式在不同 provider 之间差异很大Pi-Agent 做了一层适配这层适配是理解整个流水线的钥匙。2.2 消息对象的三种形态与状态流转在 runLoop 里流转的消息不是一种东西而是三种形态在互相转换。搞清楚这三种形态你就知道每一圈循环里数据长什么样。形态来源关键字段去向用户/系统消息外部输入、系统提示role, content作为上下文输入助手消息流式累积中LLM 流式响应role, content 增量, tool_calls 增量累积完成后入历史工具结果消息工具执行返回roletool, tool_call_id, content作为上下文输入关键在于第二种形态。流式响应不是一次性给你一个完整的助手消息而是给你一堆 delta增量片段。这些 delta 分两类文本 delta 和工具调用 delta。文本 delta 好理解就是一个个 token工具调用 delta 麻烦一点因为一个工具调用的参数是分多次传过来的你得自己拼。我见过不少人在这里踩坑以为tool_calls是一次性给全的结果拿到的是残缺的 JSON 字符串json.loads直接报错。实际上流式协议里工具调用的arguments是按片段累积的必须等流结束或者收到特定结束标志才能解析。2.3 为什么 runLoop 要设计成一圈一问而不是一次问完这里有个设计上的取舍值得说。理论上你可以让模型一次性把所有工具调用都规划好然后批量执行最后再汇总。但 Pi-Agent 选择了一圈一问每执行完一批工具就带着结果重新问一次模型。原因在于上下文依赖。很多工具调用的参数依赖前一个工具的输出比如先查数据库拿到 ID再用这个 ID 去调另一个接口。如果一次性规划模型在不知道第一个工具返回什么的情况下没法准确填第二个工具的参数。一圈一问虽然多了一次 LLM 请求但换来了工具链的正确性。代价就是延迟。每一圈都要重新走一次完整的 LLM 请求包括重新发送整个上下文。这也是为什么流式响应在这里特别重要——如果不用流式用户要等一整轮模型生成 工具执行 再生成才能看到任何输出体验会非常差。流式的价值就是让模型正在思考这件事对用户可见。3. 流式响应的消费逻辑delta 是怎么被拼成完整消息的3.1 文本 delta 与工具调用 delta 的分流处理consume_stream这个函数的核心工作是把一串异构的 delta 事件归并成一个结构化的助手消息。我把它内部的逻辑拆成三条并行处理的线第一条线处理文本内容。每收到一个文本 delta就追加到一个字符串缓冲区同时通过回调把增量推给上层前端就能实时显示。这条线最简单但要注意一个细节有些 provider 会在流结束时补一个完整的 content如果你不判断就会把内容重复追加一遍。Pi-Agent 里用了一个is_final标志来避免这个问题。第二条线处理工具调用。工具调用的 delta 通常带一个index字段标识这是第几个工具调用。同一个 index 的 delta 要合并到同一个工具调用对象上。合并的时候arguments字段是字符串拼接不是对象合并——这点特别容易搞错。我见过有人用dict.update去合并 arguments结果 JSON 结构全乱了。第三条线处理结束标志。流结束时会有一个finish_reason可能是stop正常结束、tool_calls要调工具、length超长截断。runLoop 就是靠这个字段决定下一步走向的。async def consume_stream(self, stream): content_buffer tool_calls {} # index - {id, name, arguments} finish_reason None async for chunk in stream: delta chunk.choices[0].delta # 文本增量 if delta.content: content_buffer delta.content await self.on_text_delta(delta.content) # 工具调用增量 if delta.tool_calls: for tc in delta.tool_calls: idx tc.index if idx not in tool_calls: tool_calls[idx] {id: , name: , arguments: } if tc.id: tool_calls[idx][id] tc.id if tc.function and tc.function.name: tool_calls[idx][name] tc.function.name if tc.function and tc.function.arguments: # 注意这里是字符串拼接 tool_calls[idx][arguments] tc.function.arguments if chunk.choices[0].finish_reason: finish_reason chunk.choices[0].finish_reason return build_assistant_message(content_buffer, tool_calls, finish_reason)3.2 arguments 字符串拼接的坑与解析时机上面代码里那句arguments tc.function.arguments是整段逻辑里最容易出问题的地方。我实际调试的时候遇到过三种情况第一种arguments 分片传过来每片都是合法 JSON 的一部分拼起来才完整。这是标准情况等流结束再json.loads就行。第二种某些 provider 会在每个 delta 里都带一份到目前为止的完整 arguments而不是增量。这时候你要是无脑拼接就会得到一坨重复的 JSON。判断方法是看第一个 delta 的 arguments 是不是以{开头且后续 delta 也以{开头——如果是说明是全量而非增量。第三种arguments 里包含转义字符或者多字节字符分片可能把一个 UTF-8 字符切成两半。这种情况在纯 ASCII 参数里不会出现但参数里带中文就会。解决办法是先把所有分片按字节拼接最后再统一解码而不是每片都尝试解码。提示解析 arguments 的时机一定要放在流完全结束之后不要在收到第一个 delta 时就急着解析。我早期就是急着解析结果拿到半截 JSON报了一堆莫名其妙的错。3.3 流式回调与 UI 更新的节流问题流式响应推给前端的时候如果每个 token 都触发一次 UI 更新在高频输出场景下会把渲染线程打满。Pi-Agent 默认没有做节流这是我在实际项目里自己加的一层。做法很简单用一个缓冲区累积 delta每 50 毫秒或者每累积到一定字符数才 flush 一次给 UI。这样既保持了逐字显示的观感又不会让渲染压力过大。50 毫秒这个值是我实测下来比较平衡的——低于 30 毫秒人眼基本分辨不出差异高于 100 毫秒会有明显的一顿一顿感。class ThrottledEmitter: def __init__(self, emit_fn, interval_ms50): self.emit_fn emit_fn self.interval interval_ms / 1000 self.buffer self.last_flush time.monotonic() async def push(self, delta): self.buffer delta now time.monotonic() if now - self.last_flush self.interval: await self.emit_fn(self.buffer) self.buffer self.last_flush now async def flush(self): if self.buffer: await self.emit_fn(self.buffer) self.buffer 4. 工具调用与流式输出的阻塞点为什么会出现先卡后爆4.1 阻塞的根因串行等待工具执行回到开头那个问题。为什么工具调用一插进来流式输出就卡住了因为 runLoop 的结构是串行的必须等consume_stream完全结束拿到完整的助手消息才能知道有哪些工具要调然后才能执行工具。而工具执行是同步等待的这期间没有任何新的流式数据产生前端自然就卡住了。更细一点说卡顿发生在两个地方。第一处是模型生成工具调用参数的过程——如果参数很长比如要生成一大段代码作为参数这段时间前端只能显示正在生成工具调用没有正文输出。第二处是工具真正执行的时间——如果是网络请求或者耗时计算这段时间完全是空白的。先卡后爆里的爆指的是工具执行完之后模型基于工具结果重新生成这次生成往往很快因为上下文已经确定于是一大段内容瞬间涌出来。前后对比强烈体验就很割裂。4.2 一个真实的排查链路从现象到根因我记录一下当时排查这个问题的完整过程因为这种排查思路比结论更有价值。现象是用户问一个需要查数据库的问题前端先显示正在思考然后卡住约 3 秒接着突然刷出一大段回答。第一步我先确认卡的是哪一段。在consume_stream的入口和出口各打一个时间戳发现流式消费本身只花了 800 毫秒说明卡的不是模型生成。第二步在工具执行前后打时间戳发现工具执行花了 2.2 秒。到这里定位到卡顿主要来自工具执行。第三步但 800 毫秒的模型生成时间里前端也没显示任何东西。查了一下发现是因为工具调用的参数生成阶段没有文本 delta前端没有内容可显示只能显示 loading。第四步真正的根因是runLoop 把模型生成和工具执行完全串起来了中间没有任何并行或者提前反馈的机制。用户看到的 3 秒卡顿 800 毫秒静默生成 2.2 秒工具执行。4.3 三种缓解思路的取舍定位到根因之后我试了三种改法各有取舍。第一种是提前反馈。在模型开始生成工具调用参数时就通过回调告诉前端正在准备调用 XX 工具。这样虽然还是要等但用户知道系统在干活不是死了。改动最小收益也最直接。第二种是工具预执行。如果模型生成的工具调用参数是流式到达的可以在参数拼完整的那一刻就开始执行而不是等整个流结束。但这里有个风险如果模型后面又改了参数虽然少见预执行就白做了。所以只适合参数确定后不会变的场景。第三种是并行工具执行。如果一轮里有多个工具调用且它们之间没有依赖可以并行执行。Pi-Agent 默认是串行 for 循环改成asyncio.gather就能并行。但要注意工具结果回写到消息列表的顺序必须和工具调用的顺序一致否则模型会困惑。改法改动量收益风险提前反馈小体验提升明显几乎无工具预执行中省掉参数生成等待参数变更时白做并行执行中多工具场景提速明显顺序错乱风险我最后是三种都上了但并行执行加了顺序保证用asyncio.gather拿到结果后按原始 index 排序再回写。5. 改造 runLoop让流式输出和工具执行真正并行5.1 把 consume_stream 拆成边收边发的模式原版consume_stream是收完再返回改造的核心是把它变成边收边发。具体做法是给 consume_stream 传入一个回调每收到文本 delta 就立即回调而不是等函数返回。async def consume_stream_with_callback(self, stream, on_delta, on_tool_ready): content_buffer tool_calls {} async for chunk in stream: delta chunk.choices[0].delta if delta.content: content_buffer delta.content await on_delta(delta.content) # 立即推送 if delta.tool_calls: for tc in delta.tool_calls: idx tc.index if idx not in tool_calls: tool_calls[idx] {id: , name: , arguments: } # ... 合并逻辑同上 ... # 参数拼完整时立即通知 if is_complete_json(tool_calls[idx][arguments]): await on_tool_ready(idx, tool_calls[idx]) return build_assistant_message(content_buffer, tool_calls)这里is_complete_json是个小技巧尝试解析 arguments如果成功说明参数已经完整可以提前触发工具执行。这个判断在参数是简单结构时很准复杂嵌套结构偶尔会误判所以我在实际项目里加了个参数长度稳定超过 N 个 delta 未变化的辅助判断。5.2 工具执行结果如何插队回消息流工具执行完之后结果要作为一条 tool 消息插回消息列表。这里有个顺序问题如果多个工具并行执行结果回来的顺序可能和调用顺序不一致。而 LLM 对消息顺序是敏感的tool 消息必须紧跟在对应的 assistant 消息之后且顺序要和 tool_calls 一致。我的做法是维护一个结果槽位数组每个工具执行完就填到对应 index 的槽位里等所有槽位都填满再按顺序一次性追加到消息列表。这样既享受了并行执行的提速又保证了消息顺序正确。async def execute_tools_parallel(self, tool_calls): results [None] * len(tool_calls) async def run_one(idx, call): results[idx] await self.execute_tool(call) await asyncio.gather(*[run_one(i, c) for i, c in enumerate(tool_calls)]) return results # 顺序已保证5.3 改造后的时序对比与实测数据改造前后我做了个对比测试用一个需要调用两次工具的任务先查用户 ID再查订单记录各阶段耗时。阶段改造前改造后首次模型生成800ms800ms工具1执行1200ms1200ms二次模型生成600ms600ms工具2执行1000ms1000ms三次模型生成700ms700ms用户感知首字延迟3600ms800ms总耗时4300ms4300ms注意总耗时没变因为工具执行本身的时间省不掉。但用户感知的首字延迟从 3.6 秒降到了 0.8 秒这就是流式 提前反馈的价值。用户不需要等整个任务完成第一轮模型生成的内容就能看到。注意并行执行只在工具之间无依赖时有效。如果工具2的参数依赖工具1的结果那它们本来就得串行强行并行只会拿到错误的参数。判断依赖关系这件事目前还得靠人工标注或者让模型显式声明。6. 读源码时容易忽略的几个细节6.1 异常在流式消费中的传播路径流式响应最麻烦的地方在于异常可能发生在流的任何位置。网络断了、provider 返回错误、JSON 解析失败这些异常如果处理不好会让 runLoop 直接崩掉用户看到的是一个没有任何反馈的空白。Pi-Agent 里对异常的处理是分层的网络层异常会触发重试解析层异常会记录并跳过当前 delta业务层异常比如工具执行失败会作为 tool 消息的内容返回给模型让模型自己决定怎么处理。这个分层设计很值得借鉴——不是所有异常都该往上抛有些异常应该降级成模型能理解的信息。我实际遇到过一个 case工具执行超时原版直接把异常抛出来runLoop 中断。我改成把超时信息作为 tool 结果返回模型收到后会自动换一种方式重试或者告诉用户这个操作超时了。体验好很多。6.2 上下文长度与流式截断的关系流式响应有个隐藏问题如果上下文接近模型的长度上限模型可能在生成到一半时被截断finish_reason会是length而不是stop。这时候如果你不处理会得到一个不完整的助手消息后续逻辑可能出错。Pi-Agent 的处理是检测到length时把已生成的内容作为完整消息但标记一个truncated标志。runLoop 看到这个标志可以选择继续生成把已生成内容作为前缀再问一次或者直接结束。我建议在长任务场景下开启续写逻辑否则用户会看到回答莫名其妙断掉。6.3 多轮循环下的消息去重runLoop 每转一圈都会往消息列表里追加内容。如果不小心很容易出现重复消息。我踩过一次坑在工具执行失败重试的逻辑里把同一个 assistant 消息追加了两次结果模型看到两条一样的消息开始自我重复。避免方法很简单每次追加消息前检查消息列表最后一条是不是同一条消息用消息 ID 或者内容哈希判断。这个检查开销很小但能避免很多诡异问题。7. 如果你要自己实现一套类似的流水线读完 Pi-Agent 的 runLoop我最大的感受是这套设计的精髓不在于代码多复杂而在于它把流式这件事贯穿到了整个循环里而不是只在最外层包一层。很多自研 Agent 框架的问题就是流式只做在 LLM 调用那一层工具执行、消息回写全是同步阻塞的结果就是流式的优势被后面的阻塞吃掉了。如果你要自己实现我的建议是从第一天就把 runLoop 设计成事件驱动的每个阶段都通过回调或者事件往外抛状态而不是等整个循环结束才返回。这样无论是接前端、接日志、接监控都有统一的接入点。另外别小看 delta 合并这块的复杂度。不同 provider 的流式协议差异比想象中大我建议在适配层做充分的归一化把各种奇葩格式都转成统一的内部 delta 结构上层逻辑就干净了。Pi-Agent 在这块做得不错但适配层本身也是整个代码库里最容易出 bug 的地方改的时候要格外小心。最后分享一个我调试流式问题的小技巧把所有 delta 事件按时间戳打日志导出成一份时间线。卡顿问题在时间线上一目了然——哪段时间没有事件、哪段时间事件密集比看代码猜快得多。这个习惯帮我省了至少一半的排查时间。
1. 从一次流式输出卡顿说起为什么要啃 runLoop 这条流水线如果你用过 Pi-Agent 做过稍微复杂一点的智能体任务大概率遇到过这种场景用户发一句话模型开始一个字一个字往外蹦前端看着挺流畅但中间某个工具调用一插进来整个输出就卡住了等工具返回之后才继续吐字。更诡异的是有时候工具还没执行完模型那边已经把后续内容全生成完了结果前端拿到的是一个先卡后爆的体验。这个问题我第一次遇到的时候直觉是网络抖动抓了半天包才发现问题根本不在网络上而在 Pi-Agent 内部的runLoop 消息流水线和LLM 流式响应的衔接方式上。Pi-Agent 作为一个把 LLM 调用、工具执行、消息回写串起来的智能体框架它的核心其实就是一个循环拿消息 → 调模型 → 解析流式输出 → 判断要不要调工具 → 执行工具 → 把结果塞回消息列表 → 再调模型。这个循环就是 runLoop而流式响应是这个循环里唯一一个持续产出的数据源。所以这篇东西不是那种泛泛而谈的框架介绍而是我实际读 Pi-Agent 源码、改过它的流式处理逻辑之后把 runLoop 这条流水线拆开讲清楚。核心想解决三个问题流式响应在 runLoop 里到底是怎么被消费的、工具调用和流式输出为什么会互相阻塞、以及怎么改才能让两者真正并行起来。适合已经用过 Pi-Agent、想深入定制它行为的人也适合正在自己写 Agent 框架、想参考一套成熟流式处理设计的同学。读源码这件事最怕的就是从入口一路顺着调用栈往下读读到最后忘了自己在哪。我的习惯是先找到那条主干数据流Pi-Agent 里这条主干就是消息在 runLoop 里的流转路径。抓住它其他分支都是挂在上面的枝叶。2. runLoop 的主干结构消息是怎么一圈一圈转起来的2.1 一次完整的循环到底经历了什么Pi-Agent 的 runLoop 本质上是一个while循环每一轮做四件事组装当前消息上下文、发起 LLM 请求、消费流式响应、根据响应决定是否继续循环。听起来简单但魔鬼在消费流式响应这一步因为它不是一个同步返回而是一个异步事件流。我用一个简化后的伪代码把主干画出来注意这不是源码原文是我按源码逻辑重写的结构方便你理解async def run_loop(self, messages): while True: # 1. 组装上下文注入系统提示、历史消息、工具定义 context self.build_context(messages) # 2. 发起流式请求拿到一个异步迭代器 stream await self.llm.stream_chat(context) # 3. 消费流式响应 assistant_message await self.consume_stream(stream) # 4. 把助手消息追加进历史 messages.append(assistant_message) # 5. 判断是否有工具调用 if not assistant_message.tool_calls: break # 没有工具调用循环结束 # 6. 执行工具把结果作为新消息塞回去 for tool_call in assistant_message.tool_calls: result await self.execute_tool(tool_call) messages.append(build_tool_message(tool_call, result)) # 回到 while 顶部带着工具结果再问一次模型这个结构里第 3 步consume_stream是整条流水线的心脏。它决定了流式响应怎么被拆解、怎么被累积、什么时候触发工具调用。很多人读源码卡住就是卡在这一步——因为流式响应的数据格式在不同 provider 之间差异很大Pi-Agent 做了一层适配这层适配是理解整个流水线的钥匙。2.2 消息对象的三种形态与状态流转在 runLoop 里流转的消息不是一种东西而是三种形态在互相转换。搞清楚这三种形态你就知道每一圈循环里数据长什么样。形态来源关键字段去向用户/系统消息外部输入、系统提示role, content作为上下文输入助手消息流式累积中LLM 流式响应role, content 增量, tool_calls 增量累积完成后入历史工具结果消息工具执行返回roletool, tool_call_id, content作为上下文输入关键在于第二种形态。流式响应不是一次性给你一个完整的助手消息而是给你一堆 delta增量片段。这些 delta 分两类文本 delta 和工具调用 delta。文本 delta 好理解就是一个个 token工具调用 delta 麻烦一点因为一个工具调用的参数是分多次传过来的你得自己拼。我见过不少人在这里踩坑以为tool_calls是一次性给全的结果拿到的是残缺的 JSON 字符串json.loads直接报错。实际上流式协议里工具调用的arguments是按片段累积的必须等流结束或者收到特定结束标志才能解析。2.3 为什么 runLoop 要设计成一圈一问而不是一次问完这里有个设计上的取舍值得说。理论上你可以让模型一次性把所有工具调用都规划好然后批量执行最后再汇总。但 Pi-Agent 选择了一圈一问每执行完一批工具就带着结果重新问一次模型。原因在于上下文依赖。很多工具调用的参数依赖前一个工具的输出比如先查数据库拿到 ID再用这个 ID 去调另一个接口。如果一次性规划模型在不知道第一个工具返回什么的情况下没法准确填第二个工具的参数。一圈一问虽然多了一次 LLM 请求但换来了工具链的正确性。代价就是延迟。每一圈都要重新走一次完整的 LLM 请求包括重新发送整个上下文。这也是为什么流式响应在这里特别重要——如果不用流式用户要等一整轮模型生成 工具执行 再生成才能看到任何输出体验会非常差。流式的价值就是让模型正在思考这件事对用户可见。3. 流式响应的消费逻辑delta 是怎么被拼成完整消息的3.1 文本 delta 与工具调用 delta 的分流处理consume_stream这个函数的核心工作是把一串异构的 delta 事件归并成一个结构化的助手消息。我把它内部的逻辑拆成三条并行处理的线第一条线处理文本内容。每收到一个文本 delta就追加到一个字符串缓冲区同时通过回调把增量推给上层前端就能实时显示。这条线最简单但要注意一个细节有些 provider 会在流结束时补一个完整的 content如果你不判断就会把内容重复追加一遍。Pi-Agent 里用了一个is_final标志来避免这个问题。第二条线处理工具调用。工具调用的 delta 通常带一个index字段标识这是第几个工具调用。同一个 index 的 delta 要合并到同一个工具调用对象上。合并的时候arguments字段是字符串拼接不是对象合并——这点特别容易搞错。我见过有人用dict.update去合并 arguments结果 JSON 结构全乱了。第三条线处理结束标志。流结束时会有一个finish_reason可能是stop正常结束、tool_calls要调工具、length超长截断。runLoop 就是靠这个字段决定下一步走向的。async def consume_stream(self, stream): content_buffer tool_calls {} # index - {id, name, arguments} finish_reason None async for chunk in stream: delta chunk.choices[0].delta # 文本增量 if delta.content: content_buffer delta.content await self.on_text_delta(delta.content) # 工具调用增量 if delta.tool_calls: for tc in delta.tool_calls: idx tc.index if idx not in tool_calls: tool_calls[idx] {id: , name: , arguments: } if tc.id: tool_calls[idx][id] tc.id if tc.function and tc.function.name: tool_calls[idx][name] tc.function.name if tc.function and tc.function.arguments: # 注意这里是字符串拼接 tool_calls[idx][arguments] tc.function.arguments if chunk.choices[0].finish_reason: finish_reason chunk.choices[0].finish_reason return build_assistant_message(content_buffer, tool_calls, finish_reason)3.2 arguments 字符串拼接的坑与解析时机上面代码里那句arguments tc.function.arguments是整段逻辑里最容易出问题的地方。我实际调试的时候遇到过三种情况第一种arguments 分片传过来每片都是合法 JSON 的一部分拼起来才完整。这是标准情况等流结束再json.loads就行。第二种某些 provider 会在每个 delta 里都带一份到目前为止的完整 arguments而不是增量。这时候你要是无脑拼接就会得到一坨重复的 JSON。判断方法是看第一个 delta 的 arguments 是不是以{开头且后续 delta 也以{开头——如果是说明是全量而非增量。第三种arguments 里包含转义字符或者多字节字符分片可能把一个 UTF-8 字符切成两半。这种情况在纯 ASCII 参数里不会出现但参数里带中文就会。解决办法是先把所有分片按字节拼接最后再统一解码而不是每片都尝试解码。提示解析 arguments 的时机一定要放在流完全结束之后不要在收到第一个 delta 时就急着解析。我早期就是急着解析结果拿到半截 JSON报了一堆莫名其妙的错。3.3 流式回调与 UI 更新的节流问题流式响应推给前端的时候如果每个 token 都触发一次 UI 更新在高频输出场景下会把渲染线程打满。Pi-Agent 默认没有做节流这是我在实际项目里自己加的一层。做法很简单用一个缓冲区累积 delta每 50 毫秒或者每累积到一定字符数才 flush 一次给 UI。这样既保持了逐字显示的观感又不会让渲染压力过大。50 毫秒这个值是我实测下来比较平衡的——低于 30 毫秒人眼基本分辨不出差异高于 100 毫秒会有明显的一顿一顿感。class ThrottledEmitter: def __init__(self, emit_fn, interval_ms50): self.emit_fn emit_fn self.interval interval_ms / 1000 self.buffer self.last_flush time.monotonic() async def push(self, delta): self.buffer delta now time.monotonic() if now - self.last_flush self.interval: await self.emit_fn(self.buffer) self.buffer self.last_flush now async def flush(self): if self.buffer: await self.emit_fn(self.buffer) self.buffer 4. 工具调用与流式输出的阻塞点为什么会出现先卡后爆4.1 阻塞的根因串行等待工具执行回到开头那个问题。为什么工具调用一插进来流式输出就卡住了因为 runLoop 的结构是串行的必须等consume_stream完全结束拿到完整的助手消息才能知道有哪些工具要调然后才能执行工具。而工具执行是同步等待的这期间没有任何新的流式数据产生前端自然就卡住了。更细一点说卡顿发生在两个地方。第一处是模型生成工具调用参数的过程——如果参数很长比如要生成一大段代码作为参数这段时间前端只能显示正在生成工具调用没有正文输出。第二处是工具真正执行的时间——如果是网络请求或者耗时计算这段时间完全是空白的。先卡后爆里的爆指的是工具执行完之后模型基于工具结果重新生成这次生成往往很快因为上下文已经确定于是一大段内容瞬间涌出来。前后对比强烈体验就很割裂。4.2 一个真实的排查链路从现象到根因我记录一下当时排查这个问题的完整过程因为这种排查思路比结论更有价值。现象是用户问一个需要查数据库的问题前端先显示正在思考然后卡住约 3 秒接着突然刷出一大段回答。第一步我先确认卡的是哪一段。在consume_stream的入口和出口各打一个时间戳发现流式消费本身只花了 800 毫秒说明卡的不是模型生成。第二步在工具执行前后打时间戳发现工具执行花了 2.2 秒。到这里定位到卡顿主要来自工具执行。第三步但 800 毫秒的模型生成时间里前端也没显示任何东西。查了一下发现是因为工具调用的参数生成阶段没有文本 delta前端没有内容可显示只能显示 loading。第四步真正的根因是runLoop 把模型生成和工具执行完全串起来了中间没有任何并行或者提前反馈的机制。用户看到的 3 秒卡顿 800 毫秒静默生成 2.2 秒工具执行。4.3 三种缓解思路的取舍定位到根因之后我试了三种改法各有取舍。第一种是提前反馈。在模型开始生成工具调用参数时就通过回调告诉前端正在准备调用 XX 工具。这样虽然还是要等但用户知道系统在干活不是死了。改动最小收益也最直接。第二种是工具预执行。如果模型生成的工具调用参数是流式到达的可以在参数拼完整的那一刻就开始执行而不是等整个流结束。但这里有个风险如果模型后面又改了参数虽然少见预执行就白做了。所以只适合参数确定后不会变的场景。第三种是并行工具执行。如果一轮里有多个工具调用且它们之间没有依赖可以并行执行。Pi-Agent 默认是串行 for 循环改成asyncio.gather就能并行。但要注意工具结果回写到消息列表的顺序必须和工具调用的顺序一致否则模型会困惑。改法改动量收益风险提前反馈小体验提升明显几乎无工具预执行中省掉参数生成等待参数变更时白做并行执行中多工具场景提速明显顺序错乱风险我最后是三种都上了但并行执行加了顺序保证用asyncio.gather拿到结果后按原始 index 排序再回写。5. 改造 runLoop让流式输出和工具执行真正并行5.1 把 consume_stream 拆成边收边发的模式原版consume_stream是收完再返回改造的核心是把它变成边收边发。具体做法是给 consume_stream 传入一个回调每收到文本 delta 就立即回调而不是等函数返回。async def consume_stream_with_callback(self, stream, on_delta, on_tool_ready): content_buffer tool_calls {} async for chunk in stream: delta chunk.choices[0].delta if delta.content: content_buffer delta.content await on_delta(delta.content) # 立即推送 if delta.tool_calls: for tc in delta.tool_calls: idx tc.index if idx not in tool_calls: tool_calls[idx] {id: , name: , arguments: } # ... 合并逻辑同上 ... # 参数拼完整时立即通知 if is_complete_json(tool_calls[idx][arguments]): await on_tool_ready(idx, tool_calls[idx]) return build_assistant_message(content_buffer, tool_calls)这里is_complete_json是个小技巧尝试解析 arguments如果成功说明参数已经完整可以提前触发工具执行。这个判断在参数是简单结构时很准复杂嵌套结构偶尔会误判所以我在实际项目里加了个参数长度稳定超过 N 个 delta 未变化的辅助判断。5.2 工具执行结果如何插队回消息流工具执行完之后结果要作为一条 tool 消息插回消息列表。这里有个顺序问题如果多个工具并行执行结果回来的顺序可能和调用顺序不一致。而 LLM 对消息顺序是敏感的tool 消息必须紧跟在对应的 assistant 消息之后且顺序要和 tool_calls 一致。我的做法是维护一个结果槽位数组每个工具执行完就填到对应 index 的槽位里等所有槽位都填满再按顺序一次性追加到消息列表。这样既享受了并行执行的提速又保证了消息顺序正确。async def execute_tools_parallel(self, tool_calls): results [None] * len(tool_calls) async def run_one(idx, call): results[idx] await self.execute_tool(call) await asyncio.gather(*[run_one(i, c) for i, c in enumerate(tool_calls)]) return results # 顺序已保证5.3 改造后的时序对比与实测数据改造前后我做了个对比测试用一个需要调用两次工具的任务先查用户 ID再查订单记录各阶段耗时。阶段改造前改造后首次模型生成800ms800ms工具1执行1200ms1200ms二次模型生成600ms600ms工具2执行1000ms1000ms三次模型生成700ms700ms用户感知首字延迟3600ms800ms总耗时4300ms4300ms注意总耗时没变因为工具执行本身的时间省不掉。但用户感知的首字延迟从 3.6 秒降到了 0.8 秒这就是流式 提前反馈的价值。用户不需要等整个任务完成第一轮模型生成的内容就能看到。注意并行执行只在工具之间无依赖时有效。如果工具2的参数依赖工具1的结果那它们本来就得串行强行并行只会拿到错误的参数。判断依赖关系这件事目前还得靠人工标注或者让模型显式声明。6. 读源码时容易忽略的几个细节6.1 异常在流式消费中的传播路径流式响应最麻烦的地方在于异常可能发生在流的任何位置。网络断了、provider 返回错误、JSON 解析失败这些异常如果处理不好会让 runLoop 直接崩掉用户看到的是一个没有任何反馈的空白。Pi-Agent 里对异常的处理是分层的网络层异常会触发重试解析层异常会记录并跳过当前 delta业务层异常比如工具执行失败会作为 tool 消息的内容返回给模型让模型自己决定怎么处理。这个分层设计很值得借鉴——不是所有异常都该往上抛有些异常应该降级成模型能理解的信息。我实际遇到过一个 case工具执行超时原版直接把异常抛出来runLoop 中断。我改成把超时信息作为 tool 结果返回模型收到后会自动换一种方式重试或者告诉用户这个操作超时了。体验好很多。6.2 上下文长度与流式截断的关系流式响应有个隐藏问题如果上下文接近模型的长度上限模型可能在生成到一半时被截断finish_reason会是length而不是stop。这时候如果你不处理会得到一个不完整的助手消息后续逻辑可能出错。Pi-Agent 的处理是检测到length时把已生成的内容作为完整消息但标记一个truncated标志。runLoop 看到这个标志可以选择继续生成把已生成内容作为前缀再问一次或者直接结束。我建议在长任务场景下开启续写逻辑否则用户会看到回答莫名其妙断掉。6.3 多轮循环下的消息去重runLoop 每转一圈都会往消息列表里追加内容。如果不小心很容易出现重复消息。我踩过一次坑在工具执行失败重试的逻辑里把同一个 assistant 消息追加了两次结果模型看到两条一样的消息开始自我重复。避免方法很简单每次追加消息前检查消息列表最后一条是不是同一条消息用消息 ID 或者内容哈希判断。这个检查开销很小但能避免很多诡异问题。7. 如果你要自己实现一套类似的流水线读完 Pi-Agent 的 runLoop我最大的感受是这套设计的精髓不在于代码多复杂而在于它把流式这件事贯穿到了整个循环里而不是只在最外层包一层。很多自研 Agent 框架的问题就是流式只做在 LLM 调用那一层工具执行、消息回写全是同步阻塞的结果就是流式的优势被后面的阻塞吃掉了。如果你要自己实现我的建议是从第一天就把 runLoop 设计成事件驱动的每个阶段都通过回调或者事件往外抛状态而不是等整个循环结束才返回。这样无论是接前端、接日志、接监控都有统一的接入点。另外别小看 delta 合并这块的复杂度。不同 provider 的流式协议差异比想象中大我建议在适配层做充分的归一化把各种奇葩格式都转成统一的内部 delta 结构上层逻辑就干净了。Pi-Agent 在这块做得不错但适配层本身也是整个代码库里最容易出 bug 的地方改的时候要格外小心。最后分享一个我调试流式问题的小技巧把所有 delta 事件按时间戳打日志导出成一份时间线。卡顿问题在时间线上一目了然——哪段时间没有事件、哪段时间事件密集比看代码猜快得多。这个习惯帮我省了至少一半的排查时间。
开发实战：TXT朗读器实现)
