当节点失败时——无论是外部 API 响应缓慢、瞬时网络错误还是未处理的异常——LangGraph 为您提供了三种可组合的响应机制：

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

使用 **set_node_defaults** 来一次性为所有节点配置这些机制，而不是在每个 add_node call.

这些机制按固定顺序组合：当节点尝试引发任何异常（包括超时导致的 NodeTimeoutError）时，重试策略决定是否重试。只有在重试耗尽后，错误处理程序才会运行。

如需在超级步骤边界干净地停止运行并稍后恢复，请参阅 优雅关闭.

%%{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

重试

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

传入 retry_policy= to 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 默认情况下可重试。

参数

自定义重试逻辑

传递可调用对象或异常类型到 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:  # [!code highlight]
        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 公开以下字段：

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

超时

timeout= ] 上的 add_node 参数限制了单个节点尝试的运行时间。可以传递一个数字（秒），或 timedelta, or a TimeoutPolicy 来分别设置运行和空闲限制：

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

运行超时

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
• - 流输出（yielded async stream chunks）
• - 子任务调度
• - 运行时流写入器调用
• - Any LangChain callback event from the node or its descendants (LLM tokens, tool calls, chain start/end, etc.)

心跳模式

设置 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()  # [!code highlight]
    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】并附带关于哪个限制被触发的结构化上下文：

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】时设置）。这样你可以在节点上设置默认超时，并为单个调用收紧超时。

错误处理

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

传递 error_handler= to 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()
)

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

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 是一个冻结数据类，有两个字段：

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

使用 Command 路由

错误处理器可以返回 Command to update state and route to a specific node, enabling Saga / compensation patterns:

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

可恢复的失败

行为 interrupt()

子图失败

如果节点包装了子图且子图引发未处理的异常，该异常会浮出到父节点。如果父节点有错误处理器，处理器会用子图的异常触发 error.error.

图默认值

不要重复相同的 retry_policy=, error_handler=, timeout=, or cache_policy= 在每次 add_node 调用中使用 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() 时解析，因此你可以按任意顺序调用 set_node_defaults() 之前或之后 add_node() ：

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 默认值在你希望为任何没有自己处理器的失败节点提供一个统一的恢复函数时特别有用。处理器接受相同的 (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=...)注册的错误处理器节点被排除在某些默认值之外，以防止不安全的行为：

作用域

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

函数式 API

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

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 信号或任何需要回收资源而不会丢失工作的外部监督程序很有用。

创建一个 RunControl 并将其作为 control= to invoke or stream。调用 request_drain() 从任何线程发送信号以指示运行应该停止：

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}")

语义

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

引流后恢复

使用以下方式恢复已引流的运行 invoke(None, config) 使用相同的 thread_id:

result = graph.invoke(None, config)

在节点内读取引流状态

通过以下方式访问引流状态 runtime 参数在达到超级步骤边界之前调整节点行为：

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 钩子模式

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

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

限制

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