1. 为什么偏偏拿 pi-agent 下手一次拆玩具式的学习路径我之前在好几个团队里都被问过同一个问题想学 Agent 开发到底该从 LangChain 看起还是直接啃论文每次我都建议对方先去找一个结构足够小、但又五脏俱全的开源项目把它当成乐高玩具一样拆开拼回去再拆一次。我自己就是这么过来的而 pi-agent 是我拆过最顺手的一个参照物。先说清楚我能提供的核心结论Agent 开发不等于调用 LLM 接口。它真正的复杂度藏在循环控制工具编排记忆管理这三件事上。这三件事任何一个做得稀烂模型再好都会表现成一个大号聊天机器人。参照 pi-agent 实现一个小 Agent 的好处是你能在几千行代码范围内看到整套闭环是怎么流转的而不是一头扎进几十万行的大型框架里。网上关于Agent 是什么的热搜基本常年霸榜但大家的困惑其实集中在几个具体问题上Agent 和普通对话接口的区别到底在哪Agent 的循环是谁在控制工具调用怎么才能不崩记忆怎么才能在有限上下文里塞下更多信息pi-agent 这个项目恰好每道题都给了答案只是答案需要你自己去源码里捞。这篇文章不打算逐行解读 pi-agent 的代码而是把我剥开它之后重写一个迷你版的过程完整讲一遍。项目正文为空的处境其实挺常见的——很多时候我们不是先有需求再有方案而是先接触到一个让人手痒的项目再琢磨怎么把它学透。所以本文的路线是先讲 Agent 循环的本质再给一个可运行的最小实现然后逐个攻破工具系统和记忆设计最后聊聊我跑完一轮 Demo 之后踩到的工程化坑。适合的人群是那种已经能熟练写 Python但对 Agent 内部机制还是用的时候会、看源码就懵的朋友。2. 剥到第一层Agent 循环的本质是什么、哪些环节缺一不可把 pi-agent 拆开第一眼你会发现它根本没整什么玄学。核心就一个 while 循环把当前消息列表交给模型模型决定是直接回答还是调用某个工具如果是调用工具就执行工具、把结果塞回消息列表再交给模型直到模型给出最终答案。这个再交给模型就是 Agent 和普通接口最大的分水岭。普通接口是单轮问答模型给完答案就结束了Agent 是多轮自驱模型可以在一次任务里连续决策多次每次决策都基于上一次行动的反馈。2.1 Agent 循环的最小骨架我把 pi-agent 里的循环剥出来简化成下面这个骨架def run_agent(task: str, max_steps: int 10): messages [{role: user, content: task}] step 0 while step max_steps: response llm.chat(messages, toolstool_schemas) step 1 if response.tool_calls: for call in response.tool_calls: result execute_tool(call.name, call.arguments) messages.append({ role: tool, tool_call_id: call.id, content: result }) else: # 模型不再要求调用工具输出最终答案 return response.content return 达到最大步数强制结束这段代码看着简单但它揭示了 Agent 三个隐藏的核心机制轮次上限max_steps不是可选项。模型在复杂任务里完全可能陷入死循环如果没有步数限制你的 token 费用会先把你劝退。消息历史的累积每一步的工具返回都作为新消息加入列表。这既是 Agent 能持续推理的原因也是后面上下文爆炸问题的源头。结构化工具调用tool_calls是从模型返回里单独解析出来的结构而不是让模型在文本里自己写我要调用 xx 函数。这一点很多新手会看漏导致程序无法稳定判断模型意图。2.2 模型感知与循环控制的解耦pi-agent 还有个让我印象深刻的细节它把模型能不能感知到工具和模型要不要调用工具拆成了两套配置。第一套是tools参数它决定模型在回答时知道有哪些工具可用第二套是循环里的终止条件它决定模型什么时候结束探索给出答案。这两件事混在一起写是新手最容易踩的坑。比如有人会在 system prompt 里写你可以调用以下工具却没有把工具的 JSON Schema 传给模型 API结果模型一本正经地编造出根本不存在的函数名。反过来传了工具定义但没写清必须调用工具才能回答这类约束模型又会自作聪明地猜答案。我参照 pi-agent 的实现后把这两件事彻底分开管理工具注册表管有没有循环逻辑管用不用。工具注册表决定模型能看到哪些函数的定义循环逻辑决定工具执行完之后怎么处理结果。边界清晰之后排查问题会容易得多——模型不回工具调用先查注册表工具执行报错再查循环里的异常处理。3. 动手实现前的架构取舍模型层直接裸调还是用框架封装在写第一行业务代码之前必然要回答一个问题Agent 核心逻辑到底应该基于某个现成框架还是直接用模型 SDK 裸写。我最初的冲动是直接用 LangChain因为它的AgentExecutor看起来天然支持循环、记忆、工具。但参照 pi-agent 的做法之后我改了主意。pi-agent 的依赖极轻核心逻辑里甚至没有强制要求你必选某个编排框架更像是模型 SDK 你自己的循环控制。3.1 我最终敲定的技术选型层面选择理由模型接入OpenAI SDK兼容接口生态成熟工具调用支持稳定核心循环纯 Python 自写便于理解、便于调试可复现学习过程工具执行注册表 反射调用新增工具只需一个装饰器不侵入循环代码记忆扩展自实现摘要压缩避免引入重框架突出核心原理并发场景asyncio 封装应对热搜里常见的高并发问题很多人会劝你别重复造轮子但学习型项目恰恰应该自己造一遍轮子。你亲手写一次循环再去看 LangChain 的AgentExecutor会发现它的每一步你都知道在干什么反过来直接上手框架你会被callback、intermediate_steps、memory这些抽象淹没。3.2 为什么不建议第一版就上 Dify/CrewAIDify 和 CrewAI 都很优秀但它们解决的问题和你现在要解决的问题不一样。Dify 更像工作流平台帮你把 Agent 当作应用编排起来CrewAI 则偏多角色协作几个 Agent 互相配合完成任务。当你还在纠结单个 Agent 的循环为什么跑得不对时这些框架的抽象层级反而成了障碍。pi-agent 里的做法就朴素得多一个进程里跑一个循环工具就是普通函数没有角色扮演、没有子 Agent 通信。这种朴素恰恰是学习的好土壤先练好单 Agent 的肌肉记忆再扩展多 Agent 协作路径会顺很多。3.3 关于 Rust 实现的一句话吐槽热词列表里有一项是基于 Rust 语言 AI Agent我懂这种追求极致性能的心情。但如果你连 Agent 循环都不熟Rust 的所有权模型会把你按在地上摩擦。我自己用 Python 把循环逻辑和工具系统理清楚之后才有底气去看 Rust 版的设计。语言只是载体循环和记忆的抽象不分语言。4. 从零实现核心循环消息构造、工具调用返回、终止条件这一章的代码就是你在第 2 节看到的骨架的完整版。我会补上所有细节让它可以直接运行。4.1 工具注册表先定义工具注册表让每个函数都能被模型发现并执行import inspect import json from typing import Any, Callable, Dict, List class ToolRegistry: def __init__(self): self._tools {} def register(self, func: Callable): 通过解析函数签名生成 JSON Schema sig inspect.signature(func) parameters { type: object, properties: {}, required: [] } for name, param in sig.parameters.items(): if name in (context, llm): continue parameters[properties][name] { type: string if param.annotation is str else number } if param.default is inspect.Parameter.empty: parameters[required].append(name) self._tools[func.__name__] { function: func, schema: { type: function, function: { name: func.__name__, description: func.__doc__ or , parameters: parameters } } } return func def get_schemas(self) - List[Dict[str, Any]]: return [v[schema] for v in self._tools.values()] def execute(self, name: str, arguments: str) - str: tool self._tools[name] # arguments 由模型传回是 JSON 字符串 kwargs json.loads(arguments) result tool[function](**kwargs) if not isinstance(result, str): result json.dumps(result, ensure_asciiFalse) return result这段代码的亮点在于用inspect自动生成 Schema不用手写 JSON 定义。对于学习型项目来说足够了。注意我特别跳过了名为context和llm的参数这为后面给工具注入上下文对象留了接口。4.2 Agent 类的循环class MiniAgent: def __init__(self, registry: ToolRegistry, system_prompt: str ): self.registry registry self.system_prompt system_prompt or ( 你是一个可靠的助手。如果需要获取外部信息必须调用提供的工具。 工具返回结果后根据结果继续推理最终给出简洁答案。 ) def chat(self, user_message: str, max_steps: int 10) - str: messages [{role: system, content: self.system_prompt}] messages.append({role: user, content: user_message}) for step in range(max_steps): response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolsself.registry.get_schemas(), tool_choiceauto, ) message response.choices[0].message messages.append(message) # 注意这里要保留模型的完整消息 if not message.tool_calls: return message.content # 执行所有并行工具调用 for tool_call in message.tool_calls: try: result self.registry.execute( tool_call.function.name, tool_call.function.arguments ) except Exception as e: result f工具执行异常: {e} messages.append({ role: tool, tool_call_id: tool_call.id, content: result }) return 达到最大步数提前终止。如果你跑过 OpenAI 的工具调用接口应该能感受到messages.append(message)这行有多关键。模型返回的message对象里包含tool_calls结构后续必须原样送回 API否则模型会失去上下文关联。而工具结果必须挂在tool_call_id上这是协议规定的关联方式缺失任何一个字段请求都会报错。4.3 一个带工具的完整示例registry ToolRegistry() registry.register def get_weather(city: str) - str: 查询指定城市的当前天气。 # 这里替换成真实天气 API 请求 return f{city}今天晴转多云气温 22~30℃东风 2 级。 registry.register def calculate(expression: str) - str: 计算数学表达式如 12*3。 return str(eval(expression)) agent MiniAgent(registry) print(agent.chat(北京天气怎么样))模型会先回调get_weather(北京)拿到返回后组织成自然语言回答整个流程走完大概需要 3 到 5 次模型调用。第一次调用是模型判断需要工具第二次是拿到工具结果后的总结这中间你什么都没额外配置Agent 的智能化完全是循环 结构化调用的自然涌现。4.4 终止条件为什么要单独设计我见过很多人在循环里只写一个while True结果模型在某个任务上反复调用同一个只读工具消耗大量真实 API 费用。参照 pi-agent 的做法终止条件至少要考虑三个max_steps硬限制防止死循环如果工具调用连续 N 次没有产生新的有效信息比如重复查同一个城市天气要主动中断;工具返回明显异常时要给模型一次修正机会但不能无限重试。这三个条件的判断逻辑我会放在循环的最外层因为它们属于任务生命周期控制不应该混进工具本身。5. 工具系统让 Agent 拥有手的关键设计与打磨交易所在第 4 章给出的注册表能跑通但在实际使用中马上会遇到更深层的问题。这些问题我在跑多个任务之后才陆续发现。5.1 工具描述才是模型调用准确率的胜负手同样是get_weather函数描述写查询天气和查询指定城市的当前天气返回气温、风力和降水概率是不一样的效果。description字段直接影响模型何时选择这个工具以及如何填充参数。pi-agent 源码里对每个函数的描述写得非常细致我当时不理解后来做对比测试才明白描述越具体模型的参数猜测越准确。建议把描述当成产品文案来写说明功能边界查天气 vs 查历史天气、说明参数含义city是城市中文名、说明返回内容文本便于直接回答或进一步加工。5.2 参数校验不能只靠模型的自觉模型即使看到 JSON Schema偶尔也会传错格式把int传成字符串、漏掉必填字段、传入超出枚举范围的值。这时工具执行层要兜底def execute(self, name: str, arguments: str) - str: tool self._tools[name] kwargs json.loads(arguments) # 对 kwargs 做类型和必填项校验 for req in tool[schema][function][parameters].get(required, []): if req not in kwargs: return f参数缺失: {req}请检查调用 try: result tool[function](**kwargs) except TypeError as e: return f参数类型错误: {e} ...关键点在于校验失败时返回给模型的不是抛异常而是描述性的错误文本。因为 Agent 的循环机制决定了一切结果最终都会送回模型模型读取错误文本后可以自行修正参数。这比自己写一堆 if-else 去纠错要省事得多也让 Agent 保持能自我修复的特性。5.3 并行工具调用与副作用新的 API 允许一次返回多个tool_calls这个能力非常实用。比如查询北京和上海两地天气并对比模型可能在同一轮里并行请求两个get_weather。但并行带来一个副作用风险如果这些工具里有一个是下单支付而另一个是查询余额并行执行顺序是不稳定的。pi-agent 对这个问题的处理很干脆——它给工具打上了串行/并行标注。我只在注册表里增加了一个简单字段def register(self, func: Callable, *, parallel: bool True): ...默认并行但对于写操作、有状态操作手动设为parallelFalse在执行循环里遇到非并行工具就临时退化成串行处理。这个细节让我避免了好几次并发扣款级别的灾难。5.4 工具隔离与安全边界Agent 的工具本质上是把网络请求、文件读写、命令执行能力交给模型调度。如果工具是execute_shell(cmd: str)这种你要清楚这意味着什么——模型完全可能因为 prompt 注入或幻觉执行出危险命令。我的习惯是第一版不接入任何 shell 类工具只暴露无副作用的查询函数。等到需要写文件时也严格限制路径在特定沙箱目录内。6. 记忆与上下文窗口怎么让 Agent 记住更多又不爆 token所有人跑通 Agent 第一件事都会遇到同一个问题连续对话几轮之后消息列表越来越长成本直线上升而且模型开始忽略早期信息。这就是 Agent 记忆设计的核心矛盾。6.1 消息历史的三种基本策略策略做法优点缺点整段截断只保留最近 N 条消息简单直观早期关键信息丢失摘要压缩每轮对话后生成摘要替代原始消息保留结构化信息摘要本身有信息损失且增加一次模型调用向量检索把历史消息嵌入向量库按需召回适合长程任务工程复杂对单 Agent 学习项目偏重参照 pi-agent 的进化路径我首先实现的是摘要压缩。核心做法是当消息数超过阈值比如 20 条用一次模型调用把历史消息压缩成摘要然后清空历史把摘要作为一条 system 消息塞进列表。def compact_messages(messages, summary): # 触发条件消息超过阈值 if len(messages) 20: return messages, summary # 把所有历史消息合并成一段文本 history_text \n.join( f{m[role]}: {m[content]} for m in messages ) prompt ( 将以下对话历史压缩为摘要保留关键事实、用户意图、 已经执行过的工具和结果。不要遗漏重要约束。\n\n f历史对话:\n{history_text} ) new_summary client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}] ).choices[0].message.content return [ {role: system, content: f历史摘要: {new_summary}} ], new_summary这个方案在真实场景里立竿见影——上下文从动不动多出的几千 token 降到稳定几百。6.2 长任务的跨会话记忆热搜词里的agent记忆大多指向跨会话持久化。用户上一轮问了一个问题今天重新打开应用还想接着聊这需要把摘要持久化到数据库。我在项目里用了一个轻量的 SQLite 表CREATE TABLE agent_memory ( id INTEGER PRIMARY KEY, session_id TEXT NOT NULL, summary TEXT NOT NULL, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP );每次摘要压缩后更新 session 对应的 summary 字段。新会话启动时把这个摘要作为初始 system 消息注入。这样 Agent 就有了你上次聊到哪了的基本记忆成本只需一次数据库查询。6.3 token 用量换算的实际感觉很多刚接触 Agent 的人会低估 token 消耗。一次包含工具调用的完整对话仅仅是把全部消息送去模型就已经消耗了几千 token再加上模型输出的工具调用参数、工具返回一次思考成本远超普通聊天。做多轮 Agent 时我把摘要压缩阈值调低到 12 条消息并且每轮对话之后打点记录 token 增量看到数字就会对成本有真实体感。7. 并发与稳定性网上都在问的过 Demo 阶段要面对什么热搜里有一项是AI Agent 怎么扛并发这个问题并非要把设计复杂度全部拉满但要清楚 Agent 服务和普通 HTTP 服务在并发模型上的差异并有针对性地处理。7.1 Agent 服务并发的本质瓶颈普通 Web 接口的瓶颈在数据库和业务逻辑Agent 服务的瓶颈则复杂得多模型 API 的响应时间本来就有波动一个复杂任务可能要调 3-5 次模型 API中间还穿插工具执行单次请求的总耗时可能是秒级到分钟级。如果按照传统 Flask 线程模型不加限制地接收请求很快会爆发两个问题模型 API 的并发限流被打穿大量请求 429每个请求的内存和上下文状态撑爆后端。我的做法是Agent 执行体自身通过队列调度不直接暴露给用户任意创建任务。实现方式是给每个用户会话创建一个任务队列Agent 串行消费队列里的消息不同用户之间可以并发同一个用户的请求宁愿排队也不并行处理。7.2 超时、重试与幂等工具调用天然有失败概率天气 API 可能超时、文件可能读取失败。循环里的容错办法我前面已经提过但工程化的稳定还需要额外两层。第一层是模型调用本身的超时重试。OpenAI SDK 支持timeout参数建议设置timeout60并且对网络错误做指数退避重试最多 3 次。第二层是工具侧的幂等设计写操作要带唯一请求 ID重复执行不会造成重复影响。 第三层是任务超时熔断。一个 Agent 任务如果 5 分钟还没跑完大概率卡死或者上下文爆炸了应该直接被标记为失败并释放资源。7.3 结构化日志是 Agent 排障救命稻草Agent 的每次循环都涉及模型调用、工具结果、消息变化排错难度远超普通接口。我给循环的每个关键节点加结构化日志logging.info({ step: step, model_input_tokens: usage.prompt_tokens, model_output_tokens: usage.completion_tokens, tool_calls: [c.function.name for c in message.tool_calls] if message.tool_calls else None, final_answer: message.content if not message.tool_calls else None })有了这个日志线上出问题时可以在几秒内定位——是模型进入循环了是某个工具迟迟没返回是 token 超限导致的截断顺着时间线把每步日志拉出来很多诡异问题都会现形。8. 参照 pi-agent 的进阶玩法Harness 与框架对比思考热搜词里同时出现了agent harness和agent区别以及agent框架如langchain、dify、crewai等哪个好。在学习型项目跑通之后这些概念其实可以串起来看。8.1 当你开始写 harness你才真正理解了 Agent 周期很多资料把 harness 翻译成装配或者框架我理解它就是控制 Agent 循环、记忆、工具调用的那一整套外壳。你照着 pi-agent 实现了自己的 MiniAgent其实就是在造一个微型 harness。当你开始思考我这个 harness 能不能让子 Agent 复用父 Agent 的工具能不能在循环中间插入一个人工确认环节能不能把某一步的回溯改成分支探索——这些问题的答案都写在 harness 的接口设计上。相比盲目对比 LangChain 好不好用先把 harness 拆明白再去看框架每一家设计的好与鸡肋你都能看懂个七七八八。LangChain 强在抽象生态全但学习成本高Dify 强在可视化编排和团队协作CrewAI 强在角色协作但单 Agent 的深度控制并不突出。如果你已经能徒手实现一个 Agent 循环再去选择框架就有明确标准它能让我少写哪些代码它有没有限制我的循环控制自由度它能和我现有的工具系统无障碍打通吗这几个问题比我列一堆框架特性对比表更有实际价值。8.2 Skill 机制的启发热词里有一项是agent skill我看到 pi-agent 里真的把每个工具打包成了一个技能——函数、描述、示例参数全部绑定。这种设计比裸函数更接近真实 Agent 应用技能可以被学习和复用也可以挂到特定角色下发布。如果你想做更高级的 Agent提前把工具设计升级成技能包会让后续的扩展优雅不少。9. 跑通之后的下一步我的体会与小建议整个项目玩下来我最深的体会是Agent 开发入门的关键不是模型选得多好而是你能否把循环、工具、记忆这三者的边界在代码里控制清楚。很多人被 LangChain 的 Agent 搞得很挫败根本原因不是 LangChain 差而是他们没有亲手建立过这三者的心智模型。参照 pi-agent 自己实现一遍这个心智模型就内化成了肌肉记忆。最后分享几个实际经验第一版尽量用 Python不要一边学概念一边折腾类型系统。理解循环之后再考虑 Rust 等更高性能的移植。每实现一个功能立刻写一段测试让它真的调用一次工具。不要只测模型能回答要测模型能在需要时正确选择工具并处理异常结果这两件事覆盖了 Agent 90% 的稳定性问题。日志和监控千万别省Agent 的不可预测性决定了你一定会需要一个重播机制把线上失败的完整消息轨迹拉出来复现。预算控制尽早做我在摘要压缩和步数上限都踩过坑第一次跑长任务的时候 token 消耗直接超了预期两倍后来才通过实时计数器和阈值控制找回平衡。这个迷你版 Agent 项目我至今还保留着后续我又给它加了多轮对话总结、技能包加载和新工具类型。如果正处在听说过 Agent、想动手但不知从哪下手的阶段建议你今天就写一个 while 循环的雏形出来——跑了第一个工具调用之后整个 Agent 世界的大门才算真正打开。