ARTICLE DETAIL

资讯详情

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

从单次调用到Agent Loop:构建可复用复杂智能体架构

从单次调用到Agent Loop:构建可复用复杂智能体架构 类 PI-Agent 的复杂智能体架构最核心的转变不是把模型换大而是把单次模型调用改造成一个带状态的 Loop循环。做 Agent 开发的人一开始最容易卡住的恰恰是这一步模型调用谁都会写但任务一旦变成“查资料、执行操作、根据结果继续决策”单次调用就完全不够用了。这篇文章会把模型与智能体的关系、Loop 控制流怎么设计、状态和工具怎么管理、最小可运行骨架怎么写以及常见故障的排查顺序完整拆一遍。适合两类人看一类是准备从单轮 API 调用走向 Agent 框架的开发者另一类是已经在用 Dify、扣子这类平台或其它框架但遇到死循环、上下文超限、工具调用不稳定时想搞清楚底层逻辑的人。1. 先搞清楚模型和智能体的边界在哪里1.1 模型只是函数不是流程很多初学者有个误区把“大模型能回答问题”等价于“大模型能完成任务”。实际上模型本质上是一个函数输入一段文本输出一段文本或者输出一组结构化 token。它确实能理解上下文、能分步推理、能给出工具调用的 JSON但它本身不会主动执行工具不会循环重试不会在步骤之间维护状态。模型与智能体的边界简单说模型负责“思考”生成回复、判断下一步动作、输出最终答案。智能体负责“流程”决定何时调用模型、何时调用工具、何时停止、如何把每一轮结果传回模型。这个边界不拆清楚后面写 Loop 时会出现一个典型问题把循环逻辑硬塞进 system prompt让模型自己“记住”要循环。两三轮的短任务可能能跑任务一长就会乱上下文也会迅速膨胀。我自己在实际项目里的经验是模型层永远只做一件事就是“给定消息列表返回文本”。所有循环、判断、工具执行都在模型外面做。这样模型可以被随意替换今天用这个模型明天换另一个架构主体不用动。1.2 什么场景必须上 Loop判断标准很直接如果任务只需要一次补全比如翻译、总结、单轮问答单次调用就够。如果任务包含多个步骤而且后续步骤依赖前一步的输出就必须用 Loop。如果任务需要外部工具参与比如查数据库、发请求、读文件、执行代码单次调用只能负责“建议动作”真正执行和结果回填必须由 Loop 完成。举一个最普通的例子。让 Agent“帮我查一下 A 项目的状态然后给负责人发一封进度提醒邮件”。模型自己并不知道 A 项目当前状态也没有发邮件的能力。它必须先输出一个查询动作系统执行查询后把结果喂回模型模型再根据结果生成邮件内容并调用发送工具。这个“查询—反馈—再决策—执行”的过程就是 Loop 存在的原因。所以我建议在架构设计早期就把调用层和流程层分开。调用层只负责“把消息列表发给模型拿到回复”流程层才负责循环、工具执行、终止判断。这个分层看起来只是多写几个函数但后面调试时价值极大。1.3 类 PI-Agent 架构最值得借鉴的几点要说明一下“类 PI-Agent”在我理解里并不是某一个固定框架而是一类复杂智能体架构的设计取向重点落在“多阶段循环 显式状态 工具编排”上。这类架构通常包含几个共同点有一个显式的 Agent Loop而不是靠模型自觉完成多步任务。模型输出被结构化解析通常是函数调用或 JSON 动作。工具注册表和状态管理器是独立模块。循环有明确退出条件包括成功结束、失败结束、轮次上限、超时。每一步都有日志方便回溯哪一轮出了问题。这些点看起来简单实际落地时每一条都会引出不少坑。下面从控制流、模块拆分、代码骨架、并发和排查五个方向展开。2. 从单次调用到 Loop控制流到底变了什么2.1 单次调用的标准流程单次调用大概是这样的拼一段 prompt可能包含 system 和 user 消息。调用模型接口传入消息列表。拿到返回文本展示给用户。这段流程的优点是简单缺点是一切逻辑都靠模型一次生成。比如让模型“先查今天的天气再根据天气推荐穿搭”如果模型没有真实天气数据它只能编造或者建议你另外去查。也就是说单次调用做不到“执行后反馈再决策”。2.2 Loop 的每一次迭代在做什么Loop 把单次调用扩展成一个重复执行的单元。每一次迭代至少包含四步把当前状态通常是累积的消息列表发给模型。模型返回下一动作可能是最终答案也可能是“调用某个工具参数是什么”。如果是最终答案退出循环。如果是工具调用执行工具把结果追加到消息列表里进入下一轮。这里有一个关键点必须强调模型每一次生成时看到的是前面所有轮次的累积内容而不是只有最后一个工具结果。这就是为什么 Loop 设计里消息列表的维护特别重要。每轮追加消息消息列表会不断变长模型要处理的 token 数也随之增加。很多“跑到第几轮突然变慢”的问题根源就在这里。单次调用和 Loop 的差异可以简单对比如下维度单次调用Agent Loop输入固定消息列表动态累积的消息列表输出一段文本文本或结构化动作工具执行不支持解析动作后执行状态无有每轮更新退出条件调用完即结束需要显式判断适用场景翻译、总结、单轮问答多步决策、工具调用、自主执行2.3 谁来控制循环外部循环而不是模型自己设计 Loop 时最容易犯的错是让模型自己控制“要不要继续”。你可以让模型输出“我还需要继续查”但真正的 while 循环条件必须由外部代码控制。原因是模型输出存在随机性单次生成可能漏掉结束标记也可能连续多轮输出同一个工具调用而不推进。如果外部没有轮次上限和超时任务就会死循环token 费用和延迟都会失控。所以一个稳妥的 Loop 至少要同时具备四个退出条件模型输出明确结束标记。工具调用结果为空或明确表示完成。达到最大轮次 max_steps。整体运行超过 timeout。这四个条件缺一个都不够。尤其在开发环境我会把 max_steps 设小一点比如 5 到 10先跑通再说。等逻辑稳定了再把轮次放宽到业务实际需要的范围。3. 设计 Agent Loop 时先拆好这四个模块3.1 状态管理模块状态是 Loop 的血肉。每个循环迭代之间共享什么数据决定了这个 Agent 是“有记忆”还是“每次重新开始”。最简单的方式是用一个消息列表作为唯一状态。所有发给模型的 user、assistant、tool 消息都往同一个列表里追加这样模型天然能看到完整历史。但要注意消息列表只增不减到后期 token 消耗会很高。更稳的做法是把状态拆成两层对话层存放与模型交互的消息记录。业务层存放任务相关的中间结果比如已经查到的数据、已生成的文件、用户提供的参数。业务层不一定要全部喂给模型。你可以选择只喂最近几轮的工具结果或者做一次压缩摘要。这个决策会直接影响效果和成本。我的建议是第一版先不做压缩直接全量传。等任务变长、token 涨到不可接受时再加入摘要或滑动窗口机制。过早优化反而会让排查变难。3.2 工具注册与执行模块工具模块不能只写一个 if-else 判断函数名。更工程化的做法是做一个工具注册表每个工具有名称、描述、参数 schema、执行函数、返回值结构。这样模型可以通过结构化输出明确告诉系统调用哪个工具系统也能在调用前做参数校验。一个简单的注册表结构可以是name工具名必须唯一。description给模型看的说明告诉它什么时候用这个工具。parametersJSON Schema约束参数格式。handler真实执行函数。为什么要给模型描述和参数 Schema因为模型是靠文本理解工具的。描述写得含糊模型就会在决策时犹豫或乱选。参数 Schema 写得宽松模型可能传错类型。第一次做 Agent 的人建议先跑一个只有两个工具的样例一个查询类一个写入类。把工具描述反复打磨直到模型每次都能选出正确工具。3.3 记忆与上下文窗口管理模型有上下文长度限制。Loop 每轮往消息列表追加内容迟早会触顶。触顶之后有两种常见表现请求直接报错提示超过最大 token。更隐蔽的情况模型被迫截断回答只返回一半内容。网上经常能看到“已达到输出 token 上限回答被截断”这类提示本质上都是上下文或输出长度没有预留空间。有些平台会让你发送“继续”让模型接续这算是一个兜底方案但架构设计里不能依赖这种手动续写。更好的解决办法有几种限制单轮工具结果的大小比如只保留前 1000 个字符。对中间历史做摘要用摘要取代旧消息。把无关的中间过程移出消息列表只保留结论。根据任务类型调整 max_tokens给输入或输出留空间。这里有个原则优先控制输入而不是无脑加长上下文。把工具返回结果截断、压缩比单纯扩大模型上下文窗口更省钱、更稳。3.4 策略与终止条件Loop 的终止条件不是写死在某一处而是由一组策略组成。常见策略包括模型输出以 FINISH 或类似标记开头。工具结果满足业务完成标准比如查到的数据已经写入目标表。达到轮次上限。用户中断或人工审核通过。在复杂智能体架构里我喜欢把“是否终止”单独抽成一个函数叫 should_stop(state)。这样做的好处是同一套 Loop 骨架可以适配不同任务。比如翻译类任务第一轮就能停数据分析类任务可能要跑五轮。终止策略独立后改任务时不需要动循环主体。4. 最小可运行骨架先跑通一个最简 Agent Loop4.1 环境准备与最小样例写代码之前先确认几件事你能访问某个模型接口。具体用什么模型不重要关键是模型支持结构化输出或函数调用。如果不支持也可以用“让模型输出 JSON”的普通文本方式但解析时要想好容错。你本地有 Python 环境建议 3.9 以上代码里用到了 dataclass。准备一个测试用工具。不要一上来就接十几个真实服务先用一个“返回当前时间”和一个“做加法”的假工具把 Loop 跑通再替换真实工具。4.2 核心代码示例下面这个骨架是我常用的一种写法。它不依赖任何特定框架只要求你有一个 model_call 函数输入消息列表输出文本。import json import time from dataclasses import dataclass, field from typing import Callable dataclass class AgentState: messages: list field(default_factorylist) step: int 0 tool_results: dict field(default_factorydict) finished: bool False error: str def agent_loop( model_call: Callable, tools: dict, state: AgentState, max_steps: int 8, timeout: float 60.0, finish_markers: tuple (FINISH, ANSWER:), parse_action: Callable None, ): start_time time.time() while state.step max_steps: if time.time() - start_time timeout: state.error timeout break # 1. 调用模型 response model_call(state.messages) state.messages.append({role: assistant, content: response}) # 2. 判断是否直接结束 if response.startswith(finish_markers): state.finished True break # 3. 解析动作 if parse_action is None: state.error missing parse_action break action parse_action(response) if action is None: # 解析失败记录错误信息给模型一次纠正机会 state.messages.append({ role: tool, content: ERROR: action parse failed. Please output valid action. }) state.step 1 continue tool_name action.get(tool) tool_args action.get(args, {}) # 4. 执行工具 if tool_name not in tools: state.messages.append({ role: tool, content: fERROR: unknown tool {tool_name} }) state.step 1 continue try: result tools[tool_name](**tool_args) except Exception as e: result fTOOL_ERROR: {e} state.tool_results[state.step] result state.messages.append({ role: tool, content: fRESULT: {result} }) state.step 1 return state这段代码有几个细节要注意。每次工具结果都用固定前缀拼进消息比如 RESULT:。这样做是为了让模型更好区分“这是工具结果不是用户要求”。解析失败时不直接退出而是把错误信息回填给模型让它下轮输出合法动作。这比直接报错更接近真实情况。最大轮次和超时是双保险宁可任务失败也不要无限跑。4.3 调用示例与验证方式假设有一个工具函数 get_time一个 model_call 模拟器def get_time(): return 2025-01-15 10:30:00 def fake_model_call(messages): # 开发阶段可以用固定逻辑代替真实模型方便验证 Loop last messages[-1][content] if RESULT: in last: return ANSWER: 当前时间是 last.split(RESULT:)[-1] return json.dumps({tool: get_time, args: {}}) tools {get_time: get_time} state agent_loop( model_callfake_model_call, toolstools, stateAgentState(messages[ {role: user, content: 请告诉我当前时间} ]), max_steps5, timeout10, ) print(state.messages) print(finished:, state.finished)成功的判断标准有三条state.finished 为 True。messages 里既出现了工具调用也出现了解析出的工具结果。最后一条 assistant 消息以 ANSWER: 开头。我一般会先用这种“假模型”验证 Loop 骨架确认控制流没问题再换成真实模型。这样可以把“模型输出问题”和“框架逻辑问题”分开排查。4.4 为什么建议先跑假工具和假模型很多人第一次写 Agent Loop 就把真实模型、真实工具全接上一旦报错根本分不清是模型生成的问题、工具参数的问题还是 Loop 逻辑的问题。用假模型和假工具有几个好处输出可控可以精准测试每一种分支。不花钱、不耗 token。可以快速构造异常输入比如让模型输出非法 JSON验证系统容错。等骨架稳定后再逐个替换成真实组件。这个顺序能省掉大量排查时间。真实模型延迟高、费用贵用它来调试控制流是最不划算的做法。5. 从单任务到批量任务资源估算与并发控制5.1 单个 Loop 的 token 消耗怎么估算一个 Agent Loop 跑完token 消耗约等于每一轮输入消息的 token 总和加上每一轮模型输出的 token 总和。因为消息列表是累积的第 N 轮的输入会包含前 N-1 轮的所有内容。所以总的输入 token 会随着轮数近似平方级增长。举例来说如果每一轮新产生的工具结果和模型输出大约是 1000 token跑 10 轮单单输入 token 就是 1k 2k ... 10k约 55k token实际还要加上 system 指令和 prompt 模板。这个估算方法很重要它决定了两个参数批量任务规模能开多大。每个任务的最大轮次设多少。不要用一个跑 3 轮就结束的任务去推断所有任务。有的任务轮次差异很大预算评估必须按最坏情况算。5.2 批量跑与单跑的区别单任务跑通后进入批量阶段会有几个新问题输出文件命名多个任务的结果如果都写同一个文件会互相覆盖。失败重试有一个任务跑到第 6 轮报错是整批重跑还是单独重跑。并发上限同一个 API key 同时开 20 个 Loop可能触发限流。日志关联每个任务要有独立 task_id日志里能筛出某一次完整链路。我的建议是先串行跑再看延迟是否可接受。串行确认稳定性之后再加并发。不要一开始就上高并发。5.3 并发参数怎么定并发数到底设多少和几个因素有关API 服务端的速率限制。每个任务的平均 Loop 轮次。每轮携带的 token 大小。下游工具服务的承载能力。最简单的方法先开 1 个并发跑 5 个任务记录成功率、耗时和错误码。如果平均耗时和单任务接近再逐步加到 2、4、8。每加一档观察错误率和延迟变化。如果错误率突然上升说明很可能已经接近限流阈值。不要迷信并发越大越好。很多 Agent 任务是在等模型生成不是等计算开太多并发只会放大错误和成本。本地模型或私有部署的场景还要额外关注显存和内存占用。低配置跑单条可以不代表适合批量并发这个边界一定要提前想清楚。6. 常见故障与排查顺序6.1 模型输出格式解析失败现象parse_action 一直返回 None或者工具名、参数解析错误。排查顺序先打印原始模型输出看它到底输出了什么。很多时候不是格式问题而是多了一行解释文字或者 JSON 外面包了 markdown 代码块。确认 prompt 里是否明确要求输出格式。给一个示例比给一堆规则更有效。如果是函数调用模式确认模型接口返回的结构是 content 还是 tool_calls字段名不能拿错。如果模型经常在 JSON 前加废话可以尝试在解析前做裁剪或者用正则提取最外层 JSON。加一个兜底分支解析失败时把错误信息回填给模型让它自己修正。不要一失败就换模型。多数情况下换一种输出约束或加一个示例就能解决。6.2 死循环或到达轮次上限现象任务一直不结束最后触发 max_steps。常见原因有两个模型反复调用同一个工具参数完全一样结果也一样但 Five 判断“是否完成”的逻辑没有生效。工具结果虽然是正确的但模型没有输出结束标记而是继续请求工具。排查时先看日志里每一轮的 assistant 输出。如果连续三轮都是同一个动作说明策略层缺一个“相同动作幂等检测”。更稳妥的做法是当某个工具名和参数与上一轮完全相同时直接终止并报“重复动作”。另一个思路是分析工具结果如果结果已经满足业务完成条件就主动终止不要等模型自己说 END。业务完成判断放外部比放模型更可靠。6.3 上下文超限或回答被截断现象请求报 token 超限或者模型回答到一半被截断输出不完整。排查顺序看请求的 messages 里总 token 数。很多 SDK 会打印 usage这是最直接的判断依据。看是否每一轮都追加了完整工具结果。如果工具结果很大比如几千行的数据库查询必须截断或摘要。看消息历史是否可以裁剪。旧轮次对当前决策可能不再重要可以用一句话摘要替换。看模型输出的 max_tokens。如果任务要求长回答但 max_tokens 设置过小就会出现“输出被截断”。这类问题在长 Loop 里几乎是必然遇到的。我的建议是在消息列表长度接近阈值之前主动做一次压缩不要等到报错才处理。6.4 工具执行异常没有暴露现象工具调用失败但 Loop 继续跑最终结果看起来“还行”实际数据是错的。这其实是最危险的情况。一个加法工具传了字符串执行报错结果被异常处理捕获后变成TOOL_ERROR: ...这个错误又作为工具结果喂给了模型。模型可能没意识到这是错误继续往下推理。解决方法是让工具结果带上结构化标志。比如所有工具都返回{ ok: true, data: ... }或者{ ok: false, error: reason }Loop 在执行完工具后先检查 ok 字段。如果 ok 为 false可以选择让模型重新调用也可以直接终止并标记失败。我的默认做法是允许模型重试一次但如果同一个工具错误超过两次就终止。6.5 通用排查顺序总结遇到任何 Agent 问题我建议按这个顺序排查先看现象是没输出、报错、卡死、还是结果错误。再看日志把每一步的 assistant 输出、解析结果、工具结果、终止判断全部打出来。再看输入消息列表有没有被污染工具结果有没有异常是否混入了不该出现的内容。再看参数max_steps、timeout、max_tokens、并发数是否合理。最后看环境依赖版本、API key、模型名是否对应。这些基础问题最容易在新环境翻车尤其是本地加载模型时模型路径和文件格式经常是首要排查点。踩过几次之后我发现很多问题不是工具能力不够而是前置环境和输入材料没有处理干净。Agent 架构本身可以写得很简单真正决定上线后稳不稳的是状态管理、退出条件和错误处理有没有做完整。这套从单次调用到 Loop 的改造本身并不复杂复杂的是把状态、工具、终止条件和错误处理都做成显式模块。我个人的建议一直是同一个先跑通最简骨架再逐步加工具、加任务、加并发。只要把消息列表、工具结果、退出条件、token 消耗、日志、重试策略这六件事盯住架构就不会出大的方向性问题。
返回列表