1. 从一次线上事故说起为什么要啃 runLoop 这条消息流水线去年冬天我负责的一个智能体项目在灰度阶段翻过一次车。现象很诡异用户在前端看到回复只吐了半句话就卡住了后端日志里 LLM 的流式响应明明还在持续返回 token但 Agent 侧像是聋了一样后续的 chunk 全部石沉大海。排查了整整一个下午最后定位到问题出在消息流水线的调度环节——runLoop在处理完一次工具调用之后没有正确地把控制权交还给流式响应的消费协程导致后续的 SSE 数据被缓冲区吞掉。那次事故之后我把 Pi-Agent 的runLoop相关源码从头到尾读了一遍越读越觉得这套设计有意思。它不像很多玩具级 Agent 框架那样把接收消息—调用模型—返回结果写成一条直线而是把整个交互过程抽象成了一条消息流水线用事件驱动的方式把用户输入、LLM 流式响应、工具调用、状态回写这几个环节串起来。理解这条流水线基本上就理解了 Pi-Agent 这个框架的骨架。这篇内容适合三类人看一是正在用 Pi-Agent 做二次开发、被runLoop里各种回调绕晕的工程师二是想自己从零写一个 Agent 框架、需要参考成熟调度设计的开发者三是单纯好奇大模型流式响应到底是怎么被 Agent 消费掉的技术爱好者。我会尽量把源码里的关键路径拆开讲配上我实际调试时的观察和踩过的坑让你读完能自己动手改这条流水线。需要先说明一点下面涉及的具体函数名、字段名以我手上这份源码为准不同版本可能有细微差异但核心思路是一致的。如果你手上的版本对不上重点看设计意图别死抠命名。2. 消息流水线的整体设计为什么不用一条直线搞定2.1 直线式 Agent 的三大死穴先说说为什么不能简单粗暴地写。最朴素的 Agent 实现大概长这样收到用户消息拼 prompt调 LLM等完整响应回来解析出工具调用执行工具把结果再拼进 prompt再调一次 LLM循环直到没有工具调用为止。这条直线在 demo 阶段跑得挺欢一上生产就露馅。第一个死穴是流式响应没法用。直线式实现天然是等完整响应的思维但用户要的是打字机效果token 得边生成边推。你如果在直线里硬塞流式代码会变成回调地狱可读性直接崩盘。第二个死穴是工具调用和流式输出会打架。LLM 在流式返回的过程中可能前半段是普通文本后半段突然冒出个工具调用请求。这时候你是先把文本推给用户还是先等工具执行完直线式实现很难优雅处理这种边说话边干活的场景。第三个死穴是中断和取消。用户发了消息又反悔想取消或者工具执行超时要中断整个流程直线式代码里到处是if cancelled的判断维护成本极高。Pi-Agent 的runLoop本质上是用事件循环 消息队列的模式把上面这些问题一次性解决掉。它不追求一步到位而是把整个交互拆成一个个离散的消息事件让每个环节只关心自己该处理的那类事件。2.2 runLoop 的核心抽象AgentMessage 是血液整条流水线里流动的血液就是AgentMessage。你可以把它理解成一个信封信封上写着这封信是给谁的、什么类型、里面装了什么。源码里这个结构通常包含几个关键字段类型标识是用户输入、LLM 文本增量、工具调用请求还是工具执行结果、载荷数据、以及一些元信息比如时间戳和会话 ID。我特别喜欢这个设计的一点是它把LLM 流式返回的一个 token和一次完整的工具调用结果统一成了同一种东西——都是消息。这样一来runLoop的主循环就变得极其简洁从队列里取消息根据类型分发到对应的处理器处理器可能产生新消息再塞回队列。整个系统就是一个不断吞吐消息的泵。提示如果你在调试时发现消息丢了第一件事是检查消息队列的消费端是不是有异常吞掉了消息。我踩过的坑里有一半是处理器内部抛了异常但被 catch 后静默处理导致消息凭空消失。2.3 事件驱动相比轮询的取舍有人会问为什么不用轮询比如每隔 50ms 检查一下 LLM 有没有新 token。答案是延迟和资源。轮询在流式场景下要么延迟高间隔大要么空转浪费 CPU间隔小。事件驱动是有数据才动LLM 每吐一个 token 就触发一次事件天然契合流式响应的节奏。代价是复杂度上去了。事件驱动要求你对状态管理非常小心因为处理器的执行顺序不再是你写代码的顺序而是事件到达的顺序。这也是为什么很多人读runLoop源码会觉得绕——它绕的不是逻辑是控制流。3. 核心细节拆解runLoop 主循环到底在转什么3.1 主循环的骨架结构把runLoop的主循环抽象出来大概是这么个结构伪代码帮你理解流程async def run_loop(self, session): while not session.is_done: message await self.message_queue.get() handler self.dispatcher.get_handler(message.type) if handler is None: continue try: new_messages await handler(message, session) for msg in new_messages: await self.message_queue.put(msg) except Exception as e: await self.handle_error(e, session)看着简单但魔鬼全在细节里。session.is_done什么时候置位handler返回的消息会不会造成死循环异常处理里要不要把错误也变成一条消息回灌进队列这些才是真正决定框架好不好用的地方。我实测下来Pi-Agent 在is_done的判断上做得比较克制只有当没有待处理的工具调用且LLM 已经明确结束输出且消息队列为空这三个条件同时满足时才会结束循环。这个三条件判断很关键少任何一个都会导致提前退出或死循环。3.2 消息类型与分发策略流水线里跑的消息类型按我的理解大致分这么几类用表格列出来更清楚消息类型来源典型载荷处理后的去向UserInput外部调用方用户文本触发 LLM 请求LLMDeltaLLM 流式接口单个 token 或片段推送给前端 累积到缓冲区LLMCallRequestLLM 解析器工具名 参数触发工具执行ToolResult工具执行器执行输出回灌给 LLM 作为上下文ErrorSignal任意环节异常信息决定重试或终止分发策略上Pi-Agent 用的是类型到处理器的映射表而不是一长串if-elif。这个选择在消息类型少的时候看不出优势但一旦你要扩展自定义消息类型比如加个人工介入消息映射表方案只需要注册一个新处理器不用动主循环。这是典型的开闭原则落地。3.3 流式响应的消费时机这是整条流水线里最容易出 bug 的地方。LLM 的流式响应是一个异步迭代器你async for它的时候每个 chunk 到达的时机是不确定的。Pi-Agent 的做法是在独立的协程里消费流式响应每拿到一个 chunk 就包装成 LLMDelta 消息塞进队列而不是在主循环里直接await整个流。为什么要这样因为如果主循环直接 await 流式响应那么在流式返回期间主循环是被占住的没法处理其他消息比如用户的中断请求。独立协程消费 消息队列解耦让主循环始终保持可响应状态。注意独立协程消费流式响应时一定要处理好协程的生命周期。我见过因为流式协程没被正确取消导致连接泄漏、内存缓慢增长的案例。取消逻辑要写在finally块里别偷懒。3.4 缓冲区与增量拼接LLMDelta 消息到达后不能直接推给前端就完事还得在本地维护一个累积缓冲区。原因有两个一是工具调用的参数可能是分多个 chunk 返回的你得拼完整了才能解析二是有些框架需要在流式结束后对完整文本做后处理比如敏感词过滤、格式修正。缓冲区的实现有个细节值得说Pi-Agent 用的是按消息类型分桶的缓冲区文本增量进文本桶工具调用增量进工具桶。这样拼接逻辑互不干扰也方便在流式结束后分别处理。如果所有增量都塞一个字符串里解析工具调用时就得做复杂的字符串切分很容易出错。4. 实操过程手把手跟一遍消息的完整生命周期4.1 从用户输入到 LLM 请求发出假设用户在对话框里敲了帮我查一下明天北京的天气点发送。这条文本首先被包装成 UserInput 消息塞进消息队列。主循环取到它分发给 UserInput 处理器。处理器做的事情把用户消息追加到会话历史然后构造 LLM 请求。构造请求时有个关键决策——要不要带上工具定义。如果这个会话注册了天气查询工具请求里就得带上工具 schema否则 LLM 不知道有这个工具可用。Pi-Agent 在这里会根据会话配置动态决定工具列表而不是写死。请求构造完不是直接同步调用而是启动一个流式消费协程然后把LLM 请求已发出这个状态记到 session 里。主循环继续转等待后续的 LLMDelta 消息。4.2 流式 token 的接收与推送流式消费协程开始工作LLM 每返回一个 chunk就做三件事包装成 LLMDelta 消息入队、更新 session 的累积缓冲区、记录最后活跃时间用于超时判断。主循环取到 LLMDelta分发给对应处理器。处理器把增量推给前端通过 WebSocket 或 SSE同时检查缓冲区里是否已经攒够了一个完整的工具调用。这里有个经验值工具调用的 JSON 参数通常需要攒到出现完整的闭合括号才能解析所以处理器会做一个轻量的括号配对检查而不是每个 chunk 都尝试解析。我实测下来这个攒够再解析的策略比每个 chunk 都试解析要稳得多。后者在参数被切分到多个 chunk 时会产生大量解析失败的噪音日志排查问题时非常干扰视线。4.3 工具调用的触发与执行当缓冲区检测到一个完整的工具调用处理器会生成一条 LLMCallRequest 消息。主循环取到它分发给工具执行处理器。工具执行处理器做几件事校验工具名是否已注册、校验参数是否符合 schema、执行工具、把结果包装成 ToolResult 消息。这里有个容易忽略的点——工具执行应该是异步的且要有超时控制。我见过工具因为网络问题卡死整个 Agent 会话跟着挂掉的案例。Pi-Agent 在工具执行外层包了超时超时后生成一条 ErrorSignal 消息让流水线决定是重试还是告知用户。工具执行完ToolResult 消息入队。主循环取到它把它追加到会话历史然后触发新一轮 LLM 请求——因为 LLM 需要基于工具结果继续生成回复。这就形成了一个闭环LLM 请求 → 流式响应 → 工具调用 → 工具结果 → 再请求 LLM。4.4 会话结束的判定什么时候算结束前面提过三条件没有待处理工具调用、LLM 明确结束、队列为空。实际实现里LLM 结束的标志通常是流式响应里出现了特定的结束标记比如 finish_reason 字段。这里有个坑有些 LLM 服务在流式结束时最后一个 chunk 只带 finish_reason 不带内容有些则相反。Pi-Agent 的处理是两种都兼容——只要收到 finish_reason 就标记 LLM 结束内容为空也不影响。如果你自己实现千万别假设最后一个 chunk 一定有内容。5. 常见问题与排查技巧实录5.1 流式响应卡住的排查路径这是最高频的问题。用户反馈回复到一半不动了排查顺序建议这样走先看后端日志里 LLMDelta 消息还在不在持续入队。如果入队正常但前端没更新问题在推送环节检查 WebSocket 连接状态和推送协程是否还活着。如果入队也停了问题在流式消费协程检查它是不是被异常中断了或者 LLM 服务端是不是提前关闭了连接。我遇到过一次特别隐蔽的流式消费协程因为一个未捕获的解析异常退出了但异常被上层 catch 后只记了 debug 级别日志生产环境日志级别是 info所以什么都没看到。后来把流式协程的异常处理改成必须记 error 级别问题才暴露出来。5.2 工具调用参数解析失败的常见原因参数解析失败八成是缓冲区拼接逻辑有问题。常见原因有这么几个一是增量拼接时没考虑多字节字符被切分中文尤其容易中招二是工具调用的 JSON 里嵌套了字符串字符串里又有括号简单的括号配对会误判三是 LLM 返回的参数格式本身就不合法。针对第一个拼接时要用字节流或确保按字符边界切分。针对第二个别自己写括号配对用流式 JSON 解析器。针对第三个解析失败时把原始缓冲区内容完整记下来方便事后分析是 LLM 的问题还是解析的问题。5.3 消息队列积压与死循环消息队列积压通常意味着某个处理器处理速度跟不上生产速度。比如工具执行特别慢而 LLM 还在源源不断产生新消息。这时候要么给队列设上限并做背压要么把慢处理器改成异步不阻塞主循环。死循环则是另一种情况处理器产生的消息又触发了自己。比如错误处理里生成 ErrorSignalErrorSignal 处理器又因为处理失败生成新的 ErrorSignal。防死循环的通用做法是给消息加个重试次数字段超过阈值就强制终止并上报。5.4 常见问题速查表现象可能原因排查动作回复中途卡住流式协程异常退出检查协程异常日志级别工具参数解析失败缓冲区拼接错误打印原始缓冲区内容消息队列持续增长处理器阻塞主循环检查处理器是否含同步 IO会话无法结束is_done 条件不满足打印三个条件的实时状态重复回复消息被重复消费检查队列的 ack 机制提示调试消息流水线时强烈建议加一个消息追踪 ID每条消息从产生到消费全程带着这个 ID日志里一搜就能看到完整链路。这个投入在排查复杂问题时回报极高。6. 我对这套流水线设计的一点个人看法读完 Pi-Agent 的runLoop我最大的感受是Agent 框架的难点从来不在调模型而在调度。调模型谁都会但怎么让流式响应、工具调用、中断取消、错误恢复这几件事和谐共处才是真正考验设计功力的地方。Pi-Agent 用消息流水线 事件驱动的方案把复杂度从控制流转移到了状态管理。这个转移是值得的因为状态可以观测、可以持久化、可以回放而混乱的控制流只能靠读代码去理解。如果你正在设计自己的 Agent 框架我建议认真考虑这套模式哪怕不照搬至少把消息作为一等公民这个理念吸收进去。后续如果还有机会我想再写写 Pi-Agent 里工具注册和 schema 校验那部分的实现那块也有不少值得琢磨的细节。