节点执行与容错机制

基本介绍

LangGraph 提供了三种异常处理机制
retries：重试机制。节点执行失败时，根据重试策略自动重新执行
Timeouts：超时控制。限制单次节点执行的最大等待时间
Error Handling：错误处理。重试次数耗尽后，执行特定的异常处理逻辑
三者的执行顺序如下
（1）节点开始执行
（2）如果节点执行超时或抛出异常，本次节点尝试失败
（3）重试策略 RetryPolicy 判断该异常是否需要重试
（4）如果满足重试条件，并且尚未达到最大尝试次数，则重新执行节点
（5）如果重试次数耗尽，节点仍然失败，则进入 error_handler 错误处理逻
（6）如果没有配置错误处理逻辑，则异常继续向外抛出，可能导致整个图运行失败
⚠️ 注意：超时控制和错误处理是较新版本引入的能力，要求 langgraph>=1.2

重试机制

参数配置

参数	默认值	描述
max_attempts	3	最大尝试次数（包括首次执行）
initial_interval	0.5	第一次重试前的等待时间，单位：秒
backoff_factor	2.0	每次重试后等待时间的放大倍数
max_interval	128.0	相邻两次重试之间的最大等待时间，单位：秒
jitter	True	是否为重试间隔添加随机抖动
retry_on	default_retry_on	哪些异常需要触发重试

代码示例

在添加节点时，可以通过 retry_policy= 配置节点的重试策略

python

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import RetryPolicy

from requests.exceptions import HTTPError
from loguru import logger

class EmptyState(TypedDict):
    pass

def node_a(state: EmptyState) -> EmptyState:
    logger.info(f"node a 运行")
    raise HTTPError("网络连接超时...")

builder = StateGraph(state_schema=EmptyState)
builder.add_node(
    "node_a",
    node_a,
    retry_policy=RetryPolicy(
        max_attempts=3,
        jitter=False
    )
)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", END)

graph = builder.compile()

try:
    graph.invoke({})
except HTTPError as e:
    logger.info("重试次数耗尽: {}", e)

from IPython.display import display
display(graph)

"""
运行结果如下

2026-06-01 16:44:56.684 | INFO     | __main__:node_a:12 - node a 运行
2026-06-01 16:44:57.185 | INFO     | __main__:node_a:12 - node a 运行
2026-06-01 16:44:58.186 | INFO     | __main__:node_a:12 - node a 运行
2026-06-01 16:44:58.187 | INFO     | __main__:<module>:32 - 重试次数耗尽: 网络连接超时...
"""

默认重试条件

python

def default_retry_on(exc: Exception) -> bool:
    import httpx
    import requests

    if isinstance(exc, ConnectionError):
        return True
    if isinstance(exc, httpx.HTTPStatusError):
        return 500 <= exc.response.status_code < 600
    if isinstance(exc, requests.HTTPError):
        return 500 <= exc.response.status_code < 600 if exc.response else True
    if isinstance(
        exc,
        (
            ValueError,
            TypeError,
            ArithmeticError,
            ImportError,
            LookupError,
            NameError,
            SyntaxError,
            RuntimeError,
            ReferenceError,
            StopIteration,
            StopAsyncIteration,
            OSError,
        ),
    ):
        return False
    return True

超时控制

使用前提

版本要求：langgraph>=1.2
节点级超时控制仅适用于异步节点，即使用 async def 定义的节点
同步节点一旦开始执行，通常会阻塞当前线程。Python 进程内缺少一种通用、安全的机制，可以从外部强行终止正在运行的同步函数；如果强行中断，可能导致资源未释放、锁未释放或副作用处于不一致状态。因此，LangGraph 的节点级超时更适合用于异步节点。对于同步节点，应优先在节点内部使用具体库提供的超时参数，例如 HTTP 客户端、数据库驱动或 SDK 自带的 timeout 配置

参数配置

参数	描述
run_timeout	超时时间，即单次节点运行的最长时间
idle_timeout	节点没有可观察进展（如写状态、流式输出等）时的最长空闲时间
refresh_on	空闲时间的刷新方式，可以是手动刷新或自动刷新。默认为自动刷新，此时写状态、流式输出等都会刷新

代码示例

call_model 节点单次执行最多运行 60 秒，超时后会抛出 NodeTimeoutError，该异常会交给 RetryPolicy 判断是否需要重试，如果重试次数耗尽才会进入错误处理逻辑
更多内容参考官方文档：https://docs.langchain.com/oss/python/langgraph/fault-tolerance#timeouts

python

from langgraph.types import TimeoutPolicy

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

错误处理

使用前提

版本要求：langgraph>=1.2

核心思想

节点执行失败
    ↓
是否满足 retry_policy
    ↓
满足则继续重试，直到重试次数耗尽，否则直接进行下一步
    ↓
进入 error_handler
    ↓
执行兜底恢复逻辑

代码示例

call_api 执行失败后，由 RetryPolicy 判断是否重试，最多尝试 3 次，3 次仍然失败后调用 handle_api_error，该节点可以返回状态更新，也可以通过 Command 路由到其他节点
更多内容参考官方文档：https://docs.langchain.com/oss/python/langgraph/fault-tolerance#error-handling

python

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

节点缓存

基本介绍

定义：将节点的历史运行结果保存下来，后续当节点收到相同输入时，不再重复执行节点函数，而是直接返回之前缓存的结果
适用场景：相同输入会重复出现
CachePolicy 有两个配置项
（1）key_func：根据节点输入生成缓存 Key 的函数
（2）ttl：缓存键值对的存活时间，单位为秒

代码示例

缓存失效前
（1）相同输入会命中缓存
（2）命中缓存时，节点函数不会再次执行
（3）因此不会打印 node_a 被调用
（4）输入不同则不会命中缓存，节点函数仍然会执行
缓存失效后
（1）即使输入相同，也会再次执行节点函数
（2）执行完成后，会重新写入缓存

python

import time
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.cache.memory import InMemoryCache
from langgraph.types import CachePolicy

from operator import add
from loguru import logger

class EmptyState(TypedDict):
    user: str
    invoke_counts: Annotated[int, add]

def node_a(state: EmptyState) -> EmptyState:
    logger.info("node_a 被调用, user: {}", state["user"])
    time.sleep(3) # 模拟耗时操作
    logger.info("node_a 耗时操作执行完毕")

    return {
        "invoke_counts": 1
    }

builder = StateGraph(state_schema=EmptyState)
builder.add_node(
    "node_a",
    node_a,
    cache_policy=CachePolicy(ttl=10)
)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", END)

graph = builder.compile(cache=InMemoryCache())
logger.info("首次调用图")
logger.info("运行结果: {}\n\n", graph.invoke({"user": "小明", "invoke_counts": 0}))
logger.info("相同输入再次调用图")
logger.info("运行结果: {}\n\n", graph.invoke({"user": "小明", "invoke_counts": 0}))
logger.info("不同输入再次调用图")
logger.info("运行结果: {}\n\n", graph.invoke({"user": "小花", "invoke_counts": 0}))
logger.info("不同输入再次调用图")
logger.info("运行结果: {}\n\n", graph.invoke({"user": "小花", "invoke_counts": 2}))
time.sleep(10) # 确保第一次调用缓存失效
logger.info("10秒后相同输入再次调用图，此时缓存已失效")
logger.info("运行结果: {}", graph.invoke({"user": "小明", "invoke_counts": 0}))

from IPython.display import display
display(graph)

"""
运行结果如下

2026-06-01 18:46:22.959 | INFO     | __main__:<module>:33 - 首次调用图
2026-06-01 18:46:22.960 | INFO     | __main__:node_a:15 - node_a 被调用, user: 小明
2026-06-01 18:46:25.962 | INFO     | __main__:node_a:17 - node_a 耗时操作执行完毕
2026-06-01 18:46:25.963 | INFO     | __main__:<module>:34 - 运行结果: {'user': '小明', 'invoke_counts': 1}

2026-06-01 18:46:25.964 | INFO     | __main__:<module>:35 - 相同输入再次调用图
2026-06-01 18:46:25.965 | INFO     | __main__:<module>:36 - 运行结果: {'user': '小明', 'invoke_counts': 1}

2026-06-01 18:46:25.966 | INFO     | __main__:<module>:37 - 不同输入再次调用图
2026-06-01 18:46:25.967 | INFO     | __main__:node_a:15 - node_a 被调用, user: 小花
2026-06-01 18:46:28.968 | INFO     | __main__:node_a:17 - node_a 耗时操作执行完毕
2026-06-01 18:46:28.970 | INFO     | __main__:<module>:38 - 运行结果: {'user': '小花', 'invoke_counts': 1}

2026-06-01 18:46:28.970 | INFO     | __main__:<module>:39 - 不同输入再次调用图
2026-06-01 18:46:28.971 | INFO     | __main__:node_a:15 - node_a 被调用, user: 小花
2026-06-01 18:46:31.973 | INFO     | __main__:node_a:17 - node_a 耗时操作执行完毕
2026-06-01 18:46:31.974 | INFO     | __main__:<module>:40 - 运行结果: {'user': '小花', 'invoke_counts': 3}

2026-06-01 18:46:36.976 | INFO     | __main__:<module>:42 - 11秒后相同输入再次调用图，此时缓存已失效
2026-06-01 18:46:36.977 | INFO     | __main__:node_a:15 - node_a 被调用, user: 小明
2026-06-01 18:46:39.979 | INFO     | __main__:node_a:17 - node_a 耗时操作执行完毕
2026-06-01 18:46:39.981 | INFO     | __main__:<module>:43 - 运行结果: {'user': '小明', 'invoke_counts': 1}
"""

全图默认配置

使用前提

版本要求：langgraph>=1.2

基本介绍

当多个节点都需要配置相同的执行策略时，如果每个节点都重复配置，繁琐且耗时，可以通过 set_node_defaults 为所有节点设置默认配置，即全图默认配置 Graph defaults
默认配置包括 retry_policy、timeout、error_handler、cache_policy
更多内容参考官方文档：https://docs.langchain.com/oss/python/langgraph/fault-tolerance#graph-defaults

对比总结

机制	机制类型	作用	当前课程是否重点讲解
重试机制 Retries	节点容错机制	节点失败后自动重试	是
超时设置 Timeouts	节点容错机制	限制单次节点执行时间	否，要求 langgraph>=1.2
错误处理 Error Handling	节点容错机制	重试耗尽后执行兜底逻辑	否，要求 langgraph>=1.2
节点缓存 Node Cache	节点执行优化机制	缓存节点历史运行结果	是
全图默认配置 Graph defaults	默认策略配置	为所有节点设置默认执行策略	否，要求 langgraph>=1.2

实际使用时，可以按照以下原则选择
（1）临时性失败：优先使用 RetryPolicy
（2）外部服务可能卡住：使用 timeout 控制单次节点执行时间
（3）失败后需要兜底恢复：使用 error_handler
（4）相同输入重复执行且成本较高：使用 CachePolicy
（5）多个节点使用相同策略：使用 set_node_defaults 统一配置

节点执行与容错机制

基本介绍 ​

LangGraph 提供了三种异常处理机制 ​

retries：重试机制。节点执行失败时，根据重试策略自动重新执行 ​

Timeouts：超时控制。限制单次节点执行的最大等待时间 ​

Error Handling：错误处理。重试次数耗尽后，执行特定的异常处理逻辑 ​

三者的执行顺序如下 ​

（1）节点开始执行 ​

（2）如果节点执行超时或抛出异常，本次节点尝试失败 ​

（3）重试策略 RetryPolicy 判断该异常是否需要重试 ​

（4）如果满足重试条件，并且尚未达到最大尝试次数，则重新执行节点 ​

（5）如果重试次数耗尽，节点仍然失败，则进入 error_handler 错误处理逻 ​

（6）如果没有配置错误处理逻辑，则异常继续向外抛出，可能导致整个图运行失败 ​

⚠️ 注意：超时控制和错误处理是较新版本引入的能力，要求 langgraph>=1.2 ​

重试机制 ​

参数配置 ​

代码示例 ​

在添加节点时，可以通过 retry_policy= 配置节点的重试策略 ​

默认重试条件 ​

超时控制 ​

使用前提 ​

版本要求：langgraph>=1.2 ​

节点级超时控制仅适用于异步节点，即使用 async def 定义的节点 ​

参数配置 ​

代码示例 ​

call_model 节点单次执行最多运行 60 秒，超时后会抛出 NodeTimeoutError，该异常会交给 RetryPolicy 判断是否需要重试，如果重试次数耗尽才会进入错误处理逻辑 ​

更多内容参考官方文档：https://docs.langchain.com/oss/python/langgraph/fault-tolerance#timeouts ​

错误处理 ​

使用前提 ​

版本要求：langgraph>=1.2 ​

核心思想 ​

代码示例 ​

call_api 执行失败后，由 RetryPolicy 判断是否重试，最多尝试 3 次，3 次仍然失败后调用 handle_api_error，该节点可以返回状态更新，也可以通过 Command 路由到其他节点 ​

更多内容参考官方文档：https://docs.langchain.com/oss/python/langgraph/fault-tolerance#error-handling ​

节点缓存 ​

基本介绍 ​

定义：将节点的历史运行结果保存下来，后续当节点收到相同输入时，不再重复执行节点函数，而是直接返回之前缓存的结果 ​

适用场景：相同输入会重复出现 ​

CachePolicy 有两个配置项 ​

（1）key_func：根据节点输入生成缓存 Key 的函数 ​

（2）ttl：缓存键值对的存活时间，单位为秒 ​

代码示例 ​

缓存失效前 ​

（1）相同输入会命中缓存 ​

（2）命中缓存时，节点函数不会再次执行 ​

（3）因此不会打印 node_a 被调用 ​

（4）输入不同则不会命中缓存，节点函数仍然会执行 ​

缓存失效后 ​

（1）即使输入相同，也会再次执行节点函数 ​

（2）执行完成后，会重新写入缓存 ​

全图默认配置 ​

使用前提 ​

版本要求：langgraph>=1.2 ​

基本介绍 ​

当多个节点都需要配置相同的执行策略时，如果每个节点都重复配置，繁琐且耗时，可以通过 set_node_defaults 为所有节点设置默认配置，即全图默认配置 Graph defaults ​

默认配置包括 retry_policy、timeout、error_handler、cache_policy ​

更多内容参考官方文档：https://docs.langchain.com/oss/python/langgraph/fault-tolerance#graph-defaults ​

对比总结 ​

实际使用时，可以按照以下原则选择 ​

（1）临时性失败：优先使用 RetryPolicy ​

（2）外部服务可能卡住：使用 timeout 控制单次节点执行时间 ​

（3）失败后需要兜底恢复：使用 error_handler ​

（4）相同输入重复执行且成本较高：使用 CachePolicy ​

（5）多个节点使用相同策略：使用 set_node_defaults 统一配置 ​

基本介绍

LangGraph 提供了三种异常处理机制

retries：重试机制。节点执行失败时，根据重试策略自动重新执行

Timeouts：超时控制。限制单次节点执行的最大等待时间

Error Handling：错误处理。重试次数耗尽后，执行特定的异常处理逻辑

三者的执行顺序如下

（1）节点开始执行

（2）如果节点执行超时或抛出异常，本次节点尝试失败

（3）重试策略 RetryPolicy 判断该异常是否需要重试

（4）如果满足重试条件，并且尚未达到最大尝试次数，则重新执行节点

（5）如果重试次数耗尽，节点仍然失败，则进入 error_handler 错误处理逻

（6）如果没有配置错误处理逻辑，则异常继续向外抛出，可能导致整个图运行失败

⚠️ 注意：超时控制和错误处理是较新版本引入的能力，要求 langgraph>=1.2

重试机制

参数配置

代码示例

在添加节点时，可以通过 retry_policy= 配置节点的重试策略

默认重试条件

超时控制

使用前提

版本要求：langgraph>=1.2

节点级超时控制仅适用于异步节点，即使用 async def 定义的节点

参数配置

代码示例

call_model 节点单次执行最多运行 60 秒，超时后会抛出 NodeTimeoutError，该异常会交给 RetryPolicy 判断是否需要重试，如果重试次数耗尽才会进入错误处理逻辑

更多内容参考官方文档：https://docs.langchain.com/oss/python/langgraph/fault-tolerance#timeouts

错误处理

使用前提

版本要求：langgraph>=1.2

核心思想

代码示例

call_api 执行失败后，由 RetryPolicy 判断是否重试，最多尝试 3 次，3 次仍然失败后调用 handle_api_error，该节点可以返回状态更新，也可以通过 Command 路由到其他节点

更多内容参考官方文档：https://docs.langchain.com/oss/python/langgraph/fault-tolerance#error-handling

节点缓存

基本介绍

定义：将节点的历史运行结果保存下来，后续当节点收到相同输入时，不再重复执行节点函数，而是直接返回之前缓存的结果

适用场景：相同输入会重复出现

CachePolicy 有两个配置项

（1）key_func：根据节点输入生成缓存 Key 的函数

（2）ttl：缓存键值对的存活时间，单位为秒

代码示例

缓存失效前

（1）相同输入会命中缓存

（2）命中缓存时，节点函数不会再次执行

（3）因此不会打印 node_a 被调用

（4）输入不同则不会命中缓存，节点函数仍然会执行

缓存失效后

（1）即使输入相同，也会再次执行节点函数

（2）执行完成后，会重新写入缓存

全图默认配置

使用前提

版本要求：langgraph>=1.2

基本介绍

当多个节点都需要配置相同的执行策略时，如果每个节点都重复配置，繁琐且耗时，可以通过 set_node_defaults 为所有节点设置默认配置，即全图默认配置 Graph defaults

默认配置包括 retry_policy、timeout、error_handler、cache_policy

更多内容参考官方文档：https://docs.langchain.com/oss/python/langgraph/fault-tolerance#graph-defaults

对比总结

实际使用时，可以按照以下原则选择

（1）临时性失败：优先使用 RetryPolicy

（2）外部服务可能卡住：使用 timeout 控制单次节点执行时间

（3）失败后需要兜底恢复：使用 error_handler

（4）相同输入重复执行且成本较高：使用 CachePolicy

（5）多个节点使用相同策略：使用 set_node_defaults 统一配置