很多人刚开始接触 Agent 开发时都以为最难的是让大模型“听懂人话”。等真正动手做一个能跑多轮任务的应用才发现卡住自己的根本不是 Prompt而是那个看起来非常不起眼的“循环”。同样的一个语义理解任务单次调用大模型就能完成可一旦你希望它自主规划、分步执行、根据中间结果调整策略问题就全来了循环多久结束中间结果存在哪里某一步出错了要不要重试重试几次整个循环的 Token 成本怎么控制如果大模型陷入死循环怎么强制终止这些问题不属于 Prompt 工程也不属于模型微调而是属于一套正在快速成型的工程体系——Loop Engineering。这篇文章会从底层机制讲起拆解循环设计的核心原理再用一个可运行的完整代码案例把工程落地中的六个关键难点和解决方案串联起来。如果你正准备把 Agent 从“Demo 阶段”推向“生产阶段”这篇文章值得读完。1. 这篇文章真正要解决的问题Loop Engineering 并不是一个写死循环的编程技巧而是围绕“大模型多次调用构成的执行循环”进行设计、控制、治理的一套方法论。先看一个真实场景。假设你要做一个自动写周报的 Agent流程并不复杂读取本周提交记录 → 总结重要变更 → 生成周报草稿 → 按团队模板格式化。用单次 Prompt 也能做但效果很一般因为提交记录可能是几十条零散信息大模型很难一次就完成“清洗、归纳、排序、改写”四个动作。更合理的方案是拆成多步每一步都有独立的 Prompt 和输出校验。于是代码就变成了下面这样while True: result call_llm(current_task, context) # 怎么判断 result 是否合格 # 不合格是重试还是终止 # 重试次数上限是多少 if is_done(result): break问题就从“怎么写 Prompt”变成了“怎么设计这个循环”。这个转变就是 Loop Engineering 出现的根本原因。本文要解决的四个核心问题循环的底层执行模型是什么为什么大模型循环天然存在失控风险。一个可落地的循环引擎应该包含哪些基础组件。循环的终止条件、错误处理、成本控制、可观测性怎么设计。从开发到生产哪些地方最容易被忽略。适合阅读本文的读者有三类正在用大模型 API 开发 Agent 应用的工程师想从“单次 Prompt 调用”升级到“多步任务编排”的开发者以及团队里负责 AI 应用架构设计的技术负责人。2. Loop Engineering 的核心概念与底层原理2.1 什么是 Loop EngineeringLoop Engineering 不是某个框架或者某个开源项目的名字而是一类工程实践的总称。它研究的核心对象是大模型驱动的循环执行过程。一个典型的 Loop 由四个基本部分组成。组成部分作用类比任务队列存放待完成的子任务决定下一轮做什么工厂里的待加工订单执行器调用大模型或工具完成当前子任务工位上的操作员结果评估器判断执行结果是否符合要求质检员控制策略决定继续循环、重试、终止或降级车间调度员四个部分缺一不可。很多 Demo 只实现了“执行器 简单终止条件”结果就是任务稍微复杂一点循环就失去控制。2.2 循环的底层机制从状态机视角理解理解 Loop Engineering 有一个很好的视角是有限状态机。一个 Agent 的循环过程本质上是在这几个状态之间转换READY等待执行下一个子任务。RUNNING正在调用大模型或工具。SUCCESS当前子任务执行成功结果进入下一环节。RETRY执行结果不合格准备重试。FAILED执行失败进入异常处理。TERMINATED整个循环结束。关键是状态之间的转移不是无条件进行的。每一次转移都需要一个明确的触发条件。比如从 RETRY 到 RUNNING必须满足“重试次数小于最大重试数”从 RUNNING 到 SUCCESS必须满足“结果校验通过”。如果这些条件没有设计清楚循环就等于没有刹车。这也是 loop engineering 底层原理中最核心的一点循环不是让大模型自由发挥而是用状态机约束它的行动边界。2.3 为什么大模型循环天然容易失控大模型的输出存在三个固有特性导致它驱动的循环很难稳定第一不确定性。同样的 Prompt两次调用结果可能不同。第一次校验通过第二次可能就失败了。这让循环的终止条件很难写。第二退化问题。有研究观察到大模型在长时间推理或多次迭代后可能出现性能下降生成内容质量反而变差。这种现象常被称为“退化”或“重复循环”。工程上的表现就是模型在某个错误结论上反复打转。第三上下文膨胀。每循环一次对话历史就变长一点。Token 成本上升响应延迟变高而且模型容易在长上下文中丢失早期信息。所以Loop Engineering 的本质工作就是通过工程手段对冲这三个固有风险让循环变得可控、可观测、可终止。3. 环境准备与最小工程骨架3.1 技术选型本文的完整示例使用 Python 实现因为 Python 生态对大模型应用支持最成熟而且代码直观、适合教学。环境依赖只有三个部分Python 3.9 及以上版本具体版本以你的系统环境为准。大模型 API 调用库示例中使用通用的 HTTP 调用方式方便替换为任意兼容 API。一个可用的模型服务可以在代码里配置 API Key 和 Base URL。3.2 项目结构建议按下面的结构组织工程目录loop-engineering-demo/ ├── engine/ │ ├── __init__.py │ ├── loop_engine.py # 循环引擎核心 │ ├── state.py # 状态定义 │ └── evaluator.py # 结果评估器 ├── tasks/ │ ├── __init__.py │ └── weekly_report.py # 周报示例任务 ├── config.py # 全局配置 └── main.py # 入口脚本这个结构本身就是一个最小可扩展的工程骨架。后面加新任务只需要在tasks/目录下新增一个文件加新的评估逻辑只需要扩展evaluator.py。4. 核心流程拆解从一个任务的生命周期说起在写代码之前先把一个任务在 Loop Engine 中的完整生命周期理清。这比直接看代码重要得多。4.1 第一步任务初始化任务进入引擎时不是直接丢给大模型而是先被包装成一个任务对象。任务对象包含原始目标、子任务列表、最大循环次数、超时时间、上下文容器、结果存储。这样设计的目的是让引擎在执行过程中拥有完整的“控制面板”而不是裸调 API。4.2 第二步子任务执行引擎从任务队列中取出当前子任务组装 Prompt调用大模型。这一步的工程细节在于 Prompt 组装需要把系统指令、用户目标、历史上下文、工具返回结果拼接成一次完整的调用。拼接顺序会影响模型对任务的理解建议把目标信息放在离当前指令最近的位置。4.3 第三步结果评估拿到模型输出后不能直接进入下一轮。必须经过评估器校验。评估器有三层策略基础格式校验输出是否能正确解析为 JSON是否包含必要字段语义规则校验是否满足业务约束比如金额不能为负数、日期格式必须正确。模型自评校验在某些场景下可以让模型对自己的输出打分但要注意这只能作为辅助手段不能作为唯一依据。4.4 第四步状态转移决策根据评估结果引擎决定下一步动作评估通过 → 结果写入上下文 → 进入下一个子任务。评估不通过且重试次数未超限 → 回到 RUNNING带上错误信息重新调用。评估不通过且重试次数超限 → 进入 FAILED根据配置决定是整体终止还是降级处理。循环次数达到上限 → 无论当前状态如何强制 TERMINATED。4.5 第五步收尾与返回循环终止后引擎要把最终结果、执行轨迹、Token 消耗、每步状态变迁记录一并返回。这些数据不仅用于展示更是后续优化循环策略的依据。5. 完整示例手写一个可落地的 Loop Engine5.1 定义任务状态# engine/state.py from enum import Enum class TaskState(str, Enum): READY ready RUNNING running SUCCESS success RETRY retry FAILED failed TERMINATED terminated class LoopStatus(str, Enum): COMPLETED completed FAILED failed TIMEOUT timeout MAX_ITERATIONS max_iterations这里使用str枚举有两个好处序列化方便可以直接写入日志或数据库兼容性强前后端传值时不需要额外转换。5.2 实现结果评估器# engine/evaluator.py import json from typing import Any, Dict class BaseEvaluator: 评估器基类负责校验模型输出是否满足要求。 def validate(self, output: str) - Dict[str, Any]: 返回格式{passed: bool, reason: str, parsed: Any} raise NotImplementedError class JSONEvaluator(BaseEvaluator): 校验模型输出是否为合法 JSON同时做基础字段检查。 def __init__(self, required_fieldsNone): self.required_fields required_fields or [] def validate(self, output: str) - Dict[str, Any]: try: parsed json.loads(output) except json.JSONDecodeError as e: return { passed: False, reason: fJSON 解析失败: {e}, parsed: None, } missing [field for field in self.required_fields if field not in parsed] if missing: return { passed: False, reason: f缺少必要字段: {missing}, parsed: parsed, } return { passed: True, reason: 校验通过, parsed: parsed, }评估器是一个可插拔设计。不同的任务可以传入不同的评估器。比如周报任务需要检查“本周进展”字段代码审查任务需要检查“风险等级”字段这些都是评估器的配置项而不是写死在引擎里。5.3 实现循环引擎核心# engine/loop_engine.py import time from typing import Any, Callable, Dict, List, Optional from .evaluator import BaseEvaluator from .state import LoopStatus, TaskState class LoopEngine: 一个最小可用的 Loop 引擎。 设计原则 1. 所有运行参数最大循环次数、超时时间、重试次数都通过配置注入。 2. 状态变化通过回调函数暴露给外部方便日志和监控。 3. 引擎本身不关心业务细节只负责调度和控制。 def __init__( self, llm_call: Callable[[str], str], evaluator: BaseEvaluator, max_iterations: int 10, max_retries: int 2, timeout_seconds: float 60.0, on_state_change: Optional[Callable[[str, Any], None]] None, ): self.llm_call llm_call self.evaluator evaluator self.max_iterations max_iterations self.max_retries max_retries self.timeout_seconds timeout_seconds self.on_state_change on_state_change def _notify(self, event: str, data: Any): if self.on_state_change: self.on_state_change(event, data) def run(self, task: str, system_prompt: str, context: Optional[List[Dict]] None): start_time time.time() retries 0 history list(context or []) self._notify(TaskState.READY, {task: task}) for iteration in range(1, self.max_iterations 1): # 检查整体超时 if time.time() - start_time self.timeout_seconds: self._notify(TaskState.TERMINATED, {reason: timeout}) return { status: LoopStatus.TIMEOUT, iterations: iteration - 1, result: None, history: history, } self._notify(TaskState.RUNNING, {iteration: iteration}) # 组装 Prompt把系统指令 用户任务 历史上下文拼成一次调用 messages [{role: system, content: system_prompt}] messages.extend(history) messages.append({role: user, content: task}) raw_output self.llm_call(messages) # 执行结果评估 eval_result self.evaluator.validate(raw_output) if eval_result[passed]: parsed_output eval_result[parsed] history.append({role: assistant, content: raw_output}) self._notify(TaskState.SUCCESS, { iteration: iteration, output: parsed_output, }) return { status: LoopStatus.COMPLETED, iterations: iteration, result: parsed_output, history: history, } # 评估未通过进入重试逻辑 retries 1 if retries self.max_retries: self._notify(TaskState.FAILED, { iteration: iteration, reason: eval_result[reason], }) return { status: LoopStatus.FAILED, iterations: iteration, result: None, history: history, error: eval_result[reason], } self._notify(TaskState.RETRY, { iteration: iteration, reason: eval_result[reason], }) # 把错误信息放进上下文中让模型下一轮知道哪里有问题 history.append({role: assistant, content: raw_output}) history.append({ role: user, content: f上一步输出不符合要求{eval_result[reason]}。请修正后重新输出。 }) # 达到最大迭代次数 self._notify(TaskState.TERMINATED, {reason: max_iterations}) return { status: LoopStatus.MAX_ITERATIONS, iterations: self.max_iterations, result: None, history: history, }这个引擎的核心设计有四点配置注入。max_iterations、max_retries、timeout_seconds全部通过构造参数传入而不是写死在类内部。这样同一个引擎可以服务不同任务高风险任务可以降低最大重试次数成本敏感的任务可以缩短超时时间。状态回调。on_state_change回调让引擎可以无缝接入日志系统和监控平台。在生产环境中每一次状态变化都应该有记录这是事后排查问题的基础。错误反馈进上下文。评估失败之后不是简单重试而是把失败原因拼进历史记录里告诉模型。这一条非常关键。如果不告诉模型为什么失败重试两三次往往还是同样的错误。循环边界明确。请求级超时和整体超时同时生效避免模型单次调用卡死或整个任务长时间无法结束。5.4 实现一个具体的周报任务# tasks/weekly_report.py import json def build_weekly_report_task(commits): 把原始提交记录封装成任务描述。 commit_text \n.join([f- {c} for c in commits]) task f 根据以下本周提交记录生成周报草稿。 要求 1. 将提交记录按功能模块归类。 2. 每个模块总结一句话进展。 3. 输出 JSON格式如下 {{ summary: 总体进展一句话, modules: [模块名1: 进展描述, 模块名2: 进展描述] }} 提交记录 {commit_text} return task # 示例提交数据 SAMPLE_COMMITS [ 优化用户登录接口增加验证码校验, 修复订单超时未关闭的问题, 新增商品搜索的拼音匹配, 重构订单状态机支持取消和退款流转, 修复支付回调幂等性问题, 优化首页加载速度图片改为懒加载, ]这个任务故意设计成需要“归类 → 总结 → 格式化”三个步骤比较适合演示循环的价值。单次调用也能生成结果但经过循环校验后输出的结构化程度会高很多。5.5 编写入口文件# main.py import json from engine.evaluator import JSONEvaluator from engine.loop_engine import LoopEngine from tasks.weekly_report import SAMPLE_COMMITS, build_weekly_report_task # 模拟大模型调用的函数 # 实际项目中替换为真实的 API 调用即可 def fake_llm_call(messages): 一个用于本机演示的模拟调用。 真实项目中应该调用你的模型服务。这里为了演示循环控制逻辑 故意让前两次输出不是合法 JSON展示重试机制。 global _call_count _call_count 1 last_user_message messages[-1][content] if _call_count 1: return 这个周报不太好写我先看看提交记录。 if _call_count 2: return 我认为这周主要工作内容是登录优化。没有输出 JSON。 # 第三次开始输出正确结果 return json.dumps({ summary: 本周完成了登录安全加固、订单流程修正和性能优化三项重点工作, modules: [ 用户模块增加验证码校验提升登录安全性, 订单模块修复超时未关闭问题重构状态机支持取消与退款, 搜索模块新增拼音匹配能力, 性能优化首页图片改为懒加载提升首屏速度, ], }, ensure_asciiFalse) _call_count 0 class DemoLogger: 演示用的日志回调生产环境请接入真实日志系统。 def __call__(self, event, data): print(f[状态变化] {event}) if iteration in data: print(f 迭代轮次: {data[iteration]}) if reason in data: print(f 原因: {data[reason]}) def main(): engine LoopEngine( llm_callfake_llm_call, evaluatorJSONEvaluator(required_fields[summary, modules]), max_iterations5, max_retries3, timeout_seconds120.0, on_state_changeDemoLogger(), ) task build_weekly_report_task(SAMPLE_COMMITS) result engine.run( tasktask, system_prompt你是一个严谨的技术周报生成助手。你必须严格按照要求输出 JSON。, ) print(\n 最终结果 ) print(json.dumps(result, ensure_asciiFalse, indent2)) if __name__ __main__: main()5.6 运行与验证在项目根目录执行python main.py预期输出大致如下[状态变化] ready [状态变化] running 迭代轮次: 1 [状态变化] retry 迭代轮次: 1 原因: JSON 解析失败 [状态变化] running 迭代轮次: 2 [状态变化] retry 迭代轮次: 2 原因: JSON 解析失败 [状态变化] running 迭代轮次: 3 [状态变化] success 迭代轮次: 3 最终结果 在final_result中status为completediterations为 3result中包含了summary和modules两个字段。如果运行失败优先检查三处Python 版本是否满足要求json库是否被项目中的某个文件覆盖命名fake_llm_call中的全局变量_call_count是否被重新初始化。6. 工程落地难点拆解Demo 跑通只是第一步。把 Loop Engine 放到生产环境会遇到一系列在玩具项目里完全暴露不出来的问题。下面按影响程度排序拆解。6.1 难点一循环收敛性无法保证这是 Loop Engineering 最底层的难题。大模型的输出有随机性你无法从数学上证明一个循环一定会收敛到正确结果。实践中可能遇到的场景是模型在第三步反复犯同一个错误重试三次仍然一样评估器始终不通过。应对策略有三个层次第一层是重试策略优化。简单重试往往无效更有效的是“失败原因引导重试”。这就是前面代码里把eval_result[reason]拼进上下文的原因。让模型知道具体错在哪里重试成功率会显著提升。第二层是任务拆分。如果某个子任务反复失败可以考虑把它拆成更小的子任务。例如“生成周报”这一步失败率高可以拆成“归类提交记录”和“生成周报草稿”两步。第三层是降级预案。达到最大重试次数后是直接失败还是返回一个“不完美但可用”的结果生产环境通常选择后者。对用户来说能拿到一个带警告的中间结果往往比什么都拿不到要好。6.2 难点二上下文膨胀每循环一次历史记录就增加一条。循环 10 次之后Prompt 中的历史可能已经吃掉几千个 Token。这不仅增加成本还会稀释模型对原始任务的注意力。常用方案是上下文裁剪和记忆压缩。上下文裁剪限制参与拼接的历史条数。轮次太多时只保留最近 N 条对话加上最开始的系统指令和用户原始任务。记忆压缩定期让模型把历史对话总结成一段简短的记忆摘要之后用摘要代替原始历史。这个方案效果更好但会增加一次模型调用需要在设计时评估成本。6.3 难点三成本不可控循环次数越多成本越高。生产环境必须设置成本上限。成本控制有两个维度一是轮次维度。max_iterations就是成本上限。设计任务时先估算“正常需要几轮”再留出 1.5 到 2 倍余量。二是Token 维度。每次调用大模型前可以估算输入 Token 数。如果发现单次调用已经接近模型上下文上限说明历史积累过多需要先裁剪再做下一次调用。6.4 难点四状态可观测性差Loop 是一个多步骤过程一旦中间某步出错排查难度比单次调用大得多。生产环境至少要能看到以下信息每次循环的输入 Prompt 完整内容。模型原始输出。评估器的判定结果和原因。每次状态转换的时间戳。累计 Token 消耗。这意味着on_state_change回调在生产环境中必须接上结构化日志系统而不能像 Demo 里一样只是 print 到控制台。日志格式最好采用 JSON方便接入 Elasticsearch、Loki 等日志平台。6.5 难点五幂等性在循环执行过程中如果服务重启或者网络闪断已经完成的子任务怎么办重跑一遍可能产生重复结果。解决思路是给每一步执行加一个全局唯一的trace_id和step_id。执行结果落库时以step_id为幂等键重复执行时直接复用已有结果。这个做法和普通后端接口的幂等设计完全一致只不过作用在循环的每一步上。6.6 难点六安全边界循环给了大模型更多轮次的操作机会也意味着给了它更多犯错或越权的空间。如果循环中包含工具调用、数据库访问、文件写入等操作必须设置明确的安全边界。核心原则是每一步工具调用都做权限校验不要因为“上一步通过了”就默认“这一步也安全”。涉及删除、更新、写入的操作必须在循环外先经过人工审批。落到生产环境的代码循环中每步操作都要有审计日志。7. 常见问题与排查方法问题现象可能原因排查方式解决方案循环不终止终止条件设计不完整模型总是输出不满足要求的内容查看每次评估器的失败原因连续失败是否集中在同一类错误调整评估规则或拆分任务降低单步复杂度重试多次但结果不变模型没有获得失败反馈上下文里只有原始任务检查重试分支是否把失败原因写入了上下文在重试消息中明确告知模型错误点Token 消耗增长过快历史记录无限制累积统计每个迭代后的输入 Token 数增加上下文裁剪或记忆压缩某一步依赖上一步的中间结果但拿不到状态在循环中没有正确传递检查 history 列表是否完整保存了每轮输出确保每轮结果都追加到上下文中生产环境循环日志混乱没有统一 trace_id查看日志系统中的请求关联字段为每个任务生成唯一 trace_id并在所有日志中携带超时时间设了但没生效超时只包住单次调用没有包住整个循环检查超时统计的起止时间点在循环开始前记录 start_time每轮迭代前检查总耗时8. 最佳实践与工程建议下面这些建议来自实际项目经验不是通用方法论而是可以直接落到代码里的操作策略。8.1 把评估器当作一等公民很多实现把结果校验写成一个if语句塞在循环里这是非常糟糕的做法。评估器应该是独立的、可配置的、可扩展的组件。推荐在项目早期就把评估器接口设计好之后的每一个任务都可以复用。评估器做得足够好循环的收敛性、成本、可靠性都会同步提升。反过来评估器太弱再强的循环控制策略也救不回来。8.2 默认设置合理的安全边界建议所有新建的 Loop Engine 默认开启以下参数engine LoopEngine( llm_callllm_call, evaluatorevaluator, max_iterations5, # 正常任务通常 3 轮以内 max_retries2, # 连续重试 2 次失败就终止 timeout_seconds90.0, # 单次任务总耗时上限 )如果业务场景确实需要更长的循环再针对性地调大。先严后松比先松后严安全得多。8.3 循环步数尽量控制在三步以内在 Agent 应用里一个能稳定运行的任务平均循环轮数通常只有 1 到 3 轮。超过 5 轮的复杂任务大概率是任务拆分不合理而不是循环能力不够。拆分原则是每个子任务必须可以被评估器独立校验。如果一个子任务的结果“说不清楚怎么算好”它在循环中必然会反复震荡。8.4 日志必须带上 Prompt 快照生产环境中排查 Agent 问题最痛苦的是看不到“那一轮模型到底看到了什么输入”。强烈建议在每次调用模型时将完整的 messages 列表以只读快照形式写入日志存储。这会产生大量日志但对排错的价值极高。8.5 别让循环里的工具调用裸奔如果循环中调用外部工具比如查询数据库、调 HTTP API、写文件务必给工具调用加上独立的超时控制、错误处理和审计日志。工具调用失败时不是在循环里闷头重试而是把错误信息明确写入上下文让模型根据错误决定下一步。9. 总结与后续学习方向Loop Engineering 真正解决的不是“怎么让大模型多跑几轮”的问题而是让大模型循环变得可控、可观测、可终止、可审计。它的底层是一套状态机模型外层是一组工程约束。没有这些约束循环只是 Demo 里的玩具加上这些约束循环才能成为生产环境里的可靠组件。这篇文章的核心收获有三个第一循环是 Agent 应用的骨架而循环控制策略是判断开发水平的分水岭。第二一个可落地的 Loop Engine 必须包含任务状态、结果评估、重试策略、超时控制、日志回调五个基础组件。第三生产环境真正难的不是写循环而是控制循环的收敛性、成本、可观测性和安全性。下一步建议实践路径如果你还没有接触过 Agent 开发先跑通本文的完整示例然后替换fake_llm_call为真实模型 API尝试让引擎处理一个你工作中的实际任务。如果你已经写过循环代码建议先从评估器入手把你现在的if判断重构成独立的评估器组件再接入结构化日志。这一步对后续优化效果非常明显。如果你正在设计团队级的 Agent 平台值得继续深挖几个方向循环成本的预算管理机制、多 Agent 协作时的嵌套循环控制、以及循环执行轨迹的回放与离线分析。这些方向在学术研究和工程落地两个层面都还有大量值得探索的空间。写这篇文章时一直提醒自己不要只讲“什么是循环”因为网上类似的文章已经很多了。真正有价值的部分是那些在 Demo 里看不见、上了生产环境才暴露的问题。建议收藏备用等你的 Agent 项目开始出现“循环失控”“成本超支”“排错困难”这些问题时再翻出来对照排查。 Loop Engineering：大模型Agent循环控制的核心方法论
很多人刚开始接触 Agent 开发时都以为最难的是让大模型“听懂人话”。等真正动手做一个能跑多轮任务的应用才发现卡住自己的根本不是 Prompt而是那个看起来非常不起眼的“循环”。同样的一个语义理解任务单次调用大模型就能完成可一旦你希望它自主规划、分步执行、根据中间结果调整策略问题就全来了循环多久结束中间结果存在哪里某一步出错了要不要重试重试几次整个循环的 Token 成本怎么控制如果大模型陷入死循环怎么强制终止这些问题不属于 Prompt 工程也不属于模型微调而是属于一套正在快速成型的工程体系——Loop Engineering。这篇文章会从底层机制讲起拆解循环设计的核心原理再用一个可运行的完整代码案例把工程落地中的六个关键难点和解决方案串联起来。如果你正准备把 Agent 从“Demo 阶段”推向“生产阶段”这篇文章值得读完。1. 这篇文章真正要解决的问题Loop Engineering 并不是一个写死循环的编程技巧而是围绕“大模型多次调用构成的执行循环”进行设计、控制、治理的一套方法论。先看一个真实场景。假设你要做一个自动写周报的 Agent流程并不复杂读取本周提交记录 → 总结重要变更 → 生成周报草稿 → 按团队模板格式化。用单次 Prompt 也能做但效果很一般因为提交记录可能是几十条零散信息大模型很难一次就完成“清洗、归纳、排序、改写”四个动作。更合理的方案是拆成多步每一步都有独立的 Prompt 和输出校验。于是代码就变成了下面这样while True: result call_llm(current_task, context) # 怎么判断 result 是否合格 # 不合格是重试还是终止 # 重试次数上限是多少 if is_done(result): break问题就从“怎么写 Prompt”变成了“怎么设计这个循环”。这个转变就是 Loop Engineering 出现的根本原因。本文要解决的四个核心问题循环的底层执行模型是什么为什么大模型循环天然存在失控风险。一个可落地的循环引擎应该包含哪些基础组件。循环的终止条件、错误处理、成本控制、可观测性怎么设计。从开发到生产哪些地方最容易被忽略。适合阅读本文的读者有三类正在用大模型 API 开发 Agent 应用的工程师想从“单次 Prompt 调用”升级到“多步任务编排”的开发者以及团队里负责 AI 应用架构设计的技术负责人。2. Loop Engineering 的核心概念与底层原理2.1 什么是 Loop EngineeringLoop Engineering 不是某个框架或者某个开源项目的名字而是一类工程实践的总称。它研究的核心对象是大模型驱动的循环执行过程。一个典型的 Loop 由四个基本部分组成。组成部分作用类比任务队列存放待完成的子任务决定下一轮做什么工厂里的待加工订单执行器调用大模型或工具完成当前子任务工位上的操作员结果评估器判断执行结果是否符合要求质检员控制策略决定继续循环、重试、终止或降级车间调度员四个部分缺一不可。很多 Demo 只实现了“执行器 简单终止条件”结果就是任务稍微复杂一点循环就失去控制。2.2 循环的底层机制从状态机视角理解理解 Loop Engineering 有一个很好的视角是有限状态机。一个 Agent 的循环过程本质上是在这几个状态之间转换READY等待执行下一个子任务。RUNNING正在调用大模型或工具。SUCCESS当前子任务执行成功结果进入下一环节。RETRY执行结果不合格准备重试。FAILED执行失败进入异常处理。TERMINATED整个循环结束。关键是状态之间的转移不是无条件进行的。每一次转移都需要一个明确的触发条件。比如从 RETRY 到 RUNNING必须满足“重试次数小于最大重试数”从 RUNNING 到 SUCCESS必须满足“结果校验通过”。如果这些条件没有设计清楚循环就等于没有刹车。这也是 loop engineering 底层原理中最核心的一点循环不是让大模型自由发挥而是用状态机约束它的行动边界。2.3 为什么大模型循环天然容易失控大模型的输出存在三个固有特性导致它驱动的循环很难稳定第一不确定性。同样的 Prompt两次调用结果可能不同。第一次校验通过第二次可能就失败了。这让循环的终止条件很难写。第二退化问题。有研究观察到大模型在长时间推理或多次迭代后可能出现性能下降生成内容质量反而变差。这种现象常被称为“退化”或“重复循环”。工程上的表现就是模型在某个错误结论上反复打转。第三上下文膨胀。每循环一次对话历史就变长一点。Token 成本上升响应延迟变高而且模型容易在长上下文中丢失早期信息。所以Loop Engineering 的本质工作就是通过工程手段对冲这三个固有风险让循环变得可控、可观测、可终止。3. 环境准备与最小工程骨架3.1 技术选型本文的完整示例使用 Python 实现因为 Python 生态对大模型应用支持最成熟而且代码直观、适合教学。环境依赖只有三个部分Python 3.9 及以上版本具体版本以你的系统环境为准。大模型 API 调用库示例中使用通用的 HTTP 调用方式方便替换为任意兼容 API。一个可用的模型服务可以在代码里配置 API Key 和 Base URL。3.2 项目结构建议按下面的结构组织工程目录loop-engineering-demo/ ├── engine/ │ ├── __init__.py │ ├── loop_engine.py # 循环引擎核心 │ ├── state.py # 状态定义 │ └── evaluator.py # 结果评估器 ├── tasks/ │ ├── __init__.py │ └── weekly_report.py # 周报示例任务 ├── config.py # 全局配置 └── main.py # 入口脚本这个结构本身就是一个最小可扩展的工程骨架。后面加新任务只需要在tasks/目录下新增一个文件加新的评估逻辑只需要扩展evaluator.py。4. 核心流程拆解从一个任务的生命周期说起在写代码之前先把一个任务在 Loop Engine 中的完整生命周期理清。这比直接看代码重要得多。4.1 第一步任务初始化任务进入引擎时不是直接丢给大模型而是先被包装成一个任务对象。任务对象包含原始目标、子任务列表、最大循环次数、超时时间、上下文容器、结果存储。这样设计的目的是让引擎在执行过程中拥有完整的“控制面板”而不是裸调 API。4.2 第二步子任务执行引擎从任务队列中取出当前子任务组装 Prompt调用大模型。这一步的工程细节在于 Prompt 组装需要把系统指令、用户目标、历史上下文、工具返回结果拼接成一次完整的调用。拼接顺序会影响模型对任务的理解建议把目标信息放在离当前指令最近的位置。4.3 第三步结果评估拿到模型输出后不能直接进入下一轮。必须经过评估器校验。评估器有三层策略基础格式校验输出是否能正确解析为 JSON是否包含必要字段语义规则校验是否满足业务约束比如金额不能为负数、日期格式必须正确。模型自评校验在某些场景下可以让模型对自己的输出打分但要注意这只能作为辅助手段不能作为唯一依据。4.4 第四步状态转移决策根据评估结果引擎决定下一步动作评估通过 → 结果写入上下文 → 进入下一个子任务。评估不通过且重试次数未超限 → 回到 RUNNING带上错误信息重新调用。评估不通过且重试次数超限 → 进入 FAILED根据配置决定是整体终止还是降级处理。循环次数达到上限 → 无论当前状态如何强制 TERMINATED。4.5 第五步收尾与返回循环终止后引擎要把最终结果、执行轨迹、Token 消耗、每步状态变迁记录一并返回。这些数据不仅用于展示更是后续优化循环策略的依据。5. 完整示例手写一个可落地的 Loop Engine5.1 定义任务状态# engine/state.py from enum import Enum class TaskState(str, Enum): READY ready RUNNING running SUCCESS success RETRY retry FAILED failed TERMINATED terminated class LoopStatus(str, Enum): COMPLETED completed FAILED failed TIMEOUT timeout MAX_ITERATIONS max_iterations这里使用str枚举有两个好处序列化方便可以直接写入日志或数据库兼容性强前后端传值时不需要额外转换。5.2 实现结果评估器# engine/evaluator.py import json from typing import Any, Dict class BaseEvaluator: 评估器基类负责校验模型输出是否满足要求。 def validate(self, output: str) - Dict[str, Any]: 返回格式{passed: bool, reason: str, parsed: Any} raise NotImplementedError class JSONEvaluator(BaseEvaluator): 校验模型输出是否为合法 JSON同时做基础字段检查。 def __init__(self, required_fieldsNone): self.required_fields required_fields or [] def validate(self, output: str) - Dict[str, Any]: try: parsed json.loads(output) except json.JSONDecodeError as e: return { passed: False, reason: fJSON 解析失败: {e}, parsed: None, } missing [field for field in self.required_fields if field not in parsed] if missing: return { passed: False, reason: f缺少必要字段: {missing}, parsed: parsed, } return { passed: True, reason: 校验通过, parsed: parsed, }评估器是一个可插拔设计。不同的任务可以传入不同的评估器。比如周报任务需要检查“本周进展”字段代码审查任务需要检查“风险等级”字段这些都是评估器的配置项而不是写死在引擎里。5.3 实现循环引擎核心# engine/loop_engine.py import time from typing import Any, Callable, Dict, List, Optional from .evaluator import BaseEvaluator from .state import LoopStatus, TaskState class LoopEngine: 一个最小可用的 Loop 引擎。 设计原则 1. 所有运行参数最大循环次数、超时时间、重试次数都通过配置注入。 2. 状态变化通过回调函数暴露给外部方便日志和监控。 3. 引擎本身不关心业务细节只负责调度和控制。 def __init__( self, llm_call: Callable[[str], str], evaluator: BaseEvaluator, max_iterations: int 10, max_retries: int 2, timeout_seconds: float 60.0, on_state_change: Optional[Callable[[str, Any], None]] None, ): self.llm_call llm_call self.evaluator evaluator self.max_iterations max_iterations self.max_retries max_retries self.timeout_seconds timeout_seconds self.on_state_change on_state_change def _notify(self, event: str, data: Any): if self.on_state_change: self.on_state_change(event, data) def run(self, task: str, system_prompt: str, context: Optional[List[Dict]] None): start_time time.time() retries 0 history list(context or []) self._notify(TaskState.READY, {task: task}) for iteration in range(1, self.max_iterations 1): # 检查整体超时 if time.time() - start_time self.timeout_seconds: self._notify(TaskState.TERMINATED, {reason: timeout}) return { status: LoopStatus.TIMEOUT, iterations: iteration - 1, result: None, history: history, } self._notify(TaskState.RUNNING, {iteration: iteration}) # 组装 Prompt把系统指令 用户任务 历史上下文拼成一次调用 messages [{role: system, content: system_prompt}] messages.extend(history) messages.append({role: user, content: task}) raw_output self.llm_call(messages) # 执行结果评估 eval_result self.evaluator.validate(raw_output) if eval_result[passed]: parsed_output eval_result[parsed] history.append({role: assistant, content: raw_output}) self._notify(TaskState.SUCCESS, { iteration: iteration, output: parsed_output, }) return { status: LoopStatus.COMPLETED, iterations: iteration, result: parsed_output, history: history, } # 评估未通过进入重试逻辑 retries 1 if retries self.max_retries: self._notify(TaskState.FAILED, { iteration: iteration, reason: eval_result[reason], }) return { status: LoopStatus.FAILED, iterations: iteration, result: None, history: history, error: eval_result[reason], } self._notify(TaskState.RETRY, { iteration: iteration, reason: eval_result[reason], }) # 把错误信息放进上下文中让模型下一轮知道哪里有问题 history.append({role: assistant, content: raw_output}) history.append({ role: user, content: f上一步输出不符合要求{eval_result[reason]}。请修正后重新输出。 }) # 达到最大迭代次数 self._notify(TaskState.TERMINATED, {reason: max_iterations}) return { status: LoopStatus.MAX_ITERATIONS, iterations: self.max_iterations, result: None, history: history, }这个引擎的核心设计有四点配置注入。max_iterations、max_retries、timeout_seconds全部通过构造参数传入而不是写死在类内部。这样同一个引擎可以服务不同任务高风险任务可以降低最大重试次数成本敏感的任务可以缩短超时时间。状态回调。on_state_change回调让引擎可以无缝接入日志系统和监控平台。在生产环境中每一次状态变化都应该有记录这是事后排查问题的基础。错误反馈进上下文。评估失败之后不是简单重试而是把失败原因拼进历史记录里告诉模型。这一条非常关键。如果不告诉模型为什么失败重试两三次往往还是同样的错误。循环边界明确。请求级超时和整体超时同时生效避免模型单次调用卡死或整个任务长时间无法结束。5.4 实现一个具体的周报任务# tasks/weekly_report.py import json def build_weekly_report_task(commits): 把原始提交记录封装成任务描述。 commit_text \n.join([f- {c} for c in commits]) task f 根据以下本周提交记录生成周报草稿。 要求 1. 将提交记录按功能模块归类。 2. 每个模块总结一句话进展。 3. 输出 JSON格式如下 {{ summary: 总体进展一句话, modules: [模块名1: 进展描述, 模块名2: 进展描述] }} 提交记录 {commit_text} return task # 示例提交数据 SAMPLE_COMMITS [ 优化用户登录接口增加验证码校验, 修复订单超时未关闭的问题, 新增商品搜索的拼音匹配, 重构订单状态机支持取消和退款流转, 修复支付回调幂等性问题, 优化首页加载速度图片改为懒加载, ]这个任务故意设计成需要“归类 → 总结 → 格式化”三个步骤比较适合演示循环的价值。单次调用也能生成结果但经过循环校验后输出的结构化程度会高很多。5.5 编写入口文件# main.py import json from engine.evaluator import JSONEvaluator from engine.loop_engine import LoopEngine from tasks.weekly_report import SAMPLE_COMMITS, build_weekly_report_task # 模拟大模型调用的函数 # 实际项目中替换为真实的 API 调用即可 def fake_llm_call(messages): 一个用于本机演示的模拟调用。 真实项目中应该调用你的模型服务。这里为了演示循环控制逻辑 故意让前两次输出不是合法 JSON展示重试机制。 global _call_count _call_count 1 last_user_message messages[-1][content] if _call_count 1: return 这个周报不太好写我先看看提交记录。 if _call_count 2: return 我认为这周主要工作内容是登录优化。没有输出 JSON。 # 第三次开始输出正确结果 return json.dumps({ summary: 本周完成了登录安全加固、订单流程修正和性能优化三项重点工作, modules: [ 用户模块增加验证码校验提升登录安全性, 订单模块修复超时未关闭问题重构状态机支持取消与退款, 搜索模块新增拼音匹配能力, 性能优化首页图片改为懒加载提升首屏速度, ], }, ensure_asciiFalse) _call_count 0 class DemoLogger: 演示用的日志回调生产环境请接入真实日志系统。 def __call__(self, event, data): print(f[状态变化] {event}) if iteration in data: print(f 迭代轮次: {data[iteration]}) if reason in data: print(f 原因: {data[reason]}) def main(): engine LoopEngine( llm_callfake_llm_call, evaluatorJSONEvaluator(required_fields[summary, modules]), max_iterations5, max_retries3, timeout_seconds120.0, on_state_changeDemoLogger(), ) task build_weekly_report_task(SAMPLE_COMMITS) result engine.run( tasktask, system_prompt你是一个严谨的技术周报生成助手。你必须严格按照要求输出 JSON。, ) print(\n 最终结果 ) print(json.dumps(result, ensure_asciiFalse, indent2)) if __name__ __main__: main()5.6 运行与验证在项目根目录执行python main.py预期输出大致如下[状态变化] ready [状态变化] running 迭代轮次: 1 [状态变化] retry 迭代轮次: 1 原因: JSON 解析失败 [状态变化] running 迭代轮次: 2 [状态变化] retry 迭代轮次: 2 原因: JSON 解析失败 [状态变化] running 迭代轮次: 3 [状态变化] success 迭代轮次: 3 最终结果 在final_result中status为completediterations为 3result中包含了summary和modules两个字段。如果运行失败优先检查三处Python 版本是否满足要求json库是否被项目中的某个文件覆盖命名fake_llm_call中的全局变量_call_count是否被重新初始化。6. 工程落地难点拆解Demo 跑通只是第一步。把 Loop Engine 放到生产环境会遇到一系列在玩具项目里完全暴露不出来的问题。下面按影响程度排序拆解。6.1 难点一循环收敛性无法保证这是 Loop Engineering 最底层的难题。大模型的输出有随机性你无法从数学上证明一个循环一定会收敛到正确结果。实践中可能遇到的场景是模型在第三步反复犯同一个错误重试三次仍然一样评估器始终不通过。应对策略有三个层次第一层是重试策略优化。简单重试往往无效更有效的是“失败原因引导重试”。这就是前面代码里把eval_result[reason]拼进上下文的原因。让模型知道具体错在哪里重试成功率会显著提升。第二层是任务拆分。如果某个子任务反复失败可以考虑把它拆成更小的子任务。例如“生成周报”这一步失败率高可以拆成“归类提交记录”和“生成周报草稿”两步。第三层是降级预案。达到最大重试次数后是直接失败还是返回一个“不完美但可用”的结果生产环境通常选择后者。对用户来说能拿到一个带警告的中间结果往往比什么都拿不到要好。6.2 难点二上下文膨胀每循环一次历史记录就增加一条。循环 10 次之后Prompt 中的历史可能已经吃掉几千个 Token。这不仅增加成本还会稀释模型对原始任务的注意力。常用方案是上下文裁剪和记忆压缩。上下文裁剪限制参与拼接的历史条数。轮次太多时只保留最近 N 条对话加上最开始的系统指令和用户原始任务。记忆压缩定期让模型把历史对话总结成一段简短的记忆摘要之后用摘要代替原始历史。这个方案效果更好但会增加一次模型调用需要在设计时评估成本。6.3 难点三成本不可控循环次数越多成本越高。生产环境必须设置成本上限。成本控制有两个维度一是轮次维度。max_iterations就是成本上限。设计任务时先估算“正常需要几轮”再留出 1.5 到 2 倍余量。二是Token 维度。每次调用大模型前可以估算输入 Token 数。如果发现单次调用已经接近模型上下文上限说明历史积累过多需要先裁剪再做下一次调用。6.4 难点四状态可观测性差Loop 是一个多步骤过程一旦中间某步出错排查难度比单次调用大得多。生产环境至少要能看到以下信息每次循环的输入 Prompt 完整内容。模型原始输出。评估器的判定结果和原因。每次状态转换的时间戳。累计 Token 消耗。这意味着on_state_change回调在生产环境中必须接上结构化日志系统而不能像 Demo 里一样只是 print 到控制台。日志格式最好采用 JSON方便接入 Elasticsearch、Loki 等日志平台。6.5 难点五幂等性在循环执行过程中如果服务重启或者网络闪断已经完成的子任务怎么办重跑一遍可能产生重复结果。解决思路是给每一步执行加一个全局唯一的trace_id和step_id。执行结果落库时以step_id为幂等键重复执行时直接复用已有结果。这个做法和普通后端接口的幂等设计完全一致只不过作用在循环的每一步上。6.6 难点六安全边界循环给了大模型更多轮次的操作机会也意味着给了它更多犯错或越权的空间。如果循环中包含工具调用、数据库访问、文件写入等操作必须设置明确的安全边界。核心原则是每一步工具调用都做权限校验不要因为“上一步通过了”就默认“这一步也安全”。涉及删除、更新、写入的操作必须在循环外先经过人工审批。落到生产环境的代码循环中每步操作都要有审计日志。7. 常见问题与排查方法问题现象可能原因排查方式解决方案循环不终止终止条件设计不完整模型总是输出不满足要求的内容查看每次评估器的失败原因连续失败是否集中在同一类错误调整评估规则或拆分任务降低单步复杂度重试多次但结果不变模型没有获得失败反馈上下文里只有原始任务检查重试分支是否把失败原因写入了上下文在重试消息中明确告知模型错误点Token 消耗增长过快历史记录无限制累积统计每个迭代后的输入 Token 数增加上下文裁剪或记忆压缩某一步依赖上一步的中间结果但拿不到状态在循环中没有正确传递检查 history 列表是否完整保存了每轮输出确保每轮结果都追加到上下文中生产环境循环日志混乱没有统一 trace_id查看日志系统中的请求关联字段为每个任务生成唯一 trace_id并在所有日志中携带超时时间设了但没生效超时只包住单次调用没有包住整个循环检查超时统计的起止时间点在循环开始前记录 start_time每轮迭代前检查总耗时8. 最佳实践与工程建议下面这些建议来自实际项目经验不是通用方法论而是可以直接落到代码里的操作策略。8.1 把评估器当作一等公民很多实现把结果校验写成一个if语句塞在循环里这是非常糟糕的做法。评估器应该是独立的、可配置的、可扩展的组件。推荐在项目早期就把评估器接口设计好之后的每一个任务都可以复用。评估器做得足够好循环的收敛性、成本、可靠性都会同步提升。反过来评估器太弱再强的循环控制策略也救不回来。8.2 默认设置合理的安全边界建议所有新建的 Loop Engine 默认开启以下参数engine LoopEngine( llm_callllm_call, evaluatorevaluator, max_iterations5, # 正常任务通常 3 轮以内 max_retries2, # 连续重试 2 次失败就终止 timeout_seconds90.0, # 单次任务总耗时上限 )如果业务场景确实需要更长的循环再针对性地调大。先严后松比先松后严安全得多。8.3 循环步数尽量控制在三步以内在 Agent 应用里一个能稳定运行的任务平均循环轮数通常只有 1 到 3 轮。超过 5 轮的复杂任务大概率是任务拆分不合理而不是循环能力不够。拆分原则是每个子任务必须可以被评估器独立校验。如果一个子任务的结果“说不清楚怎么算好”它在循环中必然会反复震荡。8.4 日志必须带上 Prompt 快照生产环境中排查 Agent 问题最痛苦的是看不到“那一轮模型到底看到了什么输入”。强烈建议在每次调用模型时将完整的 messages 列表以只读快照形式写入日志存储。这会产生大量日志但对排错的价值极高。8.5 别让循环里的工具调用裸奔如果循环中调用外部工具比如查询数据库、调 HTTP API、写文件务必给工具调用加上独立的超时控制、错误处理和审计日志。工具调用失败时不是在循环里闷头重试而是把错误信息明确写入上下文让模型根据错误决定下一步。9. 总结与后续学习方向Loop Engineering 真正解决的不是“怎么让大模型多跑几轮”的问题而是让大模型循环变得可控、可观测、可终止、可审计。它的底层是一套状态机模型外层是一组工程约束。没有这些约束循环只是 Demo 里的玩具加上这些约束循环才能成为生产环境里的可靠组件。这篇文章的核心收获有三个第一循环是 Agent 应用的骨架而循环控制策略是判断开发水平的分水岭。第二一个可落地的 Loop Engine 必须包含任务状态、结果评估、重试策略、超时控制、日志回调五个基础组件。第三生产环境真正难的不是写循环而是控制循环的收敛性、成本、可观测性和安全性。下一步建议实践路径如果你还没有接触过 Agent 开发先跑通本文的完整示例然后替换fake_llm_call为真实模型 API尝试让引擎处理一个你工作中的实际任务。如果你已经写过循环代码建议先从评估器入手把你现在的if判断重构成独立的评估器组件再接入结构化日志。这一步对后续优化效果非常明显。如果你正在设计团队级的 Agent 平台值得继续深挖几个方向循环成本的预算管理机制、多 Agent 协作时的嵌套循环控制、以及循环执行轨迹的回放与离线分析。这些方向在学术研究和工程落地两个层面都还有大量值得探索的空间。写这篇文章时一直提醒自己不要只讲“什么是循环”因为网上类似的文章已经很多了。真正有价值的部分是那些在 Demo 里看不见、上了生产环境才暴露的问题。建议收藏备用等你的 Agent 项目开始出现“循环失控”“成本超支”“排错困难”这些问题时再翻出来对照排查。