容错 | LangChain 中文文档

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

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

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

使用 setNodeDefaults 一次性为所有节点配置这些机制，而不必在每次 addNode 调用中重复配置。

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

如需在超级步骤（superstep）边界干净地停止一次运行并在之后恢复，请参阅优雅关闭。

INFO

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

INFO

节点级超时和节点级错误处理程序需要 @langchain/langgraph>=1.4.0。

重试

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

将 retry_policy= 传给 add_node：

python

from langgraph.types import RetryPolicy

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

将 retryPolicy 传给 addNode：

typescript

import { StateGraph } from "@langchain/langgraph";

const graph = new StateGraph(State)
  .addNode("callApi", callApi, { retryPolicy: { maxAttempts: 3 } })
  .compile();

默认行为

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

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

对于来自 requests 和 httpx 等流行 HTTP 库的异常，它只在 5xx 状态码时重试。NodeTimeoutError 默认可重试。

重试是可选加入的。只有当节点配置了 retryPolicy（直接配置，或通过 setNodeDefaults 的图默认值配置）时，节点才会重试。空策略（{}）就足够了。如果没有策略，第一次失败就会结束尝试，LangGraph 不会调用 retryOn。

如果策略省略了 retryOn，LangGraph 会使用内置处理程序来重试抛出的错误，但以下情况除外：

中止与取消错误：error.name === "AbortError"，或 error.message 以 "Cancel" 或 "AbortError" 开头
GraphValueError，通过 error.name 匹配
连接被中止：error.code === "ECONNABORTED"
状态码为 400、401、402、403、404、405、406、407 或 409 的 HTTP 客户端错误，从 error.response?.status 或 error.status 读取，适用于 fetch、Axios 及类似客户端
OpenAI 风格配额错误：error.error?.code === "insufficient_quota"

包括 408 和 5xx 响应在内的其他 HTTP 状态默认可重试，除非您覆盖 retryOn。NodeTimeoutError 不在该阻止列表中，因此当配置了重试策略时它是可重试的。

有些失败会绕过 retryOn。图控制流错误（例如 GraphInterrupt 和 Command 路由）会向上传播而不重试。中止运行的信号也会停止重试循环，即使 retryOn 会返回 true。

参数

参数	类型	默认值	说明
`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` 的可调用对象。

参数	类型	默认值	说明
`maxAttempts`	`number`	`3`	最大尝试次数，包括第一次。
`initialInterval`	`number`	`500`	第一次重试前的等待毫秒数。
`backoffFactor`	`number`	`2.0`	每次重试后应用于间隔的乘数。
`maxInterval`	`number`	`128000`	重试之间的最大毫秒数。
`jitter`	`boolean`	`true`	为间隔添加随机抖动。
`retryOn`	`(error: unknown) => boolean`	内置处理程序（设置了策略时）	对可重试异常返回 `true` 的可调用对象。仅在配置了 `retryPolicy` 时使用。
`logWarning`	`boolean`	`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),
)

将可调用对象传给 retryOn。与 Python 不同，这里没有导出的 defaultRetryOn 辅助函数——请自行实现谓词：

typescript

import { StateGraph } from "@langchain/langgraph";

class MyCustomError extends Error {}

const graph = new StateGraph(State)
  .addNode("callApi", callApi, {
    retryPolicy: {
      maxAttempts: 3,
      retryOn: (error: unknown) => {
        if (error instanceof MyCustomError) return false;
        // 对其他错误进行重试
        return true;
      },
    },
  })
  .compile();

检查重试状态

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

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。

typescript

import { StateGraph, StateSchema, START, END, type Runtime } from "@langchain/langgraph";
import * as z from "zod";

const State = new StateSchema({
  result: z.string(),
});

const myNode = async (state: typeof State.State, runtime: Runtime<typeof State>) => {
  if ((runtime.executionInfo?.nodeAttempt ?? 1) > 1) {  
    return { result: await callFallbackApi() };
  }
  return { result: await callPrimaryApi() };
};

const graph = new StateGraph(State)
  .addNode("myNode", myNode, { retryPolicy: { maxAttempts: 3 } })
  .addEdge(START, "myNode")
  .addEdge("myNode", END)
  .compile();

executionInfo 暴露以下字段：

属性	类型	说明
`nodeAttempt`	`number`	当前尝试次数（从 1 开始）。第一次尝试为 `1`，第一次重试为 `2`，依此类推。
`nodeFirstAttemptTime`	`number \| undefined`	第一次尝试开始时的 Unix 时间戳（毫秒）。在重试期间保持不变。
`threadId`	`string \| undefined`	当前执行的线程 ID。没有检查点器时为 `undefined`。
`runId`	`string \| undefined`	当前执行的运行 ID。未在配置中提供时为 `undefined`。
`checkpointId`	`string`	当前执行的检查点 ID。
`checkpointNs`	`string`	当前执行的检查点命名空间。
`taskId`	`string`	当前执行的任务 ID。

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

超时

INFO

需要 langgraph>=1.2。

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

python

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

WARNING

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

INFO

需要 @langchain/langgraph>=1.4.0。

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

typescript

import { StateGraph, type TimeoutPolicy } from "@langchain/langgraph";

// 简单的墙钟时间上限（60 秒）
new StateGraph(State).addNode("callModel", callModel, { timeout: 60_000 });

// 分别设置运行和空闲限制
new StateGraph(State).addNode("callModel", callModel, {
  timeout: { runTimeout: 120_000, idleTimeout: 30_000 },
});

运行超时

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

python

from langgraph.types import TimeoutPolicy

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

runTimeout 是对单次尝试的硬性墙钟时间上限。无论节点如何活动，它都不会被刷新：

typescript

const graph = new StateGraph(State)
  .addNode("callModel", callModel, {
    timeout: { runTimeout: 120_000 },
  })
  .compile();

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

空闲超时

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

python

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

idleTimeout 是一个可被进度重置的上限。只有当节点在指定时长内停止产生可观察的进度时才会触发——与 runTimeout 不同，每当节点产生进度信号时，计时器就会重置：

typescript

const graph = new StateGraph(State)
  .addNode("callModel", callModel, {
    timeout: { idleTimeout: 30_000 },
  })
  .compile();

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

您可以同时设置 runTimeout 和 idleTimeout。先触发的那一个会取消这次尝试。

进度信号

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

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

在默认的 refreshOn: "auto" 下，以下任一情况都会重置空闲计时器：

通过图写入路径写入状态
通过 runtime.writer 自定义流式输出
子任务调度
来自节点或其后代的任何 LangChain 回调事件（LLM token、工具调用、链开始/结束等）

心跳模式

设置 refresh_on="heartbeat" 可将刷新来源限定为显式的 runtime.heartbeat() 调用。当您想要一个严格的空闲定义，不因下属的频繁输出而被重置时，这很有用：

python

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

设置 refreshOn: "heartbeat" 可将刷新来源限定为显式的 runtime.heartbeat() 调用。当您想要一个严格的空闲定义，不因下属的频繁输出而被重置时，这很有用：

typescript

const graph = new StateGraph(State)
  .addNode("callModel", callModel, {
    timeout: { idleTimeout: 30_000, refreshOn: "heartbeat" },
  })
  .compile();

手动心跳

对于不会自然产生进度信号的长时间运行的工作，请调用 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)

typescript

import {
  StateGraph,
  StateSchema,
  START,
  END,
  type Runtime,
} from "@langchain/langgraph";
import * as z from "zod";

const State = new StateSchema({
  result: z.string(),
});

const longRunningNode = async (
  state: typeof State.State,
  runtime: Runtime<typeof State>
) => {
  for (const batch of fetchBatches()) {
    process(batch);
    runtime.heartbeat?.(); 
  }
  return { result: "done" };
};

const graph = new StateGraph(State)
  .addNode("longRunningNode", longRunningNode, {
    timeout: { idleTimeout: 30_000, refreshOn: "heartbeat" },
  })
  .addEdge(START, "longRunningNode")
  .addEdge("longRunningNode", END)
  .compile();

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

NodeTimeoutError

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

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

属性	类型	说明
`node`	`string`	执行超时的节点名称。
`elapsed`	`number`	超时触发前经过的毫秒数。
`kind`	`"idle" \| "run"`	触发了哪个超时。
`timeout`	`number`	触发的超时值（毫秒）。
`idleTimeout`	`number \| undefined`	配置的空闲超时（毫秒），如果有的话。
`runTimeout`	`number \| undefined`	配置的运行超时（毫秒），如果有的话。

在 TypeScript 中使用 isNodeTimeoutError(error) 来收窄捕获的错误。

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

typescript

const graph = new StateGraph(State)
  .addNode("callModel", callModel, {
    timeout: { idleTimeout: 30_000 },
    retryPolicy: { maxAttempts: 3 },
  })
  .compile();

使用 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"]
    ]

typescript

import { Send } from "@langchain/langgraph";

const fanOut = (state: typeof State.State) =>
  state.items.map(
    (item) =>
      new Send("processItem", { item }, { timeout: { idleTimeout: 15_000 } })
  );

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

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

错误处理

INFO

需要 langgraph>=1.2。

INFO

需要 @langchain/langgraph>=1.4.0。

错误处理程序在节点失败且所有重试都已用尽后运行。它接收当前状态，并可以使用 Command 更新状态或路由到另一个节点。这对于补偿流程（Saga 模式）非常有用，在需要优雅恢复而不是中止整个图时。

将 error_handler= 传给 add_node：

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

仅在 StateGraph 上（而不是基础 Graph 类上）将 errorHandler 传给 addNode：

typescript

import {
  StateGraph,
  StateSchema,
  START,
  Command,
  NodeError,
} from "@langchain/langgraph";
import * as z from "zod";

class ConnectionError extends Error {}

const State = new StateSchema({
  status: z.string(),
});

const chargePayment = () => {
  throw new Error("payment gateway timeout");
};

const paymentErrorHandler = (
  state: typeof State.State,
  error: NodeError
) =>
  new Command({
    update: { status: `compensated: ${error.error.message}` },
    goto: "finalize",
  });

const finalize = (state: typeof State.State) => state;

const graph = new StateGraph(State)
  .addNode("chargePayment", chargePayment, {
    retryPolicy: {
      maxAttempts: 3,
      retryOn: (err) => err instanceof ConnectionError,
    },
    errorHandler: paymentErrorHandler,
  })
  .addNode("finalize", finalize)
  .addEdge(START, "chargePayment")
  .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)。

错误处理程序通过带类型的 error: NodeError 参数接收失败上下文：

typescript

import { Command, NodeError } from "@langchain/langgraph";

const myHandler = (state: typeof State.State, error: NodeError) => {
  console.log(`Node ${error.node} failed with: ${error.error.message}`);
  return new Command({
    update: { status: "recovered" },
    goto: "nextStep",
  });
};

NodeError 是一个包含两个字段的类：

属性	类型	说明
`node`	`string`	执行失败的节点名称。
`error`	`Error`	失败节点抛出的异常。

error: NodeError 参数是可选的。不需要失败上下文的处理程序可以省略第二个参数，只接受 state。

使用 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 进行补偿，而不是中止图。

typescript

import {
  StateGraph,
  StateSchema,
  START,
  Command,
  NodeError,
} from "@langchain/langgraph";
import * as z from "zod";

class ConnectionError extends Error {}

const State = new StateSchema({
  status: z.string(),
});

const reserveInventory = () => ({ status: "reserved" });

const chargePayment = () => {
  throw new Error("payment timeout");
};

const paymentErrorHandler = (
  state: typeof State.State,
  error: NodeError
) =>
  new Command({
    update: {
      status: `compensated_after_${error.node}: ${error.error.message}`,
    },
    goto: "finalize",
  });

const finalize = (state: typeof State.State) => state;

const graph = new StateGraph(State)
  .addNode("reserveInventory", reserveInventory)
  .addNode("chargePayment", chargePayment, {
    retryPolicy: {
      maxAttempts: 3,
      retryOn: (err) => err instanceof ConnectionError,
    },
    errorHandler: paymentErrorHandler,
  })
  .addNode("finalize", finalize)
  .addEdge(START, "reserveInventory")
  .addEdge("reserveInventory", "chargePayment")
  .compile();

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

可恢复安全的失败

INFO

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

与 `interrupt()` 配合时的行为

WARNING

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

子图失败

如果节点包装了子图，而子图抛出未处理的异常，该异常会向上传播到父节点。如果父节点有错误处理程序，处理程序会触发，子图的异常位于 error.error 中。

图默认值

INFO

需要 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)                                     # 使用 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=。当某个步骤需要自己的逻辑时，节点级处理程序仍然优先：

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:
    # 将失败持久化到以 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 作为可选的第三个参数：

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`	✅	❌	缓存处理程序结果不安全

作用范围

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

图默认值

INFO

需要 @langchain/langgraph>=1.4.0。

与其在每次 addNode 调用中重复相同的 retryPolicy、errorHandler、timeout 或 cachePolicy，不如使用 setNodeDefaults 在一处配置全图默认值：

typescript

import { StateGraph, START, NodeError } from "@langchain/langgraph";

const defaultErrorHandler = (
  state: typeof State.State,
  error: NodeError
) => ({ status: `handled: ${error.error.message}` });

const graph = new StateGraph(State)
  .setNodeDefaults({
    retryPolicy: { maxAttempts: 3 },
    errorHandler: defaultErrorHandler,
    timeout: { runTimeout: 30_000 },
    cachePolicy: { ttl: 60 },
  })
  .addNode("stepA", stepA)
  .addNode("stepB", stepB)
  .addEdge(START, "stepA")
  .compile();

现在 stepA 和 stepB 共享相同的重试策略、错误处理程序、超时和缓存策略，无需任何重复。

优先级

直接传给 addNode() 的节点级值始终会覆盖 setNodeDefaults() 设置的默认值。默认值在 compile() 时解析，因此您可以按任意顺序在 addNode() 之前或之后调用 setNodeDefaults()：

typescript

import { StateGraph, START } from "@langchain/langgraph";

const graph = new StateGraph(State)
  .setNodeDefaults({ errorHandler: defaultErrorHandler })
  .addNode("stepA", stepA) // 使用 defaultErrorHandler
  .addNode("stepB", stepB, { errorHandler: customErrorHandler }) // 覆盖默认值
  .addEdge(START, "stepA")
  .compile();

默认错误处理程序

当每次图运行都对应一个外部进程（例如后台任务行），并且任何未处理的节点失败都应将该进程标记为失败时，errorHandler 默认值尤其有价值，无需在每次 addNode 中重复 errorHandler。当某个步骤需要自己的补偿逻辑时，节点级处理程序仍然优先：

typescript

import { Command, NodeError, StateGraph, START } from "@langchain/langgraph";

const markProcessFailed = (
  state: typeof State.State,
  error: NodeError
) => {
  // 将失败持久化到以 processId 为键的外部进程记录行。
  return { status: `failed at ${error.node}: ${error.error.message}` };
};

const refundPayment = (state: typeof State.State, error: NodeError) =>
  new Command({
    update: { status: `compensated after ${error.node}` },
    goto: "finalize",
  });

const graph = new StateGraph(State)
  .setNodeDefaults({
    retryPolicy: { maxAttempts: 3 },
    errorHandler: markProcessFailed,
  })
  .addNode("fetchData", fetchData) // 使用 markProcessFailed
  .addNode("chargePayment", chargePayment, {
    errorHandler: refundPayment, // 覆盖全图默认值
  })
  .addNode("finalize", finalize)
  .addEdge(START, "fetchData")
  .addEdge("fetchData", "chargePayment")
  .compile();

如果 fetchData 在重试后仍失败，markProcessFailed 会运行。如果 chargePayment 在重试后仍失败，则 refundPayment 会运行，因为节点级处理程序覆盖了默认值。

适用性矩阵

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

`setNodeDefaults` 参数	适用于普通节点	适用于错误处理节点	原因
`retryPolicy`	✅	✅	处理程序应在瞬时失败时被重试
`timeout`	✅	✅	卡住的处理程序应像卡住的普通节点一样被取消
`errorHandler`	✅	❌	处理程序绝不能捕获自身
`cachePolicy`	✅	❌	缓存处理程序结果不安全

作用范围

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

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

task 和 entrypoint 上可以使用 timeout 选项；task 还接受 retry 选项（不是 retryPolicy）：

typescript

import { entrypoint, task } from "@langchain/langgraph";

const callApi = task(
  {
    name: "callApi",
    timeout: { idleTimeout: 30_000 },
    retry: { maxAttempts: 3 },
  },
  async (url: string) => {
    const response = await fetch(url);
    return response.text();
  }
);

const myWorkflow = entrypoint(
  { name: "myWorkflow", timeout: 60_000 },
  async (inputs: { url: string }) => {
    return await callApi(inputs.url);
  }
);

其行为与 addNode 一致：超时时会抛出 NodeTimeoutError，缓冲的写入会被清除，重试策略决定是否重试。在 JavaScript/TypeScript SDK 中，task / entrypoint 上没有错误处理程序——请改用 StateGraph.addNode(..., { errorHandler })。

优雅关闭

协作式关闭让您可以在当前超级步骤完成后停止正在进行的图运行，并保存可恢复的检查点。这对于处理 SIGTERM 信号或任何需要在不丢失工作的情况下回收资源的外部监督者非常有用。

INFO

需要 langgraph>=1.2。

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

python

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:
    # 图提前停止并保存了检查点。
    # 稍后使用相同的 config 恢复。
    print(f"Drained: {e.reason}")

INFO

需要 @langchain/langgraph>=1.4.0。

创建 RunControl 并将其作为 control 传给 invoke 或 stream。从任意上下文调用 requestDrain() 以发出运行应停止的信号：

typescript

import { RunControl, GraphDrained } from "@langchain/langgraph";

const control = new RunControl();

// 在信号处理程序或监督程序中：
// control.requestDrain("sigterm");

try {
  const result = await graph.invoke(inputs, { ...config, control });
} catch (e) {
  if (e instanceof GraphDrained) {
    // 图提前停止并保存了检查点。
    // 稍后使用相同的 config 恢复。
    console.log(`Drained: ${e.reason}`);
  } else {
    throw e;
  }
}

语义

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

场景	行为
执行中的节点	运行到完成。Drain 在下一个超级步骤时生效。
正在重试且带重试策略的节点	重试循环运行到耗尽或成功为止。之后 Drain 生效。
图在与 drain 相同的时刻自然完成	正常返回。检查 `control.drain_requested` 以与正常运行区分。
仍有更多超级步骤	抛出 `GraphDrained(reason)`。检查点已保存且可恢复。
子图请求 drain	`GraphDrained` 通过父图向上传播，并在父图自己的下一个超级步骤边界将其停止。

场景	行为
执行中的节点	运行到完成。Drain 在下一个超级步骤时生效。
正在重试且带重试策略的节点	重试循环运行到耗尽或成功为止。之后 Drain 生效。
图在与 drain 相同的时刻自然完成	正常返回。检查 `control.drainRequested` 以与正常运行区分。
仍有更多超级步骤	抛出 `GraphDrained(reason)`。检查点已保存且可恢复。
子图请求 drain	`GraphDrained` 通过父图向上传播，并在父图自己的下一个超级步骤边界将其停止。

排空后恢复

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

python

result = graph.invoke(None, config)

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

typescript

const result = await graph.invoke(null, config);

在节点内部读取 drain 状态

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

python

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

typescript

import { type Runtime } from "@langchain/langgraph";

const myNode = async (state: typeof State.State, runtime: Runtime<typeof State>) => {
  if (runtime.control?.drainRequested) {
    // 跳过耗时操作并返回最简结果
    return { status: "skipped", reason: runtime.control.drainReason };
  }
  return { status: await doWork() };
};

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

INFO

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

typescript

import process from "node:process";
import { RunControl, GraphDrained } from "@langchain/langgraph";

const control = new RunControl();
process.on("SIGTERM", () => control.requestDrain("sigterm"));

try {
  const result = await graph.invoke(inputs, { ...config, control });
} catch (e) {
  if (e instanceof GraphDrained) {
    console.log(`graph drained: ${e.reason}`);
    // 下次启动时使用相同的 config 恢复
  } else {
    throw e;
  }
}

INFO

requestDrain() 不会取消进行中的异步工作。如需硬性上限，请将 drain 与优雅超时和 AbortSignal 结合使用。

局限性

超时仅适用于异步：带 timeout 的同步节点在编译时会被拒绝。
每个节点一个处理程序：每个节点最多只能有一个 error_handler。
处理程序失败会向上传播：如果错误处理程序本身抛出异常，该异常会像节点没有处理程序一样传播。
set_node_defaults 不会被子图继承：每个图独立管理自己的默认值。
setNodeDefaults 不会被子图继承：每个图独立管理自己的默认值。
错误处理程序仅适用于 StateGraph：将 errorHandler 传给 StateGraph.addNode，而不是基础 Graph 类。task / entrypoint 上没有错误处理程序。
每个节点一个处理程序：每个节点最多只能有一个 errorHandler。
处理程序失败会向上传播：如果错误处理程序本身抛出异常，该异常会像节点没有处理程序一样传播。

Graph API

Functional API

重试

默认行为

参数

自定义重试逻辑

检查重试状态

超时

运行超时

空闲超时

进度信号

心跳模式

手动心跳

NodeTimeoutError

使用 Send 的动态超时

错误处理

NodeError

使用 Command 路由

可恢复安全的失败

与 `interrupt()` 配合时的行为

子图失败

图默认值

优先级

默认错误处理程序

适用性矩阵

作用范围

图默认值

优先级

默认错误处理程序

适用性矩阵

作用范围

函数式 API

优雅关闭

语义

排空后恢复

在节点内部读取 drain 状态

SIGTERM 钩子模式

局限性

重试 ​

默认行为 ​

参数 ​

自定义重试逻辑 ​

检查重试状态 ​

超时 ​

运行超时 ​

空闲超时 ​

进度信号 ​

心跳模式 ​

手动心跳 ​

NodeTimeoutError ​

使用 Send 的动态超时 ​

错误处理 ​

NodeError ​

使用 Command 路由 ​

可恢复安全的失败 ​

与 interrupt() 配合时的行为 ​

子图失败 ​

图默认值 ​

优先级 ​

默认错误处理程序 ​

适用性矩阵 ​

作用范围 ​

图默认值 ​

优先级 ​

默认错误处理程序 ​

适用性矩阵 ​

作用范围 ​

函数式 API ​

优雅关闭 ​

语义 ​

排空后恢复 ​

在节点内部读取 drain 状态 ​

SIGTERM 钩子模式 ​

局限性 ​

重试

默认行为

参数

自定义重试逻辑

检查重试状态

超时

运行超时

空闲超时

进度信号

心跳模式

手动心跳

NodeTimeoutError

使用 Send 的动态超时

错误处理

NodeError

使用 Command 路由

可恢复安全的失败

与 `interrupt()` 配合时的行为

子图失败

图默认值

优先级

默认错误处理程序

适用性矩阵

作用范围

图默认值

优先级

默认错误处理程序

适用性矩阵

作用范围

函数式 API

优雅关闭

语义

排空后恢复

在节点内部读取 drain 状态

SIGTERM 钩子模式

局限性