容错

在 LangGraph 中配置每个节点的超时、重试和错误处理器。

当某个节点失败时 —— 无论是因为外部 API 缓慢、短暂的网络错误，还是未处理的异常 —— LangGraph 都为你提供三种可组合的机制来应对：

重试：根据异常类型和退避设置自动重新运行失败的尝试
超时：限制单次尝试可以运行多长时间
错误处理：在所有重试都耗尽后运行恢复函数

使用 set_node_defaults 为所有节点一次性配置这些机制，而不必在每次 add_node 调用时重复它们。

它们按固定顺序组合：当一次节点尝试抛出任何异常时（包括来自超时的 NodeTimeoutError），由重试策略决定是否重试。只有在重试耗尽之后，错误处理器才会运行。

关于如何在 super-step 边界干净地停止一次运行并在稍后恢复，请参见优雅关闭。

NOTE

每个节点的超时和节点级错误处理器需要 langgraph>=1.2。

%%{init:{'theme':'base','themeVariables':{'lineColor':'#40668D','primaryColor':'#E5F4FF','primaryTextColor':'#030710','primaryBorderColor':'#006DDD'}}}%%
flowchart LR
    start([Attempt starts]) --> exec[Run node]
    exec -->|"success"| done([Continue graph])
    exec -->|"any exception<br/>including NodeTimeoutError"| retry{retry_policy<br/>matches?}
    retry -->|"yes, attempts left"| exec
    retry -->|"exhausted or absent"| handler{error_handler?}
    handler -->|"yes"| run_handler["Invoke handler<br/>with NodeError"]
    run_handler --> route([Update state +<br/>Command goto])
    handler -->|"no"| bubble([Exception<br/>bubbles up])

    classDef process fill:#E5F4FF,stroke:#006DDD,stroke-width:2px,color:#030710
    classDef decision fill:#FDF3FF,stroke:#7E65AE,stroke-width:2px,color:#504B5F
    classDef alert fill:#F8E8E6,stroke:#B27D75,stroke-width:2px,color:#634643
    classDef output fill:#EBD0F0,stroke:#885270,stroke-width:2px,color:#441E33

    class exec,run_handler process
    class retry,handler decision
    class bubble alert
    class done,route,start output

重试

重试策略会根据异常类型和退避设置自动重新运行失败的节点尝试。

向 add_node 传入 retry_policy=：

python

from langgraph.types import RetryPolicy

builder.add_node(
    "call_api",
    call_api,
    retry_policy=RetryPolicy(max_attempts=3),
)

默认行为

默认情况下，retry_on 使用 default_retry_on，它会对任何异常重试，但以下异常（及其子类）除外：

ValueError
TypeError
ArithmeticError
ImportError
LookupError
NameError
SyntaxError
RuntimeError
ReferenceError
StopIteration
StopAsyncIteration
OSError

对于来自流行 HTTP 库（例如 requests 和 httpx）的异常，它仅对 5xx 状态码重试。NodeTimeoutError 默认可重试。

参数

参数	类型	默认值	描述
`max_attempts`	`int`	`3`	最大尝试次数，包括第一次。
`initial_interval`	`float`	`0.5`	第一次重试前的秒数。
`backoff_factor`	`float`	`2.0`	每次重试后应用于间隔的乘数。
`max_interval`	`float`	`128.0`	重试之间的最大秒数。
`jitter`	`bool`	`True`	为间隔添加随机抖动。
`retry_on`	`type[Exception] \| Sequence[type[Exception]] \| Callable[[Exception], bool]`	`default_retry_on`	要重试的异常，或一个对可重试异常返回 `True` 的可调用对象。

自定义重试逻辑

向 retry_on 传入一个可调用对象或异常类型。导入 default_retry_on 以扩展默认行为：

python

from langgraph.types import RetryPolicy, default_retry_on

def custom_retry_on(exc: BaseException) -> bool:
    if isinstance(exc, MyCustomError):
        return False
    return default_retry_on(exc)

builder.add_node(
    "call_api",
    call_api,
    retry_policy=RetryPolicy(max_attempts=3, retry_on=custom_retry_on),
)

检查重试状态

在节点内使用执行信息来检查当前尝试次数。当主调用持续失败时，切换到回退方案会很有用：

python

from langgraph.graph import StateGraph, START, END
from langgraph.runtime import Runtime
from langgraph.types import RetryPolicy
from typing_extensions import TypedDict

class State(TypedDict):
    result: str

def my_node(state: State, runtime: Runtime) -> State:
    if runtime.execution_info.node_attempt > 1:  
        return {"result": call_fallback_api()}
    return {"result": call_primary_api()}

builder = StateGraph(State)
builder.add_node("my_node", my_node, retry_policy=RetryPolicy(max_attempts=3))
builder.add_edge(START, "my_node")
builder.add_edge("my_node", END)

execution_info 暴露以下字段：

属性	类型	描述
`node_attempt`	`int`	当前尝试次数（从 1 开始）。第一次尝试为 `1`，第一次重试为 `2`，依此类推。
`node_first_attempt_time`	`float \| None`	第一次尝试开始时的 Unix 时间戳。在重试之间保持不变。
`thread_id`	`str \| None`	当前执行的线程 ID。没有检查点器时为 `None`。
`run_id`	`str \| None`	当前运行的 ID。未在配置中提供时为 `None`。
`checkpoint_id`	`str`	当前执行的检查点 ID。
`task_id`	`str`	当前执行的任务 ID。

即使没有重试策略，execution_info 也可用 —— node_attempt 默认为 1。

超时

NOTE

需要 langgraph>=1.2。

add_node 上的 timeout= 参数限制单次节点尝试可以运行多长时间。传入一个数字（秒）、一个 timedelta，或一个 TimeoutPolicy 以分别设置运行和空闲限制：

python

from datetime import timedelta
from langgraph.types import TimeoutPolicy

# Simple wall-clock cap
builder.add_node("call_model", call_model, timeout=60)
builder.add_node("call_model", call_model, timeout=timedelta(minutes=2))

# Separate run and idle limits
builder.add_node(
    "call_model",
    call_model,
    timeout=TimeoutPolicy(run_timeout=120, idle_timeout=30),
)

WARNING

节点超时仅适用于 async 节点。带有 timeout 的同步节点会在编译时被拒绝。若要包装阻塞式 I/O，请在 async 节点内使用 asyncio.to_thread。

运行超时

run_timeout 是对单次尝试的硬性墙钟时间上限。无论节点是否有活动，它都永远不会被刷新：

python

from langgraph.types import TimeoutPolicy

builder.add_node(
    "call_model",
    call_model,
    timeout=TimeoutPolicy(run_timeout=120),
)

当超过该限制时，LangGraph 会抛出 NodeTimeoutError，清除该失败尝试产生的所有写入，并让重试策略决定是否重试。

空闲超时

idle_timeout 是一个随进度重置的上限。仅当节点在指定时长内停止产生可观察的进度时才会触发 —— 与 run_timeout 不同，只要节点产生进度信号，时钟就会重置：

python

builder.add_node(
    "call_model",
    call_model,
    timeout=TimeoutPolicy(idle_timeout=30),
)

你可以同时设置 run_timeout 和 idle_timeout。先触发的那个会取消该次尝试。

进度信号

在默认的 refresh_on="auto" 下，空闲时钟会在以下任一情况下重置：

通过 CONFIG_KEY_SEND 进行的状态写入
流输出（产出的异步流分块）
子任务调度
运行时流写入器调用
来自该节点或其任何后代节点的任意 LangChain 回调事件（LLM token、工具调用、链开始/结束等）

心跳模式

设置 refresh_on="heartbeat" 可将刷新来源缩小为仅显式的 runtime.heartbeat() 调用。当你想要一个严格、不会被话多的下属重置的空闲定义时，这很有用：

python

builder.add_node(
    "call_model",
    call_model,
    timeout=TimeoutPolicy(idle_timeout=30, refresh_on="heartbeat"),
)

手动心跳

对于不会自然发出进度信号的长时间运行工作，调用 runtime.heartbeat() 手动重置空闲时钟：

python

from langgraph.graph import StateGraph, START, END
from langgraph.runtime import Runtime
from langgraph.types import TimeoutPolicy
from typing_extensions import TypedDict

class State(TypedDict):
    result: str

async def long_running_node(state: State, runtime: Runtime) -> State:
    for batch in fetch_batches():
        process(batch)
        runtime.heartbeat()  
    return {"result": "done"}

builder = StateGraph(State)
builder.add_node(
    "long_running_node",
    long_running_node,
    timeout=TimeoutPolicy(idle_timeout=30, refresh_on="heartbeat"),
)
builder.add_edge(START, "long_running_node")
builder.add_edge("long_running_node", END)

runtime.heartbeat() 在空闲计时的尝试之外是空操作，因此你可以无条件调用它。

NodeTimeoutError

当超时触发时，LangGraph 会抛出 NodeTimeoutError，其中包含关于触发了哪个限制的结构化上下文：

属性	类型	描述
`node`	`str`	执行超时的节点名称。
`elapsed`	`float`	超时触发前经过的秒数。
`kind`	`Literal["idle", "run"]`	触发的是哪个超时。
`idle_timeout`	`float \| None`	已配置的空闲超时（秒），如果有的话。
`run_timeout`	`float \| None`	已配置的运行超时（秒），如果有的话。

NodeTimeoutError 默认可重试。将 timeout 与重试策略结合使用可开箱即用 —— 超时时钟会在每次新的尝试时重置，且来自超时尝试的写入会在下一次重试之前被清除：

python

from langgraph.types import RetryPolicy, TimeoutPolicy

builder.add_node(
    "call_model",
    call_model,
    timeout=TimeoutPolicy(idle_timeout=30),
    retry_policy=RetryPolicy(max_attempts=3),
)

使用 Send 的动态超时

当使用 Send 动态派发节点时（例如在 map-reduce 模式中），你可以直接在 Send 上传入超时，以覆盖该特定推送中目标节点的静态超时：

python

from langgraph.types import Send, TimeoutPolicy

def fan_out(state: OverallState):
    return [
        Send("process_item", {"item": item}, timeout=TimeoutPolicy(idle_timeout=15))
        for item in state["items"]
    ]

如果 Send 上省略了超时，则适用目标节点的超时（在 add_node 时设置）。这让你可以在节点上设置默认超时，并为单个调用收紧它。

错误处理

NOTE

需要 langgraph>=1.2。

错误处理器在节点失败且所有重试都耗尽之后运行。它接收当前状态，并且可以使用 Command 更新状态或路由到其他节点。这对于补偿流程（Saga 模式）很有用，你希望优雅地恢复，而不是中止整个图。

向 add_node 传入 error_handler=：

python

from langgraph.errors import NodeError
from langgraph.types import Command, RetryPolicy
from langgraph.graph import StateGraph, START
from typing_extensions import TypedDict

class State(TypedDict):
    status: str

def charge_payment(state: State) -> State:
    raise RuntimeError("payment gateway timeout")

def payment_error_handler(state: State, error: NodeError) -> Command:
    return Command(
        update={"status": f"compensated: {error.error}"},
        goto="finalize",
    )

def finalize(state: State) -> State:
    return state

graph = (
    StateGraph(State)
    .add_node(
        "charge_payment",
        charge_payment,
        retry_policy=RetryPolicy(max_attempts=3, retry_on=ConnectionError),
        error_handler=payment_error_handler,
    )
    .add_node("finalize", finalize)
    .add_edge(START, "charge_payment")
    .compile()
)

只有在重试策略耗尽之后，处理器才会触发；如果未配置重试策略，则立即触发。重试策略和错误处理器保持解耦：可独立配置何时重试以及何时补偿。

NodeError

错误处理器通过类型化的 error: NodeError 参数接收失败上下文，该参数由类型注解注入（与 runtime: Runtime 模式相同）：

python

from langgraph.errors import NodeError

def my_handler(state: State, error: NodeError) -> Command:
    print(f"Node {error.node} failed with: {error.error}")
    return Command(update={"status": "recovered"}, goto="next_step")

NodeError 是一个包含两个字段的冻结数据类：

属性	类型	描述
`node`	`str`	执行失败的节点名称。
`error`	`BaseException`	失败节点抛出的异常。

error: NodeError 参数是可选的。不需要失败上下文的处理器可以使用更简单的签名，例如 (state) 或 (state, runtime)。

使用 Command 路由

错误处理器可以返回一个 Command 来更新状态并路由到特定节点，从而实现 Saga / 补偿模式：

python

from langgraph.errors import NodeError
from langgraph.types import Command, RetryPolicy
from langgraph.graph import StateGraph, START
from typing_extensions import TypedDict

class State(TypedDict):
    status: str

def reserve_inventory(state: State) -> State:
    return {"status": "reserved"}

def charge_payment(state: State) -> State:
    raise RuntimeError("payment timeout")

def payment_error_handler(state: State, error: NodeError) -> Command:
    return Command(
        update={"status": f"compensated_after_{error.node}: {error.error}"},
        goto="finalize",
    )

def finalize(state: State) -> State:
    return state

graph = (
    StateGraph(State)
    .add_node("reserve_inventory", reserve_inventory)
    .add_node(
        "charge_payment",
        charge_payment,
        retry_policy=RetryPolicy(max_attempts=3, retry_on=ConnectionError),
        error_handler=payment_error_handler,
    )
    .add_node("finalize", finalize)
    .add_edge(START, "reserve_inventory")
    .add_edge("reserve_inventory", "charge_payment")
    .compile()
)

charge_payment 会对 ConnectionError 最多重试 3 次。如果重试耗尽（或错误不是 ConnectionError），处理器会通过更新状态并路由到 finalize 来进行补偿，而不是中止整个图。

可安全恢复的失败

NOTE

失败的来源会被纳入检查点。如果图在节点失败之后、处理器完成之前被中断或进程崩溃，那么当图从其检查点恢复时，处理器会看到相同的 NodeError 上下文。

与 `interrupt()` 一起使用时的行为

WARNING

在节点内抛出的 interrupt() 不会被路由到错误处理器。中断使用 GraphBubbleUp 机制为人在回路工作流暂停图执行，同时绕过重试策略和错误处理器。图会照常暂停。

子图失败

如果一个节点包装了一个子图，而该子图抛出了未处理的异常，该异常会浮现到父节点。如果父节点有错误处理器，处理器会触发，子图的异常位于 error.error 中。

图默认值

NOTE

需要 langgraph>=1.2。

与其在每次 add_node 调用时重复相同的 retry_policy=、error_handler=、timeout= 或 cache_policy=，不如使用 set_node_defaults 在一个地方配置全图范围的默认值：

python

from langgraph.errors import NodeError
from langgraph.types import RetryPolicy, TimeoutPolicy
from langgraph.graph import StateGraph, START
from typing_extensions import TypedDict

class State(TypedDict):
    status: str

def default_error_handler(state: State, error: NodeError) -> State:
    return {"status": f"handled: {error.error}"}

graph = (
    StateGraph(State)
    .set_node_defaults(
        retry_policy=RetryPolicy(max_attempts=3),
        error_handler=default_error_handler,
        timeout=TimeoutPolicy(run_timeout=30),
    )
    .add_node("step_a", step_a)
    .add_node("step_b", step_b)
    .add_edge(START, "step_a")
    .compile()
)

现在 step_a 和 step_b 共享相同的重试策略、错误处理器和超时，而无需任何重复。

优先级

直接传给 add_node() 的每个节点的值总是覆盖由 set_node_defaults() 设置的默认值。默认值在 compile() 时解析，因此你可以在 add_node() 之前或之后以任意顺序调用 set_node_defaults()：

python

graph = (
    StateGraph(State)
    .set_node_defaults(error_handler=default_error_handler)
    .add_node("step_a", step_a)                                     # uses default_error_handler
    .add_node("step_b", step_b, error_handler=custom_error_handler) # uses custom_error_handler
    .add_edge(START, "step_a")
    .compile()
)

默认错误处理器

当每次图运行都映射到一个外部进程（例如后台作业记录），并且任何未处理的节点失败都应将那个进程标记为失败时，error_handler 默认值尤其有价值 —— 无需在每次 add_node 上重复 error_handler=。当某个步骤需要自己的逻辑时，每个节点的处理器仍然优先：

python

from langgraph.errors import NodeError
from langgraph.graph import StateGraph, START
from langgraph.types import Command, RetryPolicy
from typing_extensions import TypedDict

class State(TypedDict):
    process_id: str
    status: str

def fetch_data(state: State) -> State:
    return {"status": "fetched"}

def charge_payment(state: State) -> State:
    raise RuntimeError("payment timeout")

def finalize(state: State) -> State:
    return state

def mark_process_failed(state: State, error: NodeError) -> State:
    # Persist failure on the external process row keyed by process_id.
    return {"status": f"failed at {error.node}: {error.error}"}

def refund_payment(state: State, error: NodeError) -> Command:
    return Command(
        update={"status": f"compensated after {error.node}"},
        goto="finalize",
    )

graph = (
    StateGraph(State)
    .set_node_defaults(
        retry_policy=RetryPolicy(max_attempts=3),
        error_handler=mark_process_failed,
    )
    .add_node("fetch_data", fetch_data)  # uses mark_process_failed
    .add_node(
        "charge_payment",
        charge_payment,
        error_handler=refund_payment,  # overrides the graph-wide default
    )
    .add_node("finalize", finalize)
    .add_edge(START, "fetch_data")
    .add_edge("fetch_data", "charge_payment")
    .compile()
)

如果 fetch_data 在重试后失败，则运行 mark_process_failed。如果 charge_payment 在重试后失败，则改为运行 refund_payment，因为每个节点的处理器覆盖了默认值。

该处理器接受与错误处理中描述的相同的 (state, error: NodeError) 签名。如果你需要访问诸如 thread_id 之类的配置值，它还可以接受 RunnableConfig 作为可选的第三个参数：

python

from langchain_core.runnables import RunnableConfig

def mark_process_failed(
    state: State, error: NodeError, config: RunnableConfig
) -> State:
    thread_id = config["configurable"].get("thread_id")
    return {"status": f"failed on thread {thread_id}: {error.error}"}

适用性矩阵

并非所有默认值都适用于所有节点类型。错误处理器节点（通过 add_node(error_handler=...) 注册的节点）被排除在某些默认值之外，以防止不安全的行为：

`set_node_defaults` 参数	适用于普通节点	适用于错误处理器节点	理由
`retry_policy`	✅	✅	处理器在遇到短暂失败时应当重试
`timeout`	✅	✅	卡住的处理器应像卡住的普通节点一样被取消
`error_handler`	✅	❌	处理器绝不能捕获自身
`cache_policy`	✅	❌	缓存处理器的结果是不安全的

作用范围

在父图上设置的默认值不会被子图继承。每个图维护自己的默认值。

函数式 API

在函数式 API 中，@task 和 @entrypoint 上可以使用相同的 timeout= 和 retry_policy= 参数：

python

from langgraph.func import entrypoint, task
from langgraph.types import RetryPolicy, TimeoutPolicy

@task(
    timeout=TimeoutPolicy(idle_timeout=30),
    retry_policy=RetryPolicy(max_attempts=3),
)
async def call_api(url: str) -> str:
    response = await fetch(url)
    return response.text

@entrypoint(timeout=60)
async def my_workflow(inputs: dict) -> str:
    result = await call_api("https://api.example.com/data")
    return result

其行为与 add_node 完全相同：超时时抛出 NodeTimeoutError，缓冲的写入被清除，并由重试策略决定是否重试。

优雅关闭

协作式关闭让你可以在当前 super-step 完成后停止正在进行的图运行，并保存一个可恢复的检查点。这对于处理 SIGTERM 信号，或任何需要在不丢失工作的情况下回收资源的外部主管来说很有用。

NOTE

需要 langgraph>=1.2。

创建一个 RunControl，并将其作为 control= 传给 invoke 或 stream。从任意线程调用 request_drain() 来发出该运行应停止的信号：

python

from langgraph.runtime import RunControl
from langgraph.errors import GraphDrained

control = RunControl()

# In a signal handler or supervisor:
# control.request_drain("sigterm")

try:
    result = graph.invoke(inputs, config, control=control)
except GraphDrained as e:
    # The graph stopped early and saved a checkpoint.
    # Resume later with the same config.
    print(f"Drained: {e.reason}")

语义

Drain 是协作式的，在 super-step 之间运作，绝不会抢占已在运行的工作：

场景	行为
节点正在执行中	运行至完成。Drain 在下一个 super-step 生效。
带有重试策略的节点正在重试	重试循环运行至耗尽或成功。Drain 在此之后生效。
图在与 drain 相同的时刻自然结束	正常返回。检查 `control.drain_requested` 以区分普通运行。
还有更多 super-step 剩余	抛出 `GraphDrained(reason)`。检查点被保存且可恢复。
子图请求 drain	`GraphDrained` 向上冒泡穿过父图，并在其自己的下一个 super-step 边界处停止它。

drain 后恢复

使用相同的 thread_id，通过 invoke(None, config) 恢复已 drain 的运行：

python

result = graph.invoke(None, config)

在节点内读取 drain 状态

通过 runtime 参数访问 drain 状态，以在到达 super-step 边界之前调整节点行为：

python

from langgraph.runtime import Runtime

async def my_node(state: State, runtime: Runtime) -> State:
    if runtime.drain_requested:
        # Skip expensive work and return a minimal result
        return {"status": "skipped", "reason": runtime.drain_reason}
    return {"status": await do_work()}

SIGTERM 钩子模式

处理进程关闭的推荐模式：

python

import signal
from langgraph.runtime import RunControl
from langgraph.errors import GraphDrained

control = RunControl()
signal.signal(signal.SIGTERM, lambda *_: control.request_drain("sigterm"))

try:
    result = graph.invoke(inputs, config, control=control)
except GraphDrained as e:
    log.info("graph drained: %s", e.reason)
    # Resume on next startup with the same config

NOTE

request_drain() 不会取消正在运行的 asyncio 任务，也不会终止线程。若需要硬性上限，请将 drain 与优雅超时和任务取消配合使用。

限制

超时仅限 async：带有 timeout 的同步节点会在编译时被拒绝。
每个节点一个处理器：每个节点最多可以有一个 error_handler。
处理器失败会向上冒泡：如果错误处理器本身抛出异常，该异常会像节点没有处理器一样传播。
set_node_defaults 不会被子图继承：每个图独立管理自己的默认值。

数据集

创建数据集

运行评估

评估技术

定义评估目标

打分方法

实验配置

多模态评估

分析实验结果

微调

教程

常见数据类型

评估器类型

UI

SDK

框架与集成

改进评估器

反馈

追踪设置

集成

手动埋点

配置与故障排除

项目与环境设置

高级追踪技术

数据与隐私

故障排除指南

查看与管理追踪

批量导出追踪数据

轨迹视图

数据类型参考

监控与告警

在线评估器

自动化

开发你的应用

设置依赖

持久化

能力

助手

运行

并发消息处理

如何构建

鉴权与访问控制

自定义鉴权教程

服务器定制

静态加密

替换内置后端

扩展 HTTP 服务器

请求头与日志

部署指南

参考

配置

参考

提示词

创建与管理提示词

连接模型

上下文中枢

教程

全栈 Web 应用

Agent Server API

控制平面 API

计费与用量

参考

按云服务商起步

用 Terraform 部署

AWS

GCP

Azure

安装实例

配置实例

连接外部服务

平台鉴权与访问控制

管理实例

自托管可观测性

混合部署

脚本

参考

组织