引言 Agent 在执行任务时可能遇到各种错误：LLM 超时、工具返回异常、网络中断、格式解析失败。传统软件的错误处理是确定性的——写好 if-else 分支即可。Agent 的错误处理需要应对非确定性的模型输出和动态环境。2026年，“自修复"Agent 已从概念走向实践。
一、Agent 错误分类体系 ┌──────────────────────────────────────────────────────────┐ │ Agent 错误分类 │ ├──────────────┬────────────┬──────────────────────────────┤ │ 错误类型 │ 可重试性 │ 恢复策略 │ ├──────────────┼────────────┼──────────────────────────────┤ │ LLM 超时 │ 可重试 │ 指数退避 + 模型降级 │ │ LLM 限流 │ 可重试 │ 令牌桶等待 + 队列排队 │ │ LLM 格式错误 │ 可重试 │ 格式修正 + 结构化输出重试 │ │ 工具超时 │ 看情况 │ 重试 + 备用工具 + 降级 │ │ 工具异常 │ 看情况 │ 参数修正 + 自修复 │ │ 网络中断 │ 可重试 │ 自动重连 + 状态恢复 │ │ 上下文溢出 │ 不可重试 │ 上下文压缩 + 分段处理 │ │ 幻觉/错误输出│ 需判断 │ 自我验证 + 重新推理 │ │ 死循环 │ 不可重试 │ 迭代上限 + 强制终止 │ │ 权限拒绝 │ 不可重试 │ 降级 + 人工介入 │ └──────────────┴────────────┴──────────────────────────────┘ 二、基础层：智能重试 2.1 分级重试策略 from enum import Enum import asyncio import random class RetryStrategy(Enum): FIXED = "fixed" EXPONENTIAL = "exponential" LINEAR = "linear" JITTERED = "jittered" class SmartRetryPolicy: """智能重试策略""" STRATEGIES = { "timeout": RetryPolicy( max_attempts=3, strategy=RetryStrategy.EXPONENTIAL, base_delay=1.0, max_delay=30.0, jitter=True ), "rate_limit": RetryPolicy( max_attempts=5, strategy=RetryStrategy.LINEAR, base_delay=5.0, max_delay=60.0, jitter=False # 限流重试不需要抖动 ), "connection": RetryPolicy( max_attempts=5, strategy=RetryStrategy.EXPONENTIAL, base_delay=0.5, max_delay=10.0, jitter=True ), "format_error": RetryPolicy( max_attempts=2, # 格式错误重试意义有限 strategy=RetryStrategy.FIXED, base_delay=0.0, max_delay=0.0, pre_action="repair_prompt" # 重试前修复 Prompt ), } def get_policy(self, error: Exception) -> RetryPolicy: if isinstance(error, TimeoutError): return self.STRATEGIES["timeout"] elif isinstance(error, RateLimitError): return self.STRATEGIES["rate_limit"] elif isinstance(error, ConnectionError): return self.STRATEGIES["connection"] elif isinstance(error, FormatError): return self.STRATEGIES["format_error"] else: return RetryPolicy(max_attempts=1) # 不重试 def compute_delay(self, attempt: int, policy: RetryPolicy) -> float: if policy.strategy == RetryStrategy.FIXED: delay = policy.base_delay elif policy.strategy == RetryStrategy.LINEAR: delay = policy.base_delay * attempt elif policy.strategy == RetryStrategy.EXPONENTIAL: delay = policy.base_delay * (2 ** attempt) delay = min(delay, policy.max_delay) if policy.jitter: delay += random.uniform(0, delay * 0.1) return delay async def with_retry( func: callable, error_handler: callable = None, policies: SmartRetryPolicy = None ): """带智能重试的函数调用""" policies = policies or SmartRetryPolicy() last_error = None for attempt in range(10): # 安全上限 try: return await func() except Exception as e: last_error = e policy = policies.get_policy(e) if attempt >= policy.max_attempts: raise # 重试前操作（如修复 Prompt） if policy.pre_action == "repair_prompt": if error_handler: await error_handler(e, attempt) delay = policies.compute_delay(attempt, policy) logger.warning( f"Attempt {attempt+1} failed: {e}. " f"Retrying in {delay:.1f}s..." ) await asyncio.sleep(delay) raise last_error 2.2 格式错误自修复 class FormatRepairAgent: """格式错误自动修复""" REPAIR_STRATEGIES = [ "json_extract", # 从文本中提取 JSON "json_repair", # 修复常见 JSON 错误 "re_prompt", # 重新请求 LLM "structured_output", # 切换到结构化输出 ] async def repair(self, raw_output: str, expected_schema: dict) -> dict: """尝试修复格式错误""" for strategy in self.REPAIR_STRATEGIES: try: result = await getattr(self, f"_strategy_{strategy}")( raw_output, expected_schema ) if result is not None: logger.info(f"Format repaired using: {strategy}") return result except Exception: continue raise FormatRepairError( f"Could not repair output after all strategies" ) async def _strategy_json_extract(self, raw: str, schema: dict) -> dict | None: """从文本中提取 JSON""" # 尝试找到 JSON 块 import re patterns = [ r'```json\s*(.*?)\s*```', # Markdown JSON 块 r'```\s*(.*?)\s*```', # 通用代码块 r'\{[^{}]*\}', # 裸 JSON 对象 r'\[.*\]', # 裸 JSON 数组 ] for pattern in patterns: match = re.search(pattern, raw, re.DOTALL) if match: try: return json.loads(match.group(1)) except json.JSONDecodeError: continue return None async def _strategy_json_repair(self, raw: str, schema: dict) -> dict | None: """修复常见 JSON 错误""" repaired = raw # 修复尾随逗号 repaired = re.sub(r',\s*}', '}', repaired) repaired = re.sub(r',\s*]', ']', repaired) # 修复单引号 repaired = repaired.replace("'", '"') # 修复未引用的键 repaired = re.sub(r'(\w+):', r'"\1":', repaired) # 修复省略号 repaired = repaired.replace('...', 'null') try: return json.loads(repaired) except json.JSONDecodeError: return None async def _strategy_re_prompt(self, raw: str, schema: dict) -> dict | None: """使用 LLM 修复格式""" repair_prompt = f"""The following text was supposed to be valid JSON matching this schema: {json.dumps(schema, indent=2)} But it had format errors. Fix it and return ONLY valid JSON: Original text: {raw[:2000]} Return the corrected JSON:""" response = await llm.invoke( repair_prompt, response_format={"type": "json_object"}, temperature=0.0 ) try: return json.loads(response.content) except: return None async def _strategy_structured_output(self, raw: str, schema: dict) -> dict | None: """使用结构化输出 API 重新请求""" response = await llm.invoke( f"Convert this to structured data:\n{raw[:2000]}", response_format={"type": "json_schema", "json_schema": schema}, temperature=0.0 ) return json.loads(response.content) if response.content else None 三、中间层：工具错误恢复 3.1 工具执行框架 class ResilientToolExecutor: """带错误恢复的工具执行器""" def __init__(self): self.fallback_chains = {} # {tool_name: [backup_tool1, backup_tool2]} self.error_handlers = {} # {tool_name: handler} async def execute( self, tool_call: ToolCall, context: ExecutionContext ) -> ToolResult: tool_name = tool_call.tool primary = self._get_tool(tool_name) try: result = await with_retry( lambda: primary.execute(tool_call.args, context), policies=self._get_retry_policy(tool_name) ) return ToolResult(success=True, data=result) except Exception as e: # 尝试错误处理器 handler = self.error_handlers.get(tool_name) if handler: repaired_call = await handler(e, tool_call, context) if repaired_call: return await self.execute(repaired_call, context) # 尝试备用工具链 for backup_name in self.fallback_chains.get(tool_name, []): backup = self._get_tool(backup_name) try: adapted_args = await self._adapt_args( tool_call.args, primary, backup ) result = await backup.execute(adapted_args, context) return ToolResult( success=True, data=result, degraded=True, used_fallback=backup_name ) except Exception: continue return ToolResult( success=False, error=str(e), tool=tool_name ) # 注册备用工具链 executor = ResilientToolExecutor() executor.fallback_chains = { "web_search": ["bing_search", "duckduckgo_search"], "database_query": ["cache_lookup", "default_response"], "send_email": ["queue_email", "save_to_retry"], } # 注册错误处理器 async def search_error_handler( error: Exception, tool_call: ToolCall, context: ExecutionContext ) -> ToolCall | None: """搜索工具错误处理器""" if isinstance(error, TimeoutError): # 减少结果数量重试 modified_args = tool_call.args.copy() modified_args["max_results"] = 3 # 减少结果数量 return ToolCall(tool=tool_call.tool, args=modified_args) elif isinstance(error, RateLimitError): return None # 不重试，直接走备用 return None executor.error_handlers["web_search"] = search_error_handler 3.2 LLM 输出验证与自修复 class OutputValidator: """LLM 输出验证与自修复""" async def validate_and_repair( self, output: str, task: str, constraints: list[str] ) -> ValidatedOutput: # 1. 结构验证 structural = self._check_structure(output, task) if not structural.valid: repaired = await self._repair_structure(output, structural.errors) if repaired: output = repaired # 2. 约束验证 constraint_results = await self._check_constraints(output, constraints) violated = [r for r in constraint_results if not r.passed] if not violated: return ValidatedOutput(valid=True, output=output) # 3. 自修复：让 LLM 自己修正 if self._is_repairable(violated): repaired_output = await self._self_repair( output, task, violated ) if repaired_output: # 重新验证 recheck = await self._check_constraints(repaired_output, constraints) if all(r.passed for r in recheck): return ValidatedOutput( valid=True, output=repaired_output, repaired=True ) return ValidatedOutput( valid=False, output=output, violations=[r.constraint for r in violated] ) async def _self_repair( self, original_output: str, task: str, violations: list[ConstraintViolation] ) -> str | None: """让 LLM 自我修复输出""" violation_desc = "\n".join( f"- {v.constraint}: {v.detail}" for v in violations ) repair_prompt = f"""Your previous response has issues that need to be fixed. Original task: {task} Your response: {original_output[:3000]} Issues found: {violation_desc} Please fix these issues and provide the corrected response. Only output the corrected response, no explanations.""" try: response = await llm.invoke(repair_prompt, temperature=0.0) return response.content except: return None 四、高级层：自修复 Agent class SelfHealingAgent: """自修复 Agent：能识别错误并自主修复""" async def run_with_healing( self, task: str, max_heal_attempts: int = 3 ) -> str: for attempt in range(max_heal_attempts): try: result = await self._execute(task) # 自我验证 validation = await self._self_validate(task, result) if validation.confidence > 0.8: return result # 识别问题 diagnosis = await self._diagnose( task, result, validation ) # 生成修复方案 fix_plan = await self._generate_fix_plan(diagnosis) # 执行修复 task = await self._apply_fix(task, fix_plan) logger.info( f"Self-healing attempt {attempt+1}: " f"diagnosis={diagnosis.issue}, " f"fix={fix_plan.action}" ) except CircularError as e: # 检测到循环，换策略 task = await self._reframe_task(task, e) except MaxIterationsError: # 达到迭代上限，简化任务 task = await self._simplify_task(task) # 自修复失败，返回最佳结果 return result or "I was unable to complete this task." async def _diagnose( self, task: str, result: str, validation: Validation ) -> Diagnosis: """诊断输出问题""" diagnosis_prompt = f"""Analyze why the following result may be incorrect: Task: {task} Result: {result[:2000]} Validation concerns: {validation.concerns} Identify the most likely issue: - wrong_approach: Used wrong method to solve the problem - incomplete: Missing important information or steps - hallucination: Contains fabricated information - format_error: Output format doesn't match requirements - logical_error: Reasoning contains logical flaws - other: Specify Respond in JSON: {{"issue": "...", "detail": "...", "confidence": 0.0-1.0}}""" response = await llm.invoke(diagnosis_prompt, temperature=0.0) return Diagnosis(**json.loads(response.content)) async def _generate_fix_plan(self, diagnosis: Diagnosis) -> FixPlan: """生成修复计划""" FIX_STRATEGIES = { "wrong_approach": "Rethink the approach and try a different method", "incomplete": "Identify missing information and supplement it", "hallucination": "Verify all claims using tools and remove unverified ones", "format_error": "Reformat the output to match requirements", "logical_error": "Review the reasoning chain step by step", } action = FIX_STRATEGIES.get(diagnosis.issue, "Retry with more careful approach") return FixPlan( action=action, issue=diagnosis.issue, detail=diagnosis.detail, strategy="adjust_and_retry" ) 五、错误恢复编排 class ErrorRecoveryOrchestrator: """错误恢复编排器""" async def execute_with_recovery( self, task: str, agent: Agent ) -> RecoveryResult: recovery_layers = [ ("retry", RetryLayer()), ("format_repair", FormatRepairLayer()), ("tool_fallback", ToolFallbackLayer()), ("output_validation", OutputValidationLayer()), ("self_healing", SelfHealingLayer()), ("human_escalation", HumanEscalationLayer()), ] current_task = task current_output = None for layer_name, layer in recovery_layers: try: result = await layer.process( current_task, current_output, agent ) if result.success: return RecoveryResult( output=result.output, recovery_layers_used=[layer_name], success=True ) # 更新任务或输出，传给下一层 if result.modified_task: current_task = result.modified_task if result.partial_output: current_output = result.partial_output except Exception as e: logger.error(f"Recovery layer {layer_name} failed: {e}") continue return RecoveryResult( output=current_output or "Unable to complete task", success=False, recovery_layers_used=["all_failed"] ) 六、错误恢复 Checklist □ 错误分类体系明确（可重试/不可重试/需判断） □ 重试策略匹配错误类型（指数退避/线性/固定） □ 格式错误自动修复（JSON提取/修复/重新请求） □ 工具备用链配置（主工具失败→备用工具） □ LLM 输出验证+自我修复 □ 自修复 Agent 能诊断并修复自身错误 □ 循环检测器防止自修复死循环 □ 人工升级机制（自修复失败时触发） □ 错误恢复全链路追踪 □ 定期演练错误恢复流程 结语 错误恢复是 Agent 可靠性的最后一道防线。从简单的重试到复杂的自修复，每一层恢复策略都在增加系统的韧性。但要注意：自修复不是万能的——它消耗额外 Token、增加延迟、可能引入新错误。最佳策略是分层防御：基础错误用重试解决，格式错误用修复解决，逻辑错误用自修复解决，复杂错误交给人工。让 Agent 像人一样——犯错后能自我纠正，但也知道什么时候该求助。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
...
