最近在AI智能体开发中很多开发者发现传统的Workflow模式越来越力不从心。当Agent需要处理复杂任务时预定义的流程往往无法应对动态变化的环境。这正是Loop Engineering要解决的核心问题——让AI智能体具备自主决策和持续执行的能力。本文将完整介绍Loop Engineering的技术体系从基础概念到实战应用帮助开发者掌握这一AI智能体开发的新范式。无论你是刚接触AI智能体的新手还是有一定经验的开发者都能从中获得实用的技术方案。1. Loop Engineering基础概念与演进历程1.1 什么是Loop EngineeringLoop Engineering循环工程是一种让AI智能体能够自主、持续执行复杂任务的技术范式。与传统Workflow不同Loop Engineering采用目标驱动的设计思路开发者定义任务目标和约束条件智能体通过观察-思考-行动-评估-重新规划的循环过程自主决策如何达成目标。核心循环结构可以概括为Goal目标定义 → Observe观察状态 → Think分析规划 → Act执行行动 → Evaluate评估结果 → Replan调整计划 → Repeat循环直到目标达成这种模式的优势在于能够处理无法预先定义所有路径的复杂场景。比如一个代码审查Agent发现Bug时可以自主决定是先修复Bug还是继续审查这种动态决策能力是传统Workflow无法实现的。1.2 AI工程范式的三次演进要理解Loop Engineering的价值需要回顾AI工程的发展历程第一阶段Prompt Engineering提示工程核心控制模型的单次输出技术Zero-shot、Few-shot、Chain of Thought、ReAct等局限无状态、无记忆、无工具能力、无执行闭环第二阶段Harness Engineering驾驭工程核心构建Agent运行时环境组件Context Engineering、RAG Engineering、Memory Engineering、Tool Engineering、Policy Engineering价值让Agent从能说变成能做第三阶段Loop Engineering循环工程核心控制智能体的持续行为特点目标驱动、自主决策、动态调整意义让Agent从被操作的工具变成自主运行的员工这三者不是替代关系而是层层叠加的技术栈。Prompt Engineering嵌入到Harness中Harness又为Loop提供运行基础。1.3 为什么需要Loop Engineering传统Workflow在AI智能体时代面临三大挑战流程刚性问题预定义的A-B-C执行路径无法应对动态环境。比如Agent在代码审查中发现安全漏洞需要立即切换优先级但固定Workflow难以处理这种情境切换。长尾场景爆炸真实任务有无数变异情况。一个市场调研任务可能需要处理数据源不可用、格式不一致、信息缺失等多种异常维护所有分支路径的成本极高。维护成本高昂Workflow的每一步都需要人工定义和维护。当业务逻辑变化或工具升级时需要重新修改流程定义维护负担随复杂度指数级增长。Loop Engineering通过目标驱动和自主决策从根本上解决了这些问题。2. Loop Engineering核心技术架构2.1 五大核心构建块根据Addy Osmani的定义Loop Engineering包含五个关键组件自动化Automations负责定时执行和任务调度。支持选择项目、提示词、执行频率、运行环境等配置结果自动汇入分拣收件箱。工作树Worktrees提供隔离的并行开发环境。每个执行线程有独立的工作目录基于Git工作树实现版本控制。技能Skills将项目知识编码化沉淀。通过SKILL.md文件定义可复用的智能体能力支持显式调用或隐式触发。插件/连接器Plugins/Connectors对接外部工具的标准接口。基于MCPModel Context Protocol协议实现工具的统一接入。子智能体Sub-agents用于任务分解和团队协作。通过TOML格式定义在.codex/agents/或.claude/agents/目录下。2.2 七种典型Loop模式在实际应用中Loop Engineering演化出多种设计模式ReAct Loop思考-行动循环最基本的循环模式适用于路径不明确的探索性任务。Agent在每轮中先推理再行动逐步逼近目标。# ReAct Loop伪代码示例 def react_loop(goal, max_iterations10): state observe_environment() for i in range(max_iterations): thought reason(state, goal) # 思考阶段 action decide_action(thought) # 决策阶段 result execute_action(action) # 执行阶段 state update_state(state, result) # 状态更新 if goal_achieved(state, goal): return success(state) return failure(state)Plan-and-Execute规划-执行先生成完整计划再逐步执行适合结构清晰的任务。与ReAct相比效率更高但灵活性稍差。Reflection Loop反思循环加入自我评估和纠错机制。Agent完成步骤后用独立评判者评估输出质量不合格则重新执行。Tree of Thought多路径探索同时探索多条推理路径比较结果选择最优方案适用于需要创造性解决方案的场景。Graph Loop图结构控制使用有向图定义状态转换逻辑LangGraph是典型实现。在自主性和可控性之间找到平衡。Multi-Agent Loop多智能体协作主Orchestrator分解任务分发给多个Specialist Agent协调结果整合。Self-Improving Loop自改进循环最高级模式Agent从每次执行中学习优化策略实现持续改进。2.3 Loop Engineering技术栈完整的Loop Engineering技术栈包含以下层次基础设施层计算资源GPU集群、云服务容器化Docker、Kubernetes监控日志、指标、追踪系统模型层大语言模型GPT、Claude、LLaMA等微调模型领域特定的优化版本模型管理版本控制、AB测试知识层向量数据库Pinecone、Weaviate检索增强RAG管道、知识图谱数据连接器各种数据源接入运行层Harness上下文管理Context Engineering记忆系统Memory Engineering工具集成Tool Engineering策略控制Policy Engineering行为层Loop循环引擎ReAct、Plan-Execute等模式状态管理任务进度追踪编排调度多任务协调应用层Agentic应用智能ERP、CRM、Copilot用户界面聊天界面、仪表板集成接口API、Webhook3. 环境准备与工具配置3.1 基础环境要求开始Loop Engineering开发前需要准备以下环境操作系统Linux推荐Ubuntu 20.04macOS 12.0Windows 11 with WSL2Python环境# 创建虚拟环境 python -m venv loop_engineering source loop_engineering/bin/activate # Linux/macOS # loop_engineering\Scripts\activate # Windows # 安装基础依赖 pip install python-dotenv openai anthropic模型API配置# .env文件配置 OPENAI_API_KEYyour_openai_key ANTHROPIC_API_KEYyour_anthropic_key GROQ_API_KEYyour_groq_key3.2 核心开发框架LangGraph基础安装pip install langgraph langchain-openai langchain-anthropic向量数据库配置# 安装ChromaDB pip install chromadb # 或者安装Pinecone客户端 pip install pinecone-client3.3 开发工具配置VS Code配置{ extensions: [ ms-python.python, ms-toolsai.jupyter, charliermarsh.ruff ], settings: { python.defaultInterpreterPath: ./loop_engineering/bin/python } }Docker开发环境FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . CMD [python, main.py]4. Loop Engineering实战案例4.1 案例一自主代码审查Agent让我们构建一个能够自主进行代码审查的Loop Engineering应用。项目结构code_review_agent/ ├── agents/ │ ├── __init__.py │ ├── orchestrator.py # 主协调器 │ ├── security_agent.py # 安全审查专家 │ └── quality_agent.py # 代码质量专家 ├── tools/ │ ├── code_analyzer.py │ ├── git_operations.py │ └── security_scanner.py ├── memory/ │ ├── short_term.py │ └── long_term.py ├── config/ │ └── agent_config.yaml └── main.py主循环实现# agents/orchestrator.py from typing import Dict, Any, List from langgraph.graph import StateGraph, END from .security_agent import SecurityAgent from .quality_agent import QualityAgent class CodeReviewOrchestrator: def __init__(self): self.security_agent SecurityAgent() self.quality_agent QualityAgent() self.setup_workflow() def setup_workflow(self): # 定义状态图 workflow StateGraph(self.ReviewState) # 添加节点 workflow.add_node(analyze_code, self.analyze_code) workflow.add_node(security_review, self.security_review) workflow.add_node(quality_review, self.quality_review) workflow.add_node(generate_report, self.generate_report) # 定义边 workflow.set_entry_point(analyze_code) workflow.add_edge(analyze_code, security_review) workflow.add_conditional_edges( security_review, self.check_security_results, { continue: quality_review, critical: generate_report } ) workflow.add_edge(quality_review, generate_report) workflow.add_edge(generate_report, END) self.graph workflow.compile() class ReviewState: def __init__(self): self.code_content: str self.security_issues: List[Dict] [] self.quality_issues: List[Dict] [] self.final_report: str self.iteration_count: int 0 def analyze_code(self, state: ReviewState) - ReviewState: 分析代码结构 print(开始代码分析...) # 这里实现代码解析逻辑 state.iteration_count 1 return state def security_review(self, state: ReviewState) - ReviewState: 安全审查 print(进行安全审查...) state.security_issues self.security_agent.review(state.code_content) return state def quality_review(self, state: ReviewState) - ReviewState: 代码质量审查 print(进行代码质量审查...) state.quality_issues self.quality_agent.review(state.code_content) return state def generate_report(self, state: ReviewState) - ReviewState: 生成最终报告 print(生成审查报告...) report f 代码审查报告 - 迭代次数: {state.iteration_count} 安全问题: {self.format_issues(state.security_issues)} 代码质量问题: {self.format_issues(state.quality_issues)} state.final_report report return state def check_security_results(self, state: ReviewState) - str: 检查安全结果决定下一步 critical_issues [issue for issue in state.security_issues if issue.get(severity) critical] return critical if critical_issues else continue def format_issues(self, issues: List[Dict]) - str: return \n.join([f- {issue[description]} for issue in issues])运行示例# main.py from agents.orchestrator import CodeReviewOrchestrator def main(): orchestrator CodeReviewOrchestrator() # 模拟代码输入 sample_code def process_user_data(user_input): # 这里有一些安全问题 query fSELECT * FROM users WHERE name {user_input} return execute_query(query) initial_state orchestrator.ReviewState() initial_state.code_content sample_code # 执行审查循环 result orchestrator.graph.invoke(initial_state) print(result.final_report) if __name__ __main__: main()4.2 案例二多Agent任务分解系统构建一个能够将复杂任务分解为子任务的多Agent系统。任务分解Agent# agents/task_decomposer.py from typing import List, Dict from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI import json class TaskDecomposer: def __init__(self): self.llm ChatOpenAI(modelgpt-4) self.decomposition_prompt ChatPromptTemplate.from_template( 请将以下复杂任务分解为可并行执行的子任务 原始任务: {task_description} 要求: 1. 每个子任务应该相对独立 2. 子任务之间尽量减少依赖 3. 每个子任务应该有明确的完成标准 4. 估计每个子任务需要的时间 请以JSON格式返回分解结果包含子任务列表。 ) def decompose_task(self, task_description: str) - List[Dict]: 分解复杂任务 chain self.decomposition_prompt | self.llm response chain.invoke({task_description: task_description}) try: # 解析JSON响应 result json.loads(response.content) return result.get(subtasks, []) except json.JSONDecodeError: # 如果JSON解析失败使用备用方案 return self._fallback_decomposition(task_description) def _fallback_decomposition(self, task_description: str) - List[Dict]: 备用分解方案 # 简单的基于规则的任务分解 subtasks [] keywords [研究, 分析, 开发, 测试, 部署] for keyword in keywords: if keyword in task_description: subtasks.append({ name: f{keyword}阶段, description: f执行{keyword}相关任务, estimated_time: 2小时 }) return subtasks if subtasks else [{ name: 默认任务, description: task_description, estimated_time: 4小时 }]多Agent协调器# agents/multi_agent_coordinator.py import asyncio from typing import List, Dict from .task_decomposer import TaskDecomposer from concurrent.futures import ThreadPoolExecutor class MultiAgentCoordinator: def __init__(self, max_workers5): self.task_decomposer TaskDecomposer() self.executor ThreadPoolExecutor(max_workersmax_workers) self.agents {} # 注册的Agent池 def register_agent(self, agent_type: str, agent_instance): 注册特定类型的Agent self.agents[agent_type] agent_instance async def coordinate_task(self, main_task: str) - Dict: 协调执行复杂任务 print(f开始处理主任务: {main_task}) # 1. 任务分解 subtasks self.task_decomposer.decompose_task(main_task) print(f分解为 {len(subtasks)} 个子任务) # 2. 分配任务给合适的Agent task_assignments self._assign_tasks_to_agents(subtasks) # 3. 并行执行子任务 results await self._execute_subtasks_parallel(task_assignments) # 4. 结果整合 final_result self._integrate_results(results, main_task) return final_result def _assign_tasks_to_agents(self, subtasks: List[Dict]) - List[Dict]: 将任务分配给合适的Agent assignments [] for task in subtasks: # 简单的基于关键词的Agent分配 agent_type self._select_agent_by_task(task[description]) assignments.append({ task: task, agent_type: agent_type, assigned_agent: self.agents.get(agent_type) }) return assignments def _select_agent_by_task(self, task_description: str) - str: 根据任务描述选择Agent类型 if any(keyword in task_description for keyword in [研究, 分析]): return research_agent elif any(keyword in task_description for keyword in [开发, 编码]): return development_agent elif any(keyword in task_description for keyword in [测试, 验证]): return testing_agent else: return general_agent async def _execute_subtasks_parallel(self, assignments: List[Dict]): 并行执行子任务 loop asyncio.get_event_loop() # 创建任务列表 tasks [] for assignment in assignments: if assignment[assigned_agent]: task loop.run_in_executor( self.executor, assignment[assigned_agent].execute, assignment[task] ) tasks.append(task) # 等待所有任务完成 results await asyncio.gather(*tasks, return_exceptionsTrue) return results def _integrate_results(self, results: List, main_task: str) - Dict: 整合子任务结果 successful_results [] failed_results [] for i, result in enumerate(results): if isinstance(result, Exception): failed_results.append({ task_index: i, error: str(result) }) else: successful_results.append(result) return { main_task: main_task, successful_subtasks: len(successful_results), failed_subtasks: len(failed_results), integrated_result: self._merge_successful_results(successful_results), failures: failed_results } def _merge_successful_results(self, results: List) - str: 合并成功的结果 if not results: return 所有子任务都失败了 merged 任务执行结果汇总:\n\n for i, result in enumerate(results, 1): merged f子任务 {i}:\n{result}\n\n return merged5. 高级特性与优化策略5.1 记忆管理系统实现短期记忆实现# memory/short_term.py from typing import Dict, List, Any from datetime import datetime, timedelta class ShortTermMemory: def __init__(self, max_size1000, ttl_minutes60): self.memory_store {} self.max_size max_size self.ttl timedelta(minutesttl_minutes) self.access_log [] def store(self, key: str, value: Any, metadata: Dict None): 存储短期记忆 if len(self.memory_store) self.max_size: self._evict_oldest() self.memory_store[key] { value: value, timestamp: datetime.now(), metadata: metadata or {}, access_count: 0 } def retrieve(self, key: str) - Any: 检索记忆 if key not in self.memory_store: return None memory_item self.memory_store[key] # 检查是否过期 if datetime.now() - memory_item[timestamp] self.ttl: del self.memory_store[key] return None # 更新访问记录 memory_item[access_count] 1 self.access_log.append({ key: key, timestamp: datetime.now() }) return memory_item[value] def _evict_oldest(self): 淘汰最旧的记忆 if not self.memory_store: return oldest_key min(self.memory_store.keys(), keylambda k: self.memory_store[k][timestamp]) del self.memory_store[oldest_key] def get_recent_context(self, count5) - List[Any]: 获取最近的上下文 recent_items sorted( self.memory_store.items(), keylambda x: x[1][timestamp], reverseTrue )[:count] return [item[1][value] for _, item in recent_items]长期记忆实现# memory/long_term.py import sqlite3 from typing import Dict, List, Any from datetime import datetime import json class LongTermMemory: def __init__(self, db_path:memory:): self.conn sqlite3.connect(db_path) self._create_tables() def _create_tables(self): 创建记忆存储表 cursor self.conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, key TEXT UNIQUE NOT NULL, value TEXT NOT NULL, category TEXT, importance INTEGER DEFAULT 1, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_accessed TIMESTAMP DEFAULT CURRENT_TIMESTAMP, access_count INTEGER DEFAULT 0 ) ) self.conn.commit() def store(self, key: str, value: Any, category: str general, importance: int 1): 存储长期记忆 serialized_value json.dumps(value) cursor self.conn.cursor() cursor.execute( INSERT OR REPLACE INTO memories (key, value, category, importance, last_accessed, access_count) VALUES (?, ?, ?, ?, ?, COALESCE( (SELECT access_count 1 FROM memories WHERE key ?), 1 )) , (key, serialized_value, category, importance, datetime.now(), key)) self.conn.commit() def retrieve(self, key: str) - Any: 检索长期记忆 cursor self.conn.cursor() cursor.execute( SELECT value, access_count FROM memories WHERE key ? , (key,)) result cursor.fetchone() if not result: return None # 更新访问记录 cursor.execute( UPDATE memories SET access_count ?, last_accessed ? WHERE key ? , (result[1] 1, datetime.now(), key)) self.conn.commit() return json.loads(result[0]) def search_by_category(self, category: str, limit10) - List[Dict]: 按类别搜索记忆 cursor self.conn.cursor() cursor.execute( SELECT key, value, importance FROM memories WHERE category ? ORDER BY importance DESC, last_accessed DESC LIMIT ? , (category, limit)) results [] for row in cursor.fetchall(): results.append({ key: row[0], value: json.loads(row[1]), importance: row[2] }) return results5.2 工具集成框架工具管理系统# tools/tool_manager.py from typing import Dict, Callable, Any import inspect from functools import wraps class ToolManager: def __init__(self): self.tools {} self.tool_descriptions {} def register_tool(self, name: str, description: str ): 装饰器注册工具 def decorator(func: Callable): self.tools[name] func self.tool_descriptions[name] description or func.__doc__ wraps(func) def wrapper(*args, **kwargs): return func(*args, **kwargs) return wrapper return decorator def execute_tool(self, tool_name: str, **kwargs) - Any: 执行工具 if tool_name not in self.tools: raise ValueError(f工具未注册: {tool_name}) tool_func self.tools[tool_name] # 验证参数 sig inspect.signature(tool_func) bound_args sig.bind_partial(**kwargs) bound_args.apply_defaults() return tool_func(**bound_args.arguments) def get_tool_description(self, tool_name: str) - str: 获取工具描述 return self.tool_descriptions.get(tool_name, ) def list_available_tools(self) - Dict[str, str]: 列出可用工具 return self.tool_descriptions.copy() # 工具使用示例 tool_manager ToolManager() tool_manager.register_tool( namecode_analysis, description分析代码质量和潜在问题 ) def analyze_code(code: str, language: str python) - Dict: 分析给定代码的质量和潜在问题 # 实现代码分析逻辑 return { quality_score: 85, issues: [缺少异常处理, 魔法数字], suggestions: [添加输入验证, 使用常量定义] } tool_manager.register_tool( namesecurity_scan, description扫描代码安全漏洞 ) def security_scan(code: str) - Dict: 扫描代码中的安全漏洞 # 实现安全扫描逻辑 return { vulnerabilities: [SQL注入风险, 硬编码密码], risk_level: medium }6. 常见问题与解决方案6.1 循环失控问题问题现象Agent陷入无限循环任务执行时间远超预期资源消耗持续增长解决方案# 循环控制机制 class LoopController: def __init__(self, max_iterations100, timeout_seconds300): self.max_iterations max_iterations self.timeout_seconds timeout_seconds self.iteration_count 0 self.start_time None def should_continue(self) - bool: 检查是否应该继续循环 if self.start_time is None: self.start_time time.time() self.iteration_count 1 # 检查迭代次数限制 if self.iteration_count self.max_iterations: return False # 检查超时限制 if time.time() - self.start_time self.timeout_seconds: return False return True def get_progress(self) - Dict: 获取循环进度 elapsed time.time() - self.start_time if self.start_time else 0 return { iterations: self.iteration_count, elapsed_seconds: elapsed, progress_percentage: min(100, (self.iteration_count / self.max_iterations) * 100) }6.2 记忆管理问题问题现象上下文窗口溢出重要信息被遗忘记忆检索效率低解决方案# 智能记忆管理 class SmartMemoryManager: def __init__(self, short_term_memory, long_term_memory): self.stm short_term_memory self.ltm long_term_memory self.compression_threshold 0.8 # 压缩阈值 def store_important(self, key: str, value: Any, importance: int 1): 存储重要信息 # 短期记忆和长期记忆都存储 self.stm.store(key, value) self.ltm.store(key, value, importanceimportance) def compress_memory(self, current_context: str) - str: 压缩记忆以节省上下文空间 if len(current_context) 1000: # 简单的长度检查 return current_context # 实现记忆压缩逻辑 compressed self._summarize_context(current_context) return compressed def _summarize_context(self, context: str) - str: 总结上下文内容 # 使用LLM进行内容总结 # 这里是简化实现 if len(context) 500: return context[:250] ...[已压缩]... context[-250:] return context6.3 工具调用失败处理问题现象外部API调用失败工具返回异常结果依赖服务不可用解决方案# 容错工具调用 class ResilientToolExecutor: def __init__(self, max_retries3, backoff_factor1.0): self.max_retries max_retries self.backoff_factor backoff_factor def execute_with_retry(self, tool_func, *args, **kwargs): 带重试的工具执行 last_exception None for attempt in range(self.max_retries 1): try: result tool_func(*args, **kwargs) return result except Exception as e: last_exception e if attempt self.max_retries: wait_time self.backoff_factor * (2 ** attempt) print(f工具调用失败{wait_time}秒后重试...) time.sleep(wait_time) else: print(f工具调用最终失败: {e}) # 所有重试都失败后的备选方案 return self._fallback_behavior(*args, **kwargs) def _fallback_behavior(self, *args, **kwargs): 失败后的备选行为 return { status: error, message: 工具暂时不可用, fallback_data: 使用缓存数据或简化逻辑 }7. 性能优化与最佳实践7.1 循环效率优化并行执行优化import asyncio from concurrent.futures import ThreadPoolExecutor class ParallelLoopExecutor: def __init__(self, max_workersNone): self.max_workers max_workers or min(32, (os.cpu_count() or 1) 4) async def execute_parallel(self, tasks: List[Callable]): 并行执行多个任务 loop asyncio.get_event_loop() with ThreadPoolExecutor(max_workersself.max_workers) as executor: futures [ loop.run_in_executor(executor, task) for task in tasks ] results await asyncio.gather(*futures, return_exceptionsTrue) return results def batch_process(self, items: List, batch_size: int, process_func: Callable): 批量处理数据 results [] for i in range(0, len(items), batch_size): batch items[i:i batch_size] batch_tasks [lambda xitem: process_func(x) for item in batch] batch_results asyncio.run(self.execute_parallel(batch_tasks)) results.extend(batch_results) return results缓存策略优化from functools import lru_cache from datetime import datetime, timedelta class TimedCache: def __init__(self, ttl_seconds300): self.ttl timedelta(secondsttl_seconds) self._cache {} def get(self, key): 获取缓存值 if key in self._cache: value, timestamp self._cache[key] if datetime.now() - timestamp self.ttl: return value else: del self._cache[key] # 过期删除 return None def set(self, key, value): 设置缓存值 self._cache[key] (value, datetime.now()) def clear_expired(self): 清理过期缓存 current_time datetime.now() expired_keys [ key for key, (_, timestamp) in self._cache.items() if current_time - timestamp self.ttl ] for key in expired_keys: del self._cache[key] # 使用示例 lru_cache(maxsize100) def expensive_computation(x: int) - int: print(f计算 {x}...) return x * x # 模拟耗时计算7.2 资源管理最佳实践内存使用监控import psutil import resource class ResourceMonitor: def __init__(self, memory_limit_mb1024): self.memory_limit memory_limit_mb * 1024 * 1024 # 转换为字节 self.initial_memory self.get_memory_usage() def get_memory_usage(self) - int: 获取当前内存使用量 process psutil.Process() return process.memory_info().rss def check_memory_limit(self) - bool: 检查是否超过内存限制 current_usage self.get_memory_usage() return current_usage self.memory_limit def get_memory_info(self) - Dict: 获取详细内存信息 process psutil.Process() memory_info process.memory_info() return { rss_mb: memory_info.rss / 1024 / 1024, vms_mb: memory_info.vms / 1024 / 1024, percent: process.memory_percent(), limit_mb: self.memory_limit / 1024 / 1024 }优雅降级策略class GracefulDegradation: def __init__(self, primary_strategy, fallback_strategies: List[Callable]): self.primary primary_strategy self.fallbacks fallback_strategies self.current_strategy_index 0 def execute(self, *args, **kwargs): 执行策略失败时自动降级 strategies [self.primary] self.fallbacks for i, strategy in enumerate(strategies[self.current_strategy_index:], self.current_strategy_index): try: result strategy(*args, **kwargs) self.current_strategy_index i # 更新当前策略索引 return result except Exception as e: print(f策略 {i} 失败: {e}) if i len(strategies) - 1: # 最后一个策略也失败 raise raise RuntimeError(所有策略都失败了)7.3 监控与可观测性循环执行监控import time from dataclasses import dataclass from typing import Dict, Any dataclass class LoopMetrics: iteration_count: int 0 success_count: int 0 failure_count: int 0 total_duration: float 0.0 last_iteration_duration: float 0.0 class LoopMonitor: def __init__(self): self.metrics LoopMetrics() self.iteration_start_time None def start_iteration(self