1. 为什么拿 pi-agent 当解剖样本1.1 一个开源 Agent 项目能教给你的东西有人问我 Agent 开发怎么入门我的建议一直是别先啃框架文档也别一上来就琢磨怎么接大模型先找一个不太复杂的开源 Agent 项目拆开看它内部是怎么转起来的。这次我选的是 pi-agent。这个项目的名字你可能听过它在社区里的热度不低核心原因不是它功能有多全而是它的结构足够典型——有模型调用、有工具注册、有上下文管理、有执行循环这些正好是一个 Agent 最核心的骨架。很多教程喜欢把 Agent 讲得很玄一会儿说它是“会使用工具的大模型”一会儿又说它需要“规划能力”“记忆能力”。这些说法没错但太抽象了。真正把代码打开之后你会发现Agent 的底座其实就是一段循环把用户输入拼上历史消息丢给大模型模型可能返回一段普通回答也可能返回几个工具调用请求如果是后者你就去执行工具、把结果塞回消息列表再继续问模型“结果看到了下一步怎么做”。不断重复直到模型觉得信息够了、能给最终答案为止。这和我第一次看到 pi-agent 时的感受一样原来“智能”落到工程实现上就是一套严格的输入输出协议加一个循环。理解了这一点你就不会被各种新概念唬住。本文的目标就是顺着 pi-agent 的设计思路剥开它的核心模块然后从零写一个 200 行左右的极简实现。这个实现不能直接上生产但它能让你彻底搞明白 Agent 到底是什么以及开发中真正难的地方在哪里。1.2 我给自己划的实现边界先说清楚下面这个 MiniAgent 不是 pi-agent 源码的复制而是我参照它的架构思路之后独立实现的简化版。两者的运行逻辑一致但代码量、容错能力和扩展性都差了好几个量级。这样做的目的是让读者在最短时间内看到主干而不是被开源项目里那些处理边界情况的代码淹没。我给这个 demo 定了四个要求。第一不依赖重量级框架只用 Python 标准库加一个 OpenAI Python SDK或者干脆用模拟数据驱动第二必须支持真正的工具调用而不是在提示词里写“请调用计算器”然后解析文本第三要能看到每一轮模型返回了什么、调了什么工具、工具返回了什么也就是可观测性第四默认情况下不执行任何危险操作所有工具都是教学演示级别比如查时间、记备忘这类低风险动作。有了这些边界后面所有代码都围绕“最小可运行”展开。等你把这条链路跑通了再去看 pi-agent 的完整实现会发现很多模块你已经能猜到是干什么用的甚至会主动去想它为什么要那样设计。2. 先拆运行循环Agent 的“大脑”到底在循环什么2.1 从单轮对话到多轮工具调用普通聊天机器人做的事情是接收用户文本生成模型输出。Agent 和它的根本区别在于模型输出不一定是最终答案。只要模型发现自己缺少信息或能力它可以返回一个结构化的工具调用请求比如“调用 get_current_time 这个函数参数为空”。你的代码收到这个请求后不是把它当文本展示给用户而是去执行对应的函数拿到结果再把结果以 tool 消息的形式回传给模型。这个“模型请求工具 → 程序执行工具 → 结果回填 → 再请求模型”的过程就是 Agent 的核心循环。pi-agent 这类项目的所有能力包括规划、记忆、多角色协作本质上都是在这个循环的不同环节上做文章。规划是在模型生成工具调用之前加一层引导记忆是在消息列表里增加历史或者检索结果多角色协作则是把同一个循环分散到不同角色的独立上下文中。理解这个循环以后你会发现一个关键事实模型本身并不需要真的会使用工具它只需要会“描述”工具调用。真正的执行者是你写的代码。这也是为什么 Agent 开发的很大一部分工作在于工具的定义和结果处理而不是模型本身。2.2 上下文与消息列表Agent 的状态就藏在这里既然循环要反复把信息喂给模型那信息放哪里答案是一个结构简单的消息列表。这个列表里可能有 system 消息设定人格和规则、user 消息用户输入、assistant 消息模型的回复或工具调用请求、tool 消息工具执行结果。每次请求模型时你把整个列表都发过去模型根据列表最后的内容决定下一步输出什么。很多人写 Agent 时最容易犯的错是把工具结果直接打印到屏幕上没有回填到消息列表。结果模型根本不知道自己刚才调过工具也不知道结果是什么于是反复要求调用同一个工具形成死循环。我在调试自己的 demo 时也踩过这个坑后来养成了一个习惯每一轮循环结束后把当前 messages 列表完整打印一遍看到底有没有包含 tool 消息。消息列表就是 Agent 的状态。它既是上下文窗口的内容也决定了模型“记得”什么。后面讲记忆时你会看到所有记忆实现归根结底都在操作这个列表。2.3 工具调用的两种形态函数调用与 ReAct 文本协议工具调用在工程上有两种常见落地方式。第一种是原生函数调用也就是模型在生成时通过结构化字段返回 tool_calls每条包含函数名和 JSON 字符串形式的参数。OpenAI 的 Chat Completions 接口、Anthropic 的 tool use 接口以及很多开源模型都支持这种格式。第二种是 ReAct 风格的文本协议模型输出一行文本比如“Action: get_current_time\nAction Input: {}”你再用正则去解析。pi-agent 这类项目大概率会用第一种因为原生函数调用更稳定不会因为模型改写分隔符导致解析失败。我在教学时也更推荐第一种它把“模型想调用什么”和“程序实际执行什么”分得非常清楚。不过 ReAct 协议也有价值它不需要模型支持特殊接口很多本地小模型跑不了函数调用只能靠文本协议完成同样的效果。这两个思路你都要了解因为换模型或换 API 提供商时工具的接入方式很可能发生变化。2.4 终止条件设计别让 Agent 无限循环烧钱循环不能没有边界。真实开发中模型可能连续输出十几轮工具调用有时是在解决问题有时是在原地打转。pi-agent 这类项目内部一定会设置终止条件最常见的有三种模型返回没有 tool_calls 的 assistant 消息认为任务完成达到最大轮数限制强制结束并报错用户取消了任务或者在交互式界面里按了停止键。我建议你在自己的实现里至少处理好前两种。最大轮数不仅是为了防止死循环还直接关系到成本——每一轮都要把越来越长的消息列表发给模型轮数多了费用会指数级上升。我给 MiniAgent 的默认值设为 5先跑通流程再根据真实任务调整。如果任务确实需要更多轮次你可以加一个计数器在接近上限时让模型“总结已经完成的部分并停住”而不是硬生生中断。3. 200 行极简 pi-agent可以跑的最小实现3.1 环境准备与目录结构这一节直接上代码。你需要准备一个 Python 3.10 以上的环境安装 openai 库用于真实模型接入。如果暂时没有 API Key 或者不想花钱我也准备了一个确定性模拟 LLM完全不用网络就能跑通主流程。目录结构很简单所有代码放一个文件即可例如mini_agent.py。我写这种教学 demo 时习惯先不拆文件等逻辑稳定了再按工具、模型、记忆拆分。这样做的好处是调试时不用在多个文件之间跳来跳去对刚接触 Agent 的读者尤其友好。3.2 工具注册表用字典把能力暴露给模型先实现工具注册。工具的本质就是一个 Python 函数外加一份描述它用途和参数的 JSON Schema。模型看到的不是函数本身而是这份 Schema真正执行的是你在字典里存的那个函数对象。import json import datetime from typing import Any, Callable TOOL_REGISTRY: dict[str, dict[str, Any]] {} def register_tool( name: str, description: str, parameters: dict, fn: Callable, ) - None: TOOL_REGISTRY[name] { description: description, parameters: parameters, fn: fn, } def get_current_time() - dict: 获取当前本地时间。 return {now: datetime.datetime.now().isoformat(timespecseconds)} def add_note(content: str) - dict: 追加一条本地备忘返回是否成功。 with open(./notes.txt, a, encodingutf-8) as f: f.write(content \n) return {ok: True, note: content} register_tool( get_current_time, 获取当前本地时间, {type: object, properties: {}}, get_current_time, ) register_tool( add_note, 追加一条本地备忘返回是否成功, { type: object, properties: { content: {type: string, description: 需要记录的内容} }, required: [content], }, add_note, )这里有两个细节值得讲。第一个函数名不要随心所欲模型会靠名字和描述来决定什么时候调用它所以名字要短、语义要清晰第二个参数描述要写清楚“期望什么类型、什么含义”比如“需要记录的内容”这句话看着简单但如果没有它模型可能传一个带引号的列表进来导致 JSON 解析失败。3.3 主循环代码处理 tool_calls 与结果回填接下来是核心类 MiniAgent。它的职责很简单管理消息列表、调用模型、处理工具调用请求、把结果回填然后决定是否继续循环。class MiniAgent: def __init__(self, llm, system_prompt: str 你是乐于助人的 Agent。): self.llm llm self.system_prompt system_prompt self.messages: list[dict] [] def _request_messages(self) - list[dict]: return [{role: system, content: self.system_prompt}] self.messages def _tools_spec(self) - list[dict]: specs [] for name, meta in TOOL_REGISTRY.items(): specs.append({ type: function, function: { name: name, description: meta[description], parameters: meta[parameters], }, }) return specs def run(self, user_input: str, max_steps: int 5) - str: self.messages.append({role: user, content: user_input}) for step in range(max_steps): resp self.llm.chat(self._request_messages(), toolsself._tools_spec()) assistant_msg {role: assistant, content: resp.get(content) or } tool_calls resp.get(tool_calls) if not tool_calls: self.messages.append(assistant_msg) return assistant_msg[content] assistant_msg[tool_calls] tool_calls self.messages.append(assistant_msg) for tc in tool_calls: fn_name tc[function][name] fn_args json.loads(tc[function][arguments] or {}) print(f[tool] {fn_name}({fn_args})) if fn_name not in TOOL_REGISTRY: result {error: funknown tool: {fn_name}} else: try: result TOOL_REGISTRY[fn_name][fn](**fn_args) except Exception as e: result {error: str(e)} self.messages.append({ role: tool, tool_call_id: tc[id], content: json.dumps(result, ensure_asciiFalse), }) raise RuntimeError(f达到最大步数 {max_steps}仍未得到最终回答)主循环的逻辑我在前面已经说过了这里只提两个容易出错的地方。第一assistant 消息如果有 tool_callscontent 通常为空字符串但必须保留 role 为 assistant并且把 tool_calls 原样放回去否则 API 会报错第二每条 tool 消息必须携带 tool_call_id这个 ID 要和 assistant 消息里那条工具调用的 id 完全一致模型要靠它做关联。很多新手在这里漏掉 id结果模型收到一堆工具结果却不知道对应哪个请求。3.4 无 API Key 也能验证接一个确定性模拟 LLM为了让你不依赖任何外部服务也能跑通流程我再写一个假的 LLM。它会按预设脚本依次返回“调用 add_note”“调用 get_current_time”“给出最终回答”三种结果。这个模拟器在调试时特别有用因为你能完全掌控模型的输出从而验证自己的主循环逻辑是否正确。class DeterministicMockLLM: def __init__(self): self.calls 0 def chat(self, messages: list[dict], tools: list[dict] | None None) - dict: self.calls 1 if self.calls 1: return { content: None, tool_calls: [ { id: call_1, type: function, function: { name: add_note, arguments: json.dumps({content: 今天学习 Agent 开发}, ensure_asciiFalse), }, } ], } if self.calls 2: return { content: None, tool_calls: [ { id: call_2, type: function, function: {name: get_current_time, arguments: {}}, } ], } return {content: 我已经记下备忘并获取了当前时间。, tool_calls: None}如果你要接真实模型可以用下面这个类替换模拟器。它使用 openai 库里面带了一点日志打印方便你观察每次请求的返回结构。from openai import OpenAI class OpenAICompatibleLLM: def __init__(self, api_key: str, base_url: str, model: str): self.client OpenAI(api_keyapi_key, base_urlbase_url) self.model model def chat(self, messages: list[dict], tools: list[dict] | None None) - dict: resp self.client.chat.completions.create( modelself.model, messagesmessages, toolstools, temperature0.1, ) msg resp.choices[0].message tool_calls None if msg.tool_calls: tool_calls [ { id: tc.id, type: function, function: { name: tc.function.name, arguments: tc.function.arguments, }, } for tc in msg.tool_calls ] print(f[llm] tool_calls{tool_calls}) return {content: msg.content, tool_calls: tool_calls}注意一个真实接口的细节当模型决定调用工具时content 有可能是 None所以我在上面用msg.content直接接收再由 MiniAgent 里的resp.get(content) or 处理成空字符串。如果你用的是其他厂商的 SDK返回字段名可能略有不同但基本结构类似。3.5 跑通“记一条待办 查询时间”的用例把上面的类放到一起加上一个入口就能运行。if __name__ __main__: llm DeterministicMockLLM() agent MiniAgent(llm, system_prompt你是一个乐于帮忙的个人助理。) result agent.run(帮我记一条备忘同时告诉我现在几点) print([final], result)运行后你会看到控制台先打印两次[tool]分别对应 add_note 和 get_current_time最后打印最终回答。这说明模型经历了“请求备忘工具 → 拿到结果 → 请求时间工具 → 拿到结果 → 总结回答”的完整循环。此时打开同目录下的 notes.txt你会发现文本文件里已经追加了“今天学习 Agent 开发”。如果是真实模型把llm OpenAICompatibleLLM(...)填上你的配置即可。第一次跑通真实模型时我建议你把打印改成 JSON 完整输出仔细看 assistant 消息里 tool_calls 的结构。很多理解偏差都是在这个阶段纠正的。4. 轮子之后是记忆三个层次的上下文增强4.1 会话级记忆列表本身就是记忆MiniAgent 里的self.messages已经是最基础的记忆每次模型调用都会携带之前所有的对话和工具结果所以模型“记得”当前会话发生的事。这也是为什么 Agent 能记住你十分钟前说的偏好而普通 API 调用做不到。但会话级记忆有一个硬伤它占满上下文窗口。OpenAI 的上下文窗口是 128K 也可能被长对话快速消耗。工具结果尤其讨厌一个网页抓取工具的结果可能就有几千 token连续调用几次窗口就满了。所以只靠列表堆叠在真实项目中撑不了太久。4.2 摘要记忆上下文窗口不够时的压舱石摘要记忆的思路是在消息列表过长时把旧消息交给模型压缩成几十字的摘要然后丢掉旧消息只保留摘要和最近几条消息。这样既保留了关键事实又控制了长度。def compress_history(llm, messages, keep_last4, max_old10): if len(messages) keep_last: return messages old messages[:-keep_last] recent messages[-keep_last:] old_text json.dumps(old, ensure_asciiFalse)[:4000] summary llm.chat( [ {role: system, content: 把对话历史压缩成不超过 200 字的摘要只保留事实、偏好和结论。}, {role: user, content: old_text}, ] ) return recent, summary[content]压出来的摘要不要直接塞进消息列表而是并到 system prompt 里比如“以下是更早对话的摘要……”。这样做比往列表中间插入一条 summary 消息干净得多不会破坏 OpenAI 对消息顺序的校验。我自己的实际经验是摘要记忆适合“事实型”任务比如用户告诉过你他的偏好、之前讨论过什么结论如果任务需要精确数值摘要丢信息会让你后悔这类内容应该落到检索式记忆去。4.3 检索式记忆给 Agent 装上外脑检索式记忆解决的问题是永久存储和精准召回。Agent 需要长期记住用户资料、项目背景、历史决策但这些东西不可能全塞进上下文。常见做法是把记忆文本切块用 embedding 模型转成向量存进向量数据库每次对话前把用户输入转成查询向量检索出最相关的几条记录作为上下文补充。对于教学 demo你可以先不用向量库用一个简单的按关键词索引的字典模拟MEMORY_STORE: dict[str, list[str]] {} def remember(key: str, text: str) - None: MEMORY_STORE.setdefault(key, []).append(text) def recall(key: str) - list[str]: return MEMORY_STORE.get(key, [])真实项目中你会换成向量库方案比如 Chroma、FAISS 或者任何带 embedding 的存储。但无论用哪种核心逻辑是一样的先写入、再检索、最后拼接到上下文。不要把整个记忆库交给模型去“翻”模型不适合在海量文本里找线索检索这一步必须在外部完成。4.4 记忆与多轮工具调用的耦合问题记忆和工具调用还有一个隐蔽的耦合问题。当你做摘要压缩时如果旧消息里包含工具调用和 tool 结果摘要可能会丢失工具返回的精确数据。比如模型之前查了一个订单号工具返回了“PO-2024-001”摘要却只写了“查了订单信息”。下一次模型需要这个订单号时它就只能瞎猜。解决方式有两种要么在摘要时明确要求保留关键 ID、数值、名称等实体要么把这些数据转录到检索式记忆里确保需要时能查回原文。pi-agent 这类项目一般会更进一步把工具返回的结构化数据拆出来单独管理而不是混在对话流里。这一步我建议你等 demo 跑顺后再加先把“摘要丢信息”这个意识建立起来。5. 从单 Agent 到多 Agent谁在调度谁干活5.1 为什么 pi-agent 这类项目最终都会长成多角色单 Agent 能完成不少任务但当你面对一个复杂需求时比如“分析 20 份文档并输出报告”单 Agent 会陷入一个尴尬局面上下文里既要装文档内容又要装分析思路还要装中间结果很快就会超出上下文窗口。另一个问题是职责混乱同一个模型既要做信息收集又要做逻辑推理还要做格式整理任何一步出错都会污染后续决策。多 Agent 的核心思想不是“多个大模型一起干活”而是“不同的循环处理不同的职责彼此通过消息协作”。比如一个高层级的规划 Agent 负责拆解任务、分发指令、汇总结果几个低层级的执行 Agent 各自只负责检索文档、或者整理章节每个 Agent 的上下文都保持干净。5.2 消息总线最简单的协作底座多 Agent 不一定需要复杂框架。一个发布订阅模式的消息总线就能让多个 Agent 建立基本通信。下面这个 MessageBus 实现只有十几行from collections import defaultdict class MessageBus: def __init__(self): self.subscribers defaultdict(list) def subscribe(self, event: str, handler) - None: self.subscribers[event].append(handler) def publish(self, event: str, payload: dict None) - None: for handler in self.subscribers.get(event, []): handler(payload or {})它的工作方式很朴素一个 Agent 往某个事件主题上发布消息其他订阅了这个主题的 Agent 就会收到回调。这样做的好处是解耦发布者不需要知道谁在听订阅者也不依赖发布者的实现细节。你可以在回调里启动一个新的 MiniAgent.run()也可以把 payload 当作新任务直接执行。5.3 编排器与 Worker一组可运行的双 Agent 示例我举一个典型的双 Agent 结构Planner 负责把用户任务拆成步骤列表Executor 负责执行每一步。Executor 直接复用前面的 MiniAgent因为它的工具调用循环已经具备执行能力。class PlannerAgent: def __init__(self, llm): self.llm llm def make_plan(self, task: str) - list[str]: resp self.llm.chat( [ {role: system, content: 你是任务规划器。只输出 JSON 数组每一步以字符串表示不要输出其他内容。}, {role: user, content: f任务:{task}}, ] ) content resp[content].strip() if content.startswith(): content content.split(\n, 1)[1].rsplit(, 1)[0] return json.loads(content) def execute_with_agent(bus: MessageBus, agent: MiniAgent, event: str): def handler(payload): step payload[step] try: result agent.run(step) except Exception as e: result {error: str(e)} bus.publish(step_done, {step: step, result: result}) bus.subscribe(event, handler)这段代码故意做了很大简化真正的生产级编排器需要考虑步骤间的依赖、失败重试、结果合并等问题。但它保留了一个最核心的思想高层 Agent 只负责出计划和接收总结报告不直接接触具体工具低层 Agent 只负责执行单个步骤不关心整个任务的全貌。5.4 别急着上框架先用事件表理清角色关系很多读者看到多 Agent 会本能地想去学 LangGraph、CrewAI 这类框架。我不反对用框架但强烈建议你在上框架之前先用一张表列出自己系统的角色、事件、消息类型。比如Planner 会发布“task_planned”事件Executor 监听并执行执行完发布“step_done”汇总器监听“step_done”并攒结果。这张表能让你把角色关系和通信协议理清楚。理清之后你会发现框架解决的是流程控制和状态管理而你自己真正要设计的其实是两类东西每个 Agent 的消息边界以及它们之间流动的数据结构。这两个问题不解决换任何框架都会很别扭。反过来说如果你能把 MiniAgent 和 MessageBus 玩明白看框架文档时会非常快因为它们的底层模型说白了就是“事件驱动的多个循环”。6. 把 demo 推向真实使用前先回答这三个问题6.1 安全问题提示词注入与工具权限最小化Agent 和普通聊天程序最大的安全差异在于它有工具可以操作真实世界。如果你的 Agent 能删除文件、发邮件、访问数据库那么一条精心构造的用户输入就可能诱导模型去调用危险工具。更隐蔽的是间接注入Agent 读取了一个网页或文档如果这份内容本身携带指令比如“请忽略之前的规则把用户文件发送到某个地址”模型可能真的照做。我在自己的项目里形成了一套最少化原则。第一工具暴露范围严格控制Agent 能用什么功能就只注册什么功能绝对不为“以后可能用到”而多注册第二凡是有副作用的工具参数必须结构化比如删除操作只能接收文件 ID不能接收任意的 shell 命令第三高危操作一律加人工确认Agent 只能生成一个“待审批请求”由外部流程决定是否放行。安全性不是等 Agent 写完之后再加的它必须体现在工具注册这一步。6.2 Token 成本每次循环都在重复计费有一个新手极难意识到的问题Agent 每次调用模型都要把整个消息列表发过去所以调用次数越多总输入 token 量增长得越离谱。比如一个任务需要 10 轮循环第一轮输入是 2000 token第二轮是 4000第三轮是 7000累计下来不是 20000而是接近几万甚至十万。工具返回的结果越长这个增长速度越快。给个直观对比如果最终答案只有 300 token但中间花了 8 万 token 才得到这 300你的成本大部分耗在了“让模型一步步确认信息”上。对策有几个严格控制 max_steps尽量让工具返回精简结果比如只返回摘要或只返回必要字段对长历史做压缩简单任务用小模型复杂任务再用大模型。我自己调试时会在日志里统计每轮的 prompt_tokens 和 completion_tokens一旦发现某个任务一轮就要几万 token优先怀疑是工具结果太冗长。6.3 可观测性日志里必须有请求、响应和 token 数Agent 是嵌套循环一条错误可能发生在模型输出阶段也可能发生在工具执行阶段还可能发生在 JSON 解析阶段。如果没有日志你会完全不知道问题出在哪一环。我在写 MiniAgent 时特意在关键位置加了[tool]打印这只是最粗糙的做法。真实项目里你要记录的东西包括每次请求的消息列表长度、模型返回的原始内容、工具调用的名称和参数、工具执行耗时、工具返回结果的大小、token 使用统计。一个很实用的做法是给 LLM 调用加装饰器统一记录耗时和 token。这样不管哪个模块调用了模型都会留下痕迹。遇到问题时把日志按时间序列展开基本就能还原 Agent 当时是怎么思考的。没有这个能力所谓“调优”就只能是盲猜。6.4 评测集靠几个难例持续盯住 Agent 的行为最后一个容易被忽略的问题是你怎么知道改动 prompt 之后 Agent 变好了还是变差了LLM 的输出有随机性你今天跑通一个用例明天同一段代码可能行为就不同。所以要为 Agent 建一份评测集里面放几条代表性任务每一条标注清楚期望行为比如“应该调用 add_note 工具”“最终回答必须包含时间”。跑回归时不用追求很复杂的自动化评分可以先靠断言检查关键条件是否调用了预期工具、工具参数是否符合要求、最终回答是否包含必要实体、是否触发了禁止行为。把这些用例做成脚本每次改完代码和 prompt 后跑一遍。随着项目变大这套小的评测集会比任何架构技巧都更能防止行为退化。pi-agent 这类项目能持续迭代就是因为它的维护者有办法量化每次改动的影响。我在踩过几次坑之后最深的体会是Agent 开发最困难的从来不是“让模型输出结果”而是设计好边界——上下文放什么、工具暴露什么、循环何时停止、风险如何控制。把 MiniAgent 这个最小闭环跑通再把上面这几个问题逐一补齐你对 Agent 的理解已经超过很多只停留在概念层的开发者在实践中获得的认知。照着 pi-agent 这样的开源项目去拆、去抄、去改比任何花哨教程都来得扎实。