写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 Loop
2.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_factory=list) # 单独记录每步的工具调用结果，方便追踪 tool_results: List[Dict] = field(default_factory=list) # 当前迭代序号 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( question=question, state_summary=build_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_k=3) 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_iterations=8，但模型经常在检索阶段重复调用search_index，同一个关键词查了三四遍。后来我在build_state_summary里加入了“已检索关键词”列表，并把摘要给模型，重复调用才明显减少。这就是状态管理影响终止条件的一个典型例子。
另一个需要调优的是search_index的top_k。top_k太小，模型可能找不到关键片段；top_k太大，塞给模型的evidence又太多。我的经验是核心问题用top_k=2~3，复杂对比问题用top_k=5。如果你用的是向量检索，还可以把相关度低于某阈值的片段拼接成一句“未找到高相关片段”，让模型要么换关键词，要么基于已有内容回答，而不是硬编。
| 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_keys=True)) 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器”，而是真的能放在业务里跑动的自动化系统。
