容错 - Docs by LangChain中文

当节点因外部 API 响应缓慢、短暂的网络错误或未处理的异常而失败时，LangGraph 为您提供三种可组合的响应机制：

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

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

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

重试

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

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 可扩展默认行为：

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

检查重试状态

在节点内部使用 runtime.execution_info 查看当前尝试次数。当主调用持续失败时，这对于切换到备选方案非常有用：

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 令牌、工具调用、链开始/结束等）

心跳模式

设置 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= 与 retry_policy= 结合使用可直接开箱即用——每次新尝试都会重置超时时钟，且在下次重试前会清除超时尝试的写入：

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 上传递 timeout=，以覆盖该次特定推送的目标节点静态超时：

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 上省略了 timeout=，则应用目标节点的超时设置（在 add_node 时设置）。这允许您在节点上设置默认超时，并针对个别调用收紧它。

错误处理

需要 langgraph>=1.2。

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

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

处理程序仅在 retry_policy 耗尽后触发，或者在未配置重试策略时立即触发。重试策略和错误处理程序保持解耦：独立配置何时重试以及何时进行补偿。

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_handler，处理程序会以 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 默认值特别有价值。处理程序接受在错误处理中描述的相同的 (state, error: NodeError) 签名：

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

class State(TypedDict):
    status: str

def always_failing(state: State) -> State:
    raise ValueError("something went wrong")

def default_handler(state: State, error: NodeError) -> State:
    return {"status": f"recovered from {error.node}: {error.error}"}

graph = (
    StateGraph(State)
    .set_node_defaults(
        retry_policy=RetryPolicy(max_attempts=2),
        error_handler=default_handler,
    )
    .add_node("always_failing", always_failing)
    .add_edge(START, "always_failing")
    .compile()
)

节点被重试两次，然后运行 default_handler。默认处理程序也接受 RunnableConfig 作为可选的第三个参数，如果您需要访问配置值（如 thread_id）：

from langchain_core.runnables import RunnableConfig

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

适用性矩阵

并非所有默认值都适用于所有节点类型。错误处理程序节点（通过 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，清除缓冲的写入，并由重试策略决定是否重试。

优雅关闭

需要 langgraph>=1.2。

优雅关闭允许您协作式地停止正在运行的图（在当前超步完成后），并保存一个可恢复的检查点。这对于处理 SIGTERM 信号或任何需要回收资源而不丢失工作的外部监督程序很有用。创建一个 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}")

语义

Drain 是协作式的，在超步之间生效，绝不抢占正在运行的工作：

场景	行为
节点正在执行	运行至完成。Drain 在下一个超步生效。
节点当前正在重试	重试循环运行至耗尽或成功。Drain 之后生效。
图在与 drain 同一时刻自然完成	正常返回。检查 `control.drain_requested` 以区分正常运行。
还有更多超步	抛出 `GraphDrained(reason)`。检查点已保存且可恢复。
子图请求 drain	`GraphDrained` 冒泡到父图，并在其自身的下一个超步边界处停止父图。

Drain 后恢复

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

result = graph.invoke(None, config)

在节点内读取 drain 状态

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

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 任务或终止线程。如需硬性上限，请将 drain 与优雅超时和任务取消结合使用。

限制

仅限 Python：超时和错误处理程序在 JavaScript/TypeScript SDK 中不可用。重试策略在 Python 和 TypeScript 中均可使用。
超时仅限异步：带有 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.

​重试

​默认行为

​参数

​自定义重试逻辑

​检查重试状态

​超时

​运行超时

​空闲超时

​进度信号

​心跳模式

​手动心跳

​NodeTimeoutError

​使用 Send 的动态超时

​错误处理

​NodeError

​使用 Command 进行路由

​支持恢复的故障

​与 interrupt() 的行为

​子图故障

​图默认值

​优先级

​默认错误处理程序

​适用性矩阵

​作用域

​函数式 API

​优雅关闭

​语义

​Drain 后恢复

​在节点内读取 drain 状态

​SIGTERM 钩子模式

​限制

重试

默认行为

参数

自定义重试逻辑

检查重试状态

超时

运行超时

空闲超时

进度信号

心跳模式

手动心跳

NodeTimeoutError

使用 Send 的动态超时

错误处理

NodeError

使用 Command 进行路由

支持恢复的故障

与 `interrupt()` 的行为

子图故障

图默认值

优先级

默认错误处理程序

适用性矩阵

作用域

函数式 API

优雅关闭

语义

Drain 后恢复

在节点内读取 drain 状态

SIGTERM 钩子模式

限制