avatar yuguan bitou
发布日期

第 14 讲 · 重试、超时与错误路由

langgraphretryerror-handling

学习目标

  • 看懂 run_with_retry / arun_with_retry 的重试循环。
  • 掌握 RetryPolicy / TimeoutPolicy 的语义与配置。
  • 理解任务失败后的错误路由（ERROR channel + 错误处理节点）。

任务执行的最内层：run_with_retry

libs/langgraph/langgraph/pregel/_retry.py。第 12 讲 runner 提交的 run_with_retry （573–683 行，async 版 arun_with_retry 685–841 行）是节点真正被调用的地方，外面裹了重试。

精简流程：

flowchart TD
    A[run_with_retry] --> B[task.writes.clear 清空缓冲]
    B --> C[task.proc.invoke 执行节点]
    C --> D{异常?}
    D -->|无| E[成功, 返回]
    D -->|ParentCommand| F[Command 路由, 上抛]
    D -->|GraphBubbleUp/Interrupt| G[中断冒泡, 上抛 第18讲]
    D -->|普通异常| H{匹配 RetryPolicy?}
    H -->|是 且未超次数| I[指数退避 + jitter<br/>设 RESUMING=True 重试]
    I --> B
    H -->|否| J[抛出, 交 commit 路由 ERROR]

关键点：

  • 每次尝试前 task.writes.clear()：保证重试不会叠加上一次失败的半成品写（幂等性基础）。
  • 特殊异常不重试：ParentCommand（Command 路由）、GraphInterrupt/GraphBubbleUp（中断） 要原样上抛，不能当普通错误吞掉。
  • 重试是单任务粒度：一个任务重试不影响同超步其他并行任务。

RetryPolicy：重试策略

types.py，RetryPolicy（416–435 行）。配置项：

  • max_attempts：最多尝试次数。
  • initial_interval / backoff_factor / max_interval：指数退避参数。
  • jitter：随机抖动，避免重试风暴。
  • retry_on：哪些异常该重试（异常类型、元组或判定函数）。_should_retry_on（841+ 行）做匹配。

配置方式（per-node 或全图默认）：

from langgraph.types import RetryPolicy

g.add_node("call_api", call_api, retry_policy=RetryPolicy(max_attempts=3))
# 或编译期全图默认（第 8 讲 _node_defaults）

典型：给"调用外部 LLM/HTTP 工具"的节点加 retry_on=(ConnectionError, RateLimitError)， 让瞬时故障自动恢复。

TimeoutPolicy：超时

types.py，TimeoutPolicy（449–512 行）。注意一个重要限制：

同步节点不支持单任务 timeout（_retry.py 580–583 行）。

超时主要走 async 路径（arun_with_retry 内的 _arun_with_timeout），靠 _ResolvedTimeout / _TimedAttemptScope（64–273 行）实现。TimeoutPolicy 区分：

  • run_timeout：整个任务的硬超时。
  • idle_timeout：多久没进展算超时（配合 heartbeat）。
  • heartbeat：节点可主动"报活"刷新 idle 计时（流式 LLM 长输出常用）。

此外 Pregel.step_timeout（第 9 讲主循环传给 runner.tick）是整个超步的超时，与单任务 timeout 不同层。

错误路由：失败后去哪

任务最终失败（重试耗尽且非中断）后，回到第 12 讲的 commit：

  1. 写 ERROR channel（WRITES_IDX_MAP[ERROR] = -1，记录失败信息）。
  2. 若该节点配置了错误处理节点（第 8 讲 error_handler_node / node_error_handler_map）， runner 会 schedule_error_handler 调度那个 handler 节点接手。
  3. 否则异常经 _panic_or_proceed（650–698 行）汇总后向上抛，整个 run 失败。
flowchart LR
    F[任务失败] --> C[commit 写 ERROR channel]
    C --> H{有 error_handler_node?}
    H -->|有| EH[调度错误处理节点]
    H -->|无| P[_panic_or_proceed 上抛]

_should_stop_others（616–635 行）决定致命异常时是否取消同超步其他 future （避免在已经要失败的情况下浪费资源）。

使用场景

  • 健壮的工具调用：外部 API 节点 → RetryPolicy(max_attempts=3, retry_on=(TimeoutError, RateLimitError))。
  • 防卡死：async LLM 节点 → TimeoutPolicy(run_timeout=..., heartbeat=...)， 配合流式输出在长响应时报活。
  • 优雅降级：给关键节点配 error_handler 节点，失败时走兜底逻辑（如返回缓存答案），而不是整个 run 崩。
  • 幂等性提醒：因为重试会清空并重跑，节点内的外部副作用（写库、发消息）要自己保证幂等， 框架只保证 task.writes 不叠加。

动手实验

  1. 写一个前两次抛异常、第三次成功的节点，配 RetryPolicy(max_attempts=3)， 在 run_with_retry 打断点看退避与重试。
  2. 写一个 async 慢节点 + TimeoutPolicy(run_timeout=1)，观察超时行为。
  3. 给一个必失败的节点配 error_handler 节点，验证失败被兜底接管而非整体崩溃。

阅读作业

  • 精读 _retry.py 的 run_with_retry（573–683 行）与 arun_with_retry（685–841 行）。
  • 精读 types.py 的 RetryPolicy（416–435 行）与 TimeoutPolicy（449–512 行）。
  • 回看 _runner.py 的 commit（574–613 行）与 _panic_or_proceed（650–698 行）。

小结

  • run_with_retry 在最内层裹住节点：每次重试前清空 task.writes，特殊异常（Command/Interrupt）不重试。
  • RetryPolicy（退避 + retry_on）/ TimeoutPolicy（run/idle/heartbeat），同步节点不支持单任务 timeout。
  • 失败 → 写 ERROR channel → 错误处理节点接手 或 上抛；副作用需自己保证幂等。

模块四完结。下一模块进入持久化与高级控制：检查点、序列化、续跑、时间旅行、人审。