生产系统不是「不出错」,而是「出错时知道怎么办」。Agent 的失败模式比传统服务更多,因为它会自己决定下一步。
失败分类
| 类型 | 表现 | 典型原因 | 处理 |
|---|---|---|---|
| 暂时性 | 超时、429、503 | 网络、限流 | 退避重试 |
| 参数错误 | 工具返回校验失败 | 模型填错 | 修正一次 |
| 语义错误 | 调用成功但结果不对 | 理解偏差 | 验证、重规划 |
| 死循环 | 重复同一动作 | 无进展、错误信息无用 | 指纹检测、终止 |
| 资源耗尽 | token/金额/时间超限 | 无预算控制 | 硬上限、降级 |
| 级联失败 | 下游全挂 | 重试风暴 | 熔断、隔离 |
| 状态不一致 | 重试导致重复写入 | 非幂等 | 幂等键、对账 |
| 安全事件 | 越权、注入、外泄 | 权限过宽 | 阻断、审计、告警 |
先分类,再设计处理。把所有错误都变成「重试三次」是最常见的反模式。
超时:每一层都要有
总预算 120s
├─ 单次模型调用 30s
├─ 单次工具调用 10s
├─ 单步总耗时 45s
└─ 人工审批 24h(单独)没有超时,一个挂起的请求会占住连接、worker 和预算。超时后要返回可理解的失败,而不是空白。
重试:只重试值得重试的
import random, time
def retry(fn, max_attempts=3, base=0.5, retry_on=(TimeoutError, RateLimitError, ServerError)):
for attempt in range(1, max_attempts + 1):
try:
return fn()
except retry_on as e:
if attempt == max_attempts:
raise
delay = base * (2 ** (attempt - 1)) + random.uniform(0, 0.3)
time.sleep(min(delay, 8))要点:
- 指数退避 + 抖动:避免重试风暴;
- 只重试暂时性错误:参数错误、权限不足重试没有意义;
- 有总预算:重试也算时间和 token;
- 幂等:写操作必须用幂等键;
- 记录重试次数:重试率高说明下游或提示有问题。
幂等:重试的安全网
def create_refund(order_id, amount, idempotency_key):
existing = db.get_by_key(idempotency_key)
if existing:
return existing # 直接返回上次结果
result = payments.refund(order_id, amount, key=idempotency_key)
db.save(idempotency_key, result)
return result幂等键应该由服务端生成(例如 task_id + step_id),而不是让模型填。模型可能每次生成不同的 key,等于没有幂等。
熔断与隔离
当下游持续失败:
连续 5 次失败或错误率 > 50% → 打开熔断 30 秒
熔断期间 → 直接返回降级结果,不调用下游
30 秒后 → 放一个探针请求,成功则恢复隔离(bulkhead):不同工具、不同租户使用独立的并发池,避免一个坏下游拖垮全部。
预算:硬限制,不是建议
@dataclass
class Budget:
max_tokens: int = 60_000
max_cost: float = 0.50
max_seconds: int = 120
max_steps: int = 8
def check(self):
if self.tokens > self.max_tokens: raise BudgetExceeded("token")
if self.cost > self.max_cost: raise BudgetExceeded("cost")
if self.elapsed > self.max_seconds: raise BudgetExceeded("time")
if self.steps > self.max_steps: raise BudgetExceeded("steps")预算要在每一步之前检查,而不是最后算账。超限时返回已完成的部分,并说明缺什么。
死循环检测
三个信号:
- 动作指纹重复:同样的工具 + 参数连续出现;
- 观察无变化:结果哈希相同;
- 目标无进展:待办列表没有减少。
处理:
- 注入提醒:「相同调用已重复,请换方法或停止」;
- 降低温度/换模型;
- 强制结束并转人工;
- 记录为评估样本。
降级阶梯
不要只有「成功」和「失败」两种状态:
L0 正常:完整 Agent
L1 简化:减少工具、跳过重排、缩短输出
L2 规则:模板回答 / 固定流程
L3 兜底:转人工 / 稍后重试例如:
- 模型超时 → 用小模型生成简版;
- 检索失败 → 明确说资料不足,给相关链接;
- 写操作失败 → 保存草稿,提示稍后重试;
- 预算耗尽 → 返回已完成部分 + 未完成清单。
校验与修复
模型输出进入下游前必须校验:
- Schema;
- 业务规则(金额、日期、权限);
- 引用存在;
- 敏感信息;
- 幂等键。
失败时修复一次,仍失败则降级。不要无限循环「生成—校验—再生成」。
并发与竞态
多步 Agent 可能同时:
- 修改同一个任务状态;
- 调用同一个写工具;
- 读取过期缓存。
处理:
- 状态用乐观锁(version);
- 写操作串行化或加锁;
- 幂等键;
- 读己之写(read-your-writes)时绕过缓存;
- 任务级并发上限。
状态恢复
进程崩溃、部署、超时后:
- 从检查点恢复;
- 跳过已完成的副作用;
- 校验外部状态是否与记录一致;
- 对账未完成的写操作。
没有恢复能力,长任务只能从头再来,成本和风险都会翻倍。
混沌测试
主动注入故障:
- 模型超时;
- 工具 500;
- 检索返回空;
- 限流 429;
- 上下文超长;
- 人工审批超时;
- 进程在写操作后崩溃。
观察 Agent 是否:
- 停止而不是乱试;
- 给出可理解错误;
- 不重复副作用;
- 能恢复。
一个组合示例
def safe_step(step, state):
if state.budget.exceeded():
return degrade(state, "budget")
try:
with timeout(step.timeout):
result = retry(lambda: execute(step), max_attempts=2)
except CircuitOpen:
return degrade(state, "downstream")
except Exception as e:
state.errors.append(classify(e))
if is_retryable(e) and state.retry_count < 2:
return state
return escalate(state, e)
if not validate(step, result):
return escalate(state, "validation_failed")
return apply(state, result)反模式
没有次数和预算上限,一个故障能把额度烧光。
重复退款、重复发邮件、重复下单。写操作必须先幂等。
模型传错参数,重试一百次也一样。应该修复或停止。
用户不知道发生了什么,也无法采取行动。给出具体原因和下一步。
模型挂了整个产品不可用。至少要有规则兜底。
检查清单
- 每层都有超时与预算
- 只重试暂时性错误,退避 + 抖动
- 写操作幂等,键由服务端生成
- 有熔断、隔离、并发上限
- 有死循环/无进展检测
- 有 L0–L3 降级阶梯
- 输出校验 + 一次修复
- 能从检查点恢复,跳过已完成副作用
- 混沌测试覆盖主要故障
- 错误可归因、可告警
小结
- 先给失败分类,再设计处理,不要统一重试。
- 超时、重试、幂等、熔断、预算是可靠性的五件套。
- 死循环靠动作指纹与无进展检测。
- 降级要分层,失败也要给出可行动的说明。
- 用混沌测试验证,而不是等线上出事。
下一篇是安全:当 Agent 能行动时,攻击面就不只是「说错话」。