Loop Engineering：构建自主循环AI智能体的系统级工程方法 在实际 AI 应用开发中很多开发者已经熟练掌握了 Prompt Engineering能够通过精心设计的提示词让大模型完成单次问答或简单任务。但当任务变得复杂、需要多步骤执行、中间验证和持续迭代时单纯依赖人工一次次发送提示词就显得效率低下且不可靠。这正是 Loop Engineering 要解决的核心问题如何设计一个能自主、可靠、持续运行的 AI 智能体系统而不仅仅是优化单次交互。Loop Engineering 是一种系统级的 AI 工程范式它整合了上下文管理、工具调度、状态维护、验证反馈和循环控制等多个维度让 AI 智能体能够在无需人工持续干预的情况下完成从目标分解到最终达成的全过程。本文将带你从零理解 Loop Engineering 的基础概念、历史演进、核心构建块并通过一个可运行的案例实战掌握构建自主循环 AI 智能体的关键方法。1. Loop Engineering 基础概念与历史演进1.1 什么是 Loop EngineeringLoop Engineering 的核心是设计、构建和优化 AI 智能体的循环执行架构。它不是一个具体的技术或框架而是一套工程方法论关注如何让智能体在复杂任务中实现“感知-决策-行动-验证-迭代”的闭环运行。与 Prompt Engineering 相比Loop Engineering 的焦点发生了根本转变Prompt Engineering关注单次交互中如何让模型理解指令并产生预期输出。Loop Engineering关注多步任务中如何设计系统架构使智能体能够自主管理任务状态、调用工具、检查结果并决定下一步行动。通俗来说Prompt Engineering 是教模型“如何回答一个问题”而 Loop Engineering 是教系统“如何完成一个项目”。1.2 从 Prompt Engineering 到 Loop Engineering 的演进AI 应用开发范式的演进经历了三个阶段第一阶段单次提示词优化2020-2023典型技术Few-shot Learning、Chain-of-Thought、结构化输出适用场景问答、摘要、简单分类等单次交互任务局限性无法处理需要多步执行和状态维护的复杂任务第二阶段线性链式执行2023-2024典型框架LangChain、LlamaIndex 等进步实现了多步骤任务的串联执行局限性缺乏循环、分支判断和自我修正能力第三阶段循环工程范式2024-至今典型框架LangGraph、AutoGPT、CrewAI 等核心特征支持条件分支、循环迭代、多智能体协作适用场景代码生成与调试、研究分析、业务流程自动化等复杂任务这种演进背后的驱动因素是实际业务需求的复杂化。当 AI 需要处理“开发一个完整应用”而不是“写一个函数”时线性链式结构就无法满足需求了。1.3 Loop Engineering 解决的核心问题Loop Engineering 主要解决传统 AI 应用开发中的三大瓶颈人类注意力瓶颈复杂任务需要多步执行和中间验证如果每一步都需要人工确认和提供下一步指令开发效率极低。Loop Engineering 通过自动化循环解放了开发者的持续注意力。任务可靠性问题单次模型输出可能包含错误或不足缺乏自我检查和修正机制。循环架构允许智能体基于行动结果进行反思和调整提高最终输出的质量。状态维护困难长对话或多步骤任务需要记忆历史状态和中间结果。Loop Engineering 通过明确的状态管理机制确保智能体在整个执行过程中保持上下文一致性。2. Loop Engineering 五大核心构建块解析一个完整的 Loop Engineering 系统包含五个核心构建块它们共同构成了智能体的自主循环能力。2.1 目标与规划器目标与规划器是循环的起点负责将模糊的最终目标转化为清晰、可执行的子任务序列。关键技术要素目标明确化将“开发一个网站”转化为“设计数据库 schema、实现后端 API、开发前端界面、部署测试”任务分解算法基于大模型的规划能力或专用规划算法如 Plan-and-Execute依赖关系管理识别任务间的先后依赖确保执行顺序合理# 目标规划示例代码结构 class GoalPlanner: def __init__(self, llm): self.llm llm def decompose_goal(self, final_goal): 将最终目标分解为子任务列表 prompt f 请将以下目标分解为具体的执行步骤 目标{final_goal} 要求 1. 每个步骤应该是原子性的可执行任务 2. 标明步骤间的依赖关系 3. 输出格式为JSON列表 response self.llm.invoke(prompt) return self._parse_response(response) def _parse_response(self, response): # 解析LLM响应提取结构化任务列表 # 返回格式[{step: 1, task: 具体任务, dependencies: []}, ...] pass常见问题与解决方案问题规划过于理想化忽略实际执行约束解决方案在规划阶段加入可行性评估或者采用动态规划策略2.2 状态与记忆管理状态管理维护智能体执行过程中的所有上下文信息为决策提供基础。记忆类型与实现方式记忆类型存储内容技术实现适用场景短期记忆当前会话的完整历史内存中的列表或队列单次任务执行长期记忆跨会话的重要信息向量数据库关系型数据库用户偏好、学习经验工作记忆当前任务的相关上下文状态对象实时更新循环中的决策依据# 状态管理示例 class AgentState: def __init__(self): self.current_goal None self.completed_tasks [] self.pending_tasks [] self.execution_history [] # 记录每个步骤的输入输出 self.context {} # 额外上下文信息 def update_after_action(self, task, action, result, success): 执行动作后更新状态 self.execution_history.append({ task: task, action: action, result: result, success: success, timestamp: datetime.now() }) if success: self.completed_tasks.append(task) if task in self.pending_tasks: self.pending_tasks.remove(task)最佳实践定期清理无关历史避免上下文过长对重要决策点保存快照便于回溯分析实现状态序列化支持断点续执行2.3 行动与工具调度行动模块负责根据当前状态和目标选择并调用合适的工具来执行具体操作。工具分类与调用机制# 工具调度器示例 class ToolScheduler: def __init__(self): self.tools { web_search: WebSearchTool(), code_executor: CodeExecutor(), file_editor: FileEditor(), api_caller: APICaller() } def select_tool(self, task_description, current_context): 根据任务描述选择最合适的工具 # 基于任务类型、工具能力匹配度进行选择 tool_selection_prompt f 任务{task_description} 可用工具{list(self.tools.keys())} 请选择最合适的工具只返回工具名称。 selected_tool self.llm.invoke(tool_selection_prompt) return self.tools.get(selected_tool.strip()) def execute_action(self, tool_name, parameters): 执行工具调用 tool self.tools[tool_name] try: result tool.execute(parameters) return {success: True, result: result} except Exception as e: return {success: False, error: str(e)}工具设计原则每个工具功能单一、接口明确工具应该具备错误处理和重试机制重要工具操作应该记录审计日志2.4 验证与反馈机制验证机制对行动结果进行质量检查判断是否满足预期并生成改进反馈。验证层次设计class ValidationSystem: def __init__(self): self.validators { syntax_check: SyntaxValidator(), test_execution: TestExecutor(), model_self_eval: ModelSelfEval(), pattern_match: PatternValidator() } def validate_result(self, task, action, result, validation_typeauto): 多维度验证结果质量 if validation_type auto: # 根据任务类型自动选择验证策略 if code in task: validators [syntax_check, test_execution] else: validators [model_self_eval, pattern_match] else: validators [validation_type] validation_results {} for validator_name in validators: validator self.validators[validator_name] score, feedback validator.validate(task, action, result) validation_results[validator_name] { score: score, feedback: feedback } return self._aggregate_validation(validation_results) def should_continue(self, validation_results, iteration_count): 基于验证结果决定是否继续循环 overall_score validation_results[overall_score] # 达到质量阈值或超过最大迭代次数时停止 if overall_score 0.8: return False, 质量达标 elif iteration_count 10: return False, 达到最大迭代次数 else: return True, 需要继续改进验证策略选择外部验证器代码测试、编译检查、API 响应验证等模型自评让大模型对自身输出进行批判性评估规则验证基于预定义规则进行模式匹配2.5 编排与循环控制编排引擎管理整个工作流的执行顺序、条件分支和循环逻辑是 Loop Engineering 的核心控制器。基于有向图的编排实现from langgraph.graph import StateGraph, END class LoopOrchestrator: def __init__(self, planner, tool_scheduler, validator): self.planner planner self.tool_scheduler tool_scheduler self.validator validator # 构建状态图 self.workflow StateGraph(AgentState) self._build_workflow() def _build_workflow(self): # 定义工作流节点 self.workflow.add_node(plan, self._plan_node) self.workflow.add_node(execute, self._execute_node) self.workflow.add_node(validate, self._validate_node) self.workflow.add_node(reflect, self._reflect_node) # 定义边和条件转移 self.workflow.set_entry_point(plan) self.workflow.add_edge(plan, execute) self.workflow.add_edge(execute, validate) # 条件分支验证结果决定下一步 self.workflow.add_conditional_edges( validate, self._should_reflect_or_end, {reflect: reflect, end: END} ) self.workflow.add_edge(reflect, execute) self.graph self.workflow.compile() def _should_reflect_or_end(self, state): 根据验证结果决定下一步 validation_result state.get(validation_result, {}) iteration_count state.get(iteration_count, 0) should_continue, reason self.validator.should_continue( validation_result, iteration_count ) if should_continue: return reflect # 需要反思改进 else: return end # 任务完成或终止 def run(self, initial_goal): 启动循环执行 initial_state AgentState() initial_state.current_goal initial_goal final_state self.graph.invoke(initial_state) return final_state循环控制策略最大迭代次数防止无限循环超时机制控制总执行时间质量阈值达到预期质量时提前终止人工中断支持外部干预停止3. Loop Engineering 实战案例自动化代码生成与测试下面通过一个完整的实战案例演示如何构建一个能够自主完成代码开发任务的 AI 智能体。3.1 案例需求与环境准备任务目标开发一个 Python 函数接收数字列表作为输入返回排序后的列表并确保包含完整的单元测试。技术栈准备Python 3.8LangGraph 或类似编排框架大模型 APIOpenAI GPT-4、Claude 3 等代码执行环境隔离的 Docker 容器项目结构loop_engineering_demo/ ├── agents/ │ ├── planner.py # 目标规划器 │ ├── executor.py # 代码执行器 │ └── validator.py # 验证器 ├── tools/ │ ├── code_generator.py # 代码生成工具 │ ├── test_runner.py # 测试运行工具 │ └── code_editor.py # 代码编辑工具 ├── state/ │ └── agent_state.py # 状态管理 ├── orchestration/ │ └── workflow.py # 工作流编排 └── main.py # 入口文件3.2 核心组件实现状态管理实现# state/agent_state.py from typing import Dict, List, Any, Optional from dataclasses import dataclass, field from datetime import datetime dataclass class CodeGenState: 代码生成任务的状态管理 original_requirement: str current_plan: List[Dict] field(default_factorylist) generated_code: str test_code: str execution_results: List[Dict] field(default_factorylist) validation_results: Dict[str, Any] field(default_factorydict) iteration_count: int 0 error_messages: List[str] field(default_factorylist) start_time: datetime field(default_factorydatetime.now) def to_dict(self) - Dict: 转换为字典格式便于序列化 return { original_requirement: self.original_requirement, current_plan: self.current_plan, generated_code: self.generated_code, test_code: self.test_code, iteration_count: self.iteration_count, error_messages: self.error_messages }代码生成工具实现# tools/code_generator.py import openai from typing import Dict class CodeGenerator: def __init__(self, api_key: str, model: str gpt-4): self.client openai.OpenAI(api_keyapi_key) self.model model def generate_function(self, requirement: str, context: Dict None) - Dict: 根据需求生成Python函数代码 prompt self._build_prompt(requirement, context) try: response self.client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], temperature0.2 # 低温度确保代码稳定性 ) generated_code response.choices[0].message.content return { success: True, code: self._extract_code(generated_code), raw_response: generated_code } except Exception as e: return { success: False, error: str(e), code: } def _build_prompt(self, requirement: str, context: Dict) - str: 构建代码生成提示词 base_prompt f 请编写一个Python函数满足以下需求 {requirement} 要求 1. 只返回函数代码不要额外解释 2. 包含完整的函数签名和类型注解 3. 代码要简洁高效 4. 如果有边界情况请妥善处理 if context and context.get(previous_errors): base_prompt f\n\n之前出现的错误{context[previous_errors]} return base_prompt def _extract_code(self, text: str) - str: 从模型响应中提取代码块 if python in text: # 提取代码块 start text.find(python) 9 end text.find(, start) return text[start:end].strip() elif in text: # 处理其他代码块格式 start text.find() 3 end text.find(, start) code text[start:end].strip() if code.startswith(python): code code[6:].strip() return code else: # 如果没有代码块标记返回整个文本 return text.strip()测试验证器实现# tools/test_runner.py import subprocess import tempfile import os from typing import Dict, List class TestRunner: def __init__(self, timeout: int 30): self.timeout timeout def run_tests(self, code: str, test_code: str) - Dict: 在隔离环境中运行代码测试 with tempfile.TemporaryDirectory() as temp_dir: # 创建主代码文件 main_file os.path.join(temp_dir, main.py) with open(main_file, w) as f: f.write(code \n\n) # 创建测试文件 test_file os.path.join(temp_dir, test_main.py) with open(test_file, w) as f: f.write(import main\n) f.write(test_code) try: # 运行测试 result subprocess.run( [python, -m, pytest, test_file, -v], capture_outputTrue, textTrue, timeoutself.timeout, cwdtemp_dir ) return { success: result.returncode 0, returncode: result.returncode, stdout: result.stdout, stderr: result.stderr, passed: passed in result.stdout.lower() } except subprocess.TimeoutExpired: return { success: False, error: 测试执行超时, returncode: -1 } except Exception as e: return { success: False, error: str(e), returncode: -1 }3.3 工作流编排与循环执行完整工作流实现# orchestration/workflow.py from langgraph.graph import StateGraph, END from typing import Literal from state.agent_state import CodeGenState from tools.code_generator import CodeGenerator from tools.test_runner import TestRunner class CodeGenWorkflow: def __init__(self, openai_api_key: str): self.code_gen CodeGenerator(openai_api_key) self.test_runner TestRunner() self.workflow self._build_workflow() def _build_workflow(self) - StateGraph: workflow StateGraph(CodeGenState) # 添加节点 workflow.add_node(generate_code, self._generate_code_node) workflow.add_node(generate_tests, self._generate_tests_node) workflow.add_node(run_tests, self._run_tests_node) workflow.add_node(analyze_results, self._analyze_results_node) workflow.add_node(refine_code, self._refine_code_node) # 设置工作流路径 workflow.set_entry_point(generate_code) workflow.add_edge(generate_code, generate_tests) workflow.add_edge(generate_tests, run_tests) workflow.add_edge(run_tests, analyze_results) # 条件分支根据测试结果决定下一步 workflow.add_conditional_edges( analyze_results, self._should_refine_or_end, {refine: refine_code, end: END} ) workflow.add_edge(refine_code, generate_tests) # 重新生成测试 return workflow.compile() def _generate_code_node(self, state: CodeGenState) - CodeGenState: 生成主代码 print(f第{state.iteration_count 1}次迭代生成代码) context {} if state.error_messages: context[previous_errors] state.error_messages[-1] # 最近错误 result self.code_gen.generate_function( state.original_requirement, context ) if result[success]: state.generated_code result[code] state.error_messages.clear() else: state.error_messages.append(f代码生成失败{result[error]}) return state def _generate_tests_node(self, state: CodeGenState) - CodeGenState: 生成测试代码 prompt f 为以下Python函数编写完整的单元测试 {state.generated_code} 要求 1. 使用pytest框架 2. 覆盖正常情况、边界情况和异常情况 3. 测试函数名称以test_开头 4. 断言要明确具体 result self.code_gen.generate_function(prompt) if result[success]: state.test_code result[code] return state def _run_tests_node(self, state: CodeGenState) - CodeGenState: 运行测试 test_result self.test_runner.run_tests( state.generated_code, state.test_code ) state.execution_results.append(test_result) return state def _analyze_results_node(self, state: CodeGenState) - CodeGenState: 分析测试结果 latest_result state.execution_results[-1] if latest_result[success] and latest_result.get(passed, False): state.validation_results { status: success, message: 所有测试通过 } else: state.validation_results { status: failed, message: latest_result.get(stderr, 测试失败), details: latest_result } state.error_messages.append(f测试失败{latest_result.get(stderr, 未知错误)}) state.iteration_count 1 return state def _refine_code_node(self, state: CodeGenState) - CodeGenState: 基于错误信息优化代码 print(f基于错误信息优化代码当前迭代{state.iteration_count}) return state def _should_refine_or_end(self, state: CodeGenState) - Literal[refine, end]: 决定继续优化还是结束 if state.iteration_count 5: # 最大迭代次数 return end latest_validation state.validation_results if latest_validation.get(status) success: return end else: return refine def execute(self, requirement: str) - CodeGenState: 执行完整工作流 initial_state CodeGenState(original_requirementrequirement) final_state self.workflow.invoke(initial_state) return final_state3.4 运行验证与结果分析主程序入口# main.py from orchestration.workflow import CodeGenWorkflow import os def main(): # 配置API密钥 api_key os.getenv(OPENAI_API_KEY) if not api_key: print(请设置 OPENAI_API_KEY 环境变量) return # 创建工作流实例 workflow CodeGenWorkflow(api_key) # 定义代码生成需求 requirement 编写一个函数接收一个数字列表作为输入返回排序后的列表。 要求 1. 支持整数和浮点数混合排序 2. 处理空列表和单元素列表的情况 3. 时间复杂度尽可能优化 print(开始执行代码生成循环...) print(f需求{requirement}) print(- * 50) # 执行工作流 final_state workflow.execute(requirement) # 输出最终结果 print(\n * 50) print(任务执行完成) print(f总迭代次数{final_state.iteration_count}) print(f最终代码) print(final_state.generated_code) print(f最终测试结果{final_state.validation_results}) if final_state.error_messages: print(f遇到的错误{final_state.error_messages}) if __name__ __main__: main()预期输出示例开始执行代码生成循环... 需求编写一个函数接收一个数字列表作为输入返回排序后的列表... 第1次迭代生成代码 第2次迭代生成代码 任务执行完成 总迭代次数2 最终代码 def sort_numbers(numbers: List[Union[int, float]]) - List[Union[int, float]]: if not numbers: return [] if len(numbers) 1: return numbers.copy() return sorted(numbers) 最终测试结果{status: success, message: 所有测试通过}4. Loop Engineering 常见问题与排查指南在实际项目中实施 Loop Engineering 时会遇到各种问题下面列出常见问题及解决方案。4.1 循环无法终止问题问题现象智能体陷入无限循环不断重复相似操作但无法达成目标。排查步骤检查终止条件是否设置合理验证评估机制是否能准确判断任务完成分析状态更新逻辑是否正常检查最大迭代次数限制是否生效解决方案# 增强的终止条件检查 def enhanced_termination_check(state, max_iterations10, timeout_minutes5): # 迭代次数检查 if state.iteration_count max_iterations: return True, 达到最大迭代次数 # 超时检查 if (datetime.now() - state.start_time).total_seconds() timeout_minutes * 60: return True, 执行超时 # 质量达标检查 if state.validation_results.get(score, 0) 0.9: return True, 质量达标 # 重复操作检查防止在相似状态循环 recent_states state.execution_history[-3:] # 最近3次状态 if len(recent_states) 3 and self._states_are_similar(recent_states): return True, 检测到重复循环 return False, 继续执行4.2 状态管理混乱问题问题现象智能体忘记之前的关键决策或重复执行已完成的任务。排查步骤检查状态序列化和反序列化逻辑验证关键决策点是否正确记录分析上下文窗口是否过小导致历史丢失检查任务依赖关系管理解决方案# 增强的状态管理 class EnhancedAgentState(CodeGenState): def __init__(self): super().__init__() self.decision_points [] # 关键决策记录 self.completed_task_ids set() # 已完成任务ID self.task_dependencies {} # 任务依赖关系 def mark_task_completed(self, task_id): 标记任务完成并解锁依赖任务 self.completed_task_ids.add(task_id) # 解锁依赖此任务的其他任务 for dependent_task, dependencies in self.task_dependencies.items(): if task_id in dependencies: dependencies.remove(task_id) def get_available_tasks(self): 获取当前可执行的任务依赖已满足 available [] for task_id, dependencies in self.task_dependencies.items(): if task_id not in self.completed_task_ids and not dependencies: available.append(task_id) return available4.3 工具调用失败问题问题现象工具执行频繁失败导致循环中断或结果不可用。排查步骤检查工具输入参数验证验证工具执行环境配置分析错误处理和重试机制检查工具权限和资源限制解决方案# 增强的工具调用器 class RobustToolScheduler: def execute_with_retry(self, tool_name, parameters, max_retries3): 带重试机制的工具执行 last_error None for attempt in range(max_retries): try: # 参数预处理和验证 validated_params self._validate_parameters(tool_name, parameters) # 执行工具 result self.tools[tool_name].execute(validated_params) # 结果验证 if self._validate_result(tool_name, result): return {success: True, result: result, attempts: attempt 1} else: last_error 结果验证失败 except Exception as e: last_error str(e) # 指数退避重试 time.sleep(2 ** attempt) return { success: False, error: last_error, attempts: max_retries }4.4 验证机制不准确问题问题现象验证器无法准确评估任务完成质量导致过早终止或无效继续。排查步骤检查验证标准的明确性和可量化性验证多维度评估的权重设置分析验证器与任务目标的匹配度检查验证结果的一致性和可靠性解决方案# 多维度验证体系 class MultiDimensionValidator: def __init__(self): self.dimensions { functional: FunctionalValidator(), # 功能正确性 performance: PerformanceValidator(), # 性能指标 readability: ReadabilityValidator(), # 代码可读性 completeness: CompletenessValidator() # 任务完成度 } self.weights { functional: 0.4, performance: 0.2, readability: 0.2, completeness: 0.2 } def comprehensive_validate(self, state): 综合多维度验证 scores {} feedbacks {} for dim_name, validator in self.dimensions.items(): score, feedback validator.validate(state) scores[dim_name] score feedbacks[dim_name] feedback # 加权计算总分 overall_score sum(scores[dim] * self.weights[dim] for dim in scores) return { overall_score: overall_score, dimension_scores: scores, feedbacks: feedbacks, recommendation: self._generate_recommendation(scores, feedbacks) }5. Loop Engineering 最佳实践与生产环境建议将 Loop Engineering 从实验环境应用到生产环境需要遵循一系列最佳实践。5.1 设计阶段的最佳实践明确循环边界和停止条件在生产环境中必须明确界定智能体的职责范围和工作边界。避免设计开放式的循环任务每个循环都应该有清晰的完成标准和停止条件。示例代码审查智能体的边界定义# 明确的边界定义 code_review_boundaries { scope: 只审查代码质量不修改代码, max_iterations: 3, timeout: 10分钟, completion_criteria: { all_issues_addressed: True, review_comments_provided: True, quality_score: 0.8 }, escalation_conditions: [ 发现安全漏洞, 代码复杂度超过阈值, 3次迭代未达成共识 ] }设计可观测的循环架构生产环境的 Loop Engineering 系统必须具备完善的可观测性包括日志记录、指标收集和追踪能力。# 可观测性装饰器 def observable_loop(loop_function): def wrapper(*args, **kwargs): start_time time.time() loop_id str(uuid.uuid4()) # 记录循环开始 logger.info(fLoop {loop_id} started, extra{ loop_id: loop_id, timestamp: start_time, args: args }) try: result loop_function(*args, **kwargs) # 记录循环成功完成 duration time.time() - start_time logger.info(fLoop {loop_id} completed successfully, extra{ loop_id: loop_id, duration: duration, result: result }) return result except Exception as e: # 记录循环异常 duration time.time() - start_time logger.error(fLoop {loop_id} failed, extra{ loop_id: loop_id, duration: duration, error: str(e) }) raise return wrapper5.2 实施阶段的最佳实践渐进式复杂度设计不要一开始就设计复杂的多智能体协作系统。从简单的单智能体循环开始逐步增加复杂度。复杂度演进路径单智能体单任务循环ReAct 模式单智能体多任务循环带规划器多智能体协作循环角色分工分层循环架构反思执行实现成本控制机制循环执行意味着多次调用大模型成本可能快速上升。需要实现成本控制机制。class CostController: def __init__(self, budget_limit100, token_limit100000): self.budget_limit budget_limit # 美元 self.token_limit token_limit # Token数 self.current_cost 0 self.token_usage 0 def can_continue(self, estimated_cost0, estimated_tokens0): 检查是否超出成本限制 if self.current_cost estimated_cost self.budget_limit: return False, 超出预算限制 if self.token_usage estimated_tokens self.token_limit: return False, 超出Token限制 return True, 可以继续 def record_usage(self, cost, tokens): 记录资源使用情况 self.current_cost cost self.token_usage tokens5.3 运维阶段的最佳实践建立循环健康度监控生产环境需要实时监控循环系统的健康状态及时发现异常模式。关键监控指标循环执行成功率平均迭代次数分布任务完成时间分布资源消耗趋势错误类型和频率# 健康度检查 class LoopHealthMonitor: def check_