AI 多 Agent 网络中的自动错误恢复实战

在单 Agent 系统中，失败很简单：Agent 报错了，重试就是了。

在多 Agent 系统中，失败是一个图论问题。

级联失败问题

Agent A: ✅ Success
Agent B: ❌ Timeout (depends on A)
Agent C: ❌ Skipped (depends on B)
Agent D: ❌ Partial data (depends on C)

一次超时会沿着整条流水线传播。没有恢复机制，你的系统就是脆弱的。

我们的恢复策略

AgentForge 实现了三层恢复机制：

第一层：指数退避重试

@retry(max_attempts=3, backoff=exponential(base=2, max=60))
def agent_call(params):
    return llm.invoke(params)

第二层：熔断器

如果某个 Agent 在 10 分钟内失败 5 次，我们就停止调用它，并返回一个降级响应：

{
  "status": "degraded",
  "agent": "market_data",
  "fallback": "cached_data",
  "warning": "Real-time data unavailable, using 15-min delayed feed"
}

第三层：流水线重规划

当关键 Agent 失败时，编排器可以重新规划：

• 如果失败的步骤非关键，则跳过
• 用备份 Agent 替代
• 停止并携带完整上下文链路发出告警

上个月，我们的市场数据 API 在交易时段宕机了。以下是当时发生的情况：

14:32 — 市场数据 Agent 超时（第一层：3 次重试全部失败）

14:33 — 市场数据 Agent 的熔断器打开

14:33 — 流水线自动切换到缓存数据并标记警告 flag

14:35 — 生成完整报告，标注"数据延迟"免责声明

15:00 — 市场数据 API 恢复，熔断器自动关闭

零人工干预。零报告漏传。

如果你的多 Agent 系统连一个 Agent 失败都处理不了，那它还没准备好投入生产。

AgentForge 把容错做成了默认选项，而不是事后补救。