容错机制

当一个节点因外部 API 响应缓慢、瞬时网络错误或未处理的异常而失败时，LangGraph 提供了三种可组合的机制来应对：

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

使用 set_node_defaults 一次性为所有节点配置这些机制，而不是在每个 add_node 调用中重复配置。这些机制按固定顺序组合：当一个节点尝试引发任何异常（包括来自超时的 NodeTimeoutError）时，重试策略决定是否重试。只有在重试耗尽后，错误处理器才会运行。有关在超步（superstep）边界处干净地停止运行并在稍后恢复，请参阅优雅关停。

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

重试

重试策略根据异常类型和退避设置自动重新运行失败的节点尝试。向 add_node 传递 retry_policy=：

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

对于来自 requests 和 httpx 等流行 HTTP 库的异常，它只对 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 以扩展默认行为：

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),
)

检查重试状态

在节点内部使用执行信息来检查当前尝试次数。当主调用持续失败时，这对于切换到备用方案非常有用：

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。

超时

需要 langgraph>=1.2。

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

from datetime import timedelta
from langgraph.types import TimeoutPolicy

# 简单的挂钟时间限制
builder.add_node("call_model", call_model, timeout=60)
builder.add_node("call_model", call_model, timeout=timedelta(minutes=2))

# 单独的运行和空闲限制
builder.add_node(
    "call_model",
    call_model,
    timeout=TimeoutPolicy(run_timeout=120, idle_timeout=30),
)

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

运行超时

run_timeout 是对单次尝试的硬性挂钟时间限制。无论节点活动如何，它永远不会被刷新：

from langgraph.types import TimeoutPolicy

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

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

空闲超时

idle_timeout 是一个进度重置限制。它只在节点在指定持续时间内停止产生可观察的进度时才会触发——与 run_timeout 不同，当时钟重置时，节点会产生一个进度信号：

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() 调用。当您想要一个严格的空闲定义，不被话多的下属重置时，这非常有用：

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

手动心跳

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

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 与重试策略结合使用开箱即用——超时时钟在每次新尝试时重置，并且超时尝试的写入在下次重试前被清除：

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 上传递一个超时，以覆盖该特定推送的静态超时：

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 时设置）。这允许您在节点上设置默认超时，并为单个调用收紧它。

错误处理

需要 langgraph>=1.2。

错误处理器在节点失败且所有重试耗尽后运行。它接收当前状态，可以更新状态或使用 Command 路由到不同的节点。这对于补偿流程（Saga 模式）非常有用，您可以在其中优雅地恢复，而不是中止整个图。向 add_node 传递 error_handler=：

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 的模式相同）：

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 / 补偿模式：

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 来补偿，而不是中止图。

恢复安全的失败

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

与 `interrupt()` 的行为

在节点内部引发的 interrupt() 不会路由到错误处理器。中断使用 GraphBubbleUp 机制来暂停图执行，用于人在环中的工作流程，绕过重试策略和错误处理器。图照常暂停。

子图失败

如果一个节点包装了一个子图，并且子图引发了未处理的异常，该异常将浮出到父节点。如果父节点有一个错误处理器，处理器将在 error.error 中包含子图的异常并被触发。

图默认值

需要 langgraph>=1.2。

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

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()：

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

默认错误处理器

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

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:
    # 在由 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)  # 使用 mark_process_failed
    .add_node(
        "charge_payment",
        charge_payment,
        error_handler=refund_payment,  # 覆盖图形范围默认值
    )
    .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 作为可选的第三个参数：

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= 参数：

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，缓冲的写入被清除，重试策略决定是否重试。

优雅关停

协作关停允许您在当前超步完成后停止正在运行的图形运行，并保存一个可恢复的检查点。这对于处理 SIGTERM 信号或任何需要回收资源而不丢失工作的外部监督器非常有用。

需要 langgraph>=1.2。

创建一个 RunControl 并将其作为 control= 传递给 invoke 或 stream。从任何线程调用 request_drain() 以表示运行应停止：

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

control = RunControl()

# 在信号处理器或监督器中：
# control.request_drain("sigterm")

try:
    result = graph.invoke(inputs, config, control=control)
except GraphDrained as e:
    # 图提前停止并保存了一个检查点。
    # 使用相同的配置稍后恢复。
    print(f"Drained: {e.reason}")

语义

排空是协作的，并在超步之间进行，不会抢占已经运行的工作：

场景	行为
节点执行中	运行到完成。排空在下一个超步生效。
节点当前正在重试的重试策略	重试循环运行到耗尽或成功。排空在之后生效。
图在排空的同一滴答自然完成	正常返回。检查 `control.drain_requested` 以区分正常运行。
还有更多超步	引发 `GraphDrained(reason)`。检查点被保存并可恢复。
子图请求排空	`GraphDrained` 通过父级向上冒泡，并在其自身的下一个超步边界处停止。

排空后恢复

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

result = graph.invoke(None, config)

在节点内读取排空状态

通过 runtime 参数访问排空状态，以在达到超步边界之前调整节点行为：

from langgraph.runtime import Runtime

async def my_node(state: State, runtime: Runtime) -> State:
    if runtime.drain_requested:
        # 跳过昂贵的工作并返回最小结果
        return {"status": "skipped", "reason": runtime.drain_reason}
    return {"status": await do_work()}

SIGTERM 钩子模式

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

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)
    # 在下次启动时使用相同的配置恢复

request_drain() 不会取消运行中的 asyncio 任务或终止线程。要设置硬上限，请将排空与优雅超时和任务取消结合使用。

限制

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

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.

Get started

Capabilities

Production

Frontend

LangGraph APIs

重试

默认行为

参数

自定义重试逻辑

检查重试状态

超时

运行超时

空闲超时

进度信号

心跳模式

手动心跳

NodeTimeoutError

使用 Send 的动态超时

错误处理

NodeError

使用 Command 进行路由

恢复安全的失败

与 `interrupt()` 的行为

子图失败

图默认值

优先级

默认错误处理器

适用性矩阵

范围

函数式 API

优雅关停

语义

排空后恢复

在节点内读取排空状态

SIGTERM 钩子模式

限制

​重试

​默认行为

​参数

​自定义重试逻辑

​检查重试状态

​超时

​运行超时

​空闲超时

​进度信号

​心跳模式

​手动心跳

​NodeTimeoutError

​使用 Send 的动态超时

​错误处理

​NodeError

​使用 Command 进行路由

​恢复安全的失败

​与 interrupt() 的行为

​子图失败

​图默认值

​优先级

​默认错误处理器

​适用性矩阵

​范围

​函数式 API

​优雅关停

​语义

​排空后恢复

​在节点内读取排空状态

​SIGTERM 钩子模式

​限制

重试

默认行为

参数

自定义重试逻辑

检查重试状态

超时

运行超时

空闲超时

进度信号

心跳模式

手动心跳

NodeTimeoutError

使用 Send 的动态超时

错误处理

NodeError

使用 Command 进行路由

恢复安全的失败

与 `interrupt()` 的行为

子图失败

图默认值

优先级

默认错误处理器

适用性矩阵

范围

函数式 API

优雅关停

语义

排空后恢复

在节点内读取排空状态

SIGTERM 钩子模式

限制