当前位置： 首页 > news >正文

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

news 2026/5/7 22:30:46

这是生产级 Agent 必不可少的能力！我会从原理→方法→完整代码，一次性讲透如何给 Agent 加上超时控制、工具调用失败重试、模型调用异常兜底，让你的 Agent 更健壮、不会轻易崩溃。

一、核心概念与方法总览

LangGraph 的错误处理与超时控制主要分为 3 个层面：

层面解决的问题核心方法
超时控制模型 / 工具调用卡死、响应慢LLM / 工具初始化时设置timeout
节点内错误处理单次调用失败try-except捕获异常，存入 State
流程级错误处理节点失败后的流程决策重试装饰器、Fallback 节点、条件边路由

二、超时控制（防止卡死）

1. LLM 调用超时

在初始化模型时直接设置timeout参数，单位为秒：

from langchain_zhipu import ChatZhipuAI llm = ChatZhipuAI( api_key=os.getenv("ZHIPU_API_KEY"), model="glm-4.6", temperature=0, timeout=15, # ✅ 15秒超时，超过直接报错 max_retries=0 # 先关闭自动重试，方便我们手动控制 )

2. 工具调用超时

有两种方式：

方式 1：在ToolNode中设置（推荐）
from langgraph.prebuilt import ToolNode # 给所有工具设置统一超时 tool_executor = ToolNode( tools, handle_tool_errors=True, # ✅ 自动捕获工具错误 timeout=10 # ✅ 工具调用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(content=f'用户偏好：{user_preference.value}')] + messages # 调用 LLM response = llm.invoke(messages) error = None # 无错误 except Exception as e: print(f"❌ LLM 调用失败：{str(e)}") # 失败时返回友好的错误消息 response = AIMessage(content=f"抱歉，我暂时无法处理你的请求：{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_attempts=2, # 最多重试2次 wait_before=1.0, # 每次重试前等待1秒 retry_on=(Exception,) # 捕获所有异常 ) def llm_think(state: ReActState): # 节点逻辑不变 ...

3. Fallback 节点（失败时走备用方案）

当主节点失败时，自动切换到备用节点（比如用更简单的模型、或者返回固定回复）：

# 备用 LLM（比如用更轻量的模型） backup_llm = ChatZhipuAI( api_key=os.getenv("ZHIPU_API_KEY"), model="glm-4-flash", # 更轻量、更稳定的模型 temperature=0, timeout=10 ) 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_executor"

4. 条件边路由（根据错误状态决策）

修改 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(host="localhost", port=6379, db=0) checkpointer = RedisSaver("redis://localhost:6379/0") checkpointer.setup() store = RedisStore(redis_client) store.setup() print("✅ Redis 索引已创建") # ===================== 初始化 LLM（带超时控制）===================== llm = ChatZhipuAI( api_key=os.getenv("ZHIPU_API_KEY"), model=os.getenv("LLM_MODEL"), temperature=0, timeout=15, # ✅ 15秒超时 max_retries=0 ) # 备用 LLM（更轻量、更稳定） backup_llm = ChatZhipuAI( api_key=os.getenv("ZHIPU_API_KEY"), model="glm-4-flash", temperature=0, timeout=10 ) # ===================== 工具执行节点（带超时和错误处理）===================== tool_executor = ToolNode( tools, handle_tool_errors=True, # ✅ 自动捕获工具错误 timeout=10 # ✅ 工具调用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(content=f'用户偏好：{user_preference.value}')] + messages # 调用 LLM response = llm.invoke(messages) except Exception as e: print(f"❌ LLM 调用失败：{str(e)}") response = AIMessage(content=f"抱歉，我暂时遇到了一点问题：{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( checkpointer=checkpointer, store=store ) # ===================== 测试运行 ===================== 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, config=config) print(f'第一轮答案：{result1["messages"][-1].content}') # 保存长期记忆 store.put(("user_memory", "user_001"), "preference", "喜欢用摄氏度，偏好简洁回答") print("\n✅ 已保存长期记忆") # 第二轮对话 print("\n--- 第二轮对话 ---") result2 = react_agent.invoke( {"messages": [HumanMessage(content="上海天气符合我的偏好吗？")]}, config=config ) print("第二轮答案：", result2["messages"][-1].content)

五、关键要点总结

  1. 超时控制：在 LLM 和ToolNode初始化时设置timeout，防止卡死
  2. 节点内错误处理：用try-except捕获异常，把错误信息存入 State，返回友好回复
  3. 备用方案：设置 Fallback 节点，主节点失败时切换到更稳定的备用方案
  4. 条件边决策：根据 State 中的错误信息，灵活决定下一步是重试、走备用还是结束
  5. 错误记录：在 State 中增加error字段，方便后续调试和分析
http://www.jsqmd.com/news/772821/

相关文章：

  • 甘蔗切梢器电液比例位置控制系统联合仿真【附代码】
  • 红色沙漠代码 镜像
  • 别再傻傻用普通FFT了！用MATLAB手把手教你实现ZOOM-FFT频谱细化（附完整代码）
  • 如何用AD8232心电传感器在30分钟内搭建专业级心率监测系统
  • 实测绍兴3家GEO公司：服务规范与效果可验证性复盘（2026） - 花开富贵112
  • 从零搭建一个拼多多CPS返利小程序：我的踩坑记录与避坑指南
  • 大语言模型价值观对齐技术：SFT与RLHF实践对比
  • 避坑指南：IST8310磁力计I2C通信失败的7个常见原因及排查方法
  • VMware Workstation 虚拟机创建客户端系统,出现此主机不支持64位客户机操作系统问题解决
  • 3个步骤掌握AI Toolkit：从零到一的完整AI开发指南
  • 基于Hugging Face Spaces免费部署永不离线AI助手：HuggingClaw实战指南
  • 实测绍兴3家GEO公司｜服务规范与效果如何验证？（2026） - 花开富贵112
  • 3个简单步骤：使用OpenCore Legacy Patcher让旧Mac免费升级最新macOS
  • TIDAL音乐下载终极指南：如何用tidal-dl-ng轻松保存高品质无损音乐
  • 如何利用宝塔面板进行数据迁移_使用宝塔整机备份功能
  • D3.js：数据可视化的终极利器
  • 我开源了一款本地音乐播放器 —— Yeah Music，欢迎大家体验
  • 【Docker 工程实践】AI 服务容器化部署全流程
  • Synology Video Info Plugin：三步安装，让群晖Video Station影视信息更丰富
  • LangGraph 重构个人知识库问答系统（稳定 + 可扩展版）
  • 5个实战技巧：高效使用WebAssembly进行浏览器端图像处理
  • 怎么在 CloudCone VPS 上设置自动快照备份防止数据丢失
  • 月之暗面获20亿美元融资估值破200亿，成国内大模型创业累计融资最多企业
  • 从MII到RGMII：你的嵌入式网卡PCB面积是怎么省下来的？一个硬件老鸟的笔记
  • 看完100个失败私域直播案例，90%的人死在预热前
  • 用GD32F470的ADC+DMA实现高精度电流采样，附梁山派开发板实测波形
  • 零基础入门kohya_ss：在AMD GPU上轻松训练你的专属AI绘画模型
  • 终极视频分析指南：如何让AI自动理解视频内容
  • GRETNA开源工具实战指南：从零掌握MATLAB脑网络分析
  • 乘法逆元、组合数取模刷题总结