最近手头的 Agent 落地项目越来越多，光靠业务层写死 prompt 已经撑不住了，于是我开始认真啃 Pi Agent 的源码。这是最近社区里讨论不少的一个轻量级 Agent 框架，主打“可控、轻量、可插拔”的设计取向，核心抽象并不多，但很多细节设计非常见功夫。这篇是系列笔记的第一篇，主要记录整体架构、核心抽象、工具接入和记忆实现这几个关键部分，也会分享一些实际阅读过程中踩过的坑。适合打算做 Agent 二次开发、或者正在准备系统读框架源码的同学参考。
读完这套源码，我最直观的感受是：它没有堆砌一大堆“看起来很高级”的组件，而是把事情做得小而明白。Agent 的循环、消息结构、工具注册、记忆裁剪，这些东西在很多框架里都是被厚厚封装起来的黑盒，但在 Pi Agent 里你花两三个晚上就能把主链路看通。这篇文章我不打算逐文件贴注释，那样太啰嗦，我会先带你把整体地图走一遍，再挑几个核心模块展开讲原理，最后聊聊我从里面学到的设计经验和排查方法。
1. 从 README 开始：先搞清楚框架到底想解决什么问题
读源码最忌讳一上来就钻到base.py里面去。我拿到一份新框架，第一件事一定是把 README 和 examples 目录先过一遍。Pi Agent 的 README 写得不算长，但信息密度很高，从里面能读出这个项目的核心命题：它希望做一个真正能让业务方快速接入的 Agent 框架，而不是一个研究性质的玩具。框架内部默认支持多种主流大模型接口，通过一个统一的 Client 抽象接入，同时把工具调用、记忆管理、Agent 状态流转这些通用能力做成可替换的模块。
1.1 项目基因与设计取向
Pi Agent 的依赖列表非常克制。核心代码的第三方依赖主要是 pydantic、httpx、pyyaml 这类基础库，没有把 langchain 那种重的东西直接绑死。这一点我很喜欢，因为依赖少意味着更容易读、更容易改、出了问题更容易排查。它把 Agent 的核心闭环归纳成了四个抽象：Agent（决策与执行主体）、Tool（外部能力入口）、Memory（上下文与记忆存储）、LLMClient（模型调用层）。所有业务逻辑都是围绕这四个抽象展开的。
这个设计取向对阅读者非常友好。你不需要先去理解几十个抽象父类和接口再开始看业务，核心入口就是那四张牌。从工程角度看，这也体现了“组合优于继承”的思想——框架本身不强制你继承一大堆东西，而是让你在创建 Agent 时自由组合这些组件。
1.2 代码仓库结构：源码地图与阅读路径
整个仓库的结构很清晰，核心源码集中在src/pi_agent下面，大致是这样分层的：
pi_agent/ ├── agent/ │ ├── base.py # Agent 基类，核心循环在这里 │ ├── executor.py # 工具执行与结果回填 │ └── plan.py # 任务拆解与步骤规划 ├── memory/ │ ├── base.py # 记忆接口定义 │ ├── buffer.py # 滑动窗口记忆 │ └── summary.py # 摘要压缩记忆 ├── tools/ │ ├── registry.py # 工具注册中心 │ └── builtin.py # 内置工具集 ├── llm/ │ ├── client.py # 模型调用客户端抽象 │ └── protocols.py # 协议定义，比如消息格式 ├── config/ │ ├── loader.py # 配置加载 │ └── schema.py # 配置校验模型 └── api/ ├── app.py # HTTP 服务入口 └── routes.py # 路由定义我建议的阅读顺序是：先看agent/base.py把主循环看懂，再去看memory/buffer.py和tools/registry.py，最后回到llm/client.py理解它怎么把你熟悉的模型 API 变成统一接口。按这个顺序读，基本两三个晚上就能把全链路走通。如果你一上来就去看api或者examples，容易被外围代码带偏。
2. 核心抽象层：Agent 的循环到底是怎么跑起来的
Agent 框架最核心的东西，就是那个 think-act-observe 循环。Pi Agent 的主循环非常直白，不像一些框架那样引入复杂的状态机引擎，而是用了一个while循环加状态枚举来管理生命流程。这种选择我一开始觉得有点“土”，读到后面才明白它是故意为之。
2.1 Agent 基类与主循环设计
先看它的核心状态定义：
class AgentState(str, Enum): IDLE = "idle" PLANNING = "planning" EXECUTING = "executing" WAITING_TOOL = "waiting_tool" OBSERVING = "observing" FINISHED = "finished" ERROR = "error"状态不多，但每个状态都有明确边界。主循环的逻辑大致是这样：Agent 从IDLE进入PLANNING，拿到用户的输入之后规划出若干步骤，然后进入EXECUTING。执行过程中如果发现需要调用工具，就转入WAITING_TOOL等待工具结果，拿到结果后回到OBSERVING观察结果、更新上下文，再决定是继续执行还是结束。
这个循环里我认为最有价值的设计是：它允许你在每个状态流转的临界点抛事件。打个比方，就像给你家里的智能灯泡加了一个物理开关——你可以随时从外面打断它的自动流程，插入自己的逻辑。这也是为什么官方推荐你在自定义 Agent 时优先重写observe()方法而不是整个run()方法，因为观察阶段是插入业务逻辑最安全的位置。
我记得有个细节很有意思：状态机里ERROR不是用来中断的，而是用来兜底的。也就是说，当某个工具抛出异常，Agent 不会直接崩溃，而是把这个异常作为一条 tool 消息回填给模型，让模型自己决定下一步怎么处理。这在生产环境很实用，因为真实工作流里工具失败太常见了。
2.2 消息对象 Message 的设计理念
消息是 Agent 世界里最基础的载体。Pi Agent 的Message是一个 pydantic 模型，字段很少但很精准：
class Message(BaseModel): role: Literal["system", "user", "assistant", "tool"] content: str = "" tool_calls: list[ToolCall] = [] created_at: float = Field(default_factory=time.time) meta: dict = Field(default_factory=dict)其中tool_calls是模型要求调用工具的指令列表，meta用来塞一些不入消息体的附加信息，比如哪个工具执行耗时多久、是哪次重试产生的消息。这种设计聪明在哪？它把“消息内容”和“过程信息”分开，避免像某些平台那样把所有东西塞进 content 里搞成一大坨，导致后续切分和计费都变成噩梦。
如果你想自己扩展框架，我建议在Message的子类或者meta字段上做文章，而不要去动role和content的语义。因为很多下游逻辑，比如 Token 统计、消息持久化、上下文裁剪，都依赖这两个字段的稳定。
2.3 几个容易忽略的亮点细节
源码里有些小细节很容易错过，但对实际使用影响很大。比如上下文裁剪时，框架会保护第一条system消息不被删掉——这个问题我最初自己写裁剪逻辑时就踩过坑，把系统设定给裁了，结果模型聊着聊着就失忆了。再比如消息对象实现了__len__方法，返回的是消息体的 token 估算值而不是字符数，这个小细节让上层在做长度判断时特别顺手。
我举一个更具体的例子，来看它怎么估算 token 数：
def _estimate_tokens(text: str) -> int: # 按 4 字符约等于 1 个 token 估算，中文偏保守 return max(1, int(len(text) / 4))当然这个估算对中文来说不算准，但它胜在快、零依赖，而且在裁剪场景下足够用。读了这段代码我才意识到，判断一个框架设计得好不好，不一定要看它多精致，而要看它在资源有限的情况下做的取舍是否聪明。
3. 工具调用与模型接入：插件化设计的门槛在哪里
Agent 框架的扩展性主要看两块：一是工具怎么注册、怎么让模型“知道”有这些工具，二是模型接入层怎么做到“换模型不换业务代码”。Pi Agent 在这两块都做了非常典型的插件化处理，值得好好拆一拆。
3.1 工具注册机制与参数注入
Pi Agent 的工具注册用的是装饰器风格，这个模式在 FastAPI 等框架里已经很成熟了，放在这里也非常顺手：
@tool( name="calculator", description="计算四则运算表达式，输入类似 '1+2*3'", parameters={ "type": "object", "properties": { "expr": {"type": "string", "description": "数学表达式"} }, "required": ["expr"] } ) def calculator(expr: str) -> str: try: tree = ast.parse(expr, mode="eval") # 仅允许基础运算节点，防止恶意表达式 for node in ast.walk(tree): if not isinstance(node, (ast.Expression, ast.BinOp, ast.Num, ast.Operator, ast.UnaryOp, ast.Add, ast.Sub, ast.Mult, ast.Div)): return "error: invalid expression" return str(eval(compile(tree, "", "eval"), {"__builtins__": {}}, {})) except Exception as e: return f"error: {e}"这里有两个很关键的细节。第一，框架会在注册时自动读取函数的签名和 docstring，帮助模型理解这个工具是干嘛的。所以你在自己的业务代码里写工具函数时，参数名和 docstring 一定要写得规范，因为模型是按照这些信息来决定怎么给你传参的。第二，框架本身提供参数校验，但你最好在函数内部再做一道数据验证，不要信任模型生成的所有参数。
我在实际项目里犯过一个低级错误：工具函数定义了一个keyword参数，模型返回{"keyword": "排序"}，结果模型把它理解成了 Python 的保留关键字场景，传参一直报错。后来我改了参数名叫search_text才解决。这种问题不做真实调用测试根本发现不了，因为函数签名在静态检查时不会报错。
3.2 LLM 接入层的适配思路
模型接入层，Pi Agent 的做法是定义了一个协议类，而不是直接继承具体 SDK。协议类长得像这样：
class LLMClient(Protocol): def chat(self, messages: list[Message], tools: list[dict], **kwargs) -> Message: ... def stream_chat(self, messages: list[Message], tools: list[dict], **kwargs) -> Iterator[Message]: ...这样设计的好处是，只要你写的模型客户端实现了chat和stream_chat两个方法，就能无缝接入框架。框架默认提供了对主流模型 API 的适配，包括系统消息格式转换、工具调用格式转换等繁杂的工作。我第一次看client.py的时候最震惊的是：它居然在流式接口里保留了对工具调用的异步解析。流式模式天然适合打字机效果，但工具结果是一个整体 JSON，怎么把碎片拼回来是很多框架处理不好的痛点。它把内部细节完全藏住了，上层拿到Message时已经是完整对象了。
3.3 重试策略与回调钩子
模型接口是不稳定的，所以 Pi Agent 内部实现了指数退避重试策略，具体参数是初始延迟 0.5 秒、倍率 2、最大尝试 5 次。这个参数组合我实测下来，在服务偶发抖动时既不会频繁重试打爆服务，也不会因为重试次数太少导致任务直接失败。当然你可以在配置里改掉。
它还提供了几个回调钩子，比如on_tool_call、on_token_usage、on_error。我建议业务方至少接两个：on_token_usage用来记录真实消耗，on_tool_call用来做审计日志。这个其实是被现实毒打出来的经验——上线一个 Agent 应用，如果不知道每个任务花了多少 token、调了哪些工具，后面不好做成本优化和问题复盘。
4. 记忆系统：短期与长期记忆的实现细节
Agent 的记忆是最近一直有人在讨论的话题。Pi Agent 源码里对记忆的实现，我觉得它的取舍思路很有代表性：不追求全量保存，而是通过滑动窗口加摘要压缩来在“信息量”和“成本”之间找平衡。
4.1 记忆接口与内置实现
框架的记忆接口定义得非常收敛：
class BaseMemory(ABC): @abstractmethod def add(self, message: Message) -> None: ... @abstractmethod def get_messages(self, filter_system: bool = False) -> list[Message]: ... @abstractmethod def clear(self) -> None: ... @abstractmethod def compact(self) -> None: ...add负责写入消息，get_messages负责提供给 Agent 作为上下文，clear清空，compact做压缩整理。内置的记忆实现主要有两种：BufferMemory是滑动窗口方式，只保留最近 N 条消息；SummaryMemory则在超过阈值时调用模型生成摘要，把旧消息浓缩成一段背景信息放进上下文。
4.2 BufferMemory 的裁剪策略
看BufferMemory的实现时，我发现它有个很有意思的处理：裁剪不是按条数硬切的，而是按 token 估算值来约束。假设max_tokens=3000，它会从最旧的消息开始移除，直到当前 token 占用低于阈值。而且它会强制保留第一位system消息，如果你加入了一条role=system的消息但放在非首位，它依然会尽量挽留，实在留不住才给你丢。
具体代码思路大致是这样：
class BufferMemory(BaseMemory): def __init__(self, max_tokens: int = 3000): self.max_tokens = max_tokens self.messages: list[Message] = [] self._token_usage = 0 def add(self, message: Message) -> None: self.messages.append(message) self._token_usage += _estimate_tokens(message.content) self._trim() def _trim(self) -> None: if self._token_usage <= self.max_tokens: return # 保护系统消息：从去掉第一条 system 消息之外的最旧消息开始裁剪 drop_candidates = [ i for i, m in enumerate(self.messages) if m.role != "system" ] # 如果只有 system 消息且超限，则强制裁掉最旧的 system if not drop_candidates: drop_candidates = list(range(len(self.messages))) for i in drop_candidates: if self._token_usage <= self.max_tokens: break self._token_usage -= _estimate_tokens(self.messages[i].content) # 真正执行移除时，需要重新筛选，源码这里用的是一个索引标记方案这里我想多叮嘱一句：在真实的长对话场景里，只保留最近窗口很容易导致“中期遗忘”，也就是二十轮之后模型完全忘了十分钟前聊过什么。解决办法一般是把 BufferMemory 和 SummaryMemory 叠加使用——最近五轮保留原话，更早的自动生成摘要。这也是我后来在生产环境里最常用的配置组合。
4.3 摘要记忆的压缩逻辑
SummaryMemory的compact()会做三件事：把旧消息拼接成文本，调用模型让它用一段话概括核心信息，然后把这段概括以system或者user消息的形式插回记忆里。源码里有一个细节非常棒：它要求模型在压缩时必须保留“用户明确指令”“关键事实”“未完成的任务项”。这三类信息丢失了，相当于白压缩。
这个思路给了我很大启发。之前我在自己业务里做上下文压缩，一直苦恼怎么保证摘要“不丢关键信息”。后来我干脆把自己的压缩 prompt 也改成这个指令格式，效果好了非常多。你不用担心它每轮都触发压缩，因为compact()是主动调用的，你可以在每轮结束前判断 token 占用率超过 0.8 才执行，省掉了很多无效的模型调用。
5. 配置与扩展：框架给开发者留下的口子
读框架源码，除了理解它的实现，更重要的事情是从里面找到“给开发者留的口子”。Pi Agent 的配置加载和插件扩展机制虽然不复杂，但很清晰，几乎不需要额外文档就能上手。
5.1 YAML 配置与校验逻辑
框架使用 YAML 作为主配置格式，所有配置项都会通过 pydantic 做运行时校验。比如这样一个配置：
agent: name: "demo_agent" description: "一个用于测试的 Agent" memory: type: buffer max_tokens: 3000 tools: - calculator llm: provider: openai model: gpt-4o-mini temperature: 0.2 max_retries: 5配置加载逻辑里有个很贴心的细节：它支持环境变量覆盖默认值，结构是${ENV_VAR:default}。这意味着你部署时可以把 API Key 这些敏感信息放环境变量里，不用改配置文件。框架还允许传入一个自定义配置目录，默认从config/下读取。这个设计极大地降低了本地调试和云端部署的切换成本。
5.2 自定义一个简单插件的完整流程
我强烈建议你拿到框架后别急着改源码，而是先试着按官方方式写一个自定义 Agent 或者自定义工具。以自定义工具为例，整体流程非常顺：
- 在你的项目里建立一个
my_tools.py，用@tool装饰器注册你的函数。 - 在 YAML 配置的
tools列表里加上这个工具的名字。 - 在创建 Agent 时，把自定义工具模块传入工具注册中心。
- 启动你的 Agent，输入一句可以触发这个工具的话，观察工具调用日志。
我试过把公司内部的一个查询库存的接口封装成工具。整个封装过程大概只有 60 行代码，包括异常处理、结果格式化和超时设置。它最舒服的地方在于：你不需要理解模型内部的 function calling 参数是怎么组成的，框架帮你把函数签名转成 JSON Schema 的过程完全藏在底层。你只需要把参数说明写清楚，把返回值变成字符串即可。
5.3 这套框架到底值得学到什么
读一套框架，不能只看它的功能，要思考它的设计折射出的工程品味。Pi Agent 最大的优点我认为是“薄而清晰”。它没有把 Agent 复杂化成一台精密仪器，而是给了你一张基础的桌子和几个可靠的抽屉，其余的你按自己的习惯去摆放。
但它也有坑。比如多 Agent 协作、复杂的图编排能力，这个框架默认并不内置。如果你需要一个支持子 Agent 嵌套或动态拓扑的框架，Pi Agent 现阶段其实不太合适。另外一个问题是它在高并发场景下的编排能力偏弱，默认的 Executor 不太适合直接作为大规模线上服务使用。框架官方定位更偏向中小型应用和快速原型，这一点自己要心里有数。
6. 阅读源码的方法、踩坑与工程反思
最后这部分，我想把源码阅读过程本身的经验抽出来讲讲。毕竟框架内容是死的，阅读方法才是能让你持续受益的东西。我读这份源码大概花了三个晚上，中间踩过一些坑，也积累了一些好用的技巧。
6.1 我的阅读路线与工具链
我是用 IDE 配合 git 历史来读的。思路是：先读 README 和 examples 建立印象，再跑一遍测试用例，从测试代码里反推每个类的方法应该做什么，最后才读核心实现代码。读的时候我会用书签做强提醒，比如哪段逻辑用了状态模式、哪段代码做了消息重放，都单独标记出来。
这套方法源码阅读效率特别高。尤其是跑测试用例这一步，很多人会忽略，但其实测试用例是一个框架“最好的文档”，因为它把每个方法的输入、输出、边界条件全部写清楚了。我看test_memory.py的时候，几分钟就理解了 BufferMemory 的裁剪逻辑，完全不看实现代码都能写出来。
6.2 源码阅读过程中踩过的坑
我自己读这套源码时踩过的最大的坑，是 pydantic 版本差异。Pi Agent 是用新版本 pydantic 写的，而我本机环境里某些项目还在用旧版，一运行就报ValueError。后来我在本地给项目单独建了虚拟环境，问题才解决。
第二个坑是我一开始试图逐行注释所有代码，结果读到 executor 的时候已经累得不想动了。后来我调整策略：只注释那些跟主流程相关的函数，外围的工具函数和测试代码一律跳过。这个调整很有效，阅读压力瞬间小了很多。
还有一个小技巧：我在源码里临时加了日志打印，观察 Agent 实际给模型发送的完整消息列表和工具定义。跑完几个例子以后马上把日志还原干净，不然容易影响后续测试。
6.3 这套源码对日常编码的启发
读完之后，我反思了一下自己平时写业务代码的习惯，有三个收获特别想分享。
第一个是接口先行。Pi Agent 的核心模块都是先定协议再写实现，比如BaseMemory的四个方法就是整个记忆模块的通用接口。实际项目里我们也应该先把模块间交互的“协议”定清楚，再填细节，而不是边写边改接口。
第二个是薄封装优于重封装。很多框架为了追求功能全面，会把很简单的事情搞出三层抽象。Pi Agent 反其道而行之，主循环就是一个while加一个枚举，既好读又好调试。这提醒我，遇到复杂需求，先想想能不能用最简单直白的方式实现，再去考虑要不要引入重型设计模式。
第三个是错误处理要流经主链路。它在工具调用出错时，不是把异常往上抛，而是把错误转成一条 tool 消息，交给模型继续决策。这种“把错误当作数据”的思路特别适合 Agent 场景，因为模型本身具备判断力，给它更多上下文往往比让进程直接崩溃更好。
我在实际项目中，把类似思路用到了自己的工具调用层：一旦某次 API 请求失败，我会把失败原因和当前状态一起回填给模型，让模型自行决定是要重试还是换一种策略。这个小改动让我的 Agent 任务成功率提升了不少，我觉得这是读源码给我个人业务带来的最直接的收益。
接下来的计划里，我准备继续深入看它的 executor 和并发调度部分，那边应该还有更多值得拆的东西。到时候会整理成系列的第二篇，再单独聊聊。
