为什么 Agent 需要自愈能力
传统软件的错误处理是确定性的：输入 A 触发异常 B，catch 后走分支 C。但 AI Agent 面对的是概率性失败——同一个 prompt 可能这次返回合法 JSON，下次返回一段散文。2026 年的生产 Agent 通常涉及 5-20 次工具调用链，任何一环出错都会导致整条链路失败。
实测数据：一个中等复杂度的 Coding Agent（平均 12 步完成任务），单步成功率 95% 时，端到端成功率只有 0.95^12 ≈ 54%。加入错误恢复后可以拉到 85%+。
这不是”加个 retry”能解决的问题。你需要一套分层容错架构。
错误分类：先诊断再治疗
生产 Agent 的错误可以分为四类，每类需要不同的恢复策略：
| 错误类型 | 典型表现 | 可重试 | 恢复策略 |
|---|---|---|---|
| 瞬时故障 | API 429/503、网络超时 | ✅ | 指数退避重试 |
| 输出畸形 | JSON 解析失败、schema 不匹配 | ✅ | 带错误反馈重新生成 |
| 逻辑错误 | 工具调用参数错误、幻觉实体 | ⚠️ | 回滚 + 换策略重试 |
| 不可恢复 | 权限不足、资源不存在、预算耗尽 | ❌ | 优雅降级 / 人工介入 |
关键原则：不要对所有错误用同一种重试策略。对逻辑错误做简单重试只会重复犯同样的错。
分层恢复架构
┌─────────────────────────────────────┐
│ Layer 4: Circuit Breaker │ 连续失败 → 熔断 → 人工介入
├─────────────────────────────────────┤
│ Layer 3: Strategy Switch │ 同一步失败 2 次 → 换 prompt/工具
├─────────────────────────────────────┤
│ Layer 2: Corrective Retry │ 带错误上下文重新生成
├─────────────────────────────────────┤
│ Layer 1: Simple Retry │ 指数退避，处理瞬时故障
└─────────────────────────────────────┘每一层只处理自己能解决的问题，解决不了就上抛。
实现：Python 分层恢复引擎
import asyncio
import time
from enum import Enum
from dataclasses import dataclass, field
from typing import Any, Callable, Awaitable
class ErrorCategory(Enum):
TRANSIENT = "transient" # 429, 503, timeout
MALFORMED = "malformed" # 输出格式错误
LOGICAL = "logical" # 幻觉、参数错误
FATAL = "fatal" # 不可恢复
@dataclass
class RecoveryContext:
step_name: str
attempt: int = 0
errors: list[str] = field(default_factory=list)
strategy_switches: int = 0
def classify_error(error: Exception) -> ErrorCategory:
"""根据异常类型和内容分类错误"""
msg = str(error).lower()
if any(k in msg for k in ["429", "503", "timeout", "rate_limit"]):
return ErrorCategory.TRANSIENT
if any(k in msg for k in ["json", "parse", "schema", "validation"]):
return ErrorCategory.MALFORMED
if any(k in msg for k in ["not found", "permission", "quota"]):
return ErrorCategory.FATAL
return ErrorCategory.LOGICAL
async def layer1_simple_retry(
fn: Callable[[], Awaitable[Any]],
max_retries: int = 3,
base_delay: float = 1.0,
) -> Any:
"""Layer 1: 指数退避，只处理瞬时故障"""
for attempt in range(max_retries):
try:
return await fn()
except Exception as e:
if classify_error(e) != ErrorCategory.TRANSIENT:
raise
delay = base_delay * (2 ** attempt)
await asyncio.sleep(delay)
raise RuntimeError(f"瞬时故障重试 {max_retries} 次后仍失败")
async def layer2_corrective_retry(
fn: Callable[[RecoveryContext], Awaitable[Any]],
ctx: RecoveryContext,
max_retries: int = 2,
) -> Any:
"""Layer 2: 带错误反馈重试，处理输出畸形"""
for attempt in range(max_retries):
try:
ctx.attempt = attempt
return await fn(ctx)
except Exception as e:
category = classify_error(e)
if category == ErrorCategory.FATAL:
raise
ctx.errors.append(str(e))
if category != ErrorCategory.MALFORMED:
raise
raise RuntimeError(f"纠正性重试 {max_retries} 次后仍失败")核心思路：classify_error 做分诊，每层只 catch 自己负责的错误类型，其余上抛。
实现：纠正性 Prompt 注入
Layer 2 的关键是把错误信息喂回 LLM，让它自我修正：
def build_corrective_prompt(
original_prompt: str,
error_history: list[str],
) -> str:
"""构造带错误反馈的修正 prompt"""
error_block = "\n".join(
f"- Attempt {i+1}: {err}" for i, err in enumerate(error_history)
)
return f"""{original_prompt}
<previous_errors>
你之前的输出产生了以下错误，请避免重复：
{error_block}
</previous_errors>
请严格按照要求的格式输出，不要包含任何额外文本。"""实测这种”错误反馈注入”在输出格式修正场景下，第二次尝试成功率可达 90%+。
实现：策略切换（Layer 3）
当同一步连续失败 2 次，说明当前策略本身有问题，需要换一种方式：
interface StrategyOption {
name: string;
prompt: string;
tools: string[];
model?: string; // 可以降级到不同模型
}
class StrategySwitch {
private strategies: StrategyOption[];
private currentIndex = 0;
constructor(strategies: StrategyOption[]) {
this.strategies = strategies;
}
current(): StrategyOption {
return this.strategies[this.currentIndex];
}
next(): StrategyOption | null {
this.currentIndex++;
if (this.currentIndex >= this.strategies.length) return null;
return this.strategies[this.currentIndex];
}
}
// 使用示例：代码生成步骤的多策略配置
const codeGenStrategies: StrategyOption[] = [
{
name: "direct",
prompt: "直接生成完整实现",
tools: ["file_write", "run_tests"],
model: "claude-sonnet-4-20250514",
},
{
name: "plan-then-code",
prompt: "先输出伪代码计划，再逐步实现",
tools: ["file_write", "run_tests"],
model: "claude-sonnet-4-20250514",
},
{
name: "decompose",
prompt: "将任务拆成 3 个子任务分别实现",
tools: ["file_write", "run_tests"],
model: "claude-opus-4-20250514", // 升级模型
},
];策略切换的本质是：用多样性对抗确定性失败。如果一种 prompt 风格在某个任务上反复失败，换一种思路往往能突破。
熔断器：知道什么时候该停
@dataclass
class CircuitBreaker:
"""Layer 4: 连续失败超过阈值时熔断"""
failure_threshold: int = 5
reset_timeout: float = 60.0 # 秒
_failure_count: int = 0
_last_failure_time: float = 0
_state: str = "closed" # closed / open / half-open
def record_failure(self) -> None:
self._failure_count += 1
self._last_failure_time = time.time()
if self._failure_count >= self.failure_threshold:
self._state = "open"
def record_success(self) -> None:
self._failure_count = 0
self._state = "closed"
def can_proceed(self) -> bool:
if self._state == "closed":
return True
if self._state == "open":
if time.time() - self._last_failure_time > self.reset_timeout:
self._state = "half-open"
return True
return False
return True # half-open: 允许一次试探
def get_state(self) -> str:
return self._state熔断器防止 Agent 在不可恢复的场景下无限烧 token。触发熔断后应该：
- 保存当前执行状态（checkpoint）
- 通知人工介入
- 等待重置后从 checkpoint 恢复
决策矩阵：选择恢复策略
| 场景 | 推荐层级 | 最大重试 | 是否切换策略 | 是否需要 checkpoint |
|---|---|---|---|---|
| LLM API 限流 | Layer 1 | 5 次 | 否 | 否 |
| JSON 输出格式错 | Layer 2 | 2 次 | 否 | 否 |
| 工具调用参数幻觉 | Layer 2→3 | 2+2 次 | 是 | 建议 |
| 多步推理逻辑错误 | Layer 3 | 3 种策略 | 是 | 是 |
| Token 预算耗尽 | Layer 4 | 0 | 熔断 | 是 |
| 外部服务永久下线 | Layer 4 | 0 | 熔断 | 是 |
生产落地 Checklist
把这个清单贴到你的 Agent 项目 README 里，逐项检查。
- 错误分类器：所有异常经过
classify_error分诊，不要裸 catch - 重试预算：设置全局 token/时间预算，防止恢复过程本身失控
- 错误上下文传递：Layer 2 的纠正性重试必须把前次错误喂回 LLM
- 策略多样性：关键步骤至少准备 2 种备选策略（不同 prompt 或不同模型）
- 熔断阈值：根据业务 SLA 设定，不要拍脑袋写 3 次
- Checkpoint 机制：长链路任务（>5 步）必须支持中间状态持久化
- 可观测性：每次恢复动作都要记录 trace（哪一层、第几次、用了什么策略）
- 降级路径：定义每个步骤的”最低可接受输出”，实在恢复不了就降级而非崩溃
总结
Agent 错误恢复不是”加个 try-catch”的事。生产级方案需要：
- 分类：区分瞬时/畸形/逻辑/致命四类错误
- 分层：每层只处理自己能解决的问题
- 多样性：策略切换用不同方式攻克同一问题
- 边界：熔断器确保系统知道什么时候该停
这套架构的 ROI 非常明确：在我们的实践中，12 步 Agent 的端到端成功率从 54% 提升到 87%，而额外 token 开销只增加了 15%（大部分重试在 Layer 1-2 就解决了）。