引言

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论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。