OpenClaw 错误处理:让 Agent 学会"摔倒了再爬起来"
凌晨两点,你的 Agent 在跑一个关键任务。突然,API 超时了。它没有重试,没有降级,直接报错退出。第二天早上,你发现客户数据只处理了一半。这不是电影情节,这是真实的生产事故。今天,我们聊聊如何让你的 Agent 变得"抗造"——摔倒了能爬起来,走不通能换条路。
为什么错误处理如此重要?
AI Agent 和传统软件最大的不同在于:它依赖外部服务(LLM API)和不可控因素(网络、模型输出)。错误不是"会不会发生"的问题,而是"什么时候发生"的问题。
常见错误类型及发生频率:
- API 超时:约占错误的 35%,高并发时更频繁
- Rate Limit:约占 25%,免费套餐最容易触发
- 模型输出异常:约占 20%,格式不符、截断等
- 网络故障:约占 15%,DNS 解析、连接重置
- 资源耗尽:约占 5%,内存、Token 限制
OpenClaw 内置错误处理机制
1. 自动重试
# .openclaw/config.yaml
retry:
max_attempts: 3 # 最大重试次数
backoff: exponential # 退避策略:linear | exponential
base_delay_ms: 1000 # 初始延迟
max_delay_ms: 30000 # 最大延迟
retryable_errors: # 可重试的错误类型
- "timeout"
- "rate_limit"
- "server_error"
- "connection_error"
2. 降级策略
fallback:
enabled: true
strategies:
- condition: "model_error"
action: "switch_model"
target: "gpt-3.5-turbo" # 主模型挂了切换备用
- condition: "rate_limit"
action: "queue" # 排队等待
queue_ttl: 300 # 排队超时 5 分钟
- condition: "timeout"
action: "simplify_prompt" # 简化提示词重试
max_reduction: 50% # 最多削减 50%
3. 断路器(Circuit Breaker)
circuit_breaker:
enabled: true
failure_threshold: 5 # 连续 5 次失败触发断路
recovery_timeout: 60000 # 60 秒后尝试恢复
half_open_requests: 3 # 半开状态试探请求数
自定义错误处理
在 SKILL.md 中定义错误处理
# SKILL.md 片段
## 错误处理
### 自定义错误处理器
```python
@error_handler
async def handle_api_error(error, context):
if error.type == "rate_limit":
await wait_and_retry(context.delay)
return RetryAction(delay=context.delay * 2)
if error.type == "invalid_response":
# 尝试修复模型输出
fixed = await repair_output(error.response)
return FixedOutput(fixed)
if error.type == "timeout":
# 切换到快速模型
return SwitchModel("claude-3-haiku")
# 未知错误,记录并上报
log_error(error, context)
return EscalateAction()
```
错误处理最佳实践
原则一:快速失败 vs 优雅降级
不是所有错误都值得重试。区分"可恢复错误"和"致命错误":
- 可恢复:网络抖动、临时超时、Rate Limit → 重试/降级
- 致命错误:认证失败、参数错误、权限不足 → 快速失败
原则二:重试要有边界
无限重试 = 无限等待。设置合理边界:
- 最大重试次数:3-5 次
- 最大总延迟:5-10 分钟
- 指数退避 + 随机抖动(避免惊群效应)
原则三:错误要有上下文
只记录 "Error: timeout" 毫无意义。需要记录:
- 请求 ID、用户 ID
- 错误发生时的 Agent 状态
- 模型、Token 数、延迟
- 堆栈跟踪(生产环境脱敏)
错误监控与告警
配置错误监控
# .openclaw/monitoring.yaml
error_monitoring:
enabled: true
# 错误聚合
aggregation:
window: 300 # 5 分钟窗口
threshold: 10 # 同类错误超过 10 次告警
# 告警渠道
alerts:
- type: "slack"
webhook: "${SLACK_WEBHOOK}"
severity: ["critical", "high"]
- type: "email"
recipients: ["ops@company.com"]
severity: ["critical"]
- type: "webhook"
url: "https://your-service.com/alert"
severity: ["critical", "high", "medium"]
错误分级标准
| 级别 | 定义 | 响应时间 | 示例 |
|---|---|---|---|
| Critical | 服务完全不可用 | 5 分钟 | 所有 API 调用失败 |
| High | 核心功能受影响 | 30 分钟 | 特定模型持续超时 |
| Medium | 部分功能降级 | 4 小时 | 非核心 Skill 失败 |
| Low | 偶发错误 | 24 小时 | 单次请求超时后重试成功 |
实战案例:电商 Agent 的容错设计
场景:自动下单 Agent
class OrderAgent:
async def process_order(self, order):
try:
# 尝试主流程
result = await self.place_order(order)
return SuccessResult(result)
except RateLimitError as e:
# Rate Limit → 排队等待
return await self.queue_order(order, delay=60)
except TimeoutError as e:
# 超时 → 切换快速模型重试
result = await self.place_order(
order,
model="haiku",
timeout=10
)
return SuccessResult(result)
except ModelOutputError as e:
# 输出异常 → 尝试修复
fixed = await self.repair_output(e.raw_response)
if fixed:
return SuccessResult(fixed)
# 修复失败 → 人工介入
return await self.escalate_to_human(order, e)
except PaymentError as e:
# 支付失败 → 致命错误,不重试
return FailedResult(
error="payment_failed",
message=e.message,
refund_required=False
)
except Exception as e:
# 未知错误 → 记录并上报
await self.log_unknown_error(e, order)
return FailedResult(error="unknown")