写AI Agent最让人头疼的是什么不是Prompt写不好也不是模型选不对——而是你明明给了它一堆工具和任务它却在执行过程中反复调用同一个工具或者在一个环节里绕来绕去最后既没完成任务也没告诉你发生了什么。我在做自动化代理项目时几乎有一半时间都在和这种“循环”问题搏斗。后来才慢慢意识到Agent能不能稳定产出结果拼的就是循环工程也就是业内常说的Loop Engineering。Loop Engineering不是某一个开源框架而是一套关于如何设计“思考→行动→观察→再思考”循环的工程实践。它的目标很明确让Agent在复杂任务中能自主前进、及时终止、失败自愈。简单说如果Prompt是Agent的大脑工具是Agent的手脚那循环就是让手脚按照大脑指令持续工作、直到把活干完的那套动力系统。这篇文章适合正在做AI Agent开发、RAG应用、自动化流程的朋友尤其是那些写Demo没问题、一上真实场景就卡住的人。我会从循环的本质讲起再给一个最简实现然后带一个PDF多工具分析助手的实战项目最后把我在Loop Engineering里踩过的坑和排查方法整理出来。内容偏向动手建议把代码跑起来看。1. Agent循环的本质为什么说Loop Engineering是Agent的发动机1.1 单次调用与循环调用是两回事很多初学者会以为“调一次LLM就等于一个Agent”。实际不是。单次调用只能做三件事根据上下文生成文本、返回一段JSON、或者调用一次外部接口。它没有记忆不会追问自己“刚才拿到了什么结果、下一步该做什么”。以“查明天北京到深圳的机票并给出差抵用建议”为例。单次调用LLM可能只能生成一段“我建议您去官网查询”的回答。要真正完成这件事Agent得先调用航班查询工具拿到航班列表再调用日历判断明天日期可能还要调用天气工具最后把交通、住宿、行程建议汇总成报告。这个过程中每一轮都是“模型生成一个动作→工具返回结果→模型根据结果生成下一个动作”这就构成了一个循环。这个循环就是Agent Loop。在ReAct范式里它被概括为“Thought思考→ Action行动→ Observation观察”三个步骤的重复。Thought告诉模型现在所处的情况Action决定调用哪个工具或返回最终答案Observation是工具执行后的反馈。模型在循环里不断更新状态直到它能输出一个“结束”信号。1.2 为什么循环这么难设计难的不是写一个for循环而是几个工程问题什么时候循环结束循环里记忆存什么工具报错了怎么办这些问题单靠Prompt是解决不了的需要工程手段。先说终止条件。很多Agent跑飞就是因为循环缺少“刹车”。Agent任务千变万化模型有时候会误判“没完成”于是继续调用工具有时候明明已经拿到答案还要再“确认”一遍。如果你只设一个很大的max_iterations问题不是安全而是成本和耗时失控——一次循环调用一个工具可能只要几秒但累计起来一个任务跑上十几分钟也很常见。我一般在设计Loop Engine时都会要求必须有硬性终止条件且终止后必须返回可读的退出原因。再说状态维护。循环里每轮产生的Observation、工具结果、最终答案都需要有地方放。放全了上下文窗口容易爆放少了模型又缺少决策依据。这就是状态管理问题。最后是错误处理。工具调用失败、超时、返回格式不对都可能在循环中发生。如果代码不做兜底Agent可能带着错误继续跑最后给出一个看着像模像样但其实是瞎编的结论。所以Loop Engineering真正要管理的其实是“终止、状态、错误”这三个核心变量。1.3 Loop Engineering与Agent框架的关系这里需要澄清一点很多框架里已经封装了Agent Loop比如LangChain的AgentExecutor、开源社区的ReAct Agent等。那为什么还要自己理解Loop Engineering因为框架给的是通用默认值做真实项目时远远不够用。你需要知道它内部是怎么决定“调用哪个工具”的才能配置准确的停止条件、做上下文压缩、加错误重试策略。我就是从“直接用框架”到“自己写了一个小规模Loop Engine”之后才算真正入门。这篇文章后面给的实战项目也会用最少的代码来实现一个可控的循环。2. 保姆级实操从零手写一个最小可用的Agent Loop2.1 最小依赖与整体结构先把这个最小系统跑起来再看理论。环境只需要Python 3.10装一个能调LLM的客户端库再加几个函数就足够。代码结构分成四块LLM调用负责把messages发送给模型拿到文本或JSON返回。动作解析从模型输出里解析出到底是调用工具还是直接给最终答案。工具执行注册一个工具字典按名字去执行对应的Python函数。循环主体按顺序调度上述逻辑并维护循环状态。为什么不用现成框架因为最小实现能让你直观看到每个环节后面做定制时才不会两眼一抹黑。等理解了再回到框架里你会知道该覆盖哪些方法。下面是一个最简的循环主体。为了通用性我把LLM调用封装成一个mock函数实际使用时换成你自己用的SDK即可。import time from dataclasses import dataclass, field from typing import List, Dict, Callable dataclass class LoopState: # 保存整个对话历史含工具返回结果 messages: List[Dict] field(default_factorylist) # 单独记录每步的工具调用结果方便追踪 tool_results: List[Dict] field(default_factorylist) # 当前迭代序号 current_step: int 0 # 是否已正常结束 finished: bool False def llm_complete(messages: List[Dict]) - str: # 实际项目请替换为你自己的LLM客户端 # 这里的自定义函数需要能理解 messages并返回诸如 # ACTION: finish\nANSWER: 完成 # 或 # ACTION: tool\nTOOL_NAME: get_weather\nARGS: {city: 北京} raise NotImplementedError def parse_action(text: str) - Dict: # 对模型返回文本做简单解析 lines text.strip().split(\n) action_type tool_name tool_args {} answer for line in lines: if line.startswith(ACTION:): action_type line.split(:, 1)[1].strip() elif line.startswith(TOOL_NAME:): tool_name line.split(:, 1)[1].strip() elif line.startswith(ARGS:): import json tool_args json.loads(line.split(:, 1)[1]) elif line.startswith(ANSWER:): answer line.split(:, 1)[1].strip() return { type: action_type, tool_name: tool_name, tool_args: tool_args, answer: answer, } def execute_tool(tool_name: str, tool_args: Dict, tools: Dict[str, Callable]): return tools[tool_name](**tool_args) def run_agent(user_task: str, tools: Dict[str, Callable], max_iterations: int 8) - str: state LoopState() state.messages.append({role: user, content: user_task}) for _ in range(max_iterations): state.current_step 1 # 1. 调用LLM根据当前消息列表决定下一步 response llm_complete(state.messages) action parse_action(response) # 2. 判断是否完成 if action[type] finish: state.finished True return action[answer] # 3. 执行工具调用 if action[type] tool: if action[tool_name] not in tools: error_msg f工具 {action[tool_name]} 不存在 state.messages.append({role: assistant, content: response}) state.messages.append({role: user, content: f错误{error_msg}}) continue try: result execute_tool(action[tool_name], action[tool_args], tools) except Exception as e: result f工具执行异常{e} # 记录结果 state.tool_results.append({ tool_name: action[tool_name], args: action[tool_args], result: result, }) # 把结果写回消息列表 state.messages.append({role: assistant, content: response}) state.messages.append({role: user, content: f工具返回{result}}) # 如果这里还可以加入超时检测、重复动作检测等 return 达到最大循环次数任务未完成2.2 设计循环时必须想清楚的三个参数上面代码里被我偷懒跳过的几个点恰恰是Loop Engineering的关键。第一个是max_iterations。它不是随便设的。任务类型不同迭代上限差异很大。比如“查天气并整理穿衣建议”通常3次以内就够但“阅读一份多章节PDF生成财务分析报告”8次可能都太紧。我的经验是先按最坏情况估算一个上界再加2次余量。宁可让它在未完成任务时终止也不要让它在死循环里烧钱。第二个是超时设置。单次工具调用可能卡住整个循环也可能卡住。我在核心循环外层会加一个wall_clock阈值超过就终止。有的框架叫timeout有的叫deadline。给一个建议单步工具调用timeout设为30秒整轮任务timeout按max_iterations乘以单步耗时再乘1.5。第三个是错误容忍度。并不是所有工具执行失败都需要立即终止。比如天气接口偶尔超时重试一次可能就好了但你也不希望模型连续失败10次还在试。我会单独维护一个consecutive_failures变量连续失败超过2次就走“失败恢复”分支提醒模型更换工具或者直接结束。2.3 怎么让模型更稳定地返回“结束信号”这个点容易被忽略。模型在循环里输出“finish”之前往往需要你给一个明确的输出格式约定。最简单的做法是在系统提示词里写明如果已经完成任务必须输出“ACTION: finish”和“ANSWER: 最终答案”如果还需要调用工具就输出“ACTION: tool”和工具名、参数。我在实际项目里还会让模型先输出一小段“当前进展”再输出下一步动作。有了进展描述调试循环时能知道每一步模型在想什么。需要注意的是不要把所有工具的说明都堆在系统提示词里。工具多了以后模型容易把工具描述当作一般文本导致解析出错。更好的做法是动态拼接工具列表只把当前阶段可能用到的工具注入给模型后面实战部分会讲。3. 项目实战用Loop Engineering做一个多工具PDF分析助手3.1 项目需求与工具链路设计我选了一个很常见的场景给Agent一份PDF比如季度财报让它回答“这个季度研发费用是多少和上季度比有什么变化”这种问题。这类任务单次LLM调用做不了因为模型看不到PDF内容。它至少得经历“解析PDF→切块索引→检索相关段落→可能还要计算同比增长→最后汇总答案”的过程。为了演示循环我给这个PDF助手设计了四个工具pdf_parse解析PDF文件内容返回文本块列表。build_index把文本块做向量化索引。这里可以直接用faiss或者简单的内存字典只要能把“检索”动作体现出来即可。search_index用关键词或向量检索返回最相关的几个片段。calculator执行简单的数学表达式用于对比增长率。工具链路不是一开始就全部注入给模型。Agent会根据当前状态选择下一步如果还没解析PDF就先调pdf_parse如果还没有索引就调build_index如果用户的问题涉及数字计算再调calculator。这样设计的好处是避免模型在第一步就乱调用也减少上下文噪音。3.2 状态数据结构怎么设计循环体的状态我是用一个字典来维护的因为每个工具都可能往里面写入中间结果。定义一个Python字典包含pdf_bytes原始PDF字节不进入LLM上下文。chunks解析后的文本块列表尽量截断后只放摘要进上下文。index向量索引对象只存在内存里不进上下文。evidence检索到的关键片段汇总会进入LLM上下文。calc_result计算结果最后汇总时用。short_memory保存最近几轮的工具行为和结果摘要。为什么让index和pdf_bytes不进入消息历史因为它们是“二进制/庞大的中间数据”直接塞进messages会迅速撑爆上下文长度。这也是很多新手上来就炸的根因。工具产生的原始大对象应该留在内存或文件系统里循环里只传摘要。3.3 循环逻辑的核心实现下面把关键循环用伪代码写出来重点是判断下一步动作的规则以及如何把中间结果写回状态。def pdf_qa_loop(question: str, pdf_bytes: bytes) - str: state init_state(pdf_bytes) max_iterations 8 consecutive_failures 0 for i in range(max_iterations): # 让模型基于当前状态和任务目标输出下一步动作 action decide_next_action( questionquestion, state_summarybuild_state_summary(state) ) if action parse_pdf: try: chunks pdf_parse(pdf_bytes) state[chunks] chunks state[chunks_summary] summarize_chunks(chunks) consecutive_failures 0 except Exception as e: consecutive_failures 1 bind_error(state, f解析失败{e}) elif action build_index: if chunks not in state: bind_error(state, 请先解析PDF再建索引) continue state[index] build_index(state[chunks]) consecutive_failures 0 elif action search_index: query action_params.get(query) if index not in state: bind_error(state, 索引还没建好请先建索引) continue results search_index(state[index], query, top_k3) state[evidence] results consecutive_failures 0 elif action calculator: expr action_params.get(expression) state[calc_result] calculator(expr) state[messages].append({role: assistant, content: f计算结果{state[calc_result]}}) consecutive_failures 0 elif action answer: return assemble_answer(question, state[evidence], state[calc_result]) else: consecutive_failures 1 bind_error(state, f未知动作{action}) if consecutive_failures 2: return 连续失败超过阈值循环终止请检查工具配置或状态数据 return 达到最大循环次数仍未生成最终答案这里的关键设计是模型的每一步动作都对应一个状态前置条件。比如search_index前必须确保index已经建立。如果模型跳过前置条件循环会返回一条错误信息提醒模型按顺序执行。这比把“先做什么再做什么”全部写在提示词里更可靠因为即使模型判断错了代码也能拦下来。3.4 实测效果与参数调优方向这个项目我跑过几轮最明显的感受是能不能拿到准确结果很大程度取决于模型对“什么时候该停止检索”的判断。一开始我设了max_iterations8但模型经常在检索阶段重复调用search_index同一个关键词查了三四遍。后来我在build_state_summary里加入了“已检索关键词”列表并把摘要给模型重复调用才明显减少。这就是状态管理影响终止条件的一个典型例子。另一个需要调优的是search_index的top_k。top_k太小模型可能找不到关键片段top_k太大塞给模型的evidence又太多。我的经验是核心问题用top_k2~3复杂对比问题用top_k5。如果你用的是向量检索还可以把相关度低于某阈值的片段拼接成一句“未找到高相关片段”让模型要么换关键词要么基于已有内容回答而不是硬编。max_iterations典型表现我的建议3很多任务在建索引或第一次检索后就耗尽答案缺失过于紧张不建议8多数文档类任务可以完成偶有重复检索通用起步值15能应对复杂对比但成本明显上升需配合重复检测复杂任务可用4. 避坑指南我在Loop Engineering里踩过的那些坑4.1 死循环的隐形原因工具返回“成功”但无进展最容易出现的坑不是工具报错而是工具每次都返回成功但返回的内容完全一样。比如模型反复调用search_index用同一个query去找同一个index拿到的结果当然一样模型却觉得“再查一次更稳妥”。这种循环不报错但会白白消耗Token直到撞上max_iterations。我的解决办法是加一个“重复动作检测”每轮把模型要执行的tools名称和工具参数做哈希存进一个列表。如果发现最近连续N轮一般2轮出现了完全相同的动作就直接打断在返回给模型的消息里说“你刚刚执行过相同工具和参数本次不重复执行请基于已有结果继续或换个思路”。这样模型会被逼着换一个动作而不是机械循环。# 重复动作检测逻辑 last_actions [] consecutive_duplicated 0 for step in range(max_iterations): action parse_action(llm_output) action_signature (action[tool_name], json.dumps(action[tool_args], sort_keysTrue)) if last_actions and action_signature last_actions[-1]: consecutive_duplicated 1 if consecutive_duplicated 2: state.messages.append({role: user, content: 警告你连续重复执行同一个工具调用请停止重复基于已有结果继续。}) continue else: consecutive_duplicated 0 last_actions.append(action_signature)4.2 上下文爆炸每一步的Observation全塞给模型第二个高频坑是所有的Observation都原样塞进消息历史。早期我做Agent时tool_results存了很多长文本结果第二轮就把Context Window填满了。尤其是调用PDF解析或者网页抓取工具返回几万字很正常。这时模型基本丧失了对全局任务的理解之后每一轮都像“失忆”一样。我用的是两级策略。第一级叫“大结果留库”工具返回的完整内容只存到state或单独文件里不进messages第二级叫“摘要连接”往messages里塞的是工具返回的摘要比如“共解析到12个章节其中第3章涉及研发费用关键段落提取完毕共5条”。模型看到的是结构化的进展摘要而不是原始内容。这样既保留决策信息又严格控制上下文长度。4.3 重试机制不是所有失败都该重试另一个常见误区是让Agent无脑重试。我的建议是给工具分三类只读类、写操作类、外部依赖类。只读类如查询、检索失败重试风险不大写操作类如发送文件、写入数据库重试前必须考虑是否已经执行成功否则可能造成重复写入外部依赖类如第三方API要根据错误码决定是否重试4xx一般不应该重试5xx可以重试一次。在循环里我的错误处理分支会检查当前工具是否属于“不可安全重试”的类型。如果是在错误信息里明确提示模型“该工具已尝试过一次失败原因xxx。不要重复调用它请换一个方案。”这样既防止重复副作用也给模型提供了决策依据。4.4 常见问题速查表问题表现排查思路与解法死循环同一工具、同一参数反复调用加入动作签名哈希连续相同动作超过N轮强制打断上下文爆炸第2~3轮后模型开始答非所问大结果不进messages只进摘要必要时对旧observation做裁剪工具调用失败后瞎编结果模型在Observation缺失时硬答在错误分支中明确告知失败原因并禁止继续使用该工具过早结束模型还没收集足够证据就输出answer审查状态摘要补充未完成的前置条件信息降低“不知情时结束”的倾向无限等待第三方接口单步工具调用卡住整轮挂死给单步工具调用加timeout超时按失败处理重复执行写入类操作用户收到相同邮件/数据被重复写入对写操作加幂等键重试前检查是否已执行5. 让循环更聪明进阶优化思路5.1 动态注入工具而不是一次性塞全套如果Agent有十几个工具每次循环都把所有工具描述发给模型不仅浪费Token还会让模型选择困难。我建议按状态动态选择工具子集。例如PDF助手在“未解析PDF”阶段只给parse和build_index进入“回答问题”阶段才加入search_index和calculator。实现上就是每个工具声明自己依赖的前置条件循环在每次调用LLM前根据state判断哪些工具“可用”再拼装提示词。这样做的另一个好处是模型不容易调用那些“前置条件未满足”的工具循环的失败率大幅下降。工具数量多了以后这套机制很有必要。5.2 在循环里加入人工审批节点有些动作风险很高比如删除文件、转账、发布内容。如果完全交给循环里的模型决定我是不放心的。Loop Engineering可以引入“人工在环”human-in-the-loop节点当模型请求执行高影响工具时循环不会直接调用而是把待审批信息发送给一个队列等你确认后才继续执行。我的做法比较简单在动作解析后增加一个policy_check函数返回allow、deny或require_approval。如果是require_approval就暂停循环等待外部回调收到审批结果后再继续。这个设计对真实生产系统很重要能挡掉很多“模型好心办坏事”的情况。5.3 可观测性日志和追踪是Loop Engineering的命根子循环不透明是最难调试的地方。我强烈建议从第一版开始就给每次循环加上统一日志。每轮至少记录四个字段round序号、action类型、tool名称、tool_args_hash、result摘要。另外给整个任务分配一个request_id这样上下游日志能串起来。下面是一个我常用的日志格式{ request_id: 1a2b3c4d, round: 3, action: tool, tool_name: search_index, args_preview: [已过滤], result_preview: 匹配到2段第1段得分0.87..., duration_ms: 483, state_summary: index: built, evidence: 3 items }不要小看这些日志。有一次线上Agent在凌晨跑飞我靠的就是request_id串起来的日志发现它在凌晨连续调用了12次同一个查询接口。如果没有日志这种问题根本没法查。5.4 让循环成为可托管的任务而不是同步函数最后一个进阶点不要把Agent Loop写成同步阻塞的python函数。比较典型的做法是接收任务后立刻返回一个task_id把循环提交到任务队列或异步worker里执行进度通过轮询或Webhook回调暴露。这样你在生产环境可以控制并发、做失败重投、也让前端不需要一直挂起等待。我自己在项目里会用一个简单的后台worker池队列里存任务参数worker跑Loop Engine跑完把结果写回存储并回调通知。这种做法对用户来说体验更稳定也不会因为单个任务长时间执行把Web服务彻底卡死。这篇内容没有硬性结论我只说两个个人体会。第一个体会Loop Engineering不要太早追求复杂。我见过很多人一上来就上多智能体、复杂状态机结果连基本循环都不稳定。我的习惯是先写一个最笨的while循环把finished条件、错误分支、日志三件套做扎实再逐步加智能策略。第二个技巧是每次循环开始时在状态里维护一个“已完成事项”清单不要把它揉进完整对话历史。这个清单既能让模型面对长任务时不迷失方向也方便我们人工检查进度。这些年我最大的感受是Agent项目的稳定性很少取决于模型有多聪明更多取决于循环设计有多能兜底。把Loop Engineering这件事做好你写出来的Agent才不是“Demo器”而是真的能放在业务里跑动的自动化系统。
写AI Agent最让人头疼的是什么不是Prompt写不好也不是模型选不对——而是你明明给了它一堆工具和任务它却在执行过程中反复调用同一个工具或者在一个环节里绕来绕去最后既没完成任务也没告诉你发生了什么。我在做自动化代理项目时几乎有一半时间都在和这种“循环”问题搏斗。后来才慢慢意识到Agent能不能稳定产出结果拼的就是循环工程也就是业内常说的Loop Engineering。Loop Engineering不是某一个开源框架而是一套关于如何设计“思考→行动→观察→再思考”循环的工程实践。它的目标很明确让Agent在复杂任务中能自主前进、及时终止、失败自愈。简单说如果Prompt是Agent的大脑工具是Agent的手脚那循环就是让手脚按照大脑指令持续工作、直到把活干完的那套动力系统。这篇文章适合正在做AI Agent开发、RAG应用、自动化流程的朋友尤其是那些写Demo没问题、一上真实场景就卡住的人。我会从循环的本质讲起再给一个最简实现然后带一个PDF多工具分析助手的实战项目最后把我在Loop Engineering里踩过的坑和排查方法整理出来。内容偏向动手建议把代码跑起来看。1. Agent循环的本质为什么说Loop Engineering是Agent的发动机1.1 单次调用与循环调用是两回事很多初学者会以为“调一次LLM就等于一个Agent”。实际不是。单次调用只能做三件事根据上下文生成文本、返回一段JSON、或者调用一次外部接口。它没有记忆不会追问自己“刚才拿到了什么结果、下一步该做什么”。以“查明天北京到深圳的机票并给出差抵用建议”为例。单次调用LLM可能只能生成一段“我建议您去官网查询”的回答。要真正完成这件事Agent得先调用航班查询工具拿到航班列表再调用日历判断明天日期可能还要调用天气工具最后把交通、住宿、行程建议汇总成报告。这个过程中每一轮都是“模型生成一个动作→工具返回结果→模型根据结果生成下一个动作”这就构成了一个循环。这个循环就是Agent Loop。在ReAct范式里它被概括为“Thought思考→ Action行动→ Observation观察”三个步骤的重复。Thought告诉模型现在所处的情况Action决定调用哪个工具或返回最终答案Observation是工具执行后的反馈。模型在循环里不断更新状态直到它能输出一个“结束”信号。1.2 为什么循环这么难设计难的不是写一个for循环而是几个工程问题什么时候循环结束循环里记忆存什么工具报错了怎么办这些问题单靠Prompt是解决不了的需要工程手段。先说终止条件。很多Agent跑飞就是因为循环缺少“刹车”。Agent任务千变万化模型有时候会误判“没完成”于是继续调用工具有时候明明已经拿到答案还要再“确认”一遍。如果你只设一个很大的max_iterations问题不是安全而是成本和耗时失控——一次循环调用一个工具可能只要几秒但累计起来一个任务跑上十几分钟也很常见。我一般在设计Loop Engine时都会要求必须有硬性终止条件且终止后必须返回可读的退出原因。再说状态维护。循环里每轮产生的Observation、工具结果、最终答案都需要有地方放。放全了上下文窗口容易爆放少了模型又缺少决策依据。这就是状态管理问题。最后是错误处理。工具调用失败、超时、返回格式不对都可能在循环中发生。如果代码不做兜底Agent可能带着错误继续跑最后给出一个看着像模像样但其实是瞎编的结论。所以Loop Engineering真正要管理的其实是“终止、状态、错误”这三个核心变量。1.3 Loop Engineering与Agent框架的关系这里需要澄清一点很多框架里已经封装了Agent Loop比如LangChain的AgentExecutor、开源社区的ReAct Agent等。那为什么还要自己理解Loop Engineering因为框架给的是通用默认值做真实项目时远远不够用。你需要知道它内部是怎么决定“调用哪个工具”的才能配置准确的停止条件、做上下文压缩、加错误重试策略。我就是从“直接用框架”到“自己写了一个小规模Loop Engine”之后才算真正入门。这篇文章后面给的实战项目也会用最少的代码来实现一个可控的循环。2. 保姆级实操从零手写一个最小可用的Agent Loop2.1 最小依赖与整体结构先把这个最小系统跑起来再看理论。环境只需要Python 3.10装一个能调LLM的客户端库再加几个函数就足够。代码结构分成四块LLM调用负责把messages发送给模型拿到文本或JSON返回。动作解析从模型输出里解析出到底是调用工具还是直接给最终答案。工具执行注册一个工具字典按名字去执行对应的Python函数。循环主体按顺序调度上述逻辑并维护循环状态。为什么不用现成框架因为最小实现能让你直观看到每个环节后面做定制时才不会两眼一抹黑。等理解了再回到框架里你会知道该覆盖哪些方法。下面是一个最简的循环主体。为了通用性我把LLM调用封装成一个mock函数实际使用时换成你自己用的SDK即可。import time from dataclasses import dataclass, field from typing import List, Dict, Callable dataclass class LoopState: # 保存整个对话历史含工具返回结果 messages: List[Dict] field(default_factorylist) # 单独记录每步的工具调用结果方便追踪 tool_results: List[Dict] field(default_factorylist) # 当前迭代序号 current_step: int 0 # 是否已正常结束 finished: bool False def llm_complete(messages: List[Dict]) - str: # 实际项目请替换为你自己的LLM客户端 # 这里的自定义函数需要能理解 messages并返回诸如 # ACTION: finish\nANSWER: 完成 # 或 # ACTION: tool\nTOOL_NAME: get_weather\nARGS: {city: 北京} raise NotImplementedError def parse_action(text: str) - Dict: # 对模型返回文本做简单解析 lines text.strip().split(\n) action_type tool_name tool_args {} answer for line in lines: if line.startswith(ACTION:): action_type line.split(:, 1)[1].strip() elif line.startswith(TOOL_NAME:): tool_name line.split(:, 1)[1].strip() elif line.startswith(ARGS:): import json tool_args json.loads(line.split(:, 1)[1]) elif line.startswith(ANSWER:): answer line.split(:, 1)[1].strip() return { type: action_type, tool_name: tool_name, tool_args: tool_args, answer: answer, } def execute_tool(tool_name: str, tool_args: Dict, tools: Dict[str, Callable]): return tools[tool_name](**tool_args) def run_agent(user_task: str, tools: Dict[str, Callable], max_iterations: int 8) - str: state LoopState() state.messages.append({role: user, content: user_task}) for _ in range(max_iterations): state.current_step 1 # 1. 调用LLM根据当前消息列表决定下一步 response llm_complete(state.messages) action parse_action(response) # 2. 判断是否完成 if action[type] finish: state.finished True return action[answer] # 3. 执行工具调用 if action[type] tool: if action[tool_name] not in tools: error_msg f工具 {action[tool_name]} 不存在 state.messages.append({role: assistant, content: response}) state.messages.append({role: user, content: f错误{error_msg}}) continue try: result execute_tool(action[tool_name], action[tool_args], tools) except Exception as e: result f工具执行异常{e} # 记录结果 state.tool_results.append({ tool_name: action[tool_name], args: action[tool_args], result: result, }) # 把结果写回消息列表 state.messages.append({role: assistant, content: response}) state.messages.append({role: user, content: f工具返回{result}}) # 如果这里还可以加入超时检测、重复动作检测等 return 达到最大循环次数任务未完成2.2 设计循环时必须想清楚的三个参数上面代码里被我偷懒跳过的几个点恰恰是Loop Engineering的关键。第一个是max_iterations。它不是随便设的。任务类型不同迭代上限差异很大。比如“查天气并整理穿衣建议”通常3次以内就够但“阅读一份多章节PDF生成财务分析报告”8次可能都太紧。我的经验是先按最坏情况估算一个上界再加2次余量。宁可让它在未完成任务时终止也不要让它在死循环里烧钱。第二个是超时设置。单次工具调用可能卡住整个循环也可能卡住。我在核心循环外层会加一个wall_clock阈值超过就终止。有的框架叫timeout有的叫deadline。给一个建议单步工具调用timeout设为30秒整轮任务timeout按max_iterations乘以单步耗时再乘1.5。第三个是错误容忍度。并不是所有工具执行失败都需要立即终止。比如天气接口偶尔超时重试一次可能就好了但你也不希望模型连续失败10次还在试。我会单独维护一个consecutive_failures变量连续失败超过2次就走“失败恢复”分支提醒模型更换工具或者直接结束。2.3 怎么让模型更稳定地返回“结束信号”这个点容易被忽略。模型在循环里输出“finish”之前往往需要你给一个明确的输出格式约定。最简单的做法是在系统提示词里写明如果已经完成任务必须输出“ACTION: finish”和“ANSWER: 最终答案”如果还需要调用工具就输出“ACTION: tool”和工具名、参数。我在实际项目里还会让模型先输出一小段“当前进展”再输出下一步动作。有了进展描述调试循环时能知道每一步模型在想什么。需要注意的是不要把所有工具的说明都堆在系统提示词里。工具多了以后模型容易把工具描述当作一般文本导致解析出错。更好的做法是动态拼接工具列表只把当前阶段可能用到的工具注入给模型后面实战部分会讲。3. 项目实战用Loop Engineering做一个多工具PDF分析助手3.1 项目需求与工具链路设计我选了一个很常见的场景给Agent一份PDF比如季度财报让它回答“这个季度研发费用是多少和上季度比有什么变化”这种问题。这类任务单次LLM调用做不了因为模型看不到PDF内容。它至少得经历“解析PDF→切块索引→检索相关段落→可能还要计算同比增长→最后汇总答案”的过程。为了演示循环我给这个PDF助手设计了四个工具pdf_parse解析PDF文件内容返回文本块列表。build_index把文本块做向量化索引。这里可以直接用faiss或者简单的内存字典只要能把“检索”动作体现出来即可。search_index用关键词或向量检索返回最相关的几个片段。calculator执行简单的数学表达式用于对比增长率。工具链路不是一开始就全部注入给模型。Agent会根据当前状态选择下一步如果还没解析PDF就先调pdf_parse如果还没有索引就调build_index如果用户的问题涉及数字计算再调calculator。这样设计的好处是避免模型在第一步就乱调用也减少上下文噪音。3.2 状态数据结构怎么设计循环体的状态我是用一个字典来维护的因为每个工具都可能往里面写入中间结果。定义一个Python字典包含pdf_bytes原始PDF字节不进入LLM上下文。chunks解析后的文本块列表尽量截断后只放摘要进上下文。index向量索引对象只存在内存里不进上下文。evidence检索到的关键片段汇总会进入LLM上下文。calc_result计算结果最后汇总时用。short_memory保存最近几轮的工具行为和结果摘要。为什么让index和pdf_bytes不进入消息历史因为它们是“二进制/庞大的中间数据”直接塞进messages会迅速撑爆上下文长度。这也是很多新手上来就炸的根因。工具产生的原始大对象应该留在内存或文件系统里循环里只传摘要。3.3 循环逻辑的核心实现下面把关键循环用伪代码写出来重点是判断下一步动作的规则以及如何把中间结果写回状态。def pdf_qa_loop(question: str, pdf_bytes: bytes) - str: state init_state(pdf_bytes) max_iterations 8 consecutive_failures 0 for i in range(max_iterations): # 让模型基于当前状态和任务目标输出下一步动作 action decide_next_action( questionquestion, state_summarybuild_state_summary(state) ) if action parse_pdf: try: chunks pdf_parse(pdf_bytes) state[chunks] chunks state[chunks_summary] summarize_chunks(chunks) consecutive_failures 0 except Exception as e: consecutive_failures 1 bind_error(state, f解析失败{e}) elif action build_index: if chunks not in state: bind_error(state, 请先解析PDF再建索引) continue state[index] build_index(state[chunks]) consecutive_failures 0 elif action search_index: query action_params.get(query) if index not in state: bind_error(state, 索引还没建好请先建索引) continue results search_index(state[index], query, top_k3) state[evidence] results consecutive_failures 0 elif action calculator: expr action_params.get(expression) state[calc_result] calculator(expr) state[messages].append({role: assistant, content: f计算结果{state[calc_result]}}) consecutive_failures 0 elif action answer: return assemble_answer(question, state[evidence], state[calc_result]) else: consecutive_failures 1 bind_error(state, f未知动作{action}) if consecutive_failures 2: return 连续失败超过阈值循环终止请检查工具配置或状态数据 return 达到最大循环次数仍未生成最终答案这里的关键设计是模型的每一步动作都对应一个状态前置条件。比如search_index前必须确保index已经建立。如果模型跳过前置条件循环会返回一条错误信息提醒模型按顺序执行。这比把“先做什么再做什么”全部写在提示词里更可靠因为即使模型判断错了代码也能拦下来。3.4 实测效果与参数调优方向这个项目我跑过几轮最明显的感受是能不能拿到准确结果很大程度取决于模型对“什么时候该停止检索”的判断。一开始我设了max_iterations8但模型经常在检索阶段重复调用search_index同一个关键词查了三四遍。后来我在build_state_summary里加入了“已检索关键词”列表并把摘要给模型重复调用才明显减少。这就是状态管理影响终止条件的一个典型例子。另一个需要调优的是search_index的top_k。top_k太小模型可能找不到关键片段top_k太大塞给模型的evidence又太多。我的经验是核心问题用top_k2~3复杂对比问题用top_k5。如果你用的是向量检索还可以把相关度低于某阈值的片段拼接成一句“未找到高相关片段”让模型要么换关键词要么基于已有内容回答而不是硬编。max_iterations典型表现我的建议3很多任务在建索引或第一次检索后就耗尽答案缺失过于紧张不建议8多数文档类任务可以完成偶有重复检索通用起步值15能应对复杂对比但成本明显上升需配合重复检测复杂任务可用4. 避坑指南我在Loop Engineering里踩过的那些坑4.1 死循环的隐形原因工具返回“成功”但无进展最容易出现的坑不是工具报错而是工具每次都返回成功但返回的内容完全一样。比如模型反复调用search_index用同一个query去找同一个index拿到的结果当然一样模型却觉得“再查一次更稳妥”。这种循环不报错但会白白消耗Token直到撞上max_iterations。我的解决办法是加一个“重复动作检测”每轮把模型要执行的tools名称和工具参数做哈希存进一个列表。如果发现最近连续N轮一般2轮出现了完全相同的动作就直接打断在返回给模型的消息里说“你刚刚执行过相同工具和参数本次不重复执行请基于已有结果继续或换个思路”。这样模型会被逼着换一个动作而不是机械循环。# 重复动作检测逻辑 last_actions [] consecutive_duplicated 0 for step in range(max_iterations): action parse_action(llm_output) action_signature (action[tool_name], json.dumps(action[tool_args], sort_keysTrue)) if last_actions and action_signature last_actions[-1]: consecutive_duplicated 1 if consecutive_duplicated 2: state.messages.append({role: user, content: 警告你连续重复执行同一个工具调用请停止重复基于已有结果继续。}) continue else: consecutive_duplicated 0 last_actions.append(action_signature)4.2 上下文爆炸每一步的Observation全塞给模型第二个高频坑是所有的Observation都原样塞进消息历史。早期我做Agent时tool_results存了很多长文本结果第二轮就把Context Window填满了。尤其是调用PDF解析或者网页抓取工具返回几万字很正常。这时模型基本丧失了对全局任务的理解之后每一轮都像“失忆”一样。我用的是两级策略。第一级叫“大结果留库”工具返回的完整内容只存到state或单独文件里不进messages第二级叫“摘要连接”往messages里塞的是工具返回的摘要比如“共解析到12个章节其中第3章涉及研发费用关键段落提取完毕共5条”。模型看到的是结构化的进展摘要而不是原始内容。这样既保留决策信息又严格控制上下文长度。4.3 重试机制不是所有失败都该重试另一个常见误区是让Agent无脑重试。我的建议是给工具分三类只读类、写操作类、外部依赖类。只读类如查询、检索失败重试风险不大写操作类如发送文件、写入数据库重试前必须考虑是否已经执行成功否则可能造成重复写入外部依赖类如第三方API要根据错误码决定是否重试4xx一般不应该重试5xx可以重试一次。在循环里我的错误处理分支会检查当前工具是否属于“不可安全重试”的类型。如果是在错误信息里明确提示模型“该工具已尝试过一次失败原因xxx。不要重复调用它请换一个方案。”这样既防止重复副作用也给模型提供了决策依据。4.4 常见问题速查表问题表现排查思路与解法死循环同一工具、同一参数反复调用加入动作签名哈希连续相同动作超过N轮强制打断上下文爆炸第2~3轮后模型开始答非所问大结果不进messages只进摘要必要时对旧observation做裁剪工具调用失败后瞎编结果模型在Observation缺失时硬答在错误分支中明确告知失败原因并禁止继续使用该工具过早结束模型还没收集足够证据就输出answer审查状态摘要补充未完成的前置条件信息降低“不知情时结束”的倾向无限等待第三方接口单步工具调用卡住整轮挂死给单步工具调用加timeout超时按失败处理重复执行写入类操作用户收到相同邮件/数据被重复写入对写操作加幂等键重试前检查是否已执行5. 让循环更聪明进阶优化思路5.1 动态注入工具而不是一次性塞全套如果Agent有十几个工具每次循环都把所有工具描述发给模型不仅浪费Token还会让模型选择困难。我建议按状态动态选择工具子集。例如PDF助手在“未解析PDF”阶段只给parse和build_index进入“回答问题”阶段才加入search_index和calculator。实现上就是每个工具声明自己依赖的前置条件循环在每次调用LLM前根据state判断哪些工具“可用”再拼装提示词。这样做的另一个好处是模型不容易调用那些“前置条件未满足”的工具循环的失败率大幅下降。工具数量多了以后这套机制很有必要。5.2 在循环里加入人工审批节点有些动作风险很高比如删除文件、转账、发布内容。如果完全交给循环里的模型决定我是不放心的。Loop Engineering可以引入“人工在环”human-in-the-loop节点当模型请求执行高影响工具时循环不会直接调用而是把待审批信息发送给一个队列等你确认后才继续执行。我的做法比较简单在动作解析后增加一个policy_check函数返回allow、deny或require_approval。如果是require_approval就暂停循环等待外部回调收到审批结果后再继续。这个设计对真实生产系统很重要能挡掉很多“模型好心办坏事”的情况。5.3 可观测性日志和追踪是Loop Engineering的命根子循环不透明是最难调试的地方。我强烈建议从第一版开始就给每次循环加上统一日志。每轮至少记录四个字段round序号、action类型、tool名称、tool_args_hash、result摘要。另外给整个任务分配一个request_id这样上下游日志能串起来。下面是一个我常用的日志格式{ request_id: 1a2b3c4d, round: 3, action: tool, tool_name: search_index, args_preview: [已过滤], result_preview: 匹配到2段第1段得分0.87..., duration_ms: 483, state_summary: index: built, evidence: 3 items }不要小看这些日志。有一次线上Agent在凌晨跑飞我靠的就是request_id串起来的日志发现它在凌晨连续调用了12次同一个查询接口。如果没有日志这种问题根本没法查。5.4 让循环成为可托管的任务而不是同步函数最后一个进阶点不要把Agent Loop写成同步阻塞的python函数。比较典型的做法是接收任务后立刻返回一个task_id把循环提交到任务队列或异步worker里执行进度通过轮询或Webhook回调暴露。这样你在生产环境可以控制并发、做失败重投、也让前端不需要一直挂起等待。我自己在项目里会用一个简单的后台worker池队列里存任务参数worker跑Loop Engine跑完把结果写回存储并回调通知。这种做法对用户来说体验更稳定也不会因为单个任务长时间执行把Web服务彻底卡死。这篇内容没有硬性结论我只说两个个人体会。第一个体会Loop Engineering不要太早追求复杂。我见过很多人一上来就上多智能体、复杂状态机结果连基本循环都不稳定。我的习惯是先写一个最笨的while循环把finished条件、错误分支、日志三件套做扎实再逐步加智能策略。第二个技巧是每次循环开始时在状态里维护一个“已完成事项”清单不要把它揉进完整对话历史。这个清单既能让模型面对长任务时不迷失方向也方便我们人工检查进度。这些年我最大的感受是Agent项目的稳定性很少取决于模型有多聪明更多取决于循环设计有多能兜底。把Loop Engineering这件事做好你写出来的Agent才不是“Demo器”而是真的能放在业务里跑动的自动化系统。
