摘要：本文围绕 LangGraph 的节点级重试策略展开，先说明重试策略要解决的问题——工作流执行中节点异常（如数据库连接失败）会导致流程中断，引入重试后可在异常处理中重新尝试，使流程继续执行。随后介绍 RetryPolicy 类及其 retry_on 参数的作用，说明未设置 retry_on 时会回退到 default_retry_on 的默认异常列表，并通过效果图对比带异常拦截与不带异常拦截的运行差异，指出 runtime.execution_info.node_attempt 可用于实现更细致的重试、降级与熔断。最后给出完整 Python 示例代码，演示为 query_database 节点配置 retry_on、为 model 节点配置 max_attempts 的实践方式。
内容参考于：图灵AI大模型全栈
重试策略它是解决流程中出现的异常，就是说工作流在执行期间如果发生了错误没办法继续往下执行，这时候不可以让它断下来，需要让它正常的结束，重试策略就是为了让工作流正常结束，正常结束比如到了节点c，节点c需要连接数据库，但是由于网络连接的问题导致这一时刻，没有连接到数据库，然后出现连接失败异常，这时候工作流就会因为异常在这直接结束了，没办法继续往下了，加入重试策略后，在捕获的连接失败异常处理中再次连接数据，它可能就可以正常连接，从而就不会因为异常直接结束，而是继续执行后续的工作流节点
在add_node函数（添加节点函数）中有一个retry_policy参数，这个参数需要一个 RetryPolicy 类型，这个类型就是LangGraph提供的节点级别重试策略类，然后下图红框里的 retry_on 是 RetryPolicy 类中的一个参数，这个 retry_on 参数的意思是，当节点报错了要拦截的异常，下方指定了两个异常 sqlite3.OperationalError和sqlite3.IntegrityError
如果RetryPolicy中不设置retry_on参数（就是不在上图红框位置写retry_on参数），那么它会从 default_retry_on 里面获取默认的异常，也就是下方的默认异常
ValueError, TypeError, ArithmeticError, ImportError, LookupError, NameError, SyntaxError, RuntimeError, ReferenceError, StopIteration, StopAsyncIteration, OSError,看完上方的文字可能会有点迷迷瞪瞪的，看看下方的效果图就能明白了
首先下图红框抛出了 sqlite3.OperationalError 异常，然后 retry_on 里也设置了需要拦截 sqlite3.OperationalError 异常
如下图红框带异常拦截的运行结果
如下图红框，不带异常拦截的运行结果，节点会直接结束
下图红框 runtime.execution_info.node_attempt 的值是当前节点重试的次数，在LangChain中只有模型重试和节点重试中间件，在LangGraph中可以利用 runtime.execution_info.node_attempt 实现更细致的重试或降级熔断操作
出了上方同异常捕获，它还有一个次数限制，如下图红框，在 RetryPolicy 中有一个 max_attempts 的参数，它的值用来限制当前节点可以运行几次
代码
# 导入 Python 内置 SQLite 数据库模块，用于模拟数据库操作与异常触发场景 import sqlite3 # 导入 LangChain 大模型统一初始化函数，可适配多厂商模型的标准化调用 from langchain.chat_models import init_chat_model # 从 LangGraph 图模块导入核心基础组件 # END：工作流结束标记常量 # MessagesState：LangGraph 内置状态结构，自带 messages 消息列表，自动管理对话历史 # StateGraph：状态图构建器，用于搭建有向工作流 # START：工作流起始标记常量 from langgraph.graph import END, MessagesState, StateGraph, START # 导入 LangGraph 重试策略类，用于配置节点失败后的自动重试规则 from langgraph.types import RetryPolicy # 导入 LangChain 标准消息类型：AI 回复消息、用户输入消息 from langchain.messages import AIMessage, HumanMessage # 导入 LangGraph 运行时对象，节点内可通过它获取执行次数、上下文、存储等运行时元信息 from langgraph.runtime import Runtime # 导入环境变量加载工具，用于读取 .env 文件中的 API 密钥等敏感配置 from dotenv import load_dotenv # 执行环境变量加载，将 .env 文件中的配置注入系统环境 load_dotenv() # 初始化通义千问大语言模型实例 model = init_chat_model( "qwen3.8-max", # 指定调用的模型名称：通义千问 qwen3.8-max openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1", # 阿里云百炼的 OpenAI 兼容接口地址 model_provider="openai", # 指定按 OpenAI 协议规范调用接口 extra_body={"thinking": {"type": "disabled"}} # 关闭模型思考过程输出，保证返回纯文本结果 ) # 定义「数据库查询」节点函数，核心演示失败自动重试机制 # 参数1：state - 工作流全局状态，使用内置 MessagesState 结构，自带 messages 对话列表 # 参数2：runtime - LangGraph 自动注入的运行时对象，可读取当前节点执行次数、会话上下文等信息 def query_database(state: MessagesState, runtime: Runtime): # 从运行时信息中获取当前节点的尝试次数，打印当前执行轮次 print(f"正在尝试第 {runtime.execution_info.node_attempt} 次查询...") # 模拟数据库连接故障：前 2 次尝试主动抛出异常，触发重试逻辑 if runtime.execution_info.node_attempt < 3: print("模拟数据库连接失败...") # 抛出 SQLite 操作异常，该异常在重试策略的白名单中，会触发节点自动重试 raise sqlite3.OperationalError("Database connection lost") # 第 3 次尝试执行成功，返回查询结果，消息会自动追加到状态的 messages 列表中 return {"messages": [AIMessage(content=str("查询成功"))]} # 定义「大模型对话」节点函数，处理用户消息交互 def call_model(state: MessagesState): # 调用大模型，传入当前完整对话历史消息列表，获取模型响应 response = model.invoke(state["messages"]) # 将模型响应消息追加到全局状态的 messages 列表中 return {"messages": [response]} # ========== 构建工作流图 ========== # 创建状态图构建器实例，使用内置 MessagesState 作为全局状态规范 builder = StateGraph(MessagesState) # 添加数据库查询节点，并绑定自定义重试策略 builder.add_node( "query_database", # 节点唯一名称 query_database, # 节点对应的处理函数 # 配置重试策略：指定哪些异常会触发自动重试 retry_policy=RetryPolicy( retry_on=[sqlite3.OperationalError, sqlite3.IntegrityError] # 遇到这两类数据库异常时自动重试节点 ) ) # 添加模型调用节点，并配置重试次数限制 builder.add_node( "model", # 节点唯一名称 call_model, # 节点对应的处理函数 # 配置重试策略：设置最大尝试次数（含首次执行） retry_policy=RetryPolicy(max_attempts=5) # 该节点最多执行 5 次，超过则抛出异常终止 ) # 定义工作流的边连接关系（执行顺序） # 起始边：流程启动后，第一个执行 model 节点 builder.add_edge(START, "model") # 顺序边：model 节点执行完成后，进入 query_database 节点 builder.add_edge("model", "query_database") # 结束边：query_database 节点执行完成后，流程终止 builder.add_edge("query_database", END) # 编译构建好的状态图，生成可执行的工作流实例 graph = builder.compile() # ========== 测试运行 ========== # 调用工作流，传入初始用户消息，触发完整执行流程 response = graph.invoke({"messages": [HumanMessage(content="你好呀？")]}) # 打印最终的完整状态结果 print(response)
