做智能体开发时很多人第一步不是缺模型而是缺一套能讲清楚原理的最小架构。Pi Agent 就是这样一个方向上的最小实现它不追求功能堆叠而是把智能体最核心的“接收任务、组织上下文、调用工具、循环推理、返回结果”拆成可读、可改、可调试的模块。这篇文章会用 Python 从零搭建一个名为 Pi Agent 的最简智能体解释每个模块为什么存在、如何协作、遇到问题该怎么排查。文章适合三类读者刚开始接触 Agent 开发、想理解 LangChain 或 Dify 这类平台底层机制、以及想在自有项目中按需裁剪架构的开发者。读完以后你能独立实现一个最小可运行的智能体循环能看懂工具注册和消息历史的组织方式也能在模型不按预期返回时找到排查入口。1. 先理解智能体架构要解决什么问题1.1 智能体和普通 LLM 对话的本质区别普通 LLM 对话是“一问一答”用户传入一段文本模型返回一段文本。整个过程没有状态没有外部动作模型只能依靠训练时学到的知识回答。智能体则不同它的核心是“Agent Loop”也就是一个持续循环模型根据当前对话判断下一步要做什么如果发现需要查询资料或执行操作就生成一次工具调用请求程序执行工具后把结果写回对话模型再基于新信息继续推理直到它认为任务完成。这个区别决定了架构设计的起点。普通对话只需要一个 chat 接口而智能体至少需要四条链路模型接入、工具注册、消息历史、循环控制。任何一个环节缺失智能体都会退化成“套了壳的聊天机器人”。1.2 从 ReAct 模式理解 Agent 的执行本质ReActReasoning and Acting是当前大多数智能体框架的理论基础。它的核心思想是让模型交替进行“推理”和“行动”推理用于决定做什么行动用于实际执行行动的结果再作为新输入进入下一轮推理。用一句话概括智能体的能力来自“把模型输出的意图变成程序可执行的动作再把执行结果反馈给模型”。Pi Agent 的主循环就是 ReAct 的落地版本。每一轮循环做三件事把当前完整消息历史发送给模型。模型返回两种结果之一要么是工具调用请求要么是最终回答。如果是工具调用就执行工具并把结果写入消息历史继续下一轮如果是最终回答则结束循环。这个模式看起来简单但它包含了智能体最重要的设计决策模型不直接执行代码而是通过结构化参数描述“想做什么”真正执行权由程序持有。这样既能控制权限又能对每次操作做日志记录和异常处理。1.3 Pi Agent 的架构目标和适用范围Pi Agent 在这里是一个教学级最小实现不是要替代成熟框架。它的架构目标有三个每个模块职责单一能独立替换。代码量控制在数百行以内能完整阅读。依赖尽量少只保留模型调用和基础标准库。因此它适合本地学习、内部工具原型、以及给其他项目提供架构参考。如果任务涉及多用户并发、复杂工作流编排、大规模知识库检索、生产级权限管控则需要参考第 6 章的扩展路径迁移到完整框架。模块对应职责类比LLM 客户端封装模型接口统一发送消息和工具定义神经系统负责思考和表达工具注册表管理可执行函数及其参数声明四肢负责执行动作消息历史保存对话、工具调用和工具结果记忆负责维持上下文Agent 主循环控制推理与行动的交替过程大脑负责决策和调度入口配置组装以上模块并读取环境配置骨架负责连接各部分这张表也是后文的实现顺序。先写模型接入再写工具再写消息管理最后写循环逻辑上最顺。2. 环境准备、依赖与项目结构2.1 运行环境要求Pi Agent 对运行环境的要求很低学习阶段不需要 GPU也不需要复杂的分布式环境。只需要一台能访问模型服务的机器Python 3.10 或更高版本即可。项目推荐配置说明操作系统Windows 10/11、Linux、macOS本文命令以 Linux/macOS 为主Windows 用等效命令替代Python3.10 及以上使用match和类型注解时版本过低会报错模型接口OpenAI 兼容的 chat/completions 接口可用云服务也可用本地部署的兼容服务第三方库requests仅 HTTP 客户端用于调用模型接口模型方面建议选择支持 function calling工具调用的模型。不同模型的函数调用格式略有差异但主流模型基本都兼容 OpenAI 定义的tools参数结构所以下面代码统一按这套格式实现。注意落地前先确认你使用的模型服务是否支持tools参数。如果不支持工具调用模型永远不会返回tool_calls智能体只能做普通对话。2.2 项目目录结构代码按模块拆分每个文件对应文章第 1.3 节表格中的一个职责pi-agent/ ├── main.py # 入口配置模型、注册工具、启动对话 ├── agent.py # Agent 主循环 ├── llm_client.py # LLM 客户端封装 ├── tools.py # 工具注册表 ├── memory.py # 消息历史管理 └── requirements.txt # 依赖列表目录结构刻意保持扁平。智能体学习阶段最难的不是代码量而是“模块边界”不清晰。每个文件只做一件事排查问题时就只看对应文件。2.3 依赖安装创建虚拟环境并安装依赖python -m venv .venr source .venv/bin/activate pip install requestsrequests是唯一必需的外部依赖。如果你使用 Windows激活虚拟环境命令为.venv\Scripts\activate如果不使用虚拟环境直接pip install requests也可以但推荐在项目中始终使用虚拟环境。完成后生成requirements.txtpip freeze requirements.txt3. 五个核心模块的实现3.1 LLM 客户端封装统一模型接入层llm_client.py的作用是把模型 API 调用集中在一个类里。后续在任何地方需要模型能力都只调用chat()方法不需要关心 HTTP 细节。# llm_client.py import requests class LLMClient: def __init__(self, base_url, api_key, model, temperature0.7, timeout60): self.base_url base_url.rstrip(/) self.api_key api_key self.model model self.temperature temperature self.timeout timeout def chat(self, messages, toolsNone): url f{self.base_url}/v1/chat/completions headers { Authorization: fBearer {self.api_key}, Content-Type: application/json, } payload { model: self.model, messages: messages, temperature: self.temperature, } if tools: payload[tools] tools payload[tool_choice] auto resp requests.post(url, headersheaders, jsonpayload, timeoutself.timeout) resp.raise_for_status() choice resp.json()[choices][0] return choice这里几个参数值得说明base_url指向模型服务的根地址chat()内部拼接出完整的 chat/completions 路径。api_key从环境变量读取更安全不要硬编码到代码里。tool_choiceauto表示由模型自主决定是否调用工具、调用哪个工具。temperature控制随机性。工具调用类任务建议调低到 0.2 左右降低格式漂移概率。封装之后主循环不需要关心模型是云服务还是本地服务也不需要关心认证方式。3.2 工具注册表让模型具备调用真实能力工具注册表有两个职责第一维护一份“模型可见的工具声明列表”模型根据这份声明决定调用什么第二维护“程序实际执行的函数映射”收到工具调用请求后能快速定位处理函数。# tools.py import json class ToolRegistry: def __init__(self): self._definitions [] self._handlers {} def register(self, definition): def decorator(func): self._definitions.append(definition) self._handlers[definition[name]] func return func return decorator def schemas(self): return self._definitions def execute(self, name, arguments_json): if name not in self._handlers: return json.dumps({error: ftool {name} not found}, ensure_asciiFalse) try: args json.loads(arguments_json) result self._handlers[name](**args) return json.dumps(result, ensure_asciiFalse) except Exception as exc: return json.dumps({error: str(exc)}, ensure_asciiFalse)工具声明使用 OpenAI 兼容的 JSON Schema 格式。模型不直接接收 Python 函数它接收的是结构化的声明。下面注册两个示例工具# main.py片段 from tools import ToolRegistry registry ToolRegistry() registry.register( { name: get_current_time, description: 获取当前日期和时间, parameters: {type: object, properties: {}}, } ) def get_current_time(): import datetime return {time: datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S)} registry.register( { name: calc, description: 计算简单四则运算表达式例如 1 2 * 3, parameters: { type: object, properties: {expr: {type: string}}, required: [expr], }, } ) def calc(expr): return {result: safe_calc(expr)}calc工具不能使用eval。eval可以执行任意代码传入__import__(os).system(rm -rf /)这类字符串会带来严重风险。这里用ast模块实现一个只支持加减乘除的解析器# main.py片段 import ast import operator _OPERATORS { ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, } def _safe_eval(node): if isinstance(node, ast.Expression): return _safe_eval(node.body) if isinstance(node, ast.BinOp): op _OPERATORS[type(node.op)] return op(_safe_eval(node.left), _safe_eval(node.right)) if isinstance(node, ast.Constant): return node.value raise ValueError(f不支持的表达式节点: {type(node).__name__}) def safe_calc(expr): tree ast.parse(expr, modeeval) return _safe_eval(tree)这个例子说明了一个重要原则智能体框架中模型只负责“提出动作意图”程序必须自己保证动作的安全性。凡是涉及文件删除、命令执行、网络请求的工具都要加权限校验和审计日志不能因为模型说要做就直接执行。3.3 消息管理工具结果如何写回上下文LLM 是无状态接口每一轮调用都要把完整上下文重新发送。因此智能体必须自己维护消息历史并把工具调用和工具结果按协议格式写回。# memory.py class MessageHistory: def __init__(self, max_messages40): self.messages [] self.max_messages max_messages def add(self, message): self.messages.append(message) self._trim() def extend(self, messages): self.messages.extend(messages) self._trim() def _trim(self): if len(self.messages) self.max_messages: self.messages self.messages[-self.max_messages:] def to_list(self): return self.messages消息历史中的角色有三种含义不同角色来源作用system程序初始化时注入设定全局行为规则约束模型输出格式user用户输入描述任务本身assistant模型输出包含推理内容也可能包含 tool_callstool工具执行结果回填某个 tool_call 的执行结果特别注意tool消息必须包含tool_call_id用来和 assistant 消息里的tool_calls[].id对应。如果漏掉这个字段模型服务通常会返回校验错误这是新手最容易踩的坑之一。_trim用来控制消息条数避免上下文无限膨胀。但简单截尾会丢弃最早的系统提示和早期上下文生产级方案需要对消息做摘要压缩这在第 6 章展开。3.4 Agent 主循环推理与行动的交替agent.py是核心中的核心它把前面三个模块串起来。系统提示词在这里注入循环在这里控制结束条件也在这里判断。# agent.py SYSTEM_PROMPT ( 你是一个运行在程序里的智能体助手。\n 当你需要查询实时信息或执行计算时调用可用工具。\n 工具结果会以 tool 消息返回你可以基于结果继续推理。\n 如果不需要工具直接给出最终答案。 ) class PiAgent: def __init__(self, llm, tools, memory, max_steps10): self.llm llm self.tools tools self.memory memory self.max_steps max_steps def run(self, user_input): self.memory.add({role: user, content: user_input}) for step in range(1, self.max_steps 1): print(f[step {step}] 调用模型...) choice self.llm.chat( self.memory.to_list(), toolsself.tools.schemas(), ) msg choice[message] self.memory.add( { role: assistant, content: msg.get(content), tool_calls: msg.get(tool_calls), } ) if not msg.get(tool_calls): return msg.get(content) for tool_call in msg[tool_calls]: name tool_call[function][name] args tool_call[function][arguments] print(f[step {step}] 调用工具: {name}({args})) result self.tools.execute(name, args) self.memory.add( { role: tool, tool_call_id: tool_call[id], content: result, } ) return f已达到最大执行步数 {self.max_steps}任务终止。主循环的判断逻辑要重点理解每次循环先发送当前全部消息和工具声明。模型返回的 assistant 消息原样写入历史。包括content为None但带tool_calls的情况。如果tool_calls为空说明模型认为不需要调用工具直接返回内容。如果有多个tool_calls逐个执行把结果以tool角色写回。写回后进入下一轮循环让模型看到工具结果。max_steps是防止死循环的保护阀。模型在复杂任务中可能反复调用工具而不收敛必须限定轮数。默认 10 在学习和原型阶段足够。3.5 入口配置与环境变量main.py负责组装所有模块并从环境变量读取配置# main.py import os from agent import PiAgent, SYSTEM_PROMPT from llm_client import LLMClient from memory import MessageHistory from tools import ToolRegistry # 在此处继续注册工具get_current_time、calc、safe_calc def main(): base_url os.getenv(LLM_BASE_URL, https://api.example.com) api_key os.getenv(LLM_API_KEY, your-key) model os.getenv(LLM_MODEL, your-model) llm LLMClient(base_urlbase_url, api_keyapi_key, modelmodel) tools ToolRegistry() # 注册工具 tools.register(get_current_time_definition)(get_current_time) # 注册 calc... memory MessageHistory(max_messages40) memory.add({role: system, content: SYSTEM_PROMPT}) agent PiAgent(llmllm, toolstools, memorymemory) print(Pi Agent 已启动输入 exit 退出。) while True: user_input input( ).strip() if user_input.lower() in (exit, quit): break answer agent.run(user_input) print(fAI: {answer}) if __name__ __main__: main()为了让注册代码更整洁可以把工具声明和函数放在一起用装饰器注册。上面第 3.2 节已经演示了装饰器写法入口里只需要保证所有工具模块被导入一次即可。4. 运行验证与结果分析4.1 跑通一次普通对话先用最简单的方式验证架构能跑通。设置环境变量后启动export LLM_BASE_URLhttps://api.example.com export LLM_API_KEYyour-key export LLM_MODELyour-model python main.py输入一句不需要工具的问题 你好请介绍一下你自己 AI: 你好我是一个运行在程序里的智能体助手可以回答问题和调用工具完成计算、查询等任务。这里验证的是主循环的“无工具分支”模型返回的 assistant 消息没有tool_calls循环在第一步结束正常返回内容。4.2 跑通一次工具调用输入需要工具的指令 现在几点了 [step 1] 调用模型... [step 1] 调用工具: get_current_time() AI: 当前时间是 2026-05-12 14:30:25。从日志可以看到典型的 Agent 行为第一轮模型返回tool_calls程序执行时间工具把结果写回第二轮模型基于时间结果组织回答不再调用工具循环结束。再验证计算工具 计算 (3 5) * 2 的结果 [step 1] 调用模型... [step 1] 调用工具: calc({expr: (3 5) * 2}) AI: 计算结果为 16。如果模型在一个回复里产生多个工具调用比如“现在几点并计算 11”日志中会连续出现两个“调用工具”行然后再进入下一轮。4.3 中间状态和日志怎么看排查问题时最重要的不是最终答案而是中间状态。Pi Agent 的run()方法里已经打印了 step 编号和工具调用参数。建议再增加一个可选的调试模式打印每次发往模型的消息结构# agent.py增加调试模式 def run(self, user_input, debugFalse): ... if debug: for message in self.memory.to_list(): print(---- message ----) print(json.dumps(message, ensure_asciiFalse, indent2))调试模式的判断顺序是看 step 是否递增确认循环正常推进。看模型返回的消息结构确认tool_calls是否按预期生成。看工具调用参数确认模型生成的 JSON 参数是否合法。看 tool 消息是否包含正确的tool_call_id。看最终回答是否基于工具结果生成而不是凭空编造。这套观察顺序也和第 5 章的排查路径一致。5. 常见问题与排查路径5.1 模型一直不调用工具现象用户明确要求查询时间或计算模型却输出文字猜测不返回tool_calls。可能原因模型服务不支持tools参数或当前模型版本没有工具调用能力。工具声明格式与模型要求不匹配模型解析失败后干脆不调用。系统提示词没有说明工具的存在模型不知道有工具可用。tool_choice被设置为none。排查顺序直接手工构造一次带tools的 API 请求确认模型能否返回tool_calls。打印tools.schemas()检查 JSON Schema 格式是否符合 OpenAI 兼容规范。检查系统提示词中是否包含了“可以调用工具”的说明。临时把tool_choice改为required测试确认模型本身支持强制调用。5.2 工具参数解析失败现象模型返回了tool_calls但json.loads(arguments)报错常见错误是 JSON 中包含多余换行、单引号或截断内容。可能原因模型生成的参数不是合法 JSON。长参数被上下文截断。模型把参数写成 Python 字典格式而非 JSON 格式。解决方案在ToolRegistry.execute()中已经捕获异常并返回错误信息模型会在下一轮看到错误并自我修正。提高模型 temperature 的稳定性建议工具调用场景设为 0.1 到 0.3。如果频繁出现截断检查消息是否超出模型的上下文窗口。5.3 上下文超限报错现象连续多轮对话后API 返回 400 错误提示上下文长度超出模型限制。可能原因消息历史无限增长没有裁剪单次工具结果太大。处理方式调低MessageHistory的max_messages。对工具返回结果做截断例如只保留前 2000 字符。生产环境应使用“滑动窗口 摘要压缩”策略。推荐在MessageHistory._trim中加入对单条消息长度的限制MAX_MESSAGE_LENGTH 4000 def _limit_content(self, message): if isinstance(message.get(content), str) and len(message[content]) MAX_MESSAGE_LENGTH: message[content] message[content][:MAX_MESSAGE_LENGTH] ...(truncated) return message5.4 工具执行死循环或不收敛现象模型反复调用工具日志一直递增最终被max_steps截断。可能原因工具返回结果不足以支撑模型判断“任务已完成”。工具结果中包含错误信息模型尝试重试但参数没有变化。系统提示词没有给出“何时停止调用工具”的明确标准。解决方案在系统提示词中补充“如果你已经拿到足够信息直接给出最终答案不要重复调用相同工具。”对相同工具和相同参数的调用做去重连续重复超过两次就返回错误。把max_steps从 10 调低到 5快速暴露不收敛问题。5.5 典型错误速查表错误现象常见原因检查方式处理建议API 返回 401api_key 错误或环境变量未设置打印os.getenv(LLM_API_KEY)检查环境变量和 key 权限API 返回 404base_url 拼接错误检查url base_url /v1/chat/completions确认服务根路径是否正确返回 400 “tool_call_id not found”tool 消息缺少对应 id打印 memory 全部消息确保 tool_call_id 从 assistant 消息中原样复制模型输出变成了 markdown 表格系统提示词未约束格式查看原始 message.content在 system prompt 中指定输出格式Windows 下中文乱码终端编码问题检查chcp输出在main.py开头设置sys.stdout.reconfigure(encodingutf-8)5.6 通用排查顺序无论遇到什么问题都按这个顺序排查确认输入模型收到的是什么消息与预期是否一致。确认格式tools声明和消息结构是否符合模型服务协议。确认配置模型名、base_url、api_key、timeout 是否正确。确认状态消息历史是否被正确裁剪和回填。确认日志是否有工具执行异常、JSON 解析异常等信息。6. 最佳实践与生产化扩展6.1 学习环境与生产环境的差异Pi Agent 的设计目标是讲清楚原理生产环境还需要补很多内容。两套环境的差异如下维度学习环境生产环境配置管理环境变量直接读取配置中心、加密存储、灰度发布日志print 输出结构化日志、链路追踪、监控告警工具安全简单校验权限模型、操作审计、人工审批上下文简单截断摘要压缩、向量检索、长期记忆错误处理捕获后返回错误文本重试、降级、熔断、补偿并发单次调用异步、限流、多租户隔离测试手工输入验证单元测试、回归测试、评估集其中工具安全是最关键的差异点。生产环境的每一个工具调用都应该记录“谁在什么对话中调用了什么工具、参数是什么、结果是什么、耗时多少”这份审计日志既是排查依据也是安全合规要求。6.2 从最简架构到完整框架的扩展路径Pi Agent 的五个模块可以映射到成熟框架的对应能力LLM 客户端扩展为多模型适配层支持不同厂商接口和自动切换。工具注册表扩展为插件系统支持动态加载、权限分组、依赖注入。消息历史扩展为记忆体系区分短期对话记忆和长期知识记忆。Agent 主循环扩展为工作流编排支持多智能体协作和人工干预。入口配置扩展为可视化配置平台例如 Dify、Coze 这类图形化工作流产品底层也遵循“模型调用 工具节点 上下文传递”的基本模型。如果你要选择一个成熟框架迁移可以从 Dify、LangGraph、Agentscope 等项目中对照阅读查它们的会话管理、工具节点、循环控制实现会发现核心模式和本文的 Pi Agent 一致只是在工程化层面做了大量增强。6.3 可复用的开发检查清单每次开发新智能体时按这份清单自查模型支持工具调用吗用小请求验证过吗工具声明里的 description 是否足够详细模型会依赖它决定调用时机。工具参数是不是合法的 JSON Schema工具函数内部有没有做输入校验和异常捕获系统提示词是否明确了“什么时候调用工具、什么时候直接回答”消息历史是否包含 system、user、assistant、tool 四种角色且顺序正确tool 消息的 tool_call_id 是否正确关联有没有 max_steps 防死循环工具执行是否记录日志敏感工具是否加了权限校验上下文会不会超限超限后的降级策略是什么错误信息是否回传给模型让它可以自我修正这 12 条对应了本文所有关键设计点工具协议、上下文组织、循环控制、异常恢复、安全边界。拿它去复查自己的智能体项目能避免大多数“为什么模型行为不稳定”的问题。6.4 下一步学习建议如果你刚看完本文建议不要急着换框架先把 Pi Agent 跑通然后依次做四个练习新增一个中文 JSON 工具返回自定义业务数据验证模型能正确解释并引用结果。增加调试模式查看一次多轮工具调用的完整消息历史。故意注册一个返回错误格式的工具观察模型如何在下一轮自我修正。对比不同max_steps之下模型在复杂任务上的表现差异。这四个练习覆盖了智能体开发的大部分基础能力。做完以后再去看 LangGraph 的多智能体协作、Dify 的工作流节点、以及函数调用的流式输出理解速度会明显快很多。最简架构的价值不在于功能多而在于让每一个决策都透明模型什么时候做决定、程序什么时候执行、上下文里发生了什么你都能一眼看到。把这条主线理解透后面无论是扩展工具、接入知识库、还是做多智能体编排都不会偏离真正重要的架构判断。
