在上一篇笔记里，我把 Pi Agent 的整体架构和数据流梳理了一遍，从启动入口到 Session 管理、从 LLM 调用封装到配置加载，算是给这个框架画了一张全景图。但这段时间把源码翻得更细之后发现，只看整体架构是真的不够——真正决定一个 Agent 框架好不好用、能不能二次开发的，往往是那些藏在一层层封装底下的核心机制：主循环怎么调度、工具调用怎么从文本变成真实函数执行、上下文怎么在 token 预算内做取舍。
所以这篇"二"打算换个读法，不再追求覆盖面，而是挑几条最关键的代码路径往深了挖。我尽量把每一段的关键实现、设计动机和我个人的理解都写清楚，也会贴上必要的源码节选。如果你也在做 Agent 框架选型，或者自己手写 Agent 应用，这篇文章应该能给你不少可以直接抄的灵感。
1. 主控循环的执行时序：PiAgent 的"感知—规划—行动"到底怎么转起来的
1.1 一个很不起眼的while循环，其实是整个框架的心脏
我一开始读 PiAgent 源码时，最先找的就是AgentLoop或者类似名字的类。找到之后说实话有点失望——核心主循环在代码层面看起来特别简单，甚至就是一个普通的while循环加一个for遍历：
class PiAgentSession: def __init__(self, agent_config: AgentConfig, ...): self.messages: list[Message] = [] self.tool_registry = ToolRegistry() self.context_manager = ContextManager(agent_config.context) self.max_iterations = agent_config.max_iterations self.state = SessionState.READY async def run(self, user_input: str) -> AgentResult: self.messages.append(Message(role="user", content=user_input)) while self.state not in SessionState.TERMINAL: self.state = await self.step() return self.build_result() async def step(self) -> SessionState: # 1. 组织一个 view 给模型，带上上下文管理后的消息 view_messages = self.context_manager.build_view(self.messages) # 2. 调用 LLM，拿到文本回复 + 可能的 tool_calls response = await self.llm.chat_completion( messages=view_messages, tools=self.tool_registry.schemas, ) # 3. 处理响应：追加 assistant 消息，分发 tool_calls self.messages.append(self._to_message(response)) if response.tool_calls: await self._execute_tool_calls(response.tool_calls) return SessionState.RUNNING return SessionState.COMPLETED别看这段代码短，它其实就是所有 Agent 框架的核心共同点——一个"请求模型 → 模型返回文本或工具调用 → 执行工具 → 把结果喂回模型 → 再来一轮"的闭环。PiAgent 的巧妙之处在于，它把这个闭环抽成了有限状态机，而不是简单地在循环里写死各种分支。SessionState枚举了READY、RUNNING、COMPLETED、FAILED、CANCELLED等状态，每个状态决定了下一步做什么。
1.2 为什么是状态机而不是一堆if：可观测性和断点恢复都靠它
如果你只是写一个单次的调用，用状态机确实显得杀鸡用牛刀。但 PiAgent 是一个要跑长任务、要支持多轮交互、要能随时中断恢复的框架，状态机带来的收益就非常明显了。
举个例子，我在读源码时注意到每次step()返回后，框架会检查CANCELLED状态。这个看起来很小的设计，让上层应用（比如 Web 界面或 CLI 客户端）可以安全地停止一个正在执行的任务。如果像很多 demo 那样把循环写死在函数内部，外部要打断任务就只能杀进程，状态机则给了框架一个优雅停机的入口。
另一个值得注意的设计是：工具调用的结果也是作为一种Message追加进对话历史的，而不是单独存一份。这意味着模型在下一轮可以看到结构化工具结果，并且这些结果会进入上下文管理器的视线范围。源码里面对消息的拼接顺序非常讲究——永远保持user → assistant → tool → assistant → tool → ...的交替顺序，这是为了保证 LLM 在续写时能正确理解对话结构。
我之前在自己写的 Agent 应用里吃过亏：工具结果不塞回对话，只在内存里临时存，结果多轮之后模型完全"忘记"工具给出过什么数据。这个教训在 PiAgent 源码里其实是被明确规避的。
1.3 主循环的钩子位置：二次开发真正要关注的地方
主循环看着短，但它留了一排"钉子"给外部扩展。源码里有一个比较容易忽略的地方：
async def run(self, user_input: str) -> AgentResult: await self.event_bus.emit("session.start", session_id=self.id, input=user_input) ...event_bus是全局的事件总线，从会话开始、每次迭代开始、工具执行前、工具执行后、会话结束，都有对应的事件可以监听。这个设计我在读其他 Agent 框架时很少见到做得这么统一的。它让"日志埋点""限流控制""权限校验""指标上报"这些东西完全不用侵入主循环逻辑，只要往 EventBus 上挂 handler 就行。
如果要在 PiAgent 上做二次开发，我强烈建议先把主循环里这几个emit事件的位置背下来，它们是你所有扩展能力的锚点。后面我会专门用一节来讲这些事件钩子怎么用。
2. 工具注册表与函数调用封装：模型吐 JSON，凭什么能安全变成函数执行
2.1 从模型输出到函数执行的完整映射过程
Agent 框架里最神奇也最危险的一环，就是工具调用。LLM 本质上是一个文本生成模型，它输出的是字符串；但工具执行需要的是真实的函数调用。PiAgent 把这一段的逻辑拆成了三个层次：定义层、校验层、执行层。
我先贴一段工具定义的核心代码（已简化）：
@dataclass class ToolSpec: name: str description: str input_schema: dict # JSON Schema handler: Callable[..., Any] is_async: bool = False class ToolRegistry: def __init__(self): self._tools: dict[str, ToolSpec] = {} def register(self, spec: ToolSpec): if spec.name in self._tools: raise DuplicateToolError(...) self._tools[spec.name] = spec @property def schemas(self): return [{ "type": "function", "function": { "name": spec.name, "description": spec.description, "parameters": spec.input_schema, } } for spec in self._tools.values()]当模型返回tool_calls时，里面会包含function.name和function.arguments（通常是一个 JSON 格式的字符串）。框架要做的事情是：
- 用
name在ToolRegistry里找到对应的ToolSpec。 - 把
arguments用json.loads解析成 Python dict。 - 用
input_schema（JSON Schema）对参数做严格校验。 - 根据
is_async决定同步执行还是丢进事件循环。 - 把执行结果（或者异常信息）序列化成
tool角色的消息，追加回对话历史。
2.2 参数校验为何要用 JSON Schema：宁可慢一点，不能错一次
我在看教程时经常见到有人直接json.loads(arguments)然后handler(**data)，图省事。但 PiAgent 在参数校验上是砸了功夫的，它用jsonschema库做完整校验，而不是自己手写if isinstance。
原因是工具函数往往有严格的输入要求。比如一个查天气工具要求city是字符串且不能为空，如果模型偶尔抽风传了个数字，直接拿去调用可能跑出莫名其妙的结果。更糟的是某些工具会操作文件系统、网络或数据库，一旦参数错了，影响面可能不只是 BUG，而是事故。
JSON Schema 的核心价值是：声明式地描述"这个工具接受什么、拒绝什么"，而不是把校验逻辑散落在各个函数里。源码中一个典型的工具定义长这样：
class WeatherInput(BaseModel): city: str = Field(description="城市名称，例如：杭州") days: int = Field(default=1, ge=1, le=7, description="预报天数") def get_weather(params: WeatherInput) -> dict: # 实际实现省略 return {"city": params.city, "days": params.days, "temp": 28}你注意到没有，这里其实用了 Pydantic 的BaseModel做参数模型。PiAgent 的工具定义推荐做法是"Pydantic 负责参数结构和基础类型校验 + JSON Schema 负责复杂规则校验"，两者搭配，既保证了 IDE 自动补全和类型提示，又保证了运行时校验的完备性。这个组合是我见过的最舒服的参数处理方式。
2.3 执行错误不会炸死主循环：异常包装成"工具结果"
工具执行过程中大概率会出错，这个错如果直接抛出去，整个 Agent 循环就断了。PiAgent 的处理方式非常值得学习：它把异常捕获之后，包装成一条名为"工具执行失败"的消息，作为tool角色的结果塞回对话历史。
async def _execute_tool_calls(self, tool_calls): for call in tool_calls: try: result = await self._safe_invoke(call.function.name, call.function.arguments) except Exception as exc: result = {"error": f"{type(exc).__name__}: {exc}"} self.messages.append(Message( role="tool", tool_call_id=call.id, content=json.dumps(result, ensure_ascii=False) ))这个设计的高明之处在于：当模型看到错误消息后，它可以自主决定要不要换个参数重新调用、或者直接告诉用户"这个任务目前做不了"。模型成了错误处理的决策者，而不是让框架背锅。实测中，这种"把错误反馈给模型"的模式，比我们开发者自己拼命写各种 fallback 更灵活——模型甚至能做跨工具的补偿方案，比如天气接口挂了，它直接改成给你返回一个基于经验数据的参考值。
2.4 工具注册的权限签名与隔离机制
读源码到ToolRegistry内部时，我还注意到一个细节：工具会通过ToolPolicy打标签，比如TOOL_READONLY、TOOL_WRITE、TOOL_NETWORK。主循环在分发工具调用前会检查当前会话的策略限制，如果用户只授权了只读权限，那写操作工具会被直接拒绝。
这种权限设计在单机个人 Agent 上可能用不到，但一旦你把 PiAgent 接进团队协作场景——有人只该查数据、有人该能改配置——就变得非常重要。Authorization 逻辑被放在"工具调用前"这个时机，而不是放在函数内部，让开发者不需要在每个工具里写权限判断，安全策略可以统一收敛。
3. Token 预算与上下文管理：记忆的取舍不是玄学，是代码里的硬逻辑
3.1 为什么不能把全部对话历史一股脑丢给模型
新手写 Agent 时最常见的做法是把messages原封不动发给 LLM，多轮跑几次之后突然发现开始报"请求超出 token 限制"，然后就去开大上下文窗口。PiAgent 的源码告诉我，这个方向从一开始就错了。
源码里的ContextManager负责把内部的messages转换成真正发往模型的view_messages。它内部有一个非常清晰的层级化管理：
- 常驻系统提示：永远保留，不参与裁剪
- 工具定义：按需加载，使用频率低的工具会自动从请求里摘掉
- 近期对话：全部保留，这是模型最需要的短期记忆
- 早期对话：如果太长，会被摘要化或直接丢弃
- 工具结果：只保留最近几轮，详细原始结果存在磁盘或 KV 缓存里，需要用的时候再取回
这个策略如果你抽象一下，其实就是计算机体系结构里经典的"多级存储"思想。上下文窗口是内存，短期对话是 L1 缓存，早期对话的摘要走 L2，磁盘存储是外存。框架并不做"一刀切"，而是配合预算阈值动态调整。
3.2 源码里的 token 预算算法：先定总量，再分区块
我在读ContextManager.build_view()的实现时，看到了一个很舒服的分区块分配思路：
def build_view(self, messages: list[Message]) -> list[Message]: budget = self.total_budget # 从配置中读取，例如 8000 tokens # 各区块的配额：系统提示 + 工具 schema 固定占用一部分 system_block = self._build_system_block() # ~800 tokens tool_block = self._build_tool_block() # ~1200 tokens reserve = 1000 # 给输出留的余量 available = budget - system_block.estimate() - tool_block.estimate() - reserve # 按时间倒序取消息，直到塞满 available ...这段代码的执行意图很简单：先扣除"固定开销"，再留一份"输出余量"，剩下的空间才给对话历史用。源码在裁剪时是按照时间倒序保留最近消息的，也就是保留最新、最相关的对话，把老消息逐步挤出窗口。
我特别想强调那个reserve = 1000，这是给模型生成回复预留的空间。如果你把输入塞得太满，模型会因为输出空间不足而提前截断，甚至直接报错。这个道理跟操作系统给内核留保留内存一样，看起来是浪费，实际上是为了整体系统的稳定性。
3.3 三种上下文处理策略的触发条件对比
PiAgent 的上下文管理并不只有"裁剪"一种手段。读完整段代码，我整理了一张内部策略表，分享出来给你们参考：
| 策略 | 触发条件 | 具体动作 | 优缺点 |
|---|---|---|---|
TruncationPolicy | 对话总数超过窗口上限 | 丢弃最老的非关键消息 | 实现简单、响应快，但会丢失早期重要信息 |
SummarizePolicy | 对话超过窗口上限且存在可总结片段 | 把一段早期对话交给 LLM 压缩成摘要 | 信息密度高，但增加一次额外 LLM 调用，有延迟和成本 |
GroundingPolicy | 被丢弃的消息涉及工具调用 | 将关键工具结果/承诺写入 MVCC 缓存或向量库 | 防止模型"失忆"，但需要额外的检索逻辑 |
这三个策略在框架中不是互斥的，而是按优先级融合。默认情况下，先尝试SummarizePolicy，如果用户配置了禁用摘要（避免额外耗 token），就 fallback 到TruncationPolicy。而GroundingPolicy更像是一张安全网，它保证被裁剪掉的工具结果还能通过"潜伏在系统提示里的引用 ID"找回。
3.4 一个具体场景：长任务对话中"遗忘"如何被避免
看完这些策略，我还专门模拟了一个场景来验证理解对不对——假设你让 PiAgent 帮你调研一个主题，它先调用了搜索工具拿到一批结果，然后又聊了很多轮。正常情况下，那批搜索结果早就被裁剪掉了，但 PiAgent 会在摘要里显式写入类似这样的一段：
[information_retrieval_tool / retrieval_01]: 返回了 5 条与"XXX"相关的资料索引， 如需具体内容，可调用 lookup_reference(id=...)等于说，被裁剪的是"内容本体"，但留下了"线索指针"。模型在后续轮次如果想起有这回事，可以主动去调工具拉回完整内容。这个"指针代替内容"的思路，在很长上下文场景下特别有效，比如说企业知识库问答，你不会希望把几十篇文档全部塞进窗口，而是让 Agent 在需要时按需去取。
4. 一次真实调用的完整生命周期：输入、路由、执行、回填的四级链路
4.1 以"帮我记一条备忘"为例串一遍代码路径
光看静态代码容易走神，我建议所有读源码的人都要做一件事：挑一个最小的真实任务，从入口到出口完整地追一遍代码路径。我自己拿"帮我记一条备忘录"这个任务做了实验，在这里把链路拆给你们看。
- 入口层：用户输入文本进来后，先由一个
InputParser判断是直接指令（比如/memo 买牛奶）还是自然语言请求。如果是自然语言，就需要进入 Agent 循环让模型来理解意图。 - 路由层：Session 的
run()方法拿到用户消息，开始主循环。第一轮 LLM 调用后，模型返回了tool_calls: [{name: "memo.add", arguments: "{\"text\":\"买牛奶\"}"}]。 - 执行层：ToolRegistry 根据
memo.add找到对应 handler，参数校验后实际写入本地 SQLite 数据库。工具真实返回{"id": 42, "created": "..."}。 - 回填层：工具结果作为
role="tool"的消息追加到messages。下一轮 LLM 看到结果后，输出"已经帮你记好了"的确认回复。 - 收尾层：Session 状态从
RUNNING变成COMPLETED，整个调用链结束。
这四层听着机械，但每一层都有 PiAgent 自己的处理细节，下面展开讲。
4.2 异步任务队列：为什么多个工具不能一拥而上
我在源码里发现，虽然step()是在async循环里跑的，但工具执行却被放到了一个TaskQueue中，一次只跑一个。这个设计我一开始没看懂——这不浪费并发能力吗？直到我看了一段注释，才明白作者的顾虑：多个工具同时执行时，共享状态的读写顺序是不可预测的。
比如一个 Agent 同时在跑"查天气"和"记备忘"，如果它们并发执行，数据库写入顺序可能出错；更严重的是，某些工具会修改上下文缓存，并发修改可能导致下一次模型调用读取到脏数据。PiAgent 用串行化来换取确定性，下单简单粗暴，但保证了每条工具结果都能有一个清晰的因果关系。
如果你在自己做 Agent，并且你的工具都是无状态的（比如纯计算、只读查询），那可以放开并发限制来提速。但只要你有一个工具涉及写操作，我就建议学习 PiAgent 这种"写操作串行、读操作并行"的折中方案。
4.3 失败重试与格式纠错：模型吐坏了 JSON，框架怎么兜底
LLM 生成工具调用参数时，偶尔会生成非法 JSON——可能是多了一个逗号，也可能是把 key 的引号弄丢了。PiAgent 针对这种情况不是直接报错，而是实现了一个"一次自我纠正"的流程：
try: args = json.loads(call.function.arguments) except json.JSONDecodeError as e: # 第一次失败：把错误提示塞给模型，给一次重试机会 self.messages.append(Message( role="tool", tool_call_id=call.id, content=f"参数解析失败: {e}，请重新生成合法的JSON" )) # 通过一个特殊标志让主循环继续走一步 return SessionState.NEEDS_RETRY这种兜底逻辑在外面看起来是"多了一次模型调用"，但在实践中非常值：它避免了整个 Agent 任务因为一个小错误直接 FAILED。模型的 JSON 生成能力现在虽然很强，但在复杂 schema、冷门格式要求下依然会翻车，这个自纠错机制等于给系统加了一个安全带。
我实测过，一个 10 轮左右的中等任务，有大约 5% 的概率会出现一次工具参数 JSON 生成错误。如果没有自纠错机制，这 5% 的任务就会全部失败；有了它，绝大多数都能在第二轮纠正过来。
4.4 生命周期中的性能热点：我在源码里发现的三个瓶颈
追完整条链路后，我对框架的耗时分布有了新的认识。从run()到最终返回，默认情况下大部分时间都花在LLM 调用上，这是显然的。但工具执行部分也不是没有优化空间，我发现了三个容易被忽略的瓶颈：
- JSON Schema 校验性能：如果工具参数复杂、多重嵌套
anyOf规则，jsonschema校验可能需要几十毫秒甚至更多。对于高频调用的小工具，建议把复杂 schema 简化为扁平结构。 - 消息序列化开销：每一轮都要把
messages全量序列化后发给模型，如果历史消息里存了大段工具返回内容，这部分开销非常可观。合理使用上下文裁剪器能明显降低延迟。 - 事件总线 handler 的数量：挂太多事件监听器时，
emit会逐个同步调用，虽然单个 handler 很快，但累积时间不可忽视。如果只是日志需求，建议走异步 handler 或者批量上报。
5. 值得抄走的工程化设计：钩子、缓存和可观测性是怎么样落地的
5.1 事件钩子系统：为扩展留好的"钉子"
源码里EventBus的实现并不复杂，核心就是一个注册表加一个emit方法：
class EventBus: def __init__(self): self._handlers: dict[str, list[Callable]] = defaultdict(list) def on(self, event: str, handler: Callable): self._handlers[event].append(handler) async def emit(self, event: str, **kwargs): for handler in self._handlers.get(event, []): result = handler(**kwargs) if asyncio.iscoroutine(result): await result但真正有价值的不是这个类本身，而是事件被挂在哪些语义节点上。我通读一遍后发现，PiAgent 预定义的事件基本覆盖了一个 Agent 生命周期的所有重要转折点：
session.start：会话创建，可用于加载用户偏好、初始化外部连接agent.begin_step：每一轮主循环启动，可用于打印调试信息、做速率限制tool.before_invoke：工具即将执行，可用于做审批、权限校验、用量统计tool.after_invoke：工具执行完成，可用于记录耗时、上报结果llm.request_start/llm.request_end：LLM 调用前后，可用于监控 token 消耗session.end：会话结束，可用于归档数据、释放资源
我把这些事件里最常被二次开发用到的几个列了一个优先级：tool.before_invoke最高优先，因为它能实现"人在回环"（Human-in-the-Loop）审批；llm.request_end次之，因为很多团队想做 token 成本核算，这个事件是最天然的采集点。
5.2 LLM 响应缓存：幂等请求的光芒时刻
PiAgent 源码里的LLMCache很有意思——它不是对所有请求都做缓存，而是要求满足三个条件才命中：
temperature == 0（确定性输出）- 消息序列完全相同
- 工具 schema 版本没有变化
这个设计非常克制。我见过很多框架为了省成本盲目开缓存，结果模型输出因为上下文动态变化而"答非所问"。PiAgent 用严格的 key 匹配（hash(prompt + model + temperature)）几乎杜绝了这种风险。
值得一提的是，它的缓存数据存放在本地 SQLite 里，而不是简单的内存 dict。这样即使进程重启，缓存也不会丢。对于 LLM 调用成本敏感的团队来说，这种磁盘级缓存能把重复调试的成本压到几乎为零——我在本地跑实验的时候就经常命中缓存，省了不少 token。
5.3 结构化日志与可观测性：不再是"打印到控制台"
源码中有一个地方我非常欣赏：PiAgentLogger。它不是简单地把日志print出来，而是把所有关键节点都写成结构化 JSON 行。
{"event": "llm.request_end", "session_id": "abc", "model": "gpt-4o", "tokens": {"prompt": 1200, "completion": 300}, "latency_ms": 850}这种结构化日志的价值在于，它可以被任意日志平台（ELK、Loki、甚至是简单的jq命令）直接消费。你不需要专门去写解析规则，就可以直接在日志系统里做聚合查询：比如统计每个 session 的 token 消耗、计算工具调用失败率、按模型版本对比延迟。
如果团队还没有大模型应用的监控体系，我建议从 PiAgent 这套思路开始抄：事件钩子采集数据 + 结构化日志落地 + 简单看板展示。这套组合做出来一共不需要花多少时间，但能在出现线上问题时帮你省下半天排查时间。
5.4 自定义插件系统的接线方式
最后把钩子、缓存、日志三者接起来，就是一个完整的插件机制。我在本地扩展了一个"调用审批"插件，只花了十几行代码：
bus.on("tool.before_invoke", async def approval_handler(ctx): if ctx.tool_name not in ALLOWED_TOOLS: await notify_user(f"等待审批: {ctx.tool_name}") approved = await get_user_confirmation(ctx) if not approved: raise PermissionDeniedError(ctx.tool_name)注意这里emit是await的，所以 handler 里抛出的异常可以中断主循环、阻止工具执行。这个机制天然支持"拦截器"场景，不必去 hackToolRegistry或Session。框架作者一定是反复思考过扩展边界的——他们把扩展点做成了"事件 + 异常中断"这种既简单又灵活的组合。
读到这里，你会发现 PiAgent 的整体设计其实很有章法：主循环负责调度，工具注册表负责能力边界，上下文管理器负责记忆，事件总线负责解耦。每一个部分单独拿出来都算不上惊艳，但组合在一起会产生很强的工程确定性——这对于要拿 Agent 框架去支撑业务的人来说，恰恰是最重要的。
个人经验是，读这种源码不要贪多，先把主循环和工具调用这两条主链路吃透，再去碰上下文管理和插件机制会轻松很多。下一篇笔记我打算继续深入一下 Prompt 组装和多模型接入的兼容层，到时候再继续聊。
