最近手头的 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_factorytime.time) meta: dict Field(default_factorydict)其中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( namecalculator, description计算四则运算表达式输入类似 12*3, parameters{ type: object, properties: { expr: {type: string, description: 数学表达式} }, required: [expr] } ) def calculator(expr: str) - str: try: tree ast.parse(expr, modeeval) # 仅允许基础运算节点防止恶意表达式 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 ferror: {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_tokens3000它会从最旧的消息开始移除直到当前 token 占用低于阈值。而且它会强制保留第一位system消息如果你加入了一条rolesystem的消息但放在非首位它依然会尽量挽留实在留不住才给你丢。具体代码思路大致是这样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 和并发调度部分那边应该还有更多值得拆的东西。到时候会整理成系列的第二篇再单独聊聊。
最近手头的 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_factorytime.time) meta: dict Field(default_factorydict)其中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( namecalculator, description计算四则运算表达式输入类似 12*3, parameters{ type: object, properties: { expr: {type: string, description: 数学表达式} }, required: [expr] } ) def calculator(expr: str) - str: try: tree ast.parse(expr, modeeval) # 仅允许基础运算节点防止恶意表达式 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 ferror: {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_tokens3000它会从最旧的消息开始移除直到当前 token 占用低于阈值。而且它会强制保留第一位system消息如果你加入了一条rolesystem的消息但放在非首位它依然会尽量挽留实在留不住才给你丢。具体代码思路大致是这样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 和并发调度部分那边应该还有更多值得拆的东西。到时候会整理成系列的第二篇再单独聊聊。
