在实际的 Agent 项目中模型单次调用和完整智能体之间往往隔着一条比想象中更深的沟。很多开发者在本地跑通一次大模型调用之后以为下一步只需要把提示词写长一点、多问几轮就能得到一个自动执行任务的智能体。真正进入工具调用、多步推理和任务拆解之后才发现复杂性的重心根本不是“模型回答得准不准”而是循环如何启动、如何维持上下文、如何在工具异常时继续、如何在达到上限时安全终止。围绕类 PI-Agent 的复杂智能体架构设计这篇文章以一条清晰的技术主线展开先理解模型与智能体的区别再动手把一个单次模型调用改造成带工具调用的 Agent Loop最后补充生产环境必须考虑的排错和治理手段。1. 先理解模型调用和 Agent Loop 的本质区别1.1 单次调用到底解决了什么问题一次普通的大模型调用输入是一个消息列表输出是文本或结构化内容。它适合翻译、总结、改写、信息抽取这类“一次性完成”的任务。以 Python 为例使用 OpenAI 兼容客户端时最小调用形态通常是这样from openai import OpenAI client OpenAI(api_keyyour-api-key) messages [ {role: system, content: 你是一名数学助手。}, {role: user, content: 计算 23 乘以 17。}, ] response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, temperature0, ) print(response.choices[0].message.content)这段代码解决的是“把问题交给模型模型直接回答”。如果模型在训练数据里见过足够多类似问题它可以给出结果如果没见过它只能根据内部知识猜测。单次调用的边界就在这里模型没有真实计算能力没有实时数据来源也没有办法主动获取外部信息。在实际项目中单次调用的局限主要体现在三个方面。第一是知识时效性模型训练数据有截止时间无法自动获取新信息。第二是精度复杂计算、精确检索、文件操作这类任务依赖外部工具模型靠“记忆”完成不了。第三是任务连续性一个真实任务往往包含多个步骤每步的结果要成为下一步的输入单次调用无法表达这个过程。1.2 Agent 的本质是“模型 工具 循环”Agent 不是一个新模型而是一种程序架构。它仍然用大模型做决策和语言生成但把决策结果映射为可执行动作再把这些动作的执行结果重新交回给模型形成“思考-行动-观察”的循环。这个循环在英文资料里通常叫 Agent Loop广义上也包含 ReActReasoning 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 依赖和运行环境学习阶段的运行环境可以尽量简单推荐条件如下项目推荐要求说明Python3.10 及以上使用类型标注需要 3.10 的 Xopenai 客户端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_keyapi_key, base_urlbase_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 Loop4.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, modeeval) 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_iterations8, system_promptNone): 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, toolsself.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_asciiFalse) else: try: content json.dumps(func(**arguments), ensure_asciiFalse) except Exception as exc: content json.dumps({error: str(exc)}, ensure_asciiFalse) 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_iterations8单任务最多循环轮数复杂任务完成率高成本升高死循环风险增大成本可控复杂任务易被截断temperature0采样随机性回答更多样工具调用不稳定回答更确定适合工具调用max_tokens视模型而定单次输出长度上限可输出更长内容容易出现截断tool_choiceauto是否强制调用工具不设置时模型可自由选择直接回答强制 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_keyos.getenv(API_KEY), base_urlos.getenv(BASE_URL), modelos.getenv(MODEL, gpt-4o-mini), ) agent AgentLoop( llmllm, schemasTOOL_SCHEMAS, registryTOOL_REGISTRY, max_iterations8, ) if __name__ __main__: result agent.run(请计算 23*17再把结果加 5最后告诉我答案。) print(最终回答:, result).env 文件示例API_KEYyour-api-key BASE_URL MODELgpt-4o-mini如果 BASE_URL 留空客户端会连接默认的 OpenAI 地址如果使用本地模型把 BASE_URL 填成本地服务地址即可。6.2 预期流程一个正常的运行流程应该类似[step1] 调用工具后当前消息数: 4 [step2] 调用工具后当前消息数: 6 最终回答: 23 乘以 17 等于 391加 5 后是 396。第一步模型决定调用 calculate参数是 23*17第二步模型根据第一步结果继续计算 3915第三步模型给出最终答案。这里能看到模型正确理解“上一步结果要作为下一步输入”这正是 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 每一步的日志打出来看一遍比读十篇架构文章都管用。