从零搭建最简智能体：Pi Agent架构设计与Python实现

做智能体开发时，很多人第一步不是缺模型，而是缺一套能讲清楚原理的最小架构。Pi Agent 就是这样一个方向上的最小实现：它不追求功能堆叠，而是把智能体最核心的“接收任务、组织上下文、调用工具、循环推理、返回结果”拆成可读、可改、可调试的模块。这篇文章会用 Python 从零搭建一个名为 Pi Agent 的最简智能体，解释每个模块为什么存在、如何协作、遇到问题该怎么排查。

文章适合三类读者：刚开始接触 Agent 开发、想理解 LangChain 或 Dify 这类平台底层机制、以及想在自有项目中按需裁剪架构的开发者。读完以后，你能独立实现一个最小可运行的智能体循环，能看懂工具注册和消息历史的组织方式，也能在模型不按预期返回时找到排查入口。

1. 先理解智能体架构要解决什么问题

1.1 智能体和普通 LLM 对话的本质区别

普通 LLM 对话是“一问一答”：用户传入一段文本，模型返回一段文本。整个过程没有状态，没有外部动作，模型只能依靠训练时学到的知识回答。智能体则不同，它的核心是“Agent Loop”，也就是一个持续循环：模型根据当前对话判断下一步要做什么，如果发现需要查询资料或执行操作，就生成一次工具调用请求，程序执行工具后把结果写回对话，模型再基于新信息继续推理，直到它认为任务完成。

这个区别决定了架构设计的起点。普通对话只需要一个 chat 接口，而智能体至少需要四条链路：模型接入、工具注册、消息历史、循环控制。任何一个环节缺失，智能体都会退化成“套了壳的聊天机器人”。

1.2 从 ReAct 模式理解 Agent 的执行本质

ReAct（Reasoning and Acting）是当前大多数智能体框架的理论基础。它的核心思想是让模型交替进行“推理”和“行动”：推理用于决定做什么，行动用于实际执行，行动的结果再作为新输入进入下一轮推理。

用一句话概括：智能体的能力来自“把模型输出的意图变成程序可执行的动作，再把执行结果反馈给模型”。

Pi Agent 的主循环就是 ReAct 的落地版本。每一轮循环做三件事：

1. 把当前完整消息历史发送给模型。
2. 模型返回两种结果之一：要么是工具调用请求，要么是最终回答。
3. 如果是工具调用，就执行工具并把结果写入消息历史，继续下一轮；如果是最终回答，则结束循环。

这个模式看起来简单，但它包含了智能体最重要的设计决策：模型不直接执行代码，而是通过结构化参数描述“想做什么”，真正执行权由程序持有。这样既能控制权限，又能对每次操作做日志记录和异常处理。

1.3 Pi Agent 的架构目标和适用范围

Pi Agent 在这里是一个教学级最小实现，不是要替代成熟框架。它的架构目标有三个：

• 每个模块职责单一，能独立替换。
• 代码量控制在数百行以内，能完整阅读。
• 依赖尽量少，只保留模型调用和基础标准库。

因此它适合本地学习、内部工具原型、以及给其他项目提供架构参考。如果任务涉及多用户并发、复杂工作流编排、大规模知识库检索、生产级权限管控，则需要参考第 6 章的扩展路径迁移到完整框架。

这张表也是后文的实现顺序。先写模型接入，再写工具，再写消息管理，最后写循环，逻辑上最顺。

2. 环境准备、依赖与项目结构

2.1 运行环境要求

Pi Agent 对运行环境的要求很低，学习阶段不需要 GPU，也不需要复杂的分布式环境。只需要一台能访问模型服务的机器，Python 3.10 或更高版本即可。

模型方面，建议选择支持 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 requests

requests是唯一必需的外部依赖。如果你使用 Windows，激活虚拟环境命令为.venv\Scripts\activate；如果不使用虚拟环境，直接pip install requests也可以，但推荐在项目中始终使用虚拟环境。

完成后生成requirements.txt：

pip freeze > requirements.txt

3. 五个核心模块的实现

3.1 LLM 客户端封装：统一模型接入层

llm_client.py的作用是把模型 API 调用集中在一个类里。后续在任何地方需要模型能力，都只调用chat()方法，不需要关心 HTTP 细节。

# llm_client.py import requests class LLMClient: def __init__(self, base_url, api_key, model, temperature=0.7, timeout=60): self.base_url = base_url.rstrip("/") self.api_key = api_key self.model = model self.temperature = temperature self.timeout = timeout def chat(self, messages, tools=None): url = f"{self.base_url}/v1/chat/completions" headers = { "Authorization": f"Bearer {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, headers=headers, json=payload, timeout=self.timeout) resp.raise_for_status() choice = resp.json()["choices"][0] return choice

这里几个参数值得说明：

• base_url指向模型服务的根地址，chat()内部拼接出完整的 chat/completions 路径。
• api_key从环境变量读取更安全，不要硬编码到代码里。
• tool_choice="auto"表示由模型自主决定是否调用工具、调用哪个工具。
• 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": f"tool '{name}' not found"}, ensure_ascii=False) try: args = json.loads(arguments_json) result = self._handlers[name](**args) return json.dumps(result, ensure_ascii=False) except Exception as exc: return json.dumps({"error": str(exc)}, ensure_ascii=False)

工具声明使用 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, mode="eval") return _safe_eval(tree)

这个例子说明了一个重要原则：智能体框架中，模型只负责“提出动作意图”，程序必须自己保证动作的安全性。凡是涉及文件删除、命令执行、网络请求的工具，都要加权限校验和审计日志，不能因为模型说要做就直接执行。

3.3 消息管理：工具结果如何写回上下文

LLM 是无状态接口，每一轮调用都要把完整上下文重新发送。因此智能体必须自己维护消息历史，并把工具调用和工具结果按协议格式写回。

# memory.py class MessageHistory: def __init__(self, max_messages=40): 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

消息历史中的角色有三种，含义不同：

特别注意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_steps=10): 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(), tools=self.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}，任务终止。"

主循环的判断逻辑要重点理解：

1. 每次循环先发送当前全部消息和工具声明。
2. 模型返回的 assistant 消息原样写入历史。包括content为None但带tool_calls的情况。
3. 如果tool_calls为空，说明模型认为不需要调用工具，直接返回内容。
4. 如果有多个tool_calls，逐个执行，把结果以tool角色写回。
5. 写回后进入下一轮循环，让模型看到工具结果。

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_url=base_url, api_key=api_key, model=model) tools = ToolRegistry() # 注册工具 tools.register(get_current_time_definition)(get_current_time) # 注册 calc... memory = MessageHistory(max_messages=40) memory.add({"role": "system", "content": SYSTEM_PROMPT}) agent = PiAgent(llm=llm, tools=tools, memory=memory) print("Pi Agent 已启动，输入 exit 退出。") while True: user_input = input(">>> ").strip() if user_input.lower() in ("exit", "quit"): break answer = agent.run(user_input) print(f"AI: {answer}") if __name__ == "__main__": main()

为了让注册代码更整洁，可以把工具声明和函数放在一起，用装饰器注册。上面第 3.2 节已经演示了装饰器写法，入口里只需要保证所有工具模块被导入一次即可。

4. 运行验证与结果分析

4.1 跑通一次普通对话

先用最简单的方式验证架构能跑通。设置环境变量后启动：

export LLM_BASE_URL="https://api.example.com" export LLM_API_KEY="your-key" export LLM_MODEL="your-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。

如果模型在一个回复里产生多个工具调用，比如“现在几点，并计算 1+1”，日志中会连续出现两个“调用工具”行，然后再进入下一轮。

4.3 中间状态和日志怎么看

排查问题时，最重要的不是最终答案，而是中间状态。Pi Agent 的run()方法里已经打印了 step 编号和工具调用参数。建议再增加一个可选的调试模式，打印每次发往模型的消息结构：

# agent.py（增加调试模式） def run(self, user_input, debug=False): ... if debug: for message in self.memory.to_list(): print("---- message ----") print(json.dumps(message, ensure_ascii=False, indent=2))

调试模式的判断顺序是：

1. 看 step 是否递增，确认循环正常推进。
2. 看模型返回的消息结构，确认tool_calls是否按预期生成。
3. 看工具调用参数，确认模型生成的 JSON 参数是否合法。
4. 看 tool 消息是否包含正确的tool_call_id。
5. 看最终回答是否基于工具结果生成，而不是凭空编造。

这套观察顺序也和第 5 章的排查路径一致。

5. 常见问题与排查路径

5.1 模型一直不调用工具

现象：用户明确要求查询时间或计算，模型却输出文字猜测，不返回tool_calls。

可能原因：

• 模型服务不支持tools参数，或当前模型版本没有工具调用能力。
• 工具声明格式与模型要求不匹配，模型解析失败后干脆不调用。
• 系统提示词没有说明工具的存在，模型不知道有工具可用。
• tool_choice被设置为none。

排查顺序：

1. 直接手工构造一次带tools的 API 请求，确认模型能否返回tool_calls。
2. 打印tools.schemas()，检查 JSON Schema 格式是否符合 OpenAI 兼容规范。
3. 检查系统提示词中是否包含了“可以调用工具”的说明。
4. 临时把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 message

5.4 工具执行死循环或不收敛

现象：模型反复调用工具，日志一直递增，最终被max_steps截断。

可能原因：

• 工具返回结果不足以支撑模型判断“任务已完成”。
• 工具结果中包含错误信息，模型尝试重试但参数没有变化。
• 系统提示词没有给出“何时停止调用工具”的明确标准。

解决方案：

• 在系统提示词中补充：“如果你已经拿到足够信息，直接给出最终答案，不要重复调用相同工具。”
• 对相同工具和相同参数的调用做去重，连续重复超过两次就返回错误。
• 把max_steps从 10 调低到 5，快速暴露不收敛问题。

5.5 典型错误速查表

5.6 通用排查顺序

无论遇到什么问题，都按这个顺序排查：

1. 确认输入：模型收到的是什么消息，与预期是否一致。
2. 确认格式：tools声明和消息结构是否符合模型服务协议。
3. 确认配置：模型名、base_url、api_key、timeout 是否正确。
4. 确认状态：消息历史是否被正确裁剪和回填。
5. 确认日志：是否有工具执行异常、JSON 解析异常等信息。

6. 最佳实践与生产化扩展

6.1 学习环境与生产环境的差异

Pi Agent 的设计目标是讲清楚原理，生产环境还需要补很多内容。两套环境的差异如下：

其中工具安全是最关键的差异点。生产环境的每一个工具调用都应该记录“谁在什么对话中调用了什么工具、参数是什么、结果是什么、耗时多少”，这份审计日志既是排查依据，也是安全合规要求。

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 跑通，然后依次做四个练习：

1. 新增一个中文 JSON 工具，返回自定义业务数据，验证模型能正确解释并引用结果。
2. 增加调试模式，查看一次多轮工具调用的完整消息历史。
3. 故意注册一个返回错误格式的工具，观察模型如何在下一轮自我修正。
4. 对比不同max_steps之下，模型在复杂任务上的表现差异。

这四个练习覆盖了智能体开发的大部分基础能力。做完以后，再去看 LangGraph 的多智能体协作、Dify 的工作流节点、以及函数调用的流式输出，理解速度会明显快很多。

最简架构的价值不在于功能多，而在于让每一个决策都透明：模型什么时候做决定、程序什么时候执行、上下文里发生了什么，你都能一眼看到。把这条主线理解透，后面无论是扩展工具、接入知识库、还是做多智能体编排，都不会偏离真正重要的架构判断。