做 AI Agent 和自动化系统开发时,我最大的感受是:很多同学能把单个模型调用、工具封装写得很好,但一旦涉及“系统自主迭代”“根据结果自动修正”这类需求,就完全不知道从哪下手。网上讲 Loop Engineering 的资料也很少有成体系的,大多数是一笔带过,或者只讲理论不给代码。这篇文章我会把 Loop Engineering 从核心概念到代码实战完整走一遍,不讲空话,直接用可运行的 Python 工程演示一个闭环系统是怎么构建的。想深入 Agent 开发、自动化运维和自愈系统的读者,这篇能帮你省下大量自己摸索的时间。
1. 什么是 Loop Engineering:闭环思维正在改变开发方式
1.1 从“一次性开发”到“循环驱动”
传统软件开发模式可以简单概括为:需求分析、写代码、测试、发布、维护。这条流水线是“线性”的,每一步做完就进入下一步,即使后面发现问题,也要重新走一遍流程。
Loop Engineering 的核心思想完全不一样。它强调的是让系统自己形成“感知 -> 决策 -> 执行 -> 反馈 -> 迭代”的闭环。系统不是上线后就固定不变,而是在运行过程中不断根据执行结果调整自己的行为策略,从而持续逼近目标。
举个例子,传统爬虫程序如果遇到页面结构变化,通常需要人工介入修改选择器。而采用 Loop Engineering 思路的爬虫会在抓取失败后自动记录失败原因、切换备用解析策略、更新本地解析规则库,下一次遇到同类页面时就能直接用新规则处理。
Loop Engineering 不是一个开箱即用的框架,而是一套工程方法论。它融合了这些思想:
- 控制论中的反馈回路。
- DevOps 中的持续交付和监控。
- 强化学习中的奖励与策略更新。
- Agent 开发中的工具调用与结果评估。
一句话概括:Loop Engineering 就是让系统具备自我修正和持续演进能力的工程实践集合。
1.2 为什么现在 Loop Engineering 突然火了
Loop Engineering 近两年热度上升,和 AI Agent 的爆发有直接关系。
过去我们写自动化脚本,是“如果 A 就执行 B”,逻辑是写死的。但现在的 AI Agent 不同,它需要自己判断该调用哪个工具、怎么解析结果、失败了怎么办。这个过程天然就是一个循环。
大模型 LLM 的每一次推理并不是决定性的,Agent 需要通过多轮“思考 -> 行动 -> 观察 -> 再思考”来逼近正确答案。这种模式在 ReAct、AutoGPT、MetaGPT 等框架中体现得淋漓尽致。
而当循环中加入自动评估模块和记忆模块之后,Agent 就不再只是“多轮对话”,而是真正具备了“越用越聪明”的能力——这就是 Loop Engineering 最核心的价值。
1.3 Loop Engineering 的典型应用场景
| 应用领域 | 典型场景 | 闭环价值 |
|---|---|---|
| AI Agent 开发 | 多工具调用、任务拆解、失败重试 | 提高任务完成率,减少人工干预 |
| 自动化运维 | 监控告警、日志分析、故障自愈 | 缩短故障恢复时间 |
| 推荐系统 | 用户行为反馈、模型在线更新 | 推荐效果持续优化 |
| 数据管道 | 数据质量校验、异常重跑 | 保证数据产出稳定 |
| 测试平台 | 自动化用例执行、失败定位 | 降低测试维护成本 |
| 业务风控 | 风险识别、策略动态调整 | 提高风险拦截准确率 |
可以这么说:凡是“执行结果会反过来影响下一次执行策略”的系统,都在做 Loop Engineering。
2. 环境准备与项目结构
2.1 版本与环境要求
本文代码完全基于 Python 标准库实现,不依赖任何第三方框架,方便你理解闭环最底层的运行逻辑。如果你已经接触过 Agent 开发框架,也可以把本文的闭环结构迁移到 LangChain、AutoGPT 等框架中。
建议环境如下:
| 软件 | 建议版本 | 说明 |
|---|---|---|
| 操作系统 | Windows 10/11、macOS、Linux | 代码跨平台 |
| Python | 3.9+ | 推荐 3.10 以上 |
| 开发工具 | VS Code / PyCharm | 任意习惯 IDE 即可 |
| 依赖库 | 无(仅标准库) | 降低上手门槛 |
2.2 项目文件结构
为了让代码逻辑清晰,我们按职责拆分文件:
loop_engineering_demo/ ├── config.py # 全局配置参数 ├── tools.py # Agent 可调用的工具集合 ├── agent.py # Agent 核心:决策、执行、评估 ├── loop.py # 闭环主循环控制 └── main.py # 启动入口与演示脚本后面每一部分我都会给出完整代码并逐段解释。
3. 闭环系统核心原理拆解
要真正理解 Loop Engineering,光背概念是不够的。下面把闭环拆成五个关键环节,每个环节都对应具体的代码职责。这些都是后续实战的基础。
3.1 闭环五要素:感知、决策、执行、反馈、迭代
一个完整的 Engineering Loop 由五个环节组成:
- 感知:获取外部输入,包括用户请求、系统状态、监控指标、错误日志等。
- 决策:根据当前状态和目标,选择一个行动方案。这一步可以通过规则、策略权重或大模型推理完成。
- 执行:调用具体工具或函数,完成实际操作。
- 反馈:根据执行结果计算奖励或惩罚,评估本次行动是否有效。
- 迭代:将反馈结果写入记忆或策略库,更新下一次决策的参数,进入下一轮循环。
用一个生活化的例子来解释:你第一次做菜盐放少了,尝一口发现味道淡,于是第二次多加了一勺盐。这里的“尝一口”是感知,“判断咸淡”是决策,“加盐”是执行,“味道变好”是反馈,“下次按新比例放盐”就是迭代。
3.2 先看一个最小闭环示例
先不急着写完整系统,我们用一个 30 行左右的代码演示闭环的最基本形态。这段代码模拟一个“猜数字”的自适应程序:如果猜大了,系统就调低猜测范围;如果猜小了,就调高范围。
# 文件路径:minimal_loop.py import random def run_minimal_loop(target: int, max_rounds: int = 20): """ 最小闭环演示:根据反馈动态调整猜测策略 """ low, high = 1, 100 history = [] for round_idx in range(1, max_rounds + 1): guess = (low + high) // 2 # 反馈环节:比较本次执行结果与目标 if guess > target: high = guess - 1 feedback = "大了" elif guess < target: low = guess + 1 feedback = "小了" else: feedback = "命中" history.append({"round": round_idx, "guess": guess, "feedback": feedback}) print(f"第 {round_idx} 轮:猜测 {guess},反馈 {feedback}") if feedback == "命中": print(f"目标 {target} 已在 {round_idx} 轮内被找到") break else: print("超过最大轮数仍未命中") return history if __name__ == "__main__": target = random.randint(1, 100) print(f"本轮目标值为:{target}") run_minimal_loop(target)运行后输出效果如下:
本轮目标值为:73 第 1 轮:猜测 50,反馈 小了 第 2 轮:猜测 76,反馈 大了 第 3 轮:猜测 63,反馈 小了 第 4 轮:猜测 70,反馈 小了 第 5 轮:猜测 73,反馈 命中 目标 73 已在 5 轮内被找到这段代码的核心在于:每一次猜测之后,feedback都会更新low或high,而更新后的范围又会直接影响下一次猜测。这就是闭环的雏形。
3.3 从最小示例到工程化闭环
最小闭环能帮我们理解“循环反馈”的基本逻辑,但真实项目中要复杂的多。
实际工程中的 Loop Engineering 通常需要解决以下问题:
- 决策策略不是简单的二分法,而是多个工具之间的选择。
- 反馈信号来自执行结果,可能是成功标志、错误类型、响应时长、用户评分等。
- 迭代不是简单改两个边界值,而是更新策略权重、重试次数、任务队列等。
- 系统需要日志和状态记录,方便回溯问题。
- 循环必须有终止条件,否则可能陷入死循环。
带着这些思考,我们进入完整的代码实战。
4. 实战:从零实现一个带反馈优化的智能体闭环系统
下面我们来实现一个更接近实际 Agent 开发场景的闭环系统。这个系统模拟一个“自动问答并持续优化”的智能体,整体结构如下:
- 用户向 Agent 提出问题。
- Agent 根据策略权重选择工具。
- 工具执行并返回结果。
- 评估模块判断结果是否满足要求。
- 反馈模块更新策略权重,让后续决策越来越准确。
- 主循环控制整个流程,并在必要时重试。
4.1 定义配置文件
配置文件负责集中管理 Agent 的参数,方便后续调整,不需要硬编码在业务代码里。
# 文件路径:config.py """ 全局配置:集中管理闭环系统关键参数 """ class LoopConfig: def __init__(self): # Agent 名称与版本 self.agent_name = "LoopAgent" self.agent_version = "1.0.0" # 工具列表 self.tools = ["calculator", "time_service", "knowledge_base"] # 初始策略权重:值越大,被选中的概率越高 self.strategy_weights = { "calculator": 0.3, "time_service": 0.2, "knowledge_base": 0.5, } # 闭环控制参数 self.max_retry = 3 # 单次任务最大重试次数 self.max_loop = 5 # 整体闭环最大轮数 self.learning_rate = 0.1 # 反馈更新步长 # 日志开关 self.verbose = True def show_config(self): """打印当前配置""" print("=" * 50) print(f"Agent 名称:{self.agent_name}") print(f"Agent 版本:{self.agent_version}") print(f"可用工具:{', '.join(self.tools)}") print(f"初始策略权重:{self.strategy_weights}") print("=" * 50)4.2 实现工具层
工具层是 Agent 能执行的最小能力单元。每个工具都接收一个参数,并返回统一格式的结果字典。
# 文件路径:tools.py """ 工具集合:Agent 可调用的能力单元 """ import time import random def calculator(expression: str): """ 简易计算器工具:支持加减乘除和括号 """ try: # 安全校验:只允许数字、运算符号、括号、空格 allowed_chars = set("0123456789+-*/(). ") if not set(expression).issubset(allowed_chars): return { "success": False, "result": None, "error": "表达式包含非法字符" } # 使用 eval 前必须校验,生产环境建议使用更安全的解析库 result = eval(expression, {"__builtins__": {}}, {}) return { "success": True, "result": result, "error": None } except Exception as e: return { "success": False, "result": None, "error": str(e) } def time_service(query: str): """ 时间查询工具:返回当前时间信息 """ current_time = time.strftime("%Y-%m-%d %H:%M:%S") return { "success": True, "result": current_time, "error": None } def knowledge_base(query: str): """ 模拟知识库:简单关键词匹配 """ knowledge = { "python": "Python 是一种广泛使用的高级编程语言", "loop": "Loop Engineering 是一种闭环工程方法论", "agent": "Agent 是能够自主感知环境并采取行动的系统", "default": "抱歉,知识库中没有找到相关答案" } # 模拟 20% 概率查询超时,用于演示反馈优化 if random.random() < 0.2: return { "success": False, "result": None, "error": "knowledge_base timeout" } answer = knowledge.get(query.lower(), knowledge["default"]) return { "success": True, "result": answer, "error": None } def execute_tool(tool_name: str, param: str): """ 统一工具调用入口:根据工具名分发请求到具体函数 """ tool_map = { "calculator": calculator, "time_service": time_service, "knowledge_base": knowledge_base, } if tool_name not in tool_map: return { "success": False, "result": None, "error": f"unknown tool: {tool_name}" } return tool_map[tool_name](param)这里需要说明两个设计细节:
calculator工具中的eval使用存在安全风险,示例中通过字符白名单和清理__builtins__做了限制,但生产环境还是推荐使用ast.literal_eval或专门的表达式解析库。knowledge_base工具故意设置了 20% 的随机失败概率,这是为了模拟真实系统中依赖服务不稳定、超时等问题,方便演示闭环反馈机制是如何通过迭代降低失败率的。
4.3 实现 Agent 核心
Agent 类是整个闭环系统的“大脑”。它负责完成以下工作:
- 根据策略权重选择一个工具。
- 调用工具执行任务。
- 评估执行结果。
- 根据评估结果更新策略权重。
# 文件路径:agent.py """ Agent 核心:负责决策、执行、评估和策略更新 """ import random from config import LoopConfig from tools import execute_tool class LoopAgent: def __init__(self, config: LoopConfig): self.config = config self.result_memory = [] # 历史结果记忆 self.success_rates = {} # 各工具成功率统计 self.strategy_weights = dict(config.strategy_weights) self._init_success_rates() def _init_success_rates(self): """初始化各工具的成功记录""" for tool in self.config.tools: self.success_rates[tool] = {"success": 0, "fail": 0, "total": 0} def _select_tool(self): """ 决策环节:根据策略权重选择工具 权重越大的工具被选中的概率越高 """ tools = list(self.strategy_weights.keys()) weights = list(self.strategy_weights.values()) return random.choices(tools, weights=weights, k=1)[0] def _evaluate_result(self, tool_name: str, result: dict): """ 反馈环节:评估本次执行结果并更新统计 """ if result.get("success"): self.success_rates[tool_name]["success"] += 1 else: self.success_rates[tool_name]["fail"] += 1 self.success_rates[tool_name]["total"] += 1 return result.get("success", False) def _update_weight(self, tool_name: str, reward: float): """ 迭代环节:根据奖励信号调整工具权重 reward > 0 表示该工具在本轮表现良好, 增加其被选中概率;reward < 0 则降低。 """ current_weight = self.strategy_weights[tool_name] new_weight = current_weight + self.config.learning_rate * reward # 防止权重变成负数 new_weight = max(0.01, new_weight) # 权重归一化,保证总和为 1 self.strategy_weights[tool_name] = new_weight total = sum(self.strategy_weights.values()) for key in self.strategy_weights: self.strategy_weights[key] /= total def record_observation(self): """ 感知环节:记录当前环境信息(对外的接口方法) 在实际项目中,这里可以采集系统指标、用户输入等 """ return { "success_rates": self.success_rates, "strategy_weights": self.strategy_weights, } def run_task(self, task: str, param: str): """ 运行一次完整任务,带反馈优化 流程:感知 -> 决策 -> 执行 -> 反馈 -> 迭代 -> (检查是否重试) """ print(f"\n[任务开始] 任务类型:{task},参数:{param}") for loop_idx in range(1, self.config.max_loop + 1): # 感知:记录当前状态 observation = self.record_observation() if self.config.verbose: print(f"[第 {loop_idx} 轮] 当前策略权重:{observation['strategy_weights']}") # 决策:选择工具 selected_tool = self._select_tool() if self.config.verbose: print(f"[第 {loop_idx} 轮] 选定工具:{selected_tool}") # 执行:调用工具 exec_result = execute_tool(selected_tool, param) # 反馈:评估执行结果 is_success = self._evaluate_result(selected_tool, exec_result) if self.config.verbose: print(f"[第 {loop_idx} 轮] 执行结果:{exec_result}") # 记录结果 self.result_memory.append({ "loop": loop_idx, "task": task, "param": param, "tool": selected_tool, "success": is_success, "error": exec_result.get("error"), "result": exec_result.get("result"), }) # 迭代:根据结果调整策略 if is_success: # 成功,奖励该工具 self._update_weight(selected_tool, reward=1.0) print(f"[第 {loop_idx} 轮] 执行成功,已提高工具 {selected_tool} 的权重") return exec_result else: # 失败,惩罚该工具 self._update_weight(selected_tool, reward=-1.0) print(f"[第 {loop_idx} 轮] 执行失败,已降低工具 {selected_tool} 的权重") print(f"\n[任务结束] 已超过最大闭环轮数 {self.config.max_loop},任务最终失败") return { "success": False, "result": None, "error": "max_loop_exceeded" } def get_stats(self): """返回统计信息""" return { "success_rates": self.success_rates, "strategy_weights": self.strategy_weights, "memory": self.result_memory, }这段代码是全文核心,值得逐段理解:
_select_tool是决策环节,使用random.choices按权重抽样。权重大的工具更容易被选中。_evaluate_result是反馈环节,更新工具的成功/失败统计。_update_weight是迭代环节。成功时reward=1.0,失败时reward=-1.0,再通过归一化保证权重总和为 1。run_task把整个闭环串起来。注意它在成功时直接返回结果,失败时进入下一轮循环。
4.4 实现闭环主循环
主循环模块负责控制整个系统的运行节奏,同时在外部提供“持续运行直到满足退出条件”的能力。这里我们用一个简单的任务队列来模拟异步到达的请求。
# 文件路径:loop.py """ 闭环主循环控制 """ import time from config import LoopConfig from agent import LoopAgent class LoopEngine: def __init__(self, config: LoopConfig): self.config = config self.agent = LoopAgent(config) self.continue_flag = True def process_request(self, task: str, param: str): """ 处理单条请求,交给 Agent 执行 """ return self.agent.run_task(task, param) def run_demo_tasks(self): """ 按任务队列持续运行,模拟真实场景中的循环处理 """ demo_tasks = [ ("calculate", "12 + 34 * 2"), ("query_time", "now"), ("query_knowledge", "python"), ("query_knowledge", "loop"), ("query_knowledge", "agent"), ("calculate", "(10 + 5) * (8 - 3)"), ] for idx, (task, param) in enumerate(demo_tasks, 1): print(f"\n========== 处理第 {idx} 个任务 ==========") self.process_request(task, param) time.sleep(0.5) # 模拟真实请求间隔 print("\n========== 全部任务执行完毕 ==========") stats = self.agent.get_stats() print("\n最终成功率统计:") for tool, rate in stats["success_rates"].items(): print(f" {tool}: 成功 {rate['success']} 次,失败 {rate['fail']} 次") print("\n最终策略权重:") for tool, weight in stats["strategy_weights"].items(): print(f" {tool}: {weight:.4f}")4.5 编写启动入口
启动入口负责初始化配置、创建引擎、运行演示任务。
# 文件路径:main.py """ Loop Engineering 演示项目启动入口 """ from config import LoopConfig from loop import LoopEngine def main(): print("=== Loop Engineering 演示程序 ===") config = LoopConfig() config.show_config() engine = LoopEngine(config) engine.run_demo_tasks() print("\n演示结束,闭环系统已根据反馈完成策略优化。") if __name__ == "__main__": main()4.6 运行与验证
进入项目目录,执行:
python main.py预期输出部分内容如下:
=== Loop Engineering 演示程序 === ================================================== Agent 名称:LoopAgent Agent 版本:1.0.0 可用工具:calculator, time_service, knowledge_base 初始策略权重:{'calculator': 0.3, 'time_service': 0.2, 'knowledge_base': 0.5} ================================================== ========== 处理第 1 个任务 ========== [任务开始] 任务类型:calculate,参数:12 + 34 * 2 [第 1 轮] 当前策略权重:{'calculator': 0.3, 'time_service': 0.2, 'knowledge_base': 0.5} [第 1 轮] 选定工具:calculator [第 1 轮] 执行结果:{'success': True, 'result': 80, 'error': None} [第 1 轮] 执行成功,已提高工具 calculator 的权重由于knowledge_base有 20% 的随机失败率,运行过程中你有可能看到“执行失败,已降低工具 knowledge_base 的权重”的日志。这正是闭环系统在工作:它自动感知错误、调整策略、降低低效工具的使用频率。
4.7 结果分析
运行完整个 demo 后,你会发现一个非常有意思的现象:即使knowledge_base工具存在随机失败,整个 Agent 的任务完成率依然很高。
原因如下:
knowledge_base失败后,系统会降低它的权重,下一次决策时选中它的概率变小。calculator和time_service相对更稳定,权重会逐步升高。- 多轮执行后,权重分布会向高成功率工具倾斜,这就是“反馈优化”的实际效果。
这不是一个静态程序,而是一个会随着运行经验不断调整决策策略的动态系统。换句话说,它已经具备了 Loop Engineering 的核心特征。
5. 常见问题与排查思路
5.1 高危问题速查表
| 问题现象 | 常见原因 | 解决思路 |
|---|---|---|
| 系统陷入死循环,请求不终止 | 缺少合适的终止条件 | 增加最大轮数限制、超时控制、熔断机制 |
| 工具调用全部失败 | 依赖服务不可用 | 检查下游服务健康状态,增加健康检查 |
| 策略权重偏向单一工具 | 学习率过高或反馈信号不准确 | 降低学习率,优化评估函数 |
| 日志太多,无法定位问题 | 没有分级日志 | 引入 debug/info/warn/error 分级 |
| 反馈信号噪声过大 | 评估标准不稳定 | 增加人工审核或采用多次采样平均 |
| 系统状态不一致 | 缺乏记忆持久化 | 将记忆和状态写入数据库或文件 |
| 反馈迭代导致性能下降 | 权重更新频率过高 | 设置批量更新窗口,不逐条更新 |
5.2 死循环问题深度排查
Loop Engineering 系统最常见的故障就是死循环。造成死循环的原因通常有三个:
第一,终止条件缺失。比如 Agent 执行失败后不断重试同一个工具,却没有设置最大重试次数。解决方法是引入“最大轮数 + 最大重试次数”双保险,就像上面代码中的max_loop和max_retry。
第二,反馈信号没有真正影响策略。如果_update_weight函数的更新量太小,或者权重归一化逻辑错误,Agent 就可能反复选择同一个失败工具,形成轮流失败的假循环。排查时打印每个工具的成功率和权重,如果权重分布长期不变,说明反馈链路出了问题。
第三,外部依赖状态一直不恢复。比如知识库服务宕机,Agent 每次都会选择它并失败,然后降低权重,但实际可用工具也无法改写结果。这时需要引入熔断机制,连续失败 N 次后直接跳过该工具一段时间。
5.3 反馈信号失真问题
反馈评估是闭环系统的灵魂。如果评估本身不准确,整个系统会被错误信号带偏。
常见的失真情况包括:
- 把超时和业务失败混为一谈。
- 只关心结果成功与否,不关心执行效率。
- 把外部环境造成的失败归因于工具本身。
建议反馈信号至少包含三个维度:
| 维度 | 说明 | 示例 |
|---|---|---|
| 成功率 | 工具是否完成了预期功能 | success / fail |
| 时效性 | 执行是否在合理时间内完成 | 响应耗时 |
| 质量分 | 结果内容是否满足用户需求 | 关键词命中、相似度评分 |
5.4 日志与状态回溯建议
工程化闭环系统一定要有完善的状态记录。建议至少记录以下内容:
- 每次循环的轮数。
- 当前的策略权重快照。
- 选中的工具及参数。
- 执行结果和错误信息。
- 策略更新前后的权重对比。
- 触发终止条件时的上下文信息。
这样当系统行为异常时,可以通过日志复现当时的决策链路,快速定位问题发生在“感知、决策、执行、反馈、迭代”哪个环节。
6. 工程实践建议
6.1 安全边界与权限控制
Loop Engineering 系统一旦具备自主决策和执行能力,安全就必须作为第一优先级考虑。如果 Agent 能调用外部系统,必须遵循最小权限原则,只授予以完成任务所需的最小权限。
具体建议:
- 工具执行前增加参数校验,避免注入攻击。
- 涉及数据库、文件删除、远程命令等危险操作,必须经过人工审批。
- 在测试环境完整验证闭环逻辑后再发布到生产环境。
- 闭环系统要支持一键暂停和紧急停止开关。
6.2 幂等性与失败重试
闭环系统中,重试是一个非常常见的动作,但重试必须考虑幂等性。如果上一次请求实际上已经成功,只是响应超时,此时重试可能造成重复操作。
建议做法:
- 每个请求附带唯一请求 ID。
- 执行前检查是否已处理过该 ID。
- 写操作重试前增加“查询确认”步骤。
- 对于不能保证幂等的操作,宁可不自动重试,转人工处理。
6.3 异常处理与超时控制
闭环系统是长时间运行的常驻系统,每一个环节都要有异常兜底。
# 异常处理参考模式 def safe_execute(func, *args, timeout=3, **kwargs): """带超时和异常捕获的安全执行包装器""" try: # 实际项目中可使用 concurrent.futures 或 asyncio 实现超时 result = func(*args, **kwargs) return {"success": True, "result": result} except Exception as e: return {"success": False, "error": str(e)}超时控制同样关键。工具执行超时、网络请求超时、整体循环超时都应该有独立配置,防止单个慢工具拖垮整个闭环系统。
6.4 性能与成本优化
Loop Engineering 系统在运行过程中会产生大量循环调用。做 AI Agent 开发时,每一轮循环可能都涉及大模型 API 调用,成本不容忽视。
优化思路:
- 引入缓存,重复问题直接返回历史最优答案。
- 设置最大循环轮数,避免无效推理。
- 简单问题走规则路由,复杂问题才升级到大模型。
- 对循环过程进行成本监控,设置单任务成本阈值。
- 使用批量反馈更新,而不是每一条结果都立刻调整全局策略。
6.5 可观测性设计
可观测性是生产环境闭环系统的生存基础。除了日志,建议引入以下几类指标:
| 指标类型 | 示例 | 作用 |
|---|---|---|
| 吞吐量 | 每分钟处理任务数 | 判断系统负载 |
| 成功率 | 任务完成率 | 判断闭环整体效果 |
| 工具分布 | 各工具调用占比 | 判断策略是否合理 |
| 平均轮数 | 每个任务消耗的循环次数 | 判断系统效率 |
| 反馈延迟 | 从执行到反馈生效的时间 | 判断策略更新时效 |
6.6 从单体闭环到多 Agent 协作
当业务复杂到一定程度,单个闭环系统可能不够用,需要引入多 Agent 协作。比如一个 Agent 负责任务拆解,一个 Agent 负责工具执行,一个 Agent 负责质量评估,三个 Agent 之间又形成一个新的闭环。
这种架构下,每个 Agent 内部是小的 Loop,Agent 之间是更大的 Loop。设计时要注意:
- 明确每个 Agent 的职责边界。
- 定义 Agent 之间传递的消息格式。
- 加入全局协调者和冲突仲裁机制。
- 为整个系统设计统一的状态管理和审计日志。
7. 总结与下一步学习路线
这篇文章从 Loop Engineering 的概念出发,完整实现了一个带反馈优化的智能体闭环系统。你现在应该已经掌握:
- Loop Engineering 的本质是“感知 -> 决策 -> 执行 -> 反馈 -> 迭代”的闭环方法论。
- 闭环系统在 AI Agent 开发、自动化运维、推荐系统等领域有广泛应用。
- 一个最小闭环只需要几十行代码就能跑通。
- 工程化闭环需要关注终止条件、反馈信号设计、日志、安全、幂等性和可观测性。
- 通过策略权重的动态调整,系统可以自动降低低效工具的使用频率。
如果你的目标是继续深入 Agent 开发,接下来可以从这几个方向入手:
- 把本文的
_select_tool决策逻辑替换为大模型调用,用 LLM 做工具选择。 - 将
result_memory持久化到 SQLite 或 Redis,让系统重启后仍然保留历史经验。 - 引入向量数据库,让 Agent 可以检索历史相似问题的解决方案。
- 研究 LangChain、AutoGPT 等框架中 ReAct 模式的具体实现。
- 尝试把闭环系统接入真实的外部 API,比如天气查询、数据库查询、文件操作。
动手验证是最好的学习方式。建议你先把上面的代码原样跑通,然后试着修改learning_rate,观察权重收敛速度的变化;再试试把knowledge_base的失败率调高或调低,看闭环系统如何自动适应环境变化。跑通之后,再往工程化方向叠加你的真实业务逻辑。
如果这篇文章对你有帮助,可以收藏备用,下次做 Agent 开发需要设计闭环逻辑时,直接对照结构来改。