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

class S(TypedDict):
    result: str

def call_api(state: S) -> dict:
    resp = httpx.get("https://example.com/api")
    resp.raise_for_status()                 # 5xx 会抛 HTTPStatusError
    return {"result": resp.text}

g = StateGraph(S)
g.add_node("call", call_api,
           retry_policy=RetryPolicy(max_attempts=5, initial_interval=1.0))  # ← 重试
g.add_edge(START, "call"); g.add_edge("call", END)
app = g.compile()

types.py:416，一个带默认值的 NamedTuple：

# types.py:416
class RetryPolicy(NamedTuple):
    initial_interval: float = 0.5     # 首次重试前等多久（秒）
    backoff_factor: float = 2.0       # 每次重试间隔的放大倍数
    max_interval: float = 128.0       # 间隔的上限（再涨也不超过它）
    max_attempts: int = 3             # 最多尝试几次（含第一次）
    jitter: bool = True               # 是否加随机抖动
    retry_on: ... = default_retry_on  # 哪些异常触发重试

不是所有异常都值得重试。参数写错重一万次还是错。默认判断函数 _internal/_retry.py:1：

# _internal/_retry.py:1
def default_retry_on(exc: Exception) -> bool:
    import httpx, requests
    if isinstance(exc, ConnectionError):
        return True                                    # 连接问题 → 重
    if isinstance(exc, httpx.HTTPStatusError):
        return 500 <= exc.response.status_code < 600   # 只重 5xx
    if isinstance(exc, requests.HTTPError):
        return 500 <= exc.response.status_code < 600 if exc.response else True
    if isinstance(exc, (ValueError, TypeError, ArithmeticError, ImportError,
                        LookupError, NameError, SyntaxError, RuntimeError,
                        ReferenceError, StopIteration, StopAsyncIteration, OSError)):
        return False                                   # 这些是"代码/逻辑错" → 不重
    return True                                        # 其他不认识的 → 保守地重

引擎执行每个节点都过这个循环 pregel/_retry.py:573。骨架是 while True：

# pregel/_retry.py:600（精简）
while True:
    try:
        task.writes.clear()               # ① 清掉上次失败留下的半成品写
        return task.proc.invoke(task.input, config)   # ② 跑节点，成功就 return
    except GraphBubbleUp:                 # interrupt 之类的"正常中断" → 不重试，直接抛
        raise
    except Exception as exc:
        if not retry_policy:              # 没配策略 → 直接抛
            raise
        matching_policy = None            # ③ 找第一个"认这个异常"的策略（L06）
        for policy in retry_policy:
            if _should_retry_on(policy, exc):
                matching_policy = policy
                break
        if not matching_policy:           # 没有策略认它 → 抛
            raise
        attempts += 1                     # ④ 失败次数 +1
        if attempts >= matching_policy.max_attempts:   # 试够了 → 放弃
            raise
        ...                               # ⑤ 算间隔、睡、再循环（L05）
        config = patch_configurable(config, {CONFIG_KEY_RESUMING: True})

L04 步骤⑤的等待时间计算 pregel/_retry.py:663：

# pregel/_retry.py:663
interval = matching_policy.initial_interval
interval = min(
    matching_policy.max_interval,
    interval * (matching_policy.backoff_factor ** (attempts - 1)),   # 指数退避
)
sleep_time = (
    interval + random.uniform(0, 1) if matching_policy.jitter else interval  # 抖动
)
time.sleep(sleep_time)

retry_policy 可以是一个列表——引擎遇到异常时，按顺序找第一个认它的策略 pregel/_retry.py:648：

# pregel/_retry.py:648
matching_policy = None
for policy in retry_policy:
    if _should_retry_on(policy, exc):     # 这个策略认这个异常吗？
        matching_policy = policy
        break                             # 认了就用它，不再往后看

_should_retry_on 支持三种 retry_on 写法 pregel/_retry.py:841：

# pregel/_retry.py:841
def _should_retry_on(retry_policy, exc) -> bool:
    if isinstance(retry_policy.retry_on, Sequence):
        return isinstance(exc, tuple(retry_policy.retry_on))   # 异常类的列表
    elif isinstance(retry_policy.retry_on, type) and issubclass(retry_policy.retry_on, Exception):
        return isinstance(exc, retry_policy.retry_on)          # 单个异常类
    elif callable(retry_policy.retry_on):
        return retry_policy.retry_on(exc)                      # 自定义判断函数
    else:
        raise TypeError("retry_on must be an Exception class, a list/tuple of them, or a callable")

from langgraph.types import RetryPolicy
g.add_node("call", call_api, retry_policy=[
    RetryPolicy(retry_on=RateLimitError, initial_interval=10, max_attempts=5),  # 限流：等久点
    RetryPolicy(retry_on=TimeoutError,  initial_interval=0.5, max_attempts=3),  # 超时：快重
])

✋ 10 分钟动手

# 1. 读重试全链路
sed -n '416,435p'  libs/langgraph/langgraph/types.py            # RetryPolicy 字段
sed -n '1,29p'     libs/langgraph/langgraph/_internal/_retry.py  # default_retry_on
sed -n '600,682p'  libs/langgraph/langgraph/pregel/_retry.py     # run_with_retry
sed -n '841,855p'  libs/langgraph/langgraph/pregel/_retry.py     # _should_retry_on

# 2. 亲眼看退避重试
python - <<'PY'
import time
from langgraph.graph import StateGraph, START, END
from langgraph.types import RetryPolicy
from typing import TypedDict
class S(TypedDict):
    n: int
tries = {"n": 0}
def flaky(s):
    tries["n"] += 1
    print(f"  第 {tries['n']} 次尝试 @ {time.strftime('%H:%M:%S')}")
    if tries["n"] < 3:
        raise ConnectionError("temporary")   # 前两次假装网络挂
    return {"n": 42}
g = StateGraph(S)
g.add_node("f", flaky, retry_policy=RetryPolicy(initial_interval=0.5, max_attempts=5))
g.add_edge(START,"f"); g.add_edge("f",END)
print(g.compile().invoke({"n":0}))            # 前两次失败重试，第三次成功 → {'n': 42}
PY