上一篇用窄适配器接入 API 与数据库，并把下游结果转成稳定错误类别。本篇据此建立恢复策略：瞬时错误有限重试，永久错误立即停止，不确定副作用先查幂等记录，避免 Agent 用“再试一次”制造重复操作。

一、痛点：从“能演示”走向“可托管”

错误至少分四类：invalid_input 需要修正参数；forbidden 永不重试；transient 可退避重试；unknown_effect 表示请求可能已成功但响应丢失。最后一类不能直接重放写操作，应先用幂等键查询结果。重试预算属于整个任务，而不是每层各自三次，否则模型、工作流和 HTTP 客户端叠加会形成重试风暴。

真正可迁移的做法，是先写任务契约再选模型：输入从哪里来，成功产物是什么，哪些动作禁止自动执行，最多允许多少步、多少时间和多少费用，失败后由谁接管。契约一旦写进代码和测试，模型升级只是实现替换，不会悄悄改变业务责任边界。

二、原理：把概率判断放进确定性边界

指数退避降低持续压力，抖动避免多个实例同时醒来。生产环境可用随机抖动，教学示例用确定序列便于复现。每次尝试都记录 attempt、错误类别、等待时间和最终终态。达到预算后进入 degraded 或 needs_human，并保留已完成步骤，不能把暂时失败伪装成空数据。

模型输出、工具数据和记忆内容都按不可信输入处理。每个边界都执行校验、授权、裁剪和审计；所有状态变化都有明确原因。这样做看似增加一些代码，却把“偶尔答错”拆成可定位的规划错误、参数错误、工具错误或策略错误，团队才能针对性改进。

三、实现：用独立程序跑通最小闭环

下面的连接器前两次返回 503，第三次成功。重试器只对 transient 继续，并按 1、2 秒计算等待；示例不真实 sleep，因此可快速运行。替换调用函数后，这个控制器可以复用于模型 API、检索服务和业务接口。

from dataclasses import dataclass

@dataclass
class Response:
    ok: bool
    error: str = ""

attempts = 0

def flaky_call() -> Response:
    global attempts
    attempts += 1
    if attempts < 3:
        return Response(False, "transient")
    return Response(True)

max_attempts = 4
status = "failed"
for attempt in range(1, max_attempts + 1):
    response = flaky_call()
    if response.ok:
        print(f"attempt={attempt} result=ok")
        status = "completed"
        break
    if response.error != "transient":
        print(f"attempt={attempt} error={response.error} stop=true")
        break
    wait = 2 ** (attempt - 1)
    print(f"attempt={attempt} error=transient wait={wait}s")
print(f"status={status} attempts={attempts}")

运行输出：

attempt=1 error=transient wait=1s
attempt=2 error=transient wait=2s
attempt=3 result=ok
status=completed attempts=3

写操作恢复依赖幂等键。键应由稳定业务标识生成，例如 task_id 与 action，而不是随机生成后丢失。服务端在同一事务中保存键和结果；客户端超时后再次调用同一键，得到原结果而非再次扣款。若下游不支持幂等，应在本地建立 outbox，并把不可确认状态升级人工处理。

第二个程序补上生产中最容易遗漏的控制点。它与前一个示例完全独立，可单独保存运行，不依赖本系列其他文件；示例数据是内存假实现，因此不会访问真实账户或产生外部副作用。

from dataclasses import dataclass

@dataclass(frozen=True)
class Result:
    result_id: str

records: dict[str, Result] = {}
refund_count = 0

def create_refund(key: str) -> tuple[str, Result]:
    global refund_count
    if key in records:
        return "replayed", records[key]
    refund_count += 1
    result = Result("refund-R7")
    records[key] = result
    return "created", result

state1, result1 = create_refund("ticket-7:refund")
state2, result2 = create_refund("ticket-7:refund")
print(f"first_call={state1} result={result1.result_id}")
print(f"second_call={state2} result={result2.result_id}")
print(f"refund_count={refund_count}")
status = "safe" if refund_count == 1 else "duplicate"
print(f"status={status}")

运行输出：

first_call=created result=refund-R7
second_call=replayed result=refund-R7
refund_count=1
status=safe

落地时应把示例中的内存状态替换为事务型存储，把打印日志替换为结构化事件，但不要改变契约。事件至少包含 run_id、步骤、输入摘要、策略结果、耗时、错误类别和版本；密钥、完整个人信息及未经脱敏的提示不得进入日志。

四、踩坑：失败路径决定系统上限

不要对所有异常重试：鉴权失败重试只会放大流量，校验错误应该回到参数生成，限流应尊重 Retry-After。也不要让模型看到内部栈后自由决定次数。熔断器用于下游持续故障时快速失败，但半开探测要有限并发；降级结果必须显式标记，避免旧缓存被当成实时事实。

还要防止把确定逻辑模型化。金额计算、权限判断、枚举路由和状态转移应使用普通代码；模型适合处理分类、抽取、歧义消解与证据综合。每减少一个无必要的模型决策点，就减少一处延迟、成本和不可复现性，同时让测试更容易覆盖。

五、验证：用轨迹和门槛验收

通过故障注入验收：在连接前、请求后、外部成功后和本地落盘前分别中断。确认瞬时读取最终恢复，永久错误零重试，不确定写入只有一个副作用，预算耗尽进入明确终态。下一篇会把错误率、恢复率、延迟和成本纳入离线评测与线上 trace。

发布顺序固定为离线回放、影子流量、小比例灰度和有限自治。每阶段先定义通过线和回滚条件，再看结果；不能在看到分数后移动门槛。关键样本要保存初始状态、每步动作、观察、最终产物与终止原因，模型、提示和工具契约版本也必须一起记录。

可观测性必须记录恢复过程，而不是只记录最终成功；否则重试掩盖的下游退化无法进入发布决策。

参考来源

• AWS Backoff and Jitter
• HTTP Semantics
• OpenTelemetry OTLP

👍 觉得有用就点个 赞 + 收藏，方便回头查阅；有疑问直接在评论区留言，我看到都会回。

🚀 本文属于 《AI Agent 应用实战》 系列，持续更新，关注不迷路。

📌 文章里的代码都能直接跑。想要可直接 clone 的完整工程 + 配套部署脚本 / 踩坑清单？评论一声或发邮件到 cj2664@qq.com，我免费发你。
如果你正好在做类似系统、或有工程化难题想找人做，也欢迎邮件聊一句——我按实际情况评估，能落地的就接单或出方案。评论和邮件都能直接找到我，不用跳别的平台。