LangGraph 错误处理与超时控制全指南

这是生产级 Agent 必不可少的能力我会从原理→方法→完整代码一次性讲透如何给 Agent 加上超时控制、工具调用失败重试、模型调用异常兜底让你的 Agent 更健壮、不会轻易崩溃。一、核心概念与方法总览LangGraph 的错误处理与超时控制主要分为 3 个层面层面解决的问题核心方法超时控制模型 / 工具调用卡死、响应慢LLM / 工具初始化时设置timeout节点内错误处理单次调用失败try-except捕获异常存入 State流程级错误处理节点失败后的流程决策重试装饰器、Fallback 节点、条件边路由二、超时控制防止卡死1. LLM 调用超时在初始化模型时直接设置timeout参数单位为秒from langchain_zhipu import ChatZhipuAI llm ChatZhipuAI( api_keyos.getenv(ZHIPU_API_KEY), modelglm-4.6, temperature0, timeout15, # ✅ 15秒超时超过直接报错 max_retries0 # 先关闭自动重试方便我们手动控制 )2. 工具调用超时有两种方式方式 1在ToolNode中设置推荐from langgraph.prebuilt import ToolNode # 给所有工具设置统一超时 tool_executor ToolNode( tools, handle_tool_errorsTrue, # ✅ 自动捕获工具错误 timeout10 # ✅ 工具调用10秒超时 )方式 2在工具函数内部设置针对单个工具import time from langchain_core.tools import tool tool def get_weather(city: str) - str: 获取指定城市的天气 try: # 模拟可能超时的操作 time.sleep(2) if city 北京: return 北京晴25℃微风 return 未知天气 except TimeoutError: return ⚠️ 天气查询超时请稍后再试三、错误处理让 Agent 不崩溃1. 节点内try-except最基础、最常用在每个节点内部捕获异常把错误信息存入 State让流程继续走下去而不是直接崩溃from langchain_core.messages import AIMessage def llm_think(state: ReActState): print(f\n-- 第 {state[loop_count]}轮LLM思考 --) messages state[messages] user_id state[user_id] try: # 加载长期记忆 user_preference store.get((user_memory, user_id), preference) if user_preference: print(f 加载长期记忆{user_preference.value}) messages [HumanMessage(contentf用户偏好{user_preference.value})] messages # 调用 LLM response llm.invoke(messages) error None # 无错误 except Exception as e: print(f❌ LLM 调用失败{str(e)}) # 失败时返回友好的错误消息 response AIMessage(contentf抱歉我暂时无法处理你的请求{str(e)}) error str(e) # 记录错误 return { messages: [response], loop_count: state[loop_count] 1, user_id: user_id, error: error # ✅ 把错误存入 State }2. LangGraph 重试装饰器自动重试LangGraph 提供了retry装饰器可以给节点设置自动重试from langgraph.checkpoint.utils import retry # 给 llm_think 节点添加重试最多重试2次间隔1秒 retry( max_attempts2, # 最多重试2次 wait_before1.0, # 每次重试前等待1秒 retry_on(Exception,) # 捕获所有异常 ) def llm_think(state: ReActState): # 节点逻辑不变 ...3. Fallback 节点失败时走备用方案当主节点失败时自动切换到备用节点比如用更简单的模型、或者返回固定回复# 备用 LLM比如用更轻量的模型 backup_llm ChatZhipuAI( api_keyos.getenv(ZHIPU_API_KEY), modelglm-4-flash, # 更轻量、更稳定的模型 temperature0, timeout10 ) def backup_llm_think(state: ReActState): 备用节点主节点失败时调用 print(\n-- 切换到备用 LLM --) response backup_llm.invoke(state[messages]) return { messages: [response], loop_count: state[loop_count] 1, user_id: state[user_id], error: None } # 条件边根据 error 决定走主节点还是备用节点 def should_route(state: ReActState) - Literal[llm_think, backup_llm_think, end]: if state.get(error) and state[loop_count] 2: return backup_llm_think # 失败且次数少走备用 elif not state[messages][-1].tool_calls or state[loop_count] 5: return end else: return tool_executor4. 条件边路由根据错误状态决策修改 State 增加error字段然后在条件边中根据错误信息决定下一步# 修改 State 定义 class ReActState(TypedDict): messages: Annotated[Sequence[BaseMessage], add_messages] loop_count: int user_id: str error: str | None # ✅ 新增存储错误信息四、完整代码示例带超时 错误处理这是基于你之前的代码添加了超时控制、节点内错误处理、错误状态记录、防崩溃机制的完整可运行版本import os from typing import TypedDict, Sequence, Literal, Annotated import redis from dotenv import load_dotenv from langchain_core.messages import BaseMessage, HumanMessage, AIMessage from langchain_core.tools import tool from langchain_zhipu import ChatZhipuAI from langgraph.constants import START, END from langgraph.graph import StateGraph, add_messages from langgraph.prebuilt import ToolNode from langgraph.store.redis import RedisStore from langgraph.checkpoint.redis import RedisSaver # 加载环境变量 load_dotenv() # 工具定义带内部错误处理 tool def get_weather(city: str) - str: 获取指定城市的天气 try: if city 北京: return 北京晴25℃微风 elif city 上海: return 上海多云28℃东南风 else: return f{city}未知仅支持北京/上海 except Exception as e: return f⚠️ 天气查询失败{str(e)} tool def calculator(a: float, b: float, op: str) - float: 计算器支持 - * / try: if op : return a b elif op -: return a - b elif op *: return a * b elif op /: return a / b if b ! 0 else 除数不能为0 else: return 不支持的运算符 except Exception as e: return f⚠️ 计算失败{str(e)} tools [get_weather, calculator] # 状态定义新增 error 字段 class ReActState(TypedDict): messages: Annotated[Sequence[BaseMessage], add_messages] loop_count: int user_id: str error: str | None # ✅ 存储错误信息 # 初始化持久化组件 redis_client redis.Redis(hostlocalhost, port6379, db0) checkpointer RedisSaver(redis://localhost:6379/0) checkpointer.setup() store RedisStore(redis_client) store.setup() print(✅ Redis 索引已创建) # 初始化 LLM带超时控制 llm ChatZhipuAI( api_keyos.getenv(ZHIPU_API_KEY), modelos.getenv(LLM_MODEL), temperature0, timeout15, # ✅ 15秒超时 max_retries0 ) # 备用 LLM更轻量、更稳定 backup_llm ChatZhipuAI( api_keyos.getenv(ZHIPU_API_KEY), modelglm-4-flash, temperature0, timeout10 ) # 工具执行节点带超时和错误处理 tool_executor ToolNode( tools, handle_tool_errorsTrue, # ✅ 自动捕获工具错误 timeout10 # ✅ 工具调用10秒超时 ) # 节点定义带 try-except 错误处理 def llm_think(state: ReActState): print(f\n-- 第 {state[loop_count]}轮LLM思考 --) messages state[messages] user_id state[user_id] error None try: # 加载长期记忆 user_preference store.get((user_memory, user_id), preference) if user_preference: print(f 加载长期记忆{user_preference.value}) messages [HumanMessage(contentf用户偏好{user_preference.value})] messages # 调用 LLM response llm.invoke(messages) except Exception as e: print(f❌ LLM 调用失败{str(e)}) response AIMessage(contentf抱歉我暂时遇到了一点问题{str(e)}) error str(e) return { messages: [response], loop_count: state[loop_count] 1, user_id: user_id, error: error } def backup_llm_think(state: ReActState): 备用节点主 LLM 失败时调用 print(\n-- 切换到备用 LLM --) try: response backup_llm.invoke(state[messages]) error None except Exception as e: print(f❌ 备用 LLM 也失败了{str(e)}) response AIMessage(content非常抱歉服务暂时不可用请稍后再试。) error str(e) return { messages: [response], loop_count: state[loop_count] 1, user_id: state[user_id], error: error } # 条件路由根据错误状态决策 def should_continue(state: ReActState) - Literal[llm_think, backup_llm_think, tool_executor, end]: last_message state[messages][-1] # 优先处理错误 if state.get(error): if state[loop_count] 2: return backup_llm_think # 失败次数少走备用 else: return end # 失败次数多直接结束 # 正常流程 if not last_message.tool_calls or state[loop_count] 5: print(→ 结束无需工具 或 达到最大轮次) return end print(→ 继续需要调用工具) return tool_executor # 构建图 builder StateGraph(ReActState) builder.add_node(llm_think, llm_think) builder.add_node(backup_llm_think, backup_llm_think) builder.add_node(tool_executor, tool_executor) builder.add_edge(START, llm_think) builder.add_conditional_edges( llm_think, should_continue, { backup_llm_think: backup_llm_think, tool_executor: tool_executor, end: END } ) builder.add_conditional_edges( backup_llm_think, should_continue, { tool_executor: tool_executor, end: END } ) builder.add_edge(tool_executor, llm_think) # 编译图 react_agent builder.compile( checkpointercheckpointer, storestore ) # 测试运行 if __name__ __main__: print( LangGraph 带错误处理的 Agent 启动 ) config { configurable: { thread_id: session_001, user_id: user_001 } } # 第一轮对话 print(\n--- 第一轮对话 ---) initial_state { messages: [HumanMessage(content北京今天天气上海比北京高3度上海多少度)], loop_count: 0, user_id: user_001, error: None } result1 react_agent.invoke(initial_state, configconfig) print(f第一轮答案{result1[messages][-1].content}) # 保存长期记忆 store.put((user_memory, user_001), preference, 喜欢用摄氏度偏好简洁回答) print(\n✅ 已保存长期记忆) # 第二轮对话 print(\n--- 第二轮对话 ---) result2 react_agent.invoke( {messages: [HumanMessage(content上海天气符合我的偏好吗)]}, configconfig ) print(第二轮答案, result2[messages][-1].content)五、关键要点总结超时控制在 LLM 和ToolNode初始化时设置timeout防止卡死节点内错误处理用try-except捕获异常把错误信息存入 State返回友好回复备用方案设置 Fallback 节点主节点失败时切换到更稳定的备用方案条件边决策根据 State 中的错误信息灵活决定下一步是重试、走备用还是结束错误记录在 State 中增加error字段方便后续调试和分析