第 10 章 · 容错、重试与幂等性

模拟一次暂时故障，验证节点重试，而不是在所有错误上无限循环。本章不请求真实服务。

工作机制与设计选择

RetryPolicy 指定最大尝试次数和可重试异常。认证失败、参数错误通常不应按瞬时网络故障重试。检查点帮助恢复图状态，但外部系统可能已经完成写入、只丢失响应；重试时必须使用同一幂等键。官方当前文档还描述 1.2 的节点超时与错误处理器，使用前先核对安装版本。

示例

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import RetryPolicy

class TemporaryFailure(Exception):
    pass

class State(TypedDict):
    result: str

attempts = {"count": 0}

def fetch_fixture(state: State):
    attempts["count"] += 1
    if attempts["count"] == 1:
        raise TemporaryFailure("Simulated transient failure")
    return {"result": "fixture loaded"}

builder = StateGraph(State)
builder.add_node("fetch", fetch_fixture, retry_policy=RetryPolicy(
    max_attempts=2, initial_interval=0.01,
    retry_on=TemporaryFailure,
))
builder.add_edge(START, "fetch")
builder.add_edge("fetch", END)
result = builder.compile().invoke({"result": ""})
assert attempts["count"] == 2
assert result["result"] == "fixture loaded"
print(result)

运行与验收

运行应在第二次尝试成功。attempts 是仅供本地测试的计数器，不是可恢复业务状态。把错误改成 ValueError 后应直接失败，从而验证重试范围没有过宽。

容易踩的坑

避免模型 SDK、工具客户端和图节点各自重试多次，造成次数乘法膨胀。确定统一预算，记录最终异常与每层重试次数。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. max_attempts 包括第一次吗？

2. 认证错误应无限重试吗？

3. 检查点能防所有重复付款吗？

4. 多层重试有什么风险？

动手实践

1. 让每次都抛 TemporaryFailure，断言最多两次。
2. 加入不可重试异常测试，确认一次失败。
3. 设计带 request_id 的模拟写入，重复请求只产生一次记录。

官方参考

• fault-tolerance

核验日期：2026-09-17。