在实际的 Agent 项目中，模型单次调用和完整智能体之间，往往隔着一条比想象中更深的沟。很多开发者在本地跑通一次大模型调用之后，以为下一步只需要把提示词写长一点、多问几轮，就能得到一个自动执行任务的智能体。真正进入工具调用、多步推理和任务拆解之后才发现，复杂性的重心根本不是“模型回答得准不准”，而是循环如何启动、如何维持上下文、如何在工具异常时继续、如何在达到上限时安全终止。围绕类 PI-Agent 的复杂智能体架构设计，这篇文章以一条清晰的技术主线展开：先理解模型与智能体的区别，再动手把一个单次模型调用改造成带工具调用的 Agent Loop，最后补充生产环境必须考虑的排错和治理手段。
1. 先理解模型调用和 Agent Loop 的本质区别
1.1 单次调用到底解决了什么问题
一次普通的大模型调用，输入是一个消息列表，输出是文本或结构化内容。它适合翻译、总结、改写、信息抽取这类“一次性完成”的任务。以 Python 为例，使用 OpenAI 兼容客户端时，最小调用形态通常是这样：
from openai import OpenAI client = OpenAI(api_key="your-api-key") messages = [ {"role": "system", "content": "你是一名数学助手。"}, {"role": "user", "content": "计算 23 乘以 17。"}, ] response = client.chat.completions.create( model="gpt-4o-mini", messages=messages, temperature=0, ) print(response.choices[0].message.content)这段代码解决的是“把问题交给模型，模型直接回答”。如果模型在训练数据里见过足够多类似问题，它可以给出结果；如果没见过，它只能根据内部知识猜测。单次调用的边界就在这里：模型没有真实计算能力，没有实时数据来源，也没有办法主动获取外部信息。
在实际项目中，单次调用的局限主要体现在三个方面。第一是知识时效性，模型训练数据有截止时间，无法自动获取新信息。第二是精度，复杂计算、精确检索、文件操作这类任务依赖外部工具，模型靠“记忆”完成不了。第三是任务连续性，一个真实任务往往包含多个步骤，每步的结果要成为下一步的输入，单次调用无法表达这个过程。
1.2 Agent 的本质是“模型 + 工具 + 循环”
Agent 不是一个新模型，而是一种程序架构。它仍然用大模型做决策和语言生成，但把决策结果映射为可执行动作，再把这些动作的执行结果重新交回给模型，形成“思考-行动-观察”的循环。这个循环在英文资料里通常叫 Agent Loop，广义上也包含 ReAct（Reasoning and Acting）这类模式。
Agent 和直接在代码里写 if-else 流水线有本质区别。流水线的分支是预先写死的，Agent 的分支由模型在运行时决定。模型决定什么时候调用哪个工具、工具结果怎么理解、下一步做什么。程序层负责提供工具、收集结果、控制循环次数、管理上下文。开发者的工作重点从“写业务逻辑”变成了“设计模型可理解的环境”和“保证循环可控”。
1.3 Loop 为什么是复杂智能体的核心
类 PI-Agent 这类复杂智能体，核心不再是某一段提示词写得有多好，而是 Loop 设计得是否可控。可控体现在四个能力上：
- 能启动：用户输入进入循环，系统提示词和工具定义被正确注入。
- 能推进：模型每次返回动作，程序执行后把结果写回消息列表，继续下一轮。
- 能终止：模型给出最终答案、达到最大迭代次数、或检测到异常时，循环要安全退出。
- 能恢复：工具执行失败时，把错误信息作为观察结果还给模型，让模型修正计划，而不是让程序崩溃。
如果一个 Agent Loop 不能保证以上四点，模型能力再强，也会卡在死循环、上下文爆炸或者静默失败里。这也是为什么许多初学者用同一个模型，单次调用效果不错，改成复杂智能体后就频繁出问题。
2. 类 PI-Agent 架构的整体分层
2.1 模型层：模型只负责推理和动作生成
模型层是整个 Agent 的决策大脑。它可以调用云端 API，也可以使用本地部署的模型。落地时，建议用 OpenAI 兼容接口统一封装，这样本地模型和云端模型可以切换，业务代码不用跟着改。
模型层需要关注的参数不少，对 Agent 场景影响最大的是模型名称、temperature、max_tokens、top_p 这几个。工具调用和任务执行需要确定性，temperature 建议调低；max_tokens 要根据工具返回长度和最终回答长度综合设置，设置太小会导致回答被截断。
2.2 工具层：把外部能力接入循环
工具层把计算、检索、HTTP 请求、数据库查询、文件操作等能力注册给模型。每个工具要提供三样信息：名称、描述、参数结构。名称和描述决定了模型能不能正确理解“什么时候该用这个工具”，参数结构决定了模型能不能生成正确的调用参数。
这里要特别说明：工具描述不是给人看的，是给模型看的。描述越含糊，模型越容易误用。比如“计算数学表达式”和“计算两个数字的乘积”会产生完全不同的调用行为。
2.3 循环控制层：迭代、终止和容错
循环控制层负责执行主循环，维护当前步数，处理最大迭代次数，捕获工具执行异常并对齐消息格式。大多数 Agent 框架里，这一层也就是几十行代码，但却是最容易出问题的部分。
常见问题集中在：工具调用结果没有以正确的 role 和 tool_call_id 写回消息列表、异常直接抛出导致循环中断、缺少最大迭代次数导致死循环。循环控制层的设计目标，是让模型在“犯错”时仍然能继续推进，直到自然终止。
2.4 状态管理层：上下文和记忆
状态管理层保存系统提示词、用户输入、历史消息、工具调用记录和工具返回结果。复杂 Agent 还需要区分短期上下文和长期记忆。短期上下文在单次任务内部存在，长期记忆则要落到外部存储，例如向量数据库或者关系数据库。
各层职责可以用一张表概括：
| 分层 | 职责 | 典型实现 | 主要风险 |
|---|---|---|---|
| 模型层 | 推理、生成动作、生成最终回答 | 云端 API、本地模型 | 输出截断、幻觉、不调用工具 |
| 工具层 | 提供可执行能力 | 函数、HTTP 服务 | 描述不清、参数错误、接口不稳定 |
| 循环控制层 | 启动循环、推进步数、终止、容错 | 主循环代码 | 死循环、异常中断、消息格式错误 |
| 状态管理层 | 维护消息历史、长期记忆 | 内存列表、向量库、数据库 | token 超限、上下文丢失 |
这种分层的价值在于：每一层都可以独立替换。模型层换模型，工具层加工具，状态管理层换存储方案，都不会导致主循环推倒重写。搭建 Agent 时不要一开始就用重框架，先把这四层职责理清，后续扩展会顺畅很多。
3. 环境准备与最小项目结构
3.1 依赖和运行环境
学习阶段的运行环境可以尽量简单，推荐条件如下：
| 项目 | 推荐要求 | 说明 |
|---|---|---|
| Python | 3.10 及以上 | 使用类型标注需要 3.10+ 的 `X |
| openai 客户端 | 1.x | 兼容多数 OpenAI 风格接口 |
| python-dotenv | 最新稳定版 | 管理 API Key 和模型配置 |
| 模型接口 | 云端 API 或本地 OpenAI 兼容服务 | 本地可用常见推理服务提供 |
如果使用本地模型，常见做法是启动一个提供 OpenAI 兼容接口的本地推理服务，然后把 base_url 指向本地地址，模型名换成本地模型实际名称。这样本文代码无需大改，学习环境和生产环境之间切换成本很低。
安装依赖：
pip install openai python-dotenv3.2 目录结构
一个最小但可扩展的目录结构如下：
pi_agent/ ├── agent/ │ ├── __init__.py │ ├── llm.py # 统一模型调用客户端 │ ├── loop.py # Agent Loop 主循环 │ └── tools.py # 工具定义、工具注册表 ├── main.py # 入口，组装 Agent 并运行 ├── .env # API Key、模型配置 └── requirements.txt这个结构把模型、循环、工具拆开，后续加工具、换模型、加记忆都比较方便。不要把所有逻辑都写在一个文件里，否则排查 Agent 循环问题时很难定位。
3.3 统一模型调用客户端
在 agent/llm.py 中定义一个统一的客户端：
from openai import OpenAI class LLMClient: def __init__(self, api_key: str, base_url: str | None = None, model: str = "gpt-4o-mini"): self.client = OpenAI(api_key=api_key, base_url=base_url) self.model = model def chat(self, messages: list[dict], tools: list[dict] | None = None) -> dict: params = { "model": self.model, "messages": messages, } if tools: params["tools"] = tools params["tool_choice"] = "auto" response = self.client.chat.completions.create(**params) message = response.choices[0].message result = { "role": "assistant", "content": message.content, } if message.tool_calls: result["tool_calls"] = [] for tc in message.tool_calls: result["tool_calls"].append({ "id": tc.id, "type": "function", "function": { "name": tc.function.name, "arguments": tc.function.arguments, }, }) return result如果模型返回了工具调用，这里会把标准对象转成普通字典并保留 tool_call_id。这个 id 在后面写回工具结果时是必需的，丢掉了就无法和工具结果对应。base_url 参数是切换本地模型和云端模型的关键入口。
4. 从单次调用改造成 Agent Loop
4.1 先定义两个工具
在 agent/tools.py 中定义工具。第一个是安全计算器，不做裸 eval，而是用 ast 解析白名单表达式：
import ast import operator _OPERATORS = { ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, ast.Pow: operator.pow, } def _safe_eval(node): if isinstance(node, ast.Expression): return _safe_eval(node.body) if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)): return node.value if isinstance(node, ast.BinOp) and type(node.op) in _OPERATORS: left = _safe_eval(node.left) right = _safe_eval(node.right) return _OPERATORS[type(node.op)](left, right) raise ValueError("不支持的表达式") def calculate(expression: str) -> dict: try: tree = ast.parse(expression, mode="eval") result = _safe_eval(tree) return {"result": result} except Exception as exc: return {"error": f"计算失败: {exc}"} def get_weather(city: str) -> dict: # 实际项目应接入真实天气服务，这里返回模拟数据 return {"city": city, "weather": "晴", "temperature": 22}第二个是查询天气，返回模拟数据，用于演示模型如何按参数名传参。这两个工具都不复杂，但足以体现“模型生成动作、程序执行动作、结果回流”的完整链路。
4.2 定义工具 Schema 和注册表
工具 Schema 是模型理解工具的桥梁，格式是 JSON Schema 风格的数组：
TOOL_SCHEMAS = [ { "type": "function", "function": { "name": "calculate", "description": "计算数学表达式，支持加、减、乘、除、幂运算。", "parameters": { "type": "object", "properties": { "expression": { "type": "string", "description": "要计算的数学表达式，例如 23*17" } }, "required": ["expression"] } } }, { "type": "function", "function": { "name": "get_weather", "description": "查询指定城市的天气信息。", "parameters": { "type": "object", "properties": { "city": { "type": "string", "description": "城市名称" } }, "required": ["city"] } } }, ] TOOL_REGISTRY = { "calculate": calculate, "get_weather": get_weather, }这里的 description 字段不是摆设。模型会依据它决定什么时候调用工具、参数填什么，描述含糊会直接导致误调用。parameter 里的每个字段也要写清楚示例值，模型在生成参数时才有参考。
4.3 实现 Agent Loop 主循环
在 agent/loop.py 中实现主循环：
import json class AgentLoop: def __init__(self, llm, schemas, registry, max_iterations=8, system_prompt=None): self.llm = llm self.schemas = schemas self.registry = registry self.max_iterations = max_iterations self.system_prompt = system_prompt or "你是一个可以调用工具完成任务的智能体。" def run(self, user_input: str) -> str: messages = [ {"role": "system", "content": self.system_prompt}, {"role": "user", "content": user_input}, ] for step in range(1, self.max_iterations + 1): message = self.llm.chat(messages, tools=self.schemas) if not message.get("tool_calls"): return message.get("content", "") messages.append(message) for tool_call in message["tool_calls"]: name = tool_call["function"]["name"] try: arguments = json.loads(tool_call["function"]["arguments"]) except json.JSONDecodeError: arguments = {} func = self.registry.get(name) if func is None: content = json.dumps({"error": f"未知工具: {name}"}, ensure_ascii=False) else: try: content = json.dumps(func(**arguments), ensure_ascii=False) except Exception as exc: content = json.dumps({"error": str(exc)}, ensure_ascii=False) messages.append({ "role": "tool", "tool_call_id": tool_call["id"], "content": content, }) print(f"[step={step}] 调用工具后，当前消息数: {len(messages)}") return f"任务未能在 {self.max_iterations} 步内完成，请检查任务拆解或扩展迭代上限。"主循环的逻辑很直接：模型不返回工具调用，说明它认为可以直接给出最终答案，此时返回内容并退出；模型返回工具调用，程序执行并把结果以 tool 角色写回消息列表，然后进入下一轮。
关键点在于 tool_call_id 必须与模型返回的 id 一致，否则服务端会报错。工具执行的异常不能直接抛出，而要转成 JSON 字符串写回给模型，让模型有机会修正参数或更换工具。
4.4 终止条件与最大迭代次数
Loop 必须有终止条件，否则当任务复杂或模型陷入重复时，程序会无限耗下去。终止方式有三种：模型直接返回最终答案、达到 max_iterations、程序检测到异常状态。
max_iterations 是 Agent 里最需要调优的参数之一。取值太小，复杂任务会被过早截断；取值太大，单次任务成本升高，死循环风险增大。学习阶段建议设置 5 到 10，生产环境需要根据任务复杂度和成本单独压测。
参数配置速查表：
| 参数 | 默认值 | 含义 | 调大影响 | 调小影响 |
|---|---|---|---|---|
| max_iterations | 8 | 单任务最多循环轮数 | 复杂任务完成率高，成本升高，死循环风险增大 | 成本可控，复杂任务易被截断 |
| temperature | 0 | 采样随机性 | 回答更多样，工具调用不稳定 | 回答更确定，适合工具调用 |
| max_tokens | 视模型而定 | 单次输出长度上限 | 可输出更长内容 | 容易出现截断 |
| tool_choice | auto | 是否强制调用工具 | 不设置时模型可自由选择直接回答 | 强制 tool 可避免模型跳过工具 |
5. 关键机制详解
5.1 上下文管理：消息列表如何增长和裁剪
每次进入 Loop 下一轮，消息列表都会追加一条 assistant 消息和若干条 tool 消息。一个需要调用 5 次工具的任务，消息数会从 2 条增长到 12 条以上。如果工具返回内容很长，token 消耗会快速上升。
常见的处理方式有三种：滑动窗口裁剪、中间步骤摘要、外部记忆。滑动窗口只保留最近的 N 条消息，适合单轮任务；摘要适合任务需要历史上下文但不依赖完整细节的场景；外部记忆适合跨会话、跨任务的长期场景。生产环境通常组合使用。
注意：裁剪上下文时，不要只保留最后几条消息就丢弃系统提示词和关键任务指令，否则模型会丢失对任务的整体理解。
5.2 工具返回结果如何回流成观察
工具返回结果必须以 tool 角色的消息追加到消息列表，并且要带上对应的 tool_call_id。有些初学者把工具结果拼接到 user 消息里，这在简单场景偶尔能工作，但在正式 function calling 协议下会导致消息顺序错误或 id 不匹配。
正确的追加顺序是：先追加 assistant 工具调用消息，再逐条追加 tool 结果消息。顺序颠倒会被 API 拒绝，id 不匹配也会报错。可以在主循环里加一个校验函数，在发送前检查最后两条消息的角色组合，尽早发现格式问题。
5.3 输出 token 上限与截断处理
模型单次输出有长度限制。当任务要求生成长文本，或者工具返回内容很长需要模型总结时，很容易命中“达到输出 token 上限，回答被截断”的问题。判断是否截断，不能只看输出文本，要看返回对象的 finish_reason。如果 finish_reason 是 length，说明输出被截断；如果是 stop，才是正常结束。
处理方式有两种：调大 max_tokens，或者让模型分步输出。调大只适用于模型上限以内的场景；如果单次输出确实超过模型上限，就要把任务拆成多个阶段，每阶段输出一部分。
5.4 错误处理与重试策略
Agent 场景里，错误处理要分三层。第一层是模型调用层，网络超时、限流、服务端 5xx 需要重试，建议使用带指数退避的重试机制。第二层是工具执行层，工具抛出的异常不应该终止整个 Agent，而是作为观察内容返回给模型。第三层是循环控制层，当循环次数用尽或消息格式错误时，要给出明确错误信息，而不是静默失败。
对于高风险工具，还可以在调用前加一个结果检查器，用规则或小模型验证参数合法性。比如删除类操作、写数据库操作，先让检查器确认参数和权限，再执行。这个检查和执行分离的做法，能显著降低误调用带来的损失。
6. 运行验证与结果分析
6.1 组装 Agent 并运行
在 main.py 中组装：
import os from dotenv import load_dotenv from agent.llm import LLMClient from agent.loop import AgentLoop from agent.tools import TOOL_REGISTRY, TOOL_SCHEMAS load_dotenv() llm = LLMClient( api_key=os.getenv("API_KEY"), base_url=os.getenv("BASE_URL"), model=os.getenv("MODEL", "gpt-4o-mini"), ) agent = AgentLoop( llm=llm, schemas=TOOL_SCHEMAS, registry=TOOL_REGISTRY, max_iterations=8, ) if __name__ == "__main__": result = agent.run("请计算 23*17，再把结果加 5，最后告诉我答案。") print("最终回答:", result).env 文件示例：
API_KEY=your-api-key BASE_URL= MODEL=gpt-4o-mini如果 BASE_URL 留空，客户端会连接默认的 OpenAI 地址；如果使用本地模型，把 BASE_URL 填成本地服务地址即可。
6.2 预期流程
一个正常的运行流程应该类似：
[step=1] 调用工具后，当前消息数: 4 [step=2] 调用工具后，当前消息数: 6 最终回答: 23 乘以 17 等于 391，加 5 后是 396。第一步模型决定调用 calculate，参数是 23*17；第二步模型根据第一步结果继续计算 391+5；第三步模型给出最终答案。这里能看到模型正确理解“上一步结果要作为下一步输入”，这正是 Loop 相比单次调用的核心差异。
6.3 测试工具异常恢复
把 get_weather 改成对特定城市返回错误内容，而不是抛异常：
def get_weather(city: str) -> dict: if city == "不存在城市": return {"error": "未找到该城市，请检查城市名称"} return {"city": city, "weather": "晴", "temperature": 22}模型收到 error 后，应该会修正输入或向用户说明查询失败。如果直接抛异常，循环就会中断，模型没有机会修正。运行输入“查询不存在城市的天气”时，正常输出应该包含“未找到该城市”这类说明，而不是程序崩溃。
除了功能验证，还要验证发散场景：连续输入同样的问题、工具连续失败、模型反复调用同一工具。这些场景下的输出稳定性，才是 Loop 设计是否合格的真正标准。
7. 常见问题排查
7.1 现象：模型陷入死循环，一直重复调用同一工具
可能原因是任务描述不清晰、工具返回结果没有给模型“任务已完成”的信号、或者 max_iterations 设置过大。检查方式是打印每个 step 的 tool_call 名称和参数，看是否符合预期。处理建议：调低 max_iterations，在系统提示词里要求“验证结果满足任务要求后再结束”，或在循环层限制同一个工具连续调用次数。
7.2 现象：模型始终不调用工具
可能原因是工具描述不清晰、参数结构错误、或者模型本身不支持 function calling。检查方式是打印传给模型的 schemas，确认 JSON 格式正确；再确认选用的模型支持工具调用。处理建议：优化 description，字段加示例值，必要时换模型。
7.3 现象：调用工具时报 tool_call_id 不匹配
可能原因是消息顺序错误，或者把 assistant 工具调用消息漏掉了。检查方式是打印当前 messages 列表最后四条消息。处理建议：严格按照“先 assistant 工具调用，再 tool 结果”的顺序追加。
7.4 现象：输出被截断
可能原因是 max_tokens 设置过小，或单次输出超过模型上限。检查 finish_reason 是否为 length。处理建议：调大 max_tokens，或把长文本任务拆成多阶段。
7.5 现象：工具返回了错误内容但模型没有感知
可能原因是工具把异常吞掉，返回了空字符串或无关数据。处理建议：工具内部也要规范返回结构，至少包含 error 字段和可读信息；Loop 层可以在内容为空的场景下追加一条“工具未返回有效结果”的观察信息。
常见问题汇总：
| 问题现象 | 常见原因 | 检查方式 | 处理建议 |
|---|---|---|---|
| 死循环 | 缺少终止信号、max_iterations 过大 | 打印 step 日志 | 降低迭代上限，增加重复调用检测 |
| 不调用工具 | 工具描述含糊、模型不支持 | 打印 schemas | 优化 description，换支持 function calling 的模型 |
| tool_call_id 不匹配 | 消息顺序错误 | 打印最近消息列表 | 按 assistant 到 tool 的顺序追加 |
| 输出截断 | max_tokens 过小或超模型上限 | 检查 finish_reason | 调大 max_tokens 或分段生成 |
| 上下文超限 | 消息列表无限增长 | 查看 token 消耗 | 滑动窗口、摘要、外部存储 |
| 工具异常被吞掉 | 裸 except 返回空内容 | 检查工具返回 JSON | 统一错误结构，空结果单独处理 |
8. 生产环境最佳实践与扩展方向
8.1 发布前的检查清单
生产环境不能只看“程序能跑通”。至少确认以下项目：
- 配置外置：API Key、模型名、Base URL 从环境变量或配置中心读取，不硬编码。
- 日志完整：每个 step 记录模型名、token 用量、工具名称、耗时、错误信息。
- 工具权限：敏感工具要有权限控制和审计，不能让模型随意调用写操作。
- token 预算：预先评估单任务最大 token 消耗，设置预算上限。
- 异常兜底：所有工具异常都转成观察结果，循环不因单点故障崩溃。
- 监控告警：迭代次数异常升高、token 消耗突增、失败率上升都要有告警。
- 回滚方案：模型或工具版本升级后，要能快速回退到稳定版本。
8.2 记忆与外部存储
单次 Loop 里的消息列表是短期记忆，进程重启就丢失。如果 Agent 需要跨会话记忆，可以把用户偏好、历史结论、工具执行记录写入数据库或向量库；每次任务开始前检索相关记忆，拼接到系统提示词或上下文中。注意控制注入记忆的长度，避免挤占任务上下文。
8.3 可观测性：把 Loop 变成可追踪的事件流
排查 Agent 问题时，最难的是“模型为什么这么想”。生产环境建议把每步的关键信息打点成结构化日志或事件，包括输入消息摘要、模型输出、工具调用、返回结果、耗时、token 数。有了这些数据，才能定位是模型决策问题还是工具执行问题。没有可观测性的 Agent，本质上和黑盒没有区别，出了问题只能靠猜。
8.4 扩展方向
Loop 是复杂智能体的地基，往上可以扩展的方向很多：多智能体协作让不同 Agent 分别负责规划、执行和审查；引入可视化编排平台，用节点代替手写循环；集成外部知识库和长期记忆；结合模型路由做模型融合，让简单任务走低成本模型、复杂任务走强模型。无论选哪个方向，核心仍然是先保证单个 Loop 可控，再谈复杂度扩展。开源编码智能体工具也普遍采用类似的思维，本质上都是模型、工具和循环这三件事的组合。
回到最开始的问题：从单次调用到 Loop，不只是代码形态的变化，而是开发思维的变化。单次调用关心“模型输出什么”，Agent Loop 关心“模型下一步该做什么，做了之后环境怎么反馈”。对新手来说，最有价值的练习不是马上接一堆工具，而是先手写一个只有两三个工具的 Loop，把迭代、终止、异常恢复跑熟，再去接触框架和平台。把 Loop 每一步的日志打出来看一遍，比读十篇架构文章都管用。
