十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

Agent Loop核心机制:从单次模型调用到类PI-Agent复杂智能体架构实践

Agent Loop核心机制:从单次模型调用到类PI-Agent复杂智能体架构实践 很多团队在搭建智能体时容易陷入一个误区以为接一个大模型 API把 Prompt 写得足够花哨就拥有了一个“智能体”。但真正跑过复杂任务的人会发现模型只是单次推理的“大脑”它不会自己查天气、不会自己读文件、更不会在第一次回答不够好时主动修正。让一个系统从“会聊天”变成“能干活”中间隔着的正是贯穿本文的核心机制——Agent Loop。这篇文章不是泛泛地讲概念而是从可落地的代码出发带你走一遍“类 PI-Agent 复杂智能体架构”的完整设计过程。我们会从一个最朴素的单次模型调用开始逐步加入工具、循环、记忆和多阶段协同最终实现一个具备“感知—决策—行动—观察—再决策”闭环能力的复杂智能体。读完你可以直接照着搭出骨架并理解每层架构解决的是什么问题。1. 这篇文章真正要解决的问题如果你只是写一个客服问答机器人单次调用模型就足够了。但如果你要做的是“类 PI-Agent”这类复杂智能体——比如让 AI 自动完成一份市场调研报告、自动处理一批数据文件、自动编排多个 API 操作事情就会瞬间变得复杂单次调用模型模型只能基于已有知识回答无法获取实时数据。复杂任务往往需要拆成多个子任务而模型单次输出长度有限。模型在推理过程中可能需要调用工具但调用完之后还要根据结果继续思考。一旦任务链条变长上下文管理、错误恢复、死循环防护都成了绕不开的问题。这就是为什么我们要从“单次调用”走向“Loop”。AI Agent 的本质是把大模型放进一个循环结构里让它可以不断观察工具执行结果、修正计划、继续行动直到任务完成。本文面向的读者是已经熟悉大模型 API 基本调用准备从“写 Prompt 调接口”进阶到“设计 Agent 架构”的开发者。你需要有一点 Python 基础理解函数调用和 API 请求不需要有完整的机器学习背景。2. 模型与智能体的核心概念与适用场景2.1 模型单次推理的“大脑”在智能体架构里模型是原始的推理引擎。你给它一段输入它返回一段输出。模型本身没有“行动能力”也没有“记忆持久性”——它只是根据你提供的上下文做概率预测。大模型调用可以简单理解为一次函数调用response client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: 今天北京天气怎么样}] )但这会暴露一个现实问题如果你不给模型提供实时天气数据它只能靠训练数据里的信息“编”一个答案。模型的内部知识是有截止日期的也没有主动获取外部信息的能力。2.2 智能体会行动的系统智能体是在模型之上叠加了感知、规划、工具调用和记忆的系统。它的核心不再是一次生成而是围绕一个目标不断循环。现实中的类比是模型像一位刚毕业的高材生知识储备不错但你让他独立完成项目时他需要查资料、跑实验、根据实验结果调整方案。智能体的“Loop”就是把这套工作流程显式地编码到系统里。2.3 Agent Loop循环即架构Agent Loop 包含四个阶段感知Perception接收到用户任务或者获取到工具返回的新信息。决策Planning/Decision模型基于当前上下文决定下一步做什么。行动Action执行一个工具调用比如查天气、执行代码、查询数据库。观察Observation读取工具调用返回的结果把它加入上下文回到决策阶段。用户输入 - 模型决策 - 需要工具 - 执行工具 - 结果反馈给模型 - 再决策 - 输出最终答案这就是 Agent 与普通模型调用的分水岭有没有这个循环决定了系统是“一次性问答”还是“闭环执行”。2.4 类 PI-Agent 复杂智能体架构类 PI-Agent 架构可以理解为多智能体系统的工程化形态。它通常包含一个主控智能体Orchestrator负责任务拆分和调度。若干子智能体Worker分别负责检索、分析、写作、代码执行等特定能力。一个共享的记忆/上下文存储让多个智能体之间能够接力协作。一组可注册、可被模型动态调用的工具集。如果把单 Agent 比作一个全能员工那么类 PI-Agent 架构更像一个项目组项目经理拆任务组员分别执行中间通过文档和会议同步信息。维度普通模型调用单 Agent类 PI-Agent 多智能体系统任务复杂度单轮问答中等复杂高复杂度、多阶段工具使用不支持单工具多轮调用多工具、多智能体协作记忆无对话上下文共享记忆、跨任务状态上下文管理简单需控制长度需分层管理失败恢复无循环重试子任务级重试与降级从表中可以明显看出不同的复杂程度对应不同的架构设计。绝大多数场景并不需要一上来就上多智能体但理解 Loop 是理解复杂架构的基础。3. 环境准备与前置条件在动手写代码之前先准备好环境。本文以 Python 为例演示所有代码在 macOS / Linux / WindowsWSL 更佳下都可以运行。3.1 环境依赖Python 3.10 及以上版本。建议使用虚拟环境隔离依赖。安装 OpenAI Python SDK或者任何兼容 OpenAI 接口的客户端。准备一个大模型 API或使用本地模型服务比如 Ollama、LM Studio 提供的 OpenAI 兼容接口。安装python-dotenv管理环境变量。python -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activate pip install openai python-dotenv3.2 模型服务选择你可以选择任一种方式提供模型服务云端 API设置环境变量OPENAI_API_KEY和OPENAI_BASE_URL。本地模型很多本地推理工具会启动一个 OpenAI 兼容的 HTTP 服务你只需要把OPENAI_BASE_URL改成对应的本地地址比如http://localhost:1234/v1。export OPENAI_API_KEYyour-api-key export OPENAI_BASE_URLhttps://api.openai.com/v1注意不同模型对 Function Calling工具调用的支持程度不同。云端主流模型基本都支持本地模型和开源模型的兼容性差异较大。如果运行下文代码发现工具调用不生效优先确认所选模型是否支持 tools 参数。4. 核心流程拆解从单次调用到 Loop理解了概念之后我们来看架构演进的核心流程。这一步是整个设计的灵魂你不能直接跳到复杂架构而是要从一个最简版本出发逐步增加复杂度才能理解每一层设计的意义。4.1 第一层单次模型调用这是最原始的形态适合简单问答。它的特点是无工具。无循环。无记忆。模型一次输出即结束。这个版本能跑通但它只能做“纯脑力”的事。任何需要实时数据、外部操作的任务它都无力应对。4.2 第二层模型 工具调用我们在模型之上增加“工具注册表”让模型可以在回答过程中发起函数调用请求。流程变成用户输入任务。模型分析是否需要工具。如果需要返回一个结构化指令函数名 参数而不是直接生成自然语言。代码解析指令执行对应的 Python 函数。将执行结果返回给模型。这一层的进步是模型从“只能想”变成“可以通过工具做”。4.3 第三层完整 Agent Loop工具调用通常不是一次就结束的。模型查询完天气可能还需要根据天气决定是否查航班查完航班可能还要计算总价。这要求我们把第二层的单次调用包装成循环for step in range(max_steps): 模型决策 - 如果要求调用工具 - 执行工具 - 结果加入上下文 - continue - 如果给出最终答案 - break这个循环直接带来三个新问题循环次数怎么控制需要设置max_steps防止模型陷入死循环。上下文越长越多如何避免超出 token 上限需要做上下文裁剪或摘要。工具执行失败怎么办需要把错误信息返回给模型让它自行修正或放弃。4.4 第四层多智能体协作类 PI-Agent当单 Agent 的上下文过长、工具过多时系统会变得混乱。此时引入“多智能体”架构每个 Agent 负责一个领域主控 Agent 负责任务路由。典型流程接收用户任务 - 规划 Agent 拆解任务 - 分发给检索 Agent / 分析 Agent / 代码执行 Agent - 汇总各 Agent 结果 - 总结 Agent 输出最终答案这一层不是代码层面的大改而是架构层面的重组。它的前提是前几层都已经稳定否则多智能体只会放大混乱。5. 完整示例与代码实现下面我们逐步实现一个可运行的智能体。为了便于理解这一节代码可以合并成一个独立的 Python 文件也可以拆成多个模块。我会用最直白的方式实现不引入重框架。5.1 定义工具注册表在项目根目录新建tools.py# 文件路径tools.py 简单工具注册表。 每个工具都是一个普通 Python 函数通过注册表暴露给大模型。 import datetime import random def get_current_time() - str: 返回当前日期和时间格式YYYY-MM-DD HH:MM:SS now datetime.datetime.now() return now.strftime(%Y-%m-%d %H:%M:%S) def calculate(expression: str) - str: 计算数学表达式比如 1 2 * 3。 注意生产环境不要使用 eval存在安全风险。 这里只用于演示实际项目应使用更安全的解析器。 try: # 只允许基本数学字符防止执行恶意表达式 allowed set(0123456789-*/(). ) if not set(expression).issubset(allowed): return Error: invalid character in expression result eval(expression) # noqa: S307 return str(result) except Exception as e: return fError: {e} def get_weather(city: str) - str: 模拟获取城市天气。 真实项目可改成调用天气 API。 weather_list [晴, 多云, 小雨, 阴] weather random.choice(weather_list) temperature random.randint(15, 30) return f{city} 天气{weather}{temperature}℃ # 注册表key 是暴露给模型的函数名value 是对应的函数对象 TOOL_REGISTRY { get_current_time: get_current_time, calculate: calculate, get_weather: get_weather, } def get_tools_schema(): 生成 OpenAI Function Calling 所需的 tools 参数 return [ { type: function, function: { name: get_current_time, description: 获取当前日期和时间, parameters: { type: object, properties: {}, }, }, }, { type: function, function: { name: calculate, description: 计算数学表达式比如 1 2 * 3, parameters: { type: object, properties: { expression: { type: string, description: 要计算的数学表达式, } }, required: [expression], }, }, }, { type: function, function: { name: get_weather, description: 查询指定城市的天气情况, parameters: { type: object, properties: { city: { type: string, description: 城市名称比如 北京, } }, required: [city], }, }, }, ]这里真正容易踩坑的地方是工具 schema 中的参数格式必须与函数签名严格对应。如果模型生成了错误的参数类型比如把expression传成数字执行层要自己做好校验和兜底。另一个关键点是安全性。在生产环境中把任意字符串交给eval执行是非常危险的。演示代码里我加了字符白名单但真实项目建议使用ast.literal_eval或者独立的计算引擎最好是只允许调用白名单内的函数。5.2 实现 Agent Loop 核心下面实现最核心的agent_loop.py。它负责维护对话消息、循环调用模型、执行工具、将结果回填上下文。# 文件路径agent_loop.py Agent Loop 核心实现。 流程 1. 将用户问题加入 messages。 2. 调用模型传入 tools。 3. 如果模型要求调用工具执行工具并把结果加入 messages。 4. 如果模型给出最终回答结束循环。 import json import os from dotenv import load_dotenv from openai import OpenAI from tools import TOOL_REGISTRY, get_tools_schema load_dotenv() client OpenAI( api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL), ) SYSTEM_PROMPT 你是一个智能助手。你可以使用外部工具获取实时信息。 如果用户的问题需要工具辅助请优先调用工具然后根据工具返回结果作答。 如果工具调用失败请根据错误信息调整参数后重试一次。 如果不需要工具直接给出回答。 def execute_tool_call(tool_call): 根据模型返回的 tool_call 执行本地函数 function_name tool_call.function.name arguments json.loads(tool_call.function.arguments) if function_name not in TOOL_REGISTRY: return json.dumps({error: funknown tool: {function_name}}) # 调用注册表里对应的函数 func TOOL_REGISTRY[function_name] try: result func(**arguments) return json.dumps({result: result}) except TypeError as e: # 参数不匹配把错误信息返回给模型自行修正 return json.dumps({error: str(e)}) except Exception as e: return json.dumps({error: str(e)}) def run_agent(user_input: str, max_steps: int 5) - str: 运行智能体主循环。 max_steps 用于防止死循环。 messages [ {role: system, content: SYSTEM_PROMPT}, {role: user, content: user_input}, ] tools get_tools_schema() for step in range(max_steps): print(f\n--- Step {step 1} ---) response client.chat.completions.create( modelos.getenv(MODEL_NAME, gpt-4o-mini), messagesmessages, toolstools, tool_choiceauto, ) message response.choices[0].message # 情况 A模型要求调用工具 if message.tool_calls: print(f模型决定调用工具{message.tool_calls[0].function.name}) # 先把模型的 tool_call 消息加入上下文 messages.append(message.model_dump()) # 逐个执行工具调用 for tool_call in message.tool_calls: print(f执行工具{tool_call.function.name}) result execute_tool_call(tool_call) print(f工具返回{result}) # 将工具结果加入上下文 messages.append( { role: tool, tool_call_id: tool_call.id, content: result, } ) # 继续循环让模型基于工具结果进一步推理 continue # 情况 B模型直接输出文本说明任务完成 if message.content: return message.content # 情况 C既没有工具调用也没有内容可能是异常 raise RuntimeError(模型返回了空响应请检查 messages 与 tools 配置) return 已达到最大循环次数任务未在限定步数内完成请尝试细化问题或增加 max_steps。 if __name__ __main__: # 测试一时间查询 answer run_agent(现在几点了) print(\n最终回答, answer) # 测试二推理 计算 answer run_agent(帮我计算 (23 * 17) 8 的结果) print(\n最终回答, answer) # 测试三多步工具协作 answer run_agent(北京天气怎么样如果温度超过25度帮我计算 5 个汉堡的总价每个 18 元。) print(\n最终回答, answer)这段代码就是单 Agent Loop 的核心骨架。你可能会注意到我没有使用任何第三方 Agent 框架因为理解底层的消息流转比会调框架的 API 更重要。框架能帮你省不少事但当循环逻辑出现问题时不懂底层会很难排查。5.3 类 PI-Agent 多阶段协作简化示例复杂系统不会只有一个 Loop而是多个 Loop 的编排。下面是一个简化的类 PI-Agent 架构示例包含规划、执行、总结三个阶段。# 文件路径multi_agent_demo.py 类 PI-Agent 简化版规划 - 执行 - 总结 说明这是一个结构示例用于展示多智能体协作的流程。 import os from dotenv import load_dotenv from openai import OpenAI from agent_loop import run_agent load_dotenv() client OpenAI( api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL), ) MODEL_NAME os.getenv(MODEL_NAME, gpt-4o-mini) def planning_agent(task: str) - list[str]: 规划智能体把复杂任务拆解为多个可执行子任务 prompt f请将以下任务拆解为最多 3 个有序的子任务。 每个子任务用一行描述不要有多余文字不要编号之外的格式。 任务{task} response client.chat.completions.create( modelMODEL_NAME, messages[{role: user, content: prompt}], temperature0.2, ) lines response.choices[0].message.content.strip().splitlines() return [line for line in lines if line.strip()] def summary_agent(task: str, sub_results: list[str]) - str: 总结智能体汇总子任务结果输出最终答案 joined \n.join(f- {r} for r in sub_results) prompt f原始任务{task} 以下是各子任务的执行结果 {joined} 请整合这些结果输出一份完整的最终回答。 response client.chat.completions.create( modelMODEL_NAME, messages[{role: user, content: prompt}], ) return response.choices[0].message.content if __name__ __main__: task 查询当前时间然后结合当前时间给我一条工作建议 print(f原始任务{task}) # 阶段一规划 subtasks planning_agent(task) print(\n规划结果) for i, s in enumerate(subtasks, 1): print(f{i}. {s}) # 阶段二执行每个子任务都可以是一个独立 Agent Loop results [] for subtask in subtasks: print(f\n执行子任务{subtask}) try: result run_agent(subtask, max_steps4) results.append(result) except Exception as e: results.append(f子任务失败{e}) # 阶段三总结 final summary_agent(task, results) print(\n最终总结, final)这个示例展示了“多阶段编排”的基本模式先让一个模型做全局规划再用独立的 Agent Loop 执行每个子任务最后汇总成报告。真实生产环境中的类 PI-Agent 会更复杂比如每个子 Agent 有自己独立的 System Prompt 和工具集。子任务之间可能有依赖关系需要拓扑排序。需要一个消息队列来解耦任务分发和执行。需要独立的记忆存储比如向量数据库让 Agent 可以查询历史信息。但从架构演进来看你掌握了“Loop 编排”之后再去看各类 Agent 框架会轻松很多。6. 运行结果与效果验证6.1 运行准备在项目根目录创建.env文件# 文件路径.env OPENAI_API_KEYsk-xxxx OPENAI_BASE_URLhttps://api.openai.com/v1 MODEL_NAMEgpt-4o-mini如果你的模型服务不支持 Function Calling可以换用支持 OpenAI 兼容接口的模型或直接使用主流云厂商的模型服务。6.2 运行 Agent Looppython agent_loop.py预期你会看到类似输出--- Step 1 --- 模型决定调用工具get_current_time 执行工具get_current_time 工具返回{result: 2025-06-14 15:30:22} --- Step 2 --- 模型给出最终回答 最终回答 当前时间是 2025-06-14 15:30:22。注意实际输出内容取决于你使用的模型和当前时间。判断成功的关键不是模型回复的字面内容而是是否出现了至少一次模型决定调用工具。是否能看到工具返回。最终是否回到自然语言回答。如果这三个节点都出现说明 Agent Loop 跑通了。6.3 验证多智能体协作示例python multi_agent_demo.py预期会依次打印规划结果、每个子任务的执行过程、最终总结。这可以验证多阶段编排流程是通的。6.4 失败排查第一步如果代码跑不通第一件事不是改逻辑而是看模型返回的原始消息# 临时加一行调试代码放在 create 调用之后 print(response.choices[0].message.model_dump())这样你能看到模型到底返回的是tool_calls、content还是两者都有。如果模型输出里有content也有tool_calls说明模型在尝试边回答边调用工具而你的执行逻辑只处理了tool_calls分支可能丢掉一部分信息。7. 常见问题与排查思路以下是这个架构里出现频率最高的问题与排查方法建议收藏备用。问题现象可能原因排查方式解决方案模型从不调用工具模型不支持 Function Callingtools 参数格式错误Prompt 没有引导检查模型文档确认 tools 支持情况打印 response 看是否报错换一个已知支持 Function Calling 的模型升级模型检查 tools schema 是否符合 OpenAI 格式在 System Prompt 中强调优先调用工具工具调用后模型重复调用同一个工具上下文里工具结果没有正确回填模型无法理解结果打印 messages确认 tool 角色的消息是否完整确保每一条 tool_call 都有对应的 roletool 消息且 tool_call_id 匹配循环到第 N 步仍然不结束max_steps 设置太小模型一直在调用工具但方向错误观察每一步的工具调用内容看是否在重复相同动作提高 max_steps优化 Prompt要求模型避免重复调用增加“如果工具结果已满足需求则直接回答”的指令模型返回“已达到输出 token 上限回答被截断”单次模型输出过长或上下文累计超过模型限制查看调用日志中的 token 用量减少 messages 中的历史消息做上下文裁剪只保留最近的几轮对历史对话做摘要工具参数解析失败模型生成的 JSON 参数和函数签名不匹配把 tool_call.function.arguments 打印出来在 execute_tool_call 中捕获 TypeError把错误信息返回给模型让模型自己修正参数本地模型调用 tools 参数报错本地模型或推理引擎尚未兼容 Function Calling检查本地推理服务的日志确认是否识别 tools 字段改用兼容性更高的推理引擎或者退化为“让模型以 JSON 文本格式输出工具调用”自行解析任务很复杂但单个 Agent 上下文不够单 Agent 上下文长度有限塞入太多内容后丢失早期信息查看 token 计费/日志看上下文累积量引入多 Agent 架构专事专办或用向量数据库存储长期记忆其中最常见也最隐蔽的问题是“工具调用死循环”。表现为模型反复调用同一个工具答案却始终不收敛。我通常会在 System Prompt 中加一句“如果你已经获得了回答用户问题所需的全部信息请停止调用工具并直接给出最终答案。”这个约束在实际项目中往往比调大 max_steps 更有效。8. 最佳实践与工程建议8.1 安全边界工具是最高风险区域Agent 能做的事情比 Prompt 能说的事情危险得多。给模型开放工具等于给了它操作权限。以下几条是底线所有工具必须做参数白名单校验尤其是涉及文件路径、URL、Shell 命令的工具。生产环境禁止直接将模型生成的 Shell 命令交给subprocess.run。如确需执行必须经过强校验或人工审批。写操作删除、修改、发消息、转账应当设计审批机制高风险操作要默认拒绝。8.2 幂等性让重试安全工具调用可能因为网络超时而失败Agent 会重试。如果你的工具不是幂等的——比如“创建订单”被调用两次就会产生两笔订单——这就是灾难。解决方案是给每个工具请求分配一个 request_id数据库侧做去重。或者把执行结果缓存起来相同参数直接返回上次结果。8.3 上下文管理控制 token 增长Agent Loop 每多一轮messages 就会增加一轮工具调用和结果。长任务跑到 20 轮以后上下文会变得很大。可选策略完成一个子任务后把它的对话摘要成一段文字替换原有详细内容。只保留最近 N 轮对话更早的对话放入外部记忆存储。对于长期记忆场景使用向量数据库在需要时检索相关片段注入上下文。8.4 日志与可观测性写 Agent 应用和写普通 Web 应用不一样你不仅需要看请求参数和响应还需要看到“模型在哪一步调用了哪个工具参数是什么结果是什么下一步决策如何变化”。建议为每一次循环输出结构化日志{ step: 3, event: tool_call, tool: get_weather, input: {city: 北京}, output: 晴 26℃, model_response: ... }有了这些日志你才可能在用户投诉“AI 干了奇怪的事”时快速定位问题。8.5 成本控制与限流每次循环都是一次模型调用都会产生 token 消耗。如果用户问了一个简单问题而模型不在一步内回答而是绕了 5 个工具调用成本差距是数倍。建议对单次会话设置 max_steps 上限一般 5 到 10 步足够。对单用户单日调用量做配额限制。在 Prompt 中强调“能在一步内回答就不要调用工具”。8.6 从单 Agent 到多 Agent 的迁移时机不要一开始就搭建多 Agent 系统。参考下面的信号单 Agent 的 tools 数量膨胀到 15 个以上模型经常混淆该调用哪个。单次任务需要的工具调用步骤超过 10 轮上下文难以管理。不同子任务需要完全不同的 System Prompt 和工具集。团队分工导致不同人维护不同模块需要代码层面解耦。满足两个以上条件再考虑多 Agent 架构。9. 总结与后续学习方向这篇文章的核心可以浓缩成一句话模型是单次推理的大脑Agent 是带行动能力的循环系统而 Loop 是这个系统能从“会回答”走向“能干活”的关键架构。我们从单次模型调用起步逐步加入了工具注册、Function Calling、Agent Loop 和多智能体编排最终实现了一个简化版类 PI-Agent 复杂智能体骨架。整个过程没有依赖炫技框架核心逻辑不过是一个for循环加上消息管理但正因为简单才更容易理解真正的工程难点在哪里。接下来你可以往几个方向继续深入替换成更强的本地模型对比不同模型在工具调用上的稳定性和准确性。引入向量数据库作为长期记忆让 Agent 能记住历史会话和项目知识。研究真实的多智能体框架的编排源码重点看消息传递、并发调度和错误恢复的实现方式。为智能体增加前端交互层提供实时流式输出让用户能看到“AI 正在调用哪个工具”。需要提醒的是不要迷信“轮数越多越智能”。在实际项目中更高的智能往往来自更清晰的任务拆解、更完备的工具 schema 和更严格的上下文管理。Loop 只是骨架真正决定系统上限的是你如何设计每一步的输入与输出。如果你正在设计自己的类 PI-Agent 架构建议先从一个小闭环跑通再逐步扩大工具集和智能体数量。把每一步的工具返回值、模型决策日志都记录下来你会发现自己对这套系统的理解会快速提升。
返回列表