Pi Agent源码解析：执行循环、工具注册与上下文管理

先说明一下：这篇笔记是系列第二篇。第一篇我把 Pi Agent 的工程结构、模块划分和整体调用链捋了一遍，包括agent、tool、memory、context、orchestrator这几个核心包各自负责什么。当时留了个尾巴，说要把核心实现细节逐个过一遍，这篇就来填坑。

如果你是第一次接触这个框架，建议先把第一篇扫一眼，或者是直接去 GitHub 把仓库 clone 下来，边看这篇笔记边对着pi_agent的源码目录翻。没有源码在手，光看笔记会丢失很多细节，尤其是我下面要讲的这些“源码里藏着但文档里不写”的坑。

这篇笔记适合两种人：一种是正在读 Pi Agent 源码但有点摸不着头脑的，另一种是想自己写一个轻量 Agent 框架、需要参考设计思路的。不管哪种，建议你手里开着源码编辑器，遇到代码段就写个小 demo 跑一下，效果比光看强得多。

1. 核心抽象：Agent 类与执行循环

先看最核心的东西——pi_agent/agent/base.py里的Agent类。这个类承担的职责是整个框架最关键的：接收任务、调度工具、对话模型、整理结果。我拉源码时首先盯的就是它，因为只要看懂Agent.run()方法，整个框架的执行逻辑就通了七八成。

1.1 Agent 基类设计

Agent类的__init__大概是这样的：

class Agent: def __init__( self, name: str, system_prompt: str, model: BaseChatModel, tools: Optional[List[ToolEntry]] = None, memory: Optional[BaseMemory] = None, max_iterations: int = 10, temperature: float = 0.7, ): self.name = name self.system_prompt = system_prompt self.model = model self.tools = {tool.name: tool for tool in (tools or [])} self.memory = memory or ConversationBufferMemory() self.max_iterations = max_iterations self.temperature = temperature

这里有几个值得注意的设计决策。第一个是tools用了字典而不是列表，key 是工具名。这个看着不起眼，但实际是给后文“模型返回工具名 → 查表调用”的打底——模型返回的tool_call.name上来就是字符串，字典查询 O(1)，而且天然防止同名工具注册。第二个是传入的是BaseChatModel而不是直接把某个具体模型类写死，也就是说你可以在不破坏框架逻辑的前提下把模型从 GPT 换成本地部署的 Llama。这是老生常谈的依赖倒置，但源码里真正执行得彻底的框架不多，Pi Agent 算一个。

再看system_prompt。这个参数被包装过了，源码里_build_system_prompt()方法会在你传入的 system prompt 后面自动追加一段工具说明：

你是一个智能助手，可以使用以下工具完成任务： - name: 工具名 description: 工具描述 parameters: {参数名: 参数说明}

这段自动生成的提示文本不是可选的，是每次都拼的。为什么？因为 Agent 本身不知道你有没有替它把工具信息写进 system prompt，为了保证工具在模型端“可见”，框架在这一步做了一次保险。这点我觉得做得聪明：调用方只需要声明工具，不需要操心模型 API 层和 prompt 模板层的拼接，框架把这些脏活包了。

1.2 执行循环（Run Loop）源码逐行拆解

run()方法是理解 Agent 的钥匙。我直接把简化后的核心逻辑贴出来，然后逐段解释：

def run(self, task: str) -> str: messages = self.memory.get_messages() if not messages: messages = [{"role": "system", "content": self._build_system_prompt()}] messages.append({"role": "user", "content": task}) for step in range(self.max_iterations): response = self.model.chat(messages, tools=self._tool_schemas()) if response.tool_calls: messages.append(self._to_assistant_message(response)) for call in response.tool_calls: tool_result = self._execute_tool(call) messages.append( { "role": "tool", "tool_call_id": call.id, "content": tool_result, } ) continue final_answer = response.content self.memory.add_user_message(task) self.memory.add_ai_message(final_answer) return final_answer raise MaxIterationsError( f"Agent {self.name} 超过最大迭代次数 {self.max_iterations}" )

先说整体流程：用户 task 进来，先塞进消息队列，然后进入循环。循环里先调模型，看返回值里有没有工具调用。有工具调用，就执行工具，把结果回填到 messages，然后continue再调模型；没有工具调用，说明模型认为任务可以总结了，于是返回答案并写入 memory。循环上限max_iterations兜底，防止模型一直要求调工具不停下来。

看到这里你可能会问：为什么不把工具执行结果当作“最终答案”直接返回给用户，而是回填后再让模型总结一次？这个问题我在社区里见过不少人问。答案在于工具结果往往是结构化数据（JSON、状态码、日志片段），它们不是自然语言答案。比如一个搜索工具返回了 50 条 URL，你得让模型对这些结果做二次加工——挑重点、排序、组织描述——才能形成用户能看懂的答案。所以“工具执行 != 任务结束”，必须再走一轮模型调用。

另外一个细节是messages的复用方式：它没有在每轮循环里重建列表，而是直接 append。这是一个性能优化，也是一个必须严格注意的点。如果你在循环体内部错误地重建了 messages，那么前一轮工具调用的结果就丢了，极容易造成循环死转。源码里没有犯这个错，但你要是二次开发时改动了这段，千万留神。

关于MaxIterationsError，我看到有使用者吐槽这个异常不够细致。确实，只有“超过最大迭代次数”一个维度，看不出来到底是工具反复报错，还是模型死活不确定。我的排查经验是：遇到这个异常，先去查日志里最近几轮工具调用的返回内容，如果都是异常信息，那大概率是工具实现有 bug，别急着调大max_iterations。

2. 工具注册与调用链：从装饰器到 JSON Schema

工具系统是 Agent 框架里使用频率最高的子系统。Pi Agent 的工具注册走的是装饰器风格，我一开始觉得只是语法糖，深入研究后发现这里的“糖”是包了实际逻辑的。

2.1 工具协议与注册机制

先看一个最简单的工具声明：

@agent.tool(description="计算两个整数的和") def add(a: int, b: int) -> int: return a + b

agent.tool()装饰器背后做了什么？源码位于pi_agent/tool/registry.py，简化后大概是：

def tool(self, description: str = ""): def decorator(func): schema = generate_schema(func, description) entry = ToolEntry( name=func.__name__, func=func, schema=schema, ) self.tools[func.__name__] = entry return func return decorator

核心动作有两个：生成 JSON Schema 和注册。generate_schema这个函数有意思，它是靠函数签名和类型注解反推参数结构的。也就是说，你在函数里写a: int，它就能推断出a的类型是integer；写b: str，就能推断出是string。源码里用的方式是inspect.signature+ 类型注解 + 默认值推断：

def generate_schema(func, description): sig = inspect.signature(func) properties = {} required = [] for name, param in sig.parameters.items(): properties[name] = {"type": map_python_type(param.annotation)} if param.default is inspect.Parameter.empty: required.append(name) return { "type": "function", "function": { "name": func.__name__, "description": description, "parameters": { "type": "object", "properties": properties, "required": required, }, }, }

这套自动推断方案让开发者省去了手写 JSON Schema 的麻烦。但是有个坑必须提：如果函数参数没有类型注解，map_python_type会落到一个默认的"string"分支上，然后模型端就会“自由发挥”，可能传一个和预期完全不符的类型过来。我踩过这个坑，建议你的工具函数务必写全类型注解。

工具入口的__name__作为注册 key 这件事也要注意。如果同一个 Agent 下挂了两个同名的函数（比如不同模块里都写了get_data），后面注册的会静默覆盖前面注册的。源码里没有做冲突检查，这是个设计上的小缺陷，二次开发时可以加个日志。

2.2 工具调用的参数解析与结果回填

模型端发起工具调用的返回结果里，参数是一个 JSON 字符串。_execute_tool方法要做的就是从 JSON 字符串解析出参数再传给真正的函数：

def _execute_tool(self, call): tool_entry = self.tools.get(call.name) if not tool_entry: return f"错误: 工具 {call.name} 不存在" try: args = json.loads(call.arguments) except json.JSONDecodeError: # 模型偶尔会返回截断的 JSON return "错误: 工具参数不是合法 JSON" try: result = tool_entry.func(**args) except TypeError as e: return f"参数错误: {e}" return result

这一段里藏着好几个实战要点。第一，call.arguments是字符串，必须先json.loads，但模型返回的 JSON 偶尔是残缺的，特别是长参数列表时容易截断，所以JSONDecodeError要接住。第二，tool_entry.func(**args)这行会把 Python 的TypeError引出来，如果模型给的参数名不匹配，整个 Agent 循环就会断掉。源码的处理是把错误信息回传给模型，让它自己纠正，这么做的好处是至少 Agent 不会崩溃。

你要注意_execute_tool的返回值不是直接给用户看的。它会被塞进 messages 里当作role: tool的消息，然后模型会根据这个内容决定下一步。所以在设计工具返回时，尽量返回结构化、可被模型“二次引用”的文本，比如 JSON 序列化后的数据，而不是一大段渲染好的 HTML。工具可以返回 JSON 字符串，但要注意，如果返回过大，会占用上下文窗口，所以能在工具内部做好数据精简就尽量精简，比如只返回 top-10 而不是所有搜索结果。

3. 上下文管理与记忆机制：窗口不够用的日子

写 Agent 框架绕不开一个现实：大模型的上下文窗口是有限资源。Pi Agent 对这个问题的处理方案分成了两个层面：单轮内的上下文组装和跨轮次的记忆持久化。这两个层面的源码分别在context和memory包里。

3.1 上下文组装策略

继续回到run()方法中的 messages。你可能会想：messages 是不是把历史全部倒给模型？不是。Pi Agent 在context.py里有个build_context函数，它做的最重要一件事是分层：

• 系统指令层（system prompt + 工具描述）：每轮强制保留
• 任务状态层（当前的任务描述、中间变量）：每轮更新
• 动态记忆层（历史对话/过往工具结果）：根据窗口余量动态取舍

每轮循环进入模型调用前，build_context会先从 memory 里取历史消息，再拼上当前轮的 messages，最后做一个 token 估算。源码里 token 估算用的是tiktoken，如果你换成了别的模型，就得自己接入对应的 tokenizer，否则估算会偏差较大。

这个分层的思路是源码中最值得学习的设计之一。很多新手在做 Agent 时喜欢把所有内容一股脑塞进去，结果系统提示被历史对话挤到下半场，模型的“系统人格”就崩了。Pi Agent 用“分词预算”的方式避免了这个问题：系统指令永远占第一个位置，并且优先级最高。

3.2 记忆压缩与裁剪算法

跨轮次记忆是 Agent 能不能“记得住事情”的关键。Pi Agent 的默认实现是ConversationBufferMemory，就一个列表存储历史消息。但真正有意思的是memory/compressor.py里的compress_messages函数：

def compress_messages(messages, max_tokens=6000, summary_model=None): total_tokens = count_tokens(messages) if total_tokens <= max_tokens: return messages # 先裁剪大体积的工具结果 target_budget = max_tokens * 0.8 for msg in messages: if msg["role"] == "tool" and len(msg["content"]) > 500: msg["content"] = msg["content"][:500] + "...(已截断)" total_tokens = count_tokens(messages) # 再压缩最旧的对话为摘要 while total_tokens > target_budget: user_msg, ai_msg = messages.pop(0), messages.pop(0) summary = summary_model.summarize(f"{user_msg}: {ai_msg}") messages.insert(0, {"role": "system", "content": f"历史摘要: {summary}"}) total_tokens = count_tokens(messages) return messages

这里有两个反直觉但非常实用的细节。首先是裁剪优先级：工具结果优先被压缩，而不是优先删除最早的消息。原因是工具结果通常是一条巨长的 JSON，而里面大部分内容对于当前决策是冗余的——模型只需要知道“这个工具调用成功并返回了 N 条结果”就行了。相比于把几轮前的用户消息删掉，折掉工具结果对信息损失更小。

其次是“摘要替换”的操作：把最早的一组 user/assistant 消息压缩成一条 system 类型的摘要。这一手是保存长期记忆的常见策略，但位置不要搞错——把摘要插到列表头部，不要插到尾部，否则模型会把它当作当前的指令来理解，导致行为异常。

compress_messages没在每次run()里自动调用，而是暴露出来给 memory 对象在存储的时候自行选择是否调用。默认的ConversationBufferMemory不调用它，只有当你显式使用SummaryMemory时才会触发。老实说这个设计我一开始觉得有点折腾，但后来想通了：压缩是有代价的，每一次摘要都意味着细节丢失，框架把选择权交给了使用者，而不是一刀切。

另外，Pi Agent 的 memory 还有个window_size参数，用来控制单轮 messages 里保留多少条历史。说句实在话，window_size按“条数”切而不是按“token 数”切，精准度一般，但在很多本地部署的场景里够用了。如果你要做生产级记忆管理，建议在此基础上再叠加一层更细粒度的 token 预估，否则容易被对话轮次很多但每轮都挺短的场景欺骗。

4. 多 Agent 协作与任务编排：Orchestrator 的实现细节

单 Agent 能做简单任务，但稍微复杂一点的场景——比如“生成代码然后跑测试再修订”这种闭环——就需要多个角色协同。Pi Agent 在这块提供了Orchestrator，我在源码里看它的实现时觉得挺有启发。

4.1 角色定义与通信机制

多 Agent 不是把一堆 Agent 类塞进一个列表就行。Pi Agent 在orchestrator.py中定义了一个AgentMessage数据结构：

@dataclass class AgentMessage: sender: str receiver: str content: str metadata: dict = field(default_factory=dict) created_at: float = field(default_factory=time.time)

为什么要引入一个消息对象而不是直接函数调用？两个原因。第一是解耦：消息只通过“发件人/收件人”传递，每个 Agent 不知道对方内部怎么实现的；第二是可观测：所有 Agent 之间交换的信息都会经过 Orchestrator 的消息队列，你可以把完整通信链路打到日志里，哪里出了问题一目了然。我在实际调试中全靠这个日志，否则多 Agent 互相调用时你根本不知道谁在给谁传空消息。

Orchestrator 的核心调度可以简化成：

def run(self, task: str) -> str: first_agent = self.schedule_agent(None, task) result = self.agents[first_agent].run(task) for _ in range(self.max_turns): if not self.messages_out: break while self.messages_out: msg = self.messages_out.popleft() receiver = self.agents[msg.receiver] result = receiver.run(msg.content) self.results[msg.receiver] = result return result

注意schedule_agent的选路逻辑。默认实现是硬编码：第一个任务总是发给列表里的第一个 Agent，后续消息根据receiver路由。如果你想做更动态的调度，比如根据任务类型分发给不同 Agent，可以覆写schedule_agent。这个类是源码里少数几个有意“留白”给开发者扩展的地方，其他大部分模块都偏“开箱即用”。

4.2 任务分配与结果汇总的实现细节

让我用一个具体例子来说明多 Agent 编排的执行过程。假设你有一个PlannerAgent、一个CoderAgent、一个TesterAgent，任务要求“写一个计算阶乘的函数并验证它”。

Orchestrator 的流程是：

1. 把“写一个计算阶乘的函数并验证它”派发给PlannerAgent。
2. PlannerAgent返回一个拆解结果：“步骤一：实现阶乘；步骤二：编写测试用例；步骤三：运行测试”。
3. PlannerAgent产生一条AgentMessage，发给CoderAgent，内容是第一步的细化指令。
4. CoderAgent执行生成代码，返回结果；同时继续把任务推给TesterAgent。
5. TesterAgent执行测试，返回通过/失败的结论。
6. 如果测试通过，Orchestrator 直接把最终结果返回；如果失败，TesterAgent会把失败信息推回给CoderAgent，进入下一轮修订。

这个链条里的关键是消息的同步性。Orchestrator 的 while 循环是同步的——每次popleft一条消息就停下来等对应 Agent 的结果，结果回来再继续处理队列下一条。这是源码中默认的行为，单线程、容易调试，但吞吐量低。如果要并行，你得对接消息队列系统（比如 Redis Stream），这在源码里留了接口但没实现。

多 Agent 场景最容易出的问题是“流程不终止”。如果max_turns设得过大或者循环条件没写对，Agent 们会无限地互相发消息。源码用max_turns兜底，默认值 30。但如果两个 Agent 都在“修订→再评审→再修订”的循环里出不来，30 轮也会打满。我的经验是把max_turns调小一点（比如 8），宁可过早终止让用户重新发起，也不要让任务卡死在后台烧 token。

另外一个经验是给每个 Agent 设置不同的system_prompt时务必要写明“什么时候该结束并把控制权交回 Orchestrator”。这个写清楚，比在 Orchestrator 里做各种终止条件判断都有效。框架文档里没提这一点，但实际用过一轮的人应该都会有感觉。

5. 常见问题与排查技巧实录

源码看完了，实操中肯定会碰到问题。我把这几周跑 Pi Agent 时踩过的坑集中整理一下，每一条都是在真实场景里撞见、并且从源码层面定位原因之后才敢写的。

排查技巧上，我最常用的是在run()方法里临时加一行print(messages)，把每轮循环的 messages 打出来看，能直接看到模型返回了什么、工具执行结果是什么、下一轮模型是否拿到了正确的回填。这个方法虽然原始，但比任何 debugger 都直观，尤其是参数多的时候。

另外一个技巧是“最小化测试工具”。当工具出问题时，先写一个不带任何工具调用的纯对话任务，确认模型本身通畅；再挂一个只做return 1的桩工具，确认工具调用链通；然后再逐步加载真实工具。这能帮你迅速把“模型问题”和“工具问题”区分开。我每次给 Pi Agent 接入新工具都用这套流程，基本十分钟内就能定位问题。

我个人的体会是：读 Pi Agent 源码最大的收获不是学会了某个具体调用，而是理解了一个 Agent 框架在“上下文管理、工具调用、多角色协同”这三件事上是如何做权衡的。不同框架的取舍不同，但 Pi Agent 这套实现足够轻、足够清晰，适合当作研读 Agent 框架的入门蓝本。

最后分享一个扩展思路：如果你想把 Pi Agent 用在自己的项目里，不要只把它当黑盒调，试着在Agent.run()的循环体里加一个钩子函数——比如每轮调完模型后把 messages 快照写入本地——这样你就能沉淀出你自己的 Agent 行为数据集，后续做 prompt 调优、成本分析都有数据可依。这步钩子接口目前在框架里没有正式暴露，但run()方法内部加一行代码就可以实现，改动成本很低。