Agent 错误恢复与韧性

Retry 的两层结构：别把传输错误和语义错误混在一个 retry 里

背景与原理

agent 里有两种"失败"，处理方式完全相反。

• 传输层失败：HTTP 429 / 503、连接超时、ECONNRESET——确定性的瞬时故障。正确做法是程序层 retry，指数退避 + jitter，根本不该惊动模型。Marc Brooker 在 AWS 那篇经典里证明：固定 backoff 会让客户端"同步化"反复撞击同一时刻，full jitter（随机化整个退避窗口）显著降低竞争。
• 语义层失败：tool 执行了但结果是错的——文件不存在、命令报错、参数非法。这是模型需要"看到并自我修正"的信息，正确做法是把 error message 当 tool_result 喂回模型，让它换个方法。绝不能在程序层 retry——同样的参数重试一百次还是错。

头号 bug 是把两层搞反：把 429 当语义错误喂回模型（模型不会退避，继续猛打依赖），或把"文件不存在"当传输错误程序重试（死循环）。判别只有一句话：错误是否与输入无关、且只是瞬时的？是→程序退避；否→喂回模型。

实战示例

import time, random, anthropic

TRANSIENT = (anthropic.RateLimitError, anthropic.APITimeoutError,
             anthropic.InternalServerError)

def call_with_backoff(fn, max_retries=5):
    for attempt in range(max_retries):
        try:
            return fn()
        except TRANSIENT as e:        # 传输层:确定性瞬时故障 → 程序退避
            if attempt == max_retries - 1: raise
            # full jitter:随机化整个退避窗口,避免 thundering herd
            time.sleep(random.uniform(0, min(60, 2 ** attempt)))
        # 注意:语义错误(tool 报错)不在这里 catch——
        # 它们走 tool_result 喂回模型,由模型决定下一步

幂等与补偿：agent 一定会重复执行，副作用工具必须幂等

背景与原理

retry 和断点续跑有个隐藏前提：被重试的操作可以安全地执行多次。读操作天然幂等（读两遍文件没事），但写操作不是：create_order / send_email / charge_card 重试一次 = 客户被扣两次款。agent 比传统系统更危险，因为它的"重试"有两个来源：程序层 backoff，和模型层——模型看到 tool_result 含糊（超时，不知道成没成功）时，会"保险起见再调一次"。这恰恰是最危险的时刻，因为首次可能已经成功了。

两个工程手段：

• 幂等键（idempotency key）：每个有副作用的 tool call 带一个客户端生成的唯一 key，服务端用它去重——同一个 key 第二次调用直接返回首次结果，不重复执行。Stripe 等支付 API 的标准做法。关键：key 要从任务派生（确定性），而非每次随机，否则重试时 key 变了就失去去重意义。
• 补偿事务（Saga）：agent 没有数据库事务可以一键回滚。若第 3 步失败，前两步的副作用（已订票、已扣款）要靠补偿动作（退票、退款）显式撤销。给每个有副作用的 step 配一个 compensating action，失败时逆序执行。

实战示例

# 1) 幂等键:让重试安全。key 由 task 派生,而非随机
def book_flight(args):
    key = f"book:{args['trip_id']}:{args['flight_no']}"   # 确定性 key
    return api.post("/bookings", json=args,
                    headers={"Idempotency-Key": key})       # 服务端去重

# 2) 补偿:每个副作用 step 登记一个 undo,失败时逆序回滚
saga = []
try:
    r1 = book_flight(...);  saga.append(lambda: cancel_flight(r1.id))
    r2 = charge_card(...);  saga.append(lambda: refund(r2.id))
    r3 = book_hotel(...)                  # ← 这一步失败
except Exception:
    for undo in reversed(saga):     # 逆序补偿:先退款,再退票
        undo()
    raise

Checkpoint 与断点续跑：agent state 必须外置，内存 loop 是玩具

背景与原理

一个跑 40 步、半小时的 agent，在第 38 步进程崩了（OOM、部署重启、抢占式机器被回收）。如果 state 只活在内存的 messages 数组里，前 37 步全部白跑——而那 37 步的 token 和副作用已经花掉了。生产 agent 必须把 state 持久化到进程外：每完成一步就 checkpoint 一次，崩溃后从最后一个 checkpoint 恢复。这就是 durable execution 的核心。两种落地：

• 库层：LangGraph 的 checkpointer——每个 node 执行后把 state 快照到 SQLite / Postgres，用 thread_id 恢复，"loads the last snapshot and resumes exactly where it left off"。
• 平台层：Temporal 这类 durable execution engine，把整个 agent workflow 变成可崩溃恢复的——崩溃后 replay 历史、跳过已完成的 activity，"survive a crash and resume exactly where it left off"。

关键是 checkpoint 的粒度：太细（每个 token）开销大，太粗（只在结束）等于没有。实践上以"完成一次模型调用或一个有副作用的 step"为单位。且 checkpoint 必须配合幂等（§2）——恢复时最后那个 step 可能只执行了一半。

实战示例

import json, os
def agent(task, ckpt_path):
    if os.path.exists(ckpt_path):                    # 续跑:从 checkpoint 恢复
        s = json.load(open(ckpt_path)); msgs, step = s["msgs"], s["step"]
    else:
        msgs, step = [{"role":"user","content":task}], 0
    while step < MAX_STEPS:
        r = client.messages.create(model=MODEL, messages=msgs,
                                   tools=SCHEMAS, max_tokens=4096)
        msgs.append({"role":"assistant","content":r.content})
        if r.stop_reason == "end_turn": break
        results = run_tools(r.content)               # 执行工具(须幂等!)
        msgs.append({"role":"user","content":results}); step += 1
        json.dump({"msgs":msgs,"step":step}, open(ckpt_path,"w"))  # 每步落盘

超时、熔断与部分失败：区分"可降级"和"必须停"

背景与原理

三个相关机制：

• 超时（timeout）：每个 tool call 必须有上限，否则一个挂起的 HTTP 请求会让整个 agent 永久卡死。超时要分层：单个 tool 超时 < 单个 step 超时 < 整个 task 的顶层 deadline。
• 熔断（circuit breaker）：某个下游依赖（一个 API、一个 MCP server）连续失败 N 次就"跳闸"——后续调用直接快速失败，不再傻等超时，过一段时间再"半开"放一个试探。这防止 agent 把已经挂掉的依赖反复打，既省 token 又防止级联拖慢。经典三态：closed → open → half-open。
• 部分失败（partial failure）：并行 fan-out（Day 13 的 orchestrator-worker）里 5 个 worker，1 个挂了。反模式是整批 fail；成熟做法是收集成功的 4 个，把"1 个失败"作为信息交给 orchestrator 决策——降级用 4 个结果继续，还是这个子任务关键、必须停。

实战示例

class CircuitBreaker:
    def __init__(self, fail_max=5, reset_after=30):
        self.fail_max, self.reset_after = fail_max, reset_after
        self.fails, self.opened_at = 0, None
    def call(self, fn):
        if self.opened_at:                          # OPEN
            if time.time() - self.opened_at < self.reset_after:
                raise CircuitOpen("依赖熔断中,快速失败")
            self.opened_at = None                   # → HALF-OPEN:放一个试探
        try:
            out = fn(); self.fails = 0              # 成功 → CLOSED
            return out
        except Exception:
            self.fails += 1
            if self.fails >= self.fail_max:
                self.opened_at = time.time()       # → OPEN:跳闸
            raise