错误处理、重试与超时策略

错误处理、重试与超时策略

学习目标

本章要解决什么问题：

• RetryPolicy 的配置项与退避策略实现
• TimeoutPolicy 的两种超时类型（run_timeout / idle_timeout）
• 节点级 error_handler 的执行流程
• 同步任务为何不支持超时

项目实践

RetryPolicy 配置

RetryPolicy（libs/langgraph/langgraph/types.py）定义重试策略：

@dataclass
class RetryPolicy:
    retry_on: type[Exception] | tuple[type[Exception], ...] | Callable  # 可重试的异常类型
    max_attempts: int = 3       # 最大重试次数
    initial_interval: float = 0.5  # 初始间隔（秒）
    max_interval: float = 128.0    # 最大间隔
    exponential_base: float = 2.0  # 指数底数
    jitter: bool = True            # 是否添加随机抖动

退避策略实现（_retry.py，37KB）：

interval = min(initial * (exponential_base ^ attempt), max_interval)
if jitter:
    interval *= random.uniform(0.5, 1.5)

Jitter 的作用：防止多个节点同时重试导致”重试风暴”——所有节点在同一时刻重新执行，再次同时失败。

TimeoutPolicy 两种超时类型

@dataclass
class TimeoutPolicy:
    run_timeout: float | None = None    # 硬墙超时：从开始到结束的总时长
    idle_timeout: float | None = None   # 空闲超时：无进展的时长

• run_timeout：硬墙超时，从任务开始计时，超时即终止。不随进度刷新。
• idle_timeout：空闲超时，每次有进展（checkpoint、写入、心跳）时刷新计时器。

对于长时间运行但不持续输出进度的任务，可通过 runtime.heartbeat() 手动刷新 idle_timeout。

节点级 error_handler

def error_handler(state, exc: Exception) -> dict:
    return {"error": str(exc), "status": "failed"}

graph.add_node("process", process_node, error_handler=error_handler)

当 process_node 抛出异常时：

1. 异常被捕获并传递给 error_handler
2. error_handler 接收异常和当前状态
3. error_handler 的返回值写入状态
4. 执行继续（而非中断）

限制：

• error_handler 自身的异常不会被再次处理（直接向上抛出）
• error_handler 不能被缓存（缓存错误结果不安全）

节点默认策略

graph.set_node_defaults(
    retry_policy=RetryPolicy(max_attempts=3),
    timeout=60.0,
    error_handler=default_error_handler,
)

设置后，所有未显式配置的节点自动继承这些策略。

陷阱与对策

同步任务不支持超时

问题：同步函数无法在进程内安全取消，timeout 对同步任务抛出异常。

规避：对有时间要求的任务使用 async def。

重试后 checkpoint 状态不一致

问题：重试前节点可能已经部分写入状态，重试后状态包含重复数据。

规避：节点设计为幂等——多次执行相同输入产生相同输出。

设计取舍

优势

• 灵活配置：节点级 vs 图级、重试 vs 超时 vs 错误处理
• 指数退避 + jitter：防止重试风暴
• idle_timeout：适合长时间运行但间歇性输出的任务

代价

• 同步无超时：无法安全取消同步操作
• 幂等要求：重试场景下节点必须幂等

替代方案

• 全局重试：在 invoke() 层重试整个图，而非单个节点
• 外部监控：通过 LangSmith 等外部系统监控失败率

参考来源

• 源码验证: libs/langgraph/langgraph/types.py — RetryPolicy / TimeoutPolicy
• 源码验证: libs/langgraph/langgraph/pregel/_retry.py — 重试实现