最近我把一个开源仓库里的 pi-agent 翻了个底朝天又参照它的思路自己实现了一个只有几百行的小 Agent。这件事做完之后很多以前想不明白的问题一下子通了。过去我写 Agent 总是先接一个比较重的框架配各种记忆、编排和插件结果出了问题根本不知道是哪一层装出来的。剥开 pi-agent 之后我发现所谓 Agent核心不是模型也不是框架而是一个朴素到极点的循环让模型看到历史和工具结果决定下一步做什么执行完再回到模型面前。这篇文章不打算完整复述源码而是想分享我参照它做最小实现的过程先拆结构再写代码最后补上实测中必然踩到的坑。适合已经会一点 Python、想自己造一个 Agent 而不是只会调包的开发者看。1. 先拆清楚一件事pi-agent 里哪个部分是灵魂1.1 剥开之后看到的四块积木第一次读完 pi-agent我发现整个仓库的代码量比想象中少得多。它的核心逻辑可以拆成四块模型封装层、工具仓库、消息状态容器、主循环。这四块并不是什么新概念但分开看非常清晰我当时在纸上画了一张草图二十秒就理解了这个 Agent 的完整执行路径。用一个生活化的类比来解释模型封装层是接线员所有对模型的请求都从它这里进出工具仓库是工具箱每把工具都有一个名字和一份使用说明消息状态容器是工作台上的记事本每一轮聊了什么、工具返回了什么都会被记录下来主循环则是一个不断重复的想、做、看流程。这四块里前三个都相对被动真正推动 Agent 往前跑的只有主循环。pi-agent 让我最舒服的一点是它没有把 Agent 抽象成一大堆类。它的核心代码读起来就像一段普通的业务逻辑发请求、拿结果、执行工具、把观察结果写回消息列表。这种直白的设计反而比那些绕了好几层的框架更容易启发人因为你可以从一个最小骨架出发往任意方向扩展。1.2 主循环是整个 Agent 的发动机普通聊天机器人和 Agent 的区别就体现在主循环上。普通聊天机器人每次调用模型返回一段文字就结束了模型没有任何动手的空间Agent 则把模型的输出当成一个动作建议执行完动作再把结果喂回去循环直到模型认为任务完成。pi-agent 的主循环比我预想中还要薄本质就是一段伪代码repeat: 把对话消息发给模型 模型返回文本或工具调用指令 如果是工具调用指令 执行对应函数 把结果作为工具消息追加到对话里 继续下一轮 否则 返回最终文本这不是什么新发明很多年前就有人用类似的方式做自动化但大模型的出现让这个循环变得非常强。模型负责产生意图本地代码负责执行两者通过消息列表完成通信。我在复刻时时刻提醒自己这个循环就是 Agent 的全部骨架后续所有工作都要长在这个骨架上不能反过来被各种外围模块绑架。1.3 工具调用在这套循环里的真正位置很多人以为工具调用是模型的能力模型只要把参数填好函数就会自动执行。实际完全不是这样。模型只负责输出一个意图它可能说我想调用 calculator参数是 expression3.5*60真正执行四则运算的是你本地写好的 Python 函数。pi-agent 的处理方式很扎实它把工具调用解析出来用本地函数执行再把执行结果以一条普通消息的形式塞回对话历史里。模型在下一轮看到这条工具消息就知道刚才的动作有没有成功、结果是什么。正是因为工具调用被这样显式处理了开发者才能在任何一步插入校验、缓存、日志和权限控制。如果把这些逻辑全部藏到模型 SDK 内部一旦出问题就只能对着黑盒发呆。2. 动手前先设计我要复刻的最小 Agent 到底包含哪几个文件2.1 先给能用划一条线在动手之前我给自己定了一条边界这次复刻的目标不是再造一个通用 Agent 平台而是做一个能跑通完整闭环的最小实现。我给它定的能力门槛是能进行多轮对话能记住自己刚刚做过的动作能调用两个自定义工具一个查当前时间一个做四则运算能处理模型偶尔返回的错误格式不至于一崩到底不接向量数据库不做复杂长期记忆不做流式输出之所以把范围压得这么小是因为我在过去调试大型 Agent 项目时吃够了苦头功能越多变量越多你很难判断到底是 prompt 问题、工具问题还是循环问题。先让主干彻底跑通再来加枝节这是最稳妥的路径。pi-agent 给我们的启示也是克制的它只做执行链不负责把全世界都集成进去。2.2 目录结构与每个文件的职责我的目录结构完全是照着 pi-agent 的模块划分简化出来的一共只有五个文件mini-agent/ ├── config.py # API 地址、模型名、密钥 ├── models.py # 模型请求封装 ├── tools.py # 工具注册表与执行函数 ├── agent.py # 主循环 └── run.py # 命令行入口每个文件的职责非常单一。config.py单独拆出来是因为调试时经常要换模型和接口地址集中管理可以避免到处改字符串models.py只负责和模型服务商通信不涉及任何业务逻辑tools.py提供注册机制和执行机制agent.py是唯一有循环的地方run.py只做一件事把用户输入交给 Agent再把最终结果打出来。这五个文件不是一上来就全部写完的。我平时写代码的习惯是先让主循环跑起来哪怕工具只有一个hello函数也要先把骨架走通再往里加东西。骨架通了之后后续的每个新功能都会变得非常自然。2.3 一个关键决定先放弃流式输出也先放弃异步我设计这个最小实现时故意砍掉了两样东西流式输出和异步并发。流式输出看起来很酷但它需要处理增量事件主循环里每个响应分片都要拼回去工具中间穿插流式文本后渲染状态会非常难控。异步并发则会让工具执行结果回到正确对话轮次这件事变得复杂工具 A 跑得慢工具 B 跑得快谁先写回消息列表都可能影响结果。同步单轮请求虽然笨但有一个不可替代的优点每次调用模型都能等到一个完整的结构化响应之后再解析就非常省心。pi-agent 这类轻量实现基本也是这个路数。我建议所有想自己实现 Agent 的人第一版都先做同步、非流式把循环逻辑吃透之后再优化体验。3. 核心代码逐个落地模型接口、工具注册、主循环3.1 models.py只做发消息、收响应这一件事模型接口的封装不需要很复杂我用requests直接调兼容接口。为什么不接模型服务商官方 SDK因为我希望这个实现能随时切换不同模型或本地服务只要对方提供兼容的聊天补全接口改掉config.py里的地址和模型名就能跑。# models.py import requests import json def chat_once(messages, toolsNone, configNone): url config[base_url].rstrip(/) /chat/completions headers { Content-Type: application/json, Authorization: fBearer {config[api_key]}, } payload { model: config[model], messages: messages, temperature: 0.2, } if tools: payload[tools] tools resp requests.post(url, headersheaders, jsonpayload, timeout60) resp.raise_for_status() data resp.json() choice data[choices][0][message] text choice.get(content) tool_calls choice.get(tool_calls, []) return {text: text, tool_calls: tool_calls}这个函数不关心业务逻辑只负责把消息列表发出去再把模型返回的内容拆成文本和工具调用两部分。text有可能是None因为模型选择了工具调用时它可能不输出自然语言tool_calls则可能是空列表。主循环要做的事情就是根据这两种情况的组合决定下一步动作。我特意把temperature调成 0.2。Agent 场景下我们希望模型尽量稳定、少发散温度太高容易在工具参数上产生随机性导致同样的输入两次返回不同的expression。如果你用的模型对工具调用支持不够好可以把这个值调到 0稳定性会进一步提升。3.2 tools.py用注册表而不是 if-else 去管理工具pi-agent 里工具的管理方式非常朴素就是一个全局字典REGISTRYkey 是工具名value 包含函数、描述和一串参数说明。模型看到的其实是这些说明执行时才会真正映射到函数。我参照这个思路实现了一个注册表。# tools.py import datetime TOOL_REGISTRY {} def register_tool(name, description, parameters_schema): def decorator(fn): TOOL_REGISTRY[name] { fn: fn, description: description, parameters: parameters_schema, } return fn return decorator register_tool( nameget_current_time, description获取当前本地时间用于需要知道现在几点、几号、星期几的场景, parameters_schema{ type: object, properties: {}, }, ) def get_current_time(): now datetime.datetime.now() return f{now:%Y-%m-%d %H:%M:%S 星期%w} register_tool( namecalculate, description计算四则运算表达式输入必须是合法的数学表达式例如 (3.5*60)120, parameters_schema{ type: object, properties: { expression: { type: string, description: 数学表达式比如 3.5 * 60, } }, required: [expression], }, ) def calculate(expression): # 这里做了一个最简单的安全过滤只允许数字和运算符 allowed set(0123456789-*/.() ) if set(expression) - allowed: raise ValueError(表达式包含非法字符) return str(eval(expression))这两个工具加在一起还不到四十行但已经足够展示注册表的优势。之后想加get_weather、search只需要再写一个函数加一个装饰器主循环完全不用改。工具描述写得好不好直接影响模型调用准确率。你写description时不能太抽象最好告诉模型什么场景下用、期望拿到什么返回值就像教一个实习生使用新工具一样。calculate里我用了一个简单字符白名单来限制表达式内容。生产环境直接eval是危险行为但在本地最小实现里这个白名单已经能挡住绝大多数意外输入。如果你要把 Agent 推广到多人使用建议换成正则解析或安全的表达式计算库。3.3 agent.py一个最薄的 Agent 主循环主循环是整个项目的心脏。我参照 pi-agent 的做法把它压缩成一个run_agent函数。函数接收用户输入然后在一个for循环里反复调用模型、执行工具直到模型输出最终文本或达到最大轮数。# agent.py import json from models import chat_once from tools import TOOL_REGISTRY SYSTEM_PROMPT 你是一个能调用工具的智能助手。 当你需要获得实时信息或进行精确计算时请调用工具。 得到工具结果后基于结果回答用户。 如果任务已经完成直接用自然语言回复不要再重复调用工具。 def build_tool_schemas(): schemas [] for name, meta in TOOL_REGISTRY.items(): schemas.append({ type: function, function: { name: name, description: meta[description], parameters: meta[parameters], }, }) return schemas def execute_tool(name, arguments): meta TOOL_REGISTRY.get(name) if meta is None: return f错误工具 {name} 不存在 try: args json.loads(arguments) if isinstance(arguments, str) else arguments result meta[fn](**args) return str(result) except Exception as e: return f工具执行出错{e} def run_agent(user_input, max_steps10): messages [{role: system, content: SYSTEM_PROMPT}] messages.append({role: user, content: user_input}) for step in range(max_steps): resp chat_once(messages, toolsbuild_tool_schemas(), configCONFIG) text resp[text] tool_calls resp[tool_calls] if not tool_calls: return text messages.append({ role: assistant, content: text, tool_calls: tool_calls, }) for call in tool_calls: fn_name call[function][name] fn_args call[function][arguments] result execute_tool(fn_name, fn_args) messages.append({ role: tool, tool_call_id: call.get(id, fcall_{step}_{fn_name}), content: result, }) return 已达到最大轮数任务未完成。这段代码有两点需要特别说明。第一当模型发起工具调用时我们必须先把 assistant 这条消息原样追加到历史里再追加对应的 tool 消息。这样模型将来回顾上下文时才能理解我上次调用了一个工具得到了这样一个结果。如果漏掉 assistant 的消息很多接口会直接报错或产生上下文错乱。第二max_steps不是可有可无的保险而是必备的最后一道防线。模型在幻觉状态下会反复调用工具没有轮数限制这个 Agent 可能永远也结束不了。我故意没有在这个循环里做复杂的异常处理只让异常以普通文本形式回到消息列表。这样设计的好处是模型看到工具执行出错之后有可能会自我纠错换个参数重新调用而不是整个流程直接崩溃。3.4 一条消息是怎么被循环改写的光看代码可能还是不够直观。我用一个真实的交互序列来展示消息列表是如何一步步变长的。假设用户输入是现在是几点过 3.5 个小时后是几点第一步messages里只有 system 和 user 两条。模型看到用户要时间决定调用get_current_time返回一个空的文本和一个工具调用。执行循环将 assistant 消息和 tool 消息追加进去此时messages变成四条。第二步模型看到当前时间开始做加法它可能再次调用calculate(21:30 对应的时间换算)不过这里为了简单我们让它输出最终回答现在是 21:303.5 小时后是次日 01:00。tool_calls为空循环结束。这个状态流转过程我建议你在调试时把每条消息的 role、content 都打印出来看。pi-agent 内部没有魔法模型能看到的全部信息都躺在messages数组里。你只要盯着这个数组就一定能定位出绝大多数逻辑问题。4. 本地跑通之后我踩过的三个坑和对应解法4.1 坑一模型返回的 JSON 参数不是真的 JSON第一个让我头疼的问题是解析工具参数。模型返回的call[function][arguments]是一个字符串但当我用json.loads去解析时经常报错。错误类型五花八门参数里出现单引号、多余换行、前后的花括号数量不对、甚至混进了类似思考过程的碎碎念。原因在于虽然接口约定返回的是合法 JSON但模型生成时依然可能带出各种不规范的文本片段尤其是在切换不同模型的时候。我的解决办法是写一个更宽容的解析函数def safe_json_loads(text): try: return json.loads(text) except json.JSONDecodeError: start text.find({) end text.rfind(}) if start -1 or end -1: raise ValueError(找不到合法 JSON) return json.loads(text[start:end 1])先用标准解析失败后再手动提取最外层的花括号范围重新尝试。如果这一步还是失败我会把错误信息以工具消息的形式返回给模型让它重新生成参数。这个策略实际跑下来大部分错误都能在第一轮自我修复中解决。Agent 开发的第一课就是任何从模型回来的文本都要按不可信输入处理。它返回的内容不是你写的代码而是一个概率采样结果永远不要假设它一定符合格式约定。4.2 坑二上下文被工具结果撑爆第二三轮开始乱答第二个问题出现在工具返回内容比较长的时候。我的某个工具会返回一段较长的状态信息第一次调用没问题但到了第三轮模型突然开始重复之前的回答或者忽略新工具结果。我打印出消息列表一看模型上下文窗口被一大堆历史工具输出占满了关键信息被淹没。解决办法是控制工具结果的长度。我写了一个truncate_result函数限制单条工具返回不能超过 800 个字符超过的部分用省略提示代替def truncate_result(content, limit800): content str(content) if len(content) limit: return content return content[:limit] \n...[结果过长已截断]这个操作虽然粗暴但在最小实现阶段非常有效。模型需要的是工具结果里的关键数字、状态、结论而不需要整个原始日志。类似的思路还包括定期丢弃最早的历史消息只保留最近几轮。pi-agent 这类轻量项目通常会把上下文管理交给调用方也就是说你要自己权衡保留多少历史。4.3 坑三Agent 陷入死循环反复调用同一个工具第三个坑也最有代表性模型反复调用同一个工具而且每次参数完全一样。比如它连续五次调用calculate(11)翻来覆去就是不给最终答案。我一开始以为是工具实现出了问题后来才意识到是主循环缺少重复动作检测。原因通常有两种。第一种是 prompt 里没有说清楚拿到结果后必须结束模型误以为每个工具调用都要配一个后续动作第二种是模型产生了某种生成退化陷入局部循环。解决思路也是两层第一层在 system prompt 里加强约束明确告诉模型完成用户目标后立即用自然语言回复禁止重复相同工具调用。第二层在循环里加一个简单的重复检测连续三次相同工具加相同参数就直接中断并返回当前信息if len(tool_calls) 0: sig f{fn_name}_{fn_args} if sig last_signature: same_count 1 else: same_count 0 last_signature sig if same_count 3: return 检测到重复工具调用已自动停止这个保护机制让我免于一次非常尴尬的演示事故。现在你会发现Agent 开发的稳定性问题很多都不是模型的问题而是循环控制的问题。一个轻量主循环看似简单真正让它稳定运行需要反复打磨边界条件。5. 从能跑到能用我在复刻之后进一步加了三个模块5.1 短期记忆给 Agent 一个刚做过什么的便签基础循环跑通之后我很快遇到一个体验问题如果用户输入帮我记录下来 3.5 小时后的闹钟再追问我说的是几小时后模型可能已经完全不知道了。原因很简单我的循环只在内部处理工具调用没有把用户目标沉淀成短期记忆。我参照 pi-agent 的扩展思路在 Agent 里加了一个简单的内部状态字典用来记录最近一次用户请求和最近一次工具结果memory { last_user_intent: user_input, last_tool_result: None, }每次工具执行完成后除了把结果写入消息列表也同步更新这个字典。下一轮对话时我可以把记忆内容拼进 system prompt让模型知道用户刚才想做的是 X上次工具得到的结果是 Y。这个设计比直接堆整个历史消息更省 token也比完全依赖消息窗口更稳定。真正的长期记忆需要向量数据库但大部分场景下一张便签已经够用。5.2 规划与子任务拆解先列清单再做执行第二个让我觉得质变的模块是规划。pi-agent 的做法给了我很大启发在真正执行之前让模型先输出一个任务清单然后按清单逐步执行。这个思想被称为 plan-then-execute它在面对复杂问题时能显著降低模型在中途迷失的概率。我的实现很粗糙在 system prompt 之后增加一个planner阶段要求模型先输出步骤列表我解析这个列表再把它转换成若干独立的子任务逐个喂给原来的主循环。比如用户问如果现在是晚上 23:00帮我算一下 8 小时后是什么时候并判断是否适合吃早饭。原始循环可能会在时间换算和推理之间来回切换加了规划后模型先输出1. 获取当前时间2. 用计算时间3. 根据结果判断。规划层再把这三步依次交给执行循环每步完成后把结果带回到规划层。这样整个行为变得可控得多。坏消息是代码量会增加不少所以它更适合作为第二阶段的优化不建议第一次实现就加。5.3 约束与权限防止模型乱调用危险函数最后加上去的模块是约束与权限。本地实现看着无关紧要但只要你给模型接入任何能写文件、发消息、执行系统命令的工具就必须认真对待权限问题。我在工具注册表里加了一个字段dangerous并且在执行入口做了一次检查def execute_tool(name, arguments, require_confirmFalse): meta TOOL_REGISTRY.get(name) if meta.get(dangerous) and require_confirm: confirm input(f确认执行危险工具 {name}输入 y 继续) if confirm.lower() ! y: return 用户已取消这次执行 # 继续原有执行逻辑这只是最基础的一层防护生产环境还需要更细的审计日志和角色鉴权。但这个机制让我意识到Agent 工具调用本质上是模型替你下指令、代码替你动手。你在任何框架里都应该保留一层人类可控的闸门不要把执行权完全交给模型的概率输出。整个复刻做完之后我最深的感觉是Agent 开发真正难的地方不在循环本身而在于你愿意花多少功夫去处理模型输出的不可靠性和上下文的膨胀。pi-agent 的价值不是代码有多惊艳而是它把 Agent 最核心的执行链用一种容易被模仿的方式摆在你面前。如果你也想试建议直接照今天这份骨架先跑通再往里面加你真正需要的工具。还有一个很实用的调试小技巧在循环里把每一步messages数组打印出来你会立刻看清模型是看到了什么才做出这个决定这个习惯能帮你省下大量瞎猜的时间。