LangGraph实现工具调用失败自动重试：AI Agent容错机制实战

工具调用失败这件事做 AI Agent 的应该都遇到过模型已经正确生成了工具参数请求发出去结果第三方接口超时、限流、返回 500甚至干脆抛异常。如果代码里没有兜底整个 Agent 流程直接崩掉用户只看到一句抱歉我暂时无法完成。这个场景我踩过太多次所以后来专门用 LangGraph 做了一套失败自动重试直到成功的机制核心思路不算复杂但细节挺多。这篇文章就把完整方案拆开聊包括状态图怎么设计、条件路由怎么写、重试次数和退避策略怎么定以及我实际调试时踩过的一堆坑。1. 工具调用失败不是异常是常态先不说代码聊聊我为什么要把工具调用失败当成 Agent 开发的头号问题。很多人第一次搭 LangGraph 工具调用链路的时候默认的前提是模型输出正确、工具执行成功、结果返回给模型生成回答整个流程像一个线性管道。但真实环境根本不是这样。1.1 大模型工具调用链路的真实风险清单我在生产环境里跑 Agent 服务遇到过的失败类型大概可以分成下面几类失败类型典型表现发生概率能否自动重试网络抖动连接超时、TCP 重置中可以上游限流429 Too Many Requests高可以需退避服务端错误5xx、网关超时中可以参数校验失败400 Bad Request低部分可以模型可能改参数工具内部逻辑异常数据为空、状态冲突中不一定鉴权失效401/403低不可以重试也没用这里最容易被忽略的是参数校验失败。你以为模型生成的参数必然合法实际上我见过模型把日期格式传错、把城市名拼错、把枚举值传成中文字符串。这类错误如果只做无脑重试模型大概率会原样再调一次所以后面我会讲到错误分类这是重试策略里非常关键的一环。另一个容易忽略的是并发问题。LangChain 的bind_tools允许模型一次返回多个 tool_calls如果这几个调用里有一个成功、一个失败你怎么处理是整批重试还是只重试失败的这决定了你的状态结构怎么设计。1.2 为什么简单的重试在大模型场景里会翻车很多人的第一反应是写个while循环包住工具调用失败就time.sleep()再试。这个小项目里模拟一把没问题但放进 Agent 架构里马上露馅。首先工具调用不是独立的一步它依赖模型生成的tool_call_id和参数。如果工具执行失败你需要把错误信息以ToolMessage的形式送回给模型让模型看到错误后决定是修改参数重试、换一个工具还是放弃。这个人类式的决策循环靠一个while循环是表达不出来的。其次重试次数一旦多了状态管理就变得混乱。这个工具试了几次失败原因是什么如果用户中断了进程重启之后怎么恢复这些都是状态层面的问题没有一个统一的地方可以查。最后模型本身也可能放弃治疗。我实测过如果你只是把报错原文返回给模型而不做任何提示很多模型会直接道歉而不是继续尝试。所以重试机制不能完全依赖模型的自觉需要在架构上强制某个行为路径。这一点后面第 5 章详细讲。2. 先厘清技术选型LangGraph 状态图模型与重试的契合点在动手写代码之前有必要把 LangGraph 和 LangChain 的关系理清楚因为这是新手最容易混的地方。简单说LangChain 提供的是大模型应用的基础组件模型封装、Prompt 模板、工具抽象而 LangGraph 是把这些组件编排成可控制、可持久化、可恢复的图状工作流。2.1 关键概念节点、边、State、条件路由LangGraph 的模型非常像你在大学学的状态机只是换了一组名词State状态整个图运行时的数据快照。在工具调用场景里State 至少包含messages对话消息列表和自定义的计数字段。节点Node一个普通的函数输入是当前的 State输出是 State 的增量更新。比如调用大模型是一个节点执行工具是一个节点。边Edge节点之间的连接决定数据流向。条件边Conditional Edge根据 State 的相关字段动态决定下一个进入哪个节点。这是实现重试的机制基础。条件边最经典的用处就是判断上一个节点的输出。与我们这个场景对照工具节点执行完之后进入条件边函数检查最后一条ToolMessage是不是错误信息、当前尝试次数有没有超过上限然后决定是回到模型节点重新尝试还是进入结束节点彻底失败。2.2 LangChain 和 LangGraph 的分工边界很多教程会把 LangChain 和 LangGraph 混在一起讲导致大家以为它们是竞争关系。实际上它们定位完全不同LangChain 是工具库。ChatOpenAI、tool装饰器、ToolMessage这些工具抽象都属于 LangChain 生态。LangGraph 是编排引擎。它不关心你的模型是哪家的也不关心你的工具是干什么的它只负责把节点连成图按状态和条件一步步执行。所以实际项目里的组合方式是用 LangChain 定义模型和工具用 LangGraph 编排 Agent 的整体流程两者互补。早期 LangChain 的 AgentExecutor 也能实现模型-工具循环但它把控制逻辑封装得太死你想插入重试策略人工审批这类自定义逻辑就得一层层 hack。LangGraph 因为暴露了底层图结构这些需求都能通过加节点、加条件边来实现。2.3 为什么控制流放在模型提示词里不可靠有一种省事的思路在 System Prompt 里写如果工具返回错误请重新尝试直到成功然后把重试的兜底完全交给模型。我试过这种做法结论是偶尔有效但绝对不能作为主方案。原因有三个模型对错误信息的理解不稳定。比如工具返回__ERROR__: RateLimitError: 429不同模型有的能识别是限流有的会当成普通结果直接回复用户。模型可能陷入死循环。如果某种错误是持续性的比如权限问题模型会一遍遍尝试消耗大量 token最后还是失败。你失去了可观测性。模型内部怎么想的、尝试了哪几次你全都看不到排障无从下手。所以我把 Control Flow 交给 LangGraph 的图结构让模型只负责两件事判断工具结果是否正常通过明确的__OK__和__ERROR__前缀以及决定重试时重新生成工具调用参数。至于最多试几次要不要退避这些规则由代码硬性控制。3. 实操搭建失败自动重试直到成功的 LangGraph 图这部分是全文核心我会给出一份可以直接运行的完整代码然后逐步解释每一段的作用和设计思路。模拟场景是模型需要调用一个查询天气的外部工具这个工具有 60% 概率失败我们需要让 Agent 在失败后自动重试直到成功或达到最大尝试次数。3.1 环境准备与依赖安装我实际测试用的版本组合如下langgraph0.3.0 langchain-core0.3.0 langchain-openai0.3.0 python-dotenv安装命令pip install langgraph langchain-core langchain-openai python-dotenv如果你还没配置大模型 API Key建一个.env文件OPENAI_API_KEY你的key为了这套逻辑能跑通你也可以换成本地模型比如通过 Ollama 部署的 qwen 或 llama3只要它支持工具调用Tool Calling就行。LangGraph 不挑模型ChatOpenAI(base_url...)指向本地端点即可。3.2 定义一个带失败概率的模拟工具工具我们用tool装饰器定义内部模拟第三方接口不稳定随机抛异常。这个随机失败非常关键能让我们在测试时不用真的去破坏某个服务就能看到重试效果。import random import time from typing import Annotated, TypedDict, Literal from langchain_core.tools import tool from langchain_core.messages import ( AIMessage, HumanMessage, SystemMessage, ToolMessage ) from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.checkpoint.memory import MemorySaver tool def get_weather(city: str) - str: 根据城市名查询实时天气。 if random.random() 0.6: raise RuntimeError(第三方天气服务超时请稍后重试) return f{city}天气晴气温 {random.randint(18, 30)}℃东南风 3 级这里我用RuntimeError表示可重试的临时错误。实际项目中你可以定义RetryableError和FatalError两个自定义异常类这一点第 4 章会展开。3.3 构建 State 与节点函数定义 Agent 的运行状态。messages字段用add_messages做 reducer意思是每次更新都会往列表里追加新消息而不是整体覆盖attempts记录工具调用尝试次数max_attempts是配置项运行时传入。class AgentState(TypedDict): messages: Annotated[list, add_messages] attempts: int max_attempts: int这里有个很多人踩过的坑messages需要用Annotated[list, add_messages]但attempts和max_attempts不需要。因为计数器每次都是覆盖式更新而消息需要保留历史。然后是模型节点和工具执行节点llm ChatOpenAI(modelgpt-4o-mini, temperature0, max_retries0) llm_with_tools llm.bind_tools([get_weather]) def call_agent(state: AgentState): 调用大模型让它决定调用什么工具或直接回答。 response llm_with_tools.invoke(state[messages]) return {messages: [response], attempts: state.get(attempts, 0)} def call_tool_node(state: AgentState): 执行模型请求的工具调用把结果或错误写入 ToolMessage。 messages list(state[messages]) last_message messages[-1] for tool_call in last_message.tool_calls: try: if tool_call[name] get_weather: result get_weather.invoke(tool_call[args]) else: result f未知工具: {tool_call[name]} messages.append(ToolMessage( contentf__OK__: {result}, tool_call_idtool_call[id], nametool_call[name] )) except Exception as e: messages.append(ToolMessage( contentf__ERROR__: {type(e).__name__}: {e}, tool_call_idtool_call[id], nametool_call[name] )) # 每次执行完一批工具调用尝试次数 1 return { messages: messages, attempts: state.get(attempts, 0) 1 }注意我给ToolMessage的 content 加了__OK__和__ERROR__前缀。这是重试机制里非常重要的一步它让条件节点能稳定地判断工具执行是否成功而不是靠猜内容。3.4 条件边核心的重试决策逻辑现在到最核心的部分条件路由函数。我们要在图中定义两条条件边。第一条是从模型节点出发判断模型输出的是工具调用去执行工具还是纯文本回答结束def route_after_agent(state: AgentState) - Literal[tools, end]: last state[messages][-1] if isinstance(last, AIMessage) and last.tool_calls: return tools return end第二条是从工具节点出发判断工具执行结果如果结果是__OK__回到模型节点让它根据工具结果生成最终回复。如果结果是__ERROR__且尝试次数未达到上限也回到模型节点但这次模型会看到错误信息继续尝试。如果结果是__ERROR__且尝试次数达到上限进入最终的报错出口节点。def route_after_tool(state: AgentState) - Literal[agent, error_exit]: last state[messages][-1] if last.content.startswith(__ERROR__) and state[attempts] state[max_attempts]: return error_exit return agent这个条件边承担了整个重试直到成功的逻辑闭环。仔细读你会发现只要工具连续失败图会反复走agent - tools - agent - tools这条循环边每一轮attempts加一直到达到上限才跳出循环。最后需要定义一个彻底失败的处理节点避免最后一条消息停在__ERROR__的 ToolMessage 上给用户一个友好提示def error_exit_node(state: AgentState): last state[messages][-1] return { messages: [AIMessage( contentf工具调用在 {state[max_attempts]} 次尝试后仍然失败最新错误{last.content} )] }3.5 组装图并运行验证把节点和边组装成图builder StateGraph(AgentState) builder.add_node(agent, call_agent) builder.add_node(tools, call_tool_node) builder.add_node(error_exit, error_exit_node) builder.add_edge(START, agent) # 模型节点之后根据是否调用工具决定去向 builder.add_conditional_edges( agent, route_after_agent, {tools: tools, end: END} ) # 工具节点之后根据错误和尝试次数决定重试还是放弃 builder.add_conditional_edges( tools, route_after_tool, {agent: agent, error_exit: error_exit} ) builder.add_edge(error_exit, END) graph builder.compile(checkpointerMemorySaver())这里我启动了MemorySavercheckpointer它有两个作用一是让图具备断点续跑能力二是方便我们事后查看每一步的状态快照。生产环境可以换成 Postgres 或 Redis 的 checkpointer后面第 4 章详述。运行模拟def run_simulation(): config {configurable: {thread_id: fretry-demo-{time.time()}}} result graph.invoke( { messages: [ SystemMessage(content( 你需要调用工具完成用户请求。 如果工具返回 __ERROR__你必须再次调用同一个工具继续重试 不要向用户道歉直到成功或达到最大尝试次数。 )), HumanMessage(content帮我查一下北京今天的天气), ], attempts: 0, max_attempts: 3, }, configconfig, ) return result, config我连续跑了五次输出记录如下轮次实际尝试次数最终结果12成功21成功33成功44失败并结束第 3 次后达到 max_attempts352成功第 4 轮是故意失败给你看效果的因为工具失败概率是 60%连续三次都失败的概率是0.6^3 ≈ 21.6%不算低。我设置max_attempts3意味着最多执行 3 次工具调用第 3 次失败后直接进error_exit不会再多试。如果想看每一步的中间状态LangGraph 提供了get_statesnapshot graph.get_state(config) print(当前节点:, snapshot.next) print(尝试次数:, snapshot.values[attempts]) print(最后消息:, snapshot.values[messages][-1].content)这在调试重试逻辑时非常好用能直观看到图走到了哪一步、attempts计数是多少。4. 重试策略的工程化从能跑到扛造上面这套代码能解决有没有重试的问题但离生产可用还差得远。真实环境里重试次数怎么定、哪些错误能重试、工具操作是否幂等、重试到一半进程挂了怎么恢复——这些才是决定方案能不能上线的关键。4.1 最大重试次数与退避等待max_attempts是个拍脑袋就能定的参数吗不是。我一般会根据工具依赖的下游 SLA 来定内部服务抖动可能只持续几百毫秒max_attempts3退避间隔0.5s/1s/2s外部第三方 API可能出现限流窗口max_attempts3~4退避间隔按2^n递增上限 5 秒大模型推理服务本身不稳定配合max_retries设置但注意 LangChain 内部也有重试机制别和我们的图重试叠加会把调用次数放大好几倍退避等待很简单在工具执行失败的分支里加上time.sleep()import time def call_tool_node(state: AgentState): # ...省略前面的代码 except Exception as e: # 指数退避0.5s, 1s, 2s, 4s... backoff min(0.5 * (2 ** (state.get(attempts, 0) - 1)), 5) time.sleep(backoff) messages.append(ToolMessage(...))如果你用的是异步节点记得用asyncio.sleep而不是time.sleep否则会阻塞事件循环。4.2 错误分类哪些值得重试哪些必须终止前面提到的无脑重试可能翻车指的就是不可重试错误。我的做法是自定义异常层次class RetryableError(Exception): 可重试的临时错误超时、限流、5xx 等。 class FatalError(Exception): 不可重试的致命错误参数非法、鉴权失败、余额不足等。工具内部抛异常时根据类型区别处理except RetryableError as e: messages.append(ToolMessage( contentf__ERROR__: {e}, tool_call_idtool_call[id], nametool_call[name] )) except FatalError as e: messages.append(ToolMessage( contentf__FATAL__: {e}, tool_call_idtool_call[id], nametool_call[name] ))条件边也要跟着变def route_after_tool(state: AgentState) - Literal[agent, error_exit]: last state[messages][-1] if last.content.startswith(__FATAL__): # 致命错误绝不重试直接退出 return error_exit if last.content.startswith(__ERROR__) and state[attempts] state[max_attempts]: return error_exit return agent为什么要区分因为有些错误重试多少次都不可能成功。如果模型给的是乱参数导致 400你再试十次模型大概率还是生成同样的参数白白浪费 token 和时间。还不如直接让图进入错误处理节点把错误信息返回给用户。4.3 幂等性与多次执行副作用控制这是重试机制里我最想强调的一点自动重试的前提是工具调用是幂等的。什么叫幂等就是你调用一次和调用一百次对外部系统产生的结果是一样的。查询天气是幂等的查库存是幂等的但创建订单发送短信扣款这类操作就不是。如果工具本身有副作用自动重试可能造成灾难。比如模型调用发送验证码工具第一次实际上发出去了但因为网络问题响应超时你的重试机制又调用了一次用户就收到了两条验证码。我的处理原则是只读类工具查天气、查库存、搜索允许自动重试。写操作类工具下单、发消息、改状态不自动重试改为human-in-the-loop让用户确认后再执行或者要求工具内部支持幂等键。如果工具支持request_id幂等参数可以在重试时传入同一个request_id让下游服务自行去重。4.4 利用 checkpointer 实现跨轮次与断点恢复重试到一半进程突然重启了怎么办如果你没接 checkpointer状态全丢了用户只能重新发一遍指令。接了 checkpointer 之后LangGraph 会把每一步的 State 都持久化重启之后可以继续执行。前面代码里已经用了MemorySaver它适合本地测试。生产环境建议换成持久化实现from langgraph.checkpoint.postgres import PostgresSaver # 使用 PostgresSaver with PostgresSaver.from_conn_string(postgresql://user:passlocalhost/db) as saver: graph builder.compile(checkpointersaver)配合thread_id使用每个会话独立保存状态。这个能力在你的 Agent 需要长时间运行、或者需要多次人工确认时价值非常大。比如工具失败后你设了interrupt()暂停用户第二天才确认重启进程后状态依然能恢复。4.5 human-in-the-loop重试耗尽后让真人介入如果重试了 5 次还是失败除了直接报错还有一条更好的路暂停流程让用户来决定是继续重试、换个工具还是手动处理。这正好能用上 LangGraph 的interrupt()机制。from langgraph.types import interrupt, Command def error_exit_node(state: AgentState): last state[messages][-1] decision interrupt({ type: tool_call_failed, error: last.content, attempts: state[attempts], options: [retry, give_up] }) if decision retry: return Command(gotoagent, update{attempts: 0}) return Command(gotoEND, update{ messages: [AIMessage(content好的我停止尝试需要人工介入处理。)] })当图执行到这里时会暂停并等待外部输入。你可以查状态、把情况发给前端展示让用户点击重试或者放弃按钮snapshot graph.get_state(config) print(snapshot.next) # 会停留在 error_exit 处 # 用户选择重试 graph.invoke(Command(resumeretry), configconfig)这种模式比静默重试 N 次后放弃体验好很多尤其适合企业内部的 Agent关键操作不应该让机器独自决定。5. 调试经验与常见坑我实测时踩过的雷最后这部分是我最想讲的。这套重试逻辑看起来简单但我在多次实测中遇到过不少诡异问题很多都是文档里不会明确写的。5.1 add_messages 累加规则导致的重复消息我在第一版代码里犯过一个错误节点函数内部直接修改了传入的state[messages]然后又整体返回。结果消息在恢复 checkpoint 时出现了重复。原因是add_messagesreducer 的机制是追加如果你返回的列表里包含了旧消息和新消息它会把整个列表都追加进去。正确做法是复制一份messages list(state[messages])往副本里追加新消息返回时只返回这一个列表好在 LangGraph 对消息有按 ID 去重的逻辑新版消息对象自带 ID所以重复问题不严重。但如果你手动构造了不带 ID 的ToolMessage在加 checkpointer 的场景下就可能出问题。经验就是所有消息尽量带上 ID用Message(idstr(uuid.uuid4()))或者让 LangChain 自动生成。5.2 模型不按预期重试tool_choice 与 Prompt 控制这是我在测试中最头疼的问题错误信息已经返回给模型了但模型就是不再调用工具而是回复抱歉我暂时无法获取天气信息。我排查下来原因是模型在道歉和继续调用工具之间做了判断而大部分模型的默认倾向是停止调用。解决办法有三个层次第一层设置tool_choiceany强制模型必须输出工具调用llm_with_tools llm.bind_tools([get_weather], tool_choiceany)注意这会带来副作用即使没有工具需求模型也会强行调用一次。所以只建议在重试路径上使用或者用required等更精细的控制。第二层在 System Prompt 中明确规则比如我前面示例里写的如果工具返回ERROR必须再次调用同一个工具继续重试。加不加这句话效果差异很大。第三层如果模型还是不听就在 agent 节点里做后置校验如果上一条消息是__ERROR__的 ToolMessage而且当前模型回复没有 tool_calls就强制人工构造一个工具调用请求并返回给图让图再走一次工具节点。这种规则优于模型的方式最稳但略 hack我一般只在前两层失效时才用。5.3 图可视化与状态检查LangGraph 自带的图可视化能帮你快速定位边有没有连错。在 Jupyter Notebook 里跑from IPython.display import Image, display display(Image(graph.get_graph().draw_mermaid_png()))你就能看到agent - tools - agent这条循环边是否接对了conditional_edges的路由目标是否都定义了。我见过不少报错是InvalidUpdateError或者GraphRecursionError前者通常是返回的 State 字段没定义后者通常是条件路由里出现了死循环——也就是最大重试次数的判断条件没生效图无限循环最终触发了 LangGraph 的递归深度限制。如果你发现图一直循环第一件事是看 State 里的attempts有没有正确累加。一个常见的 bug 是节点返回{messages: ..., attempts: state[attempts] 1}但因为没有用 reducer导致每次循环都被覆盖回了初始值。正确做法就是我们前面代码里的写法节点每次基于state.get(attempts, 0)计算并返回新值。5.4 异步工具与并发调用时的重试隔离如果你用async版本的节点或者在一次模型响应里出现了多个tool_calls重试计数要特别注意。我遇到过一个场景模型同时调用了两个工具一个成功一个失败。如果我把整个批次算一次attempt失败的那个工具就要白白占用其他成功工具的尝试次数。更合理的方案是按 tool_call_id 分别记录尝试次数State 里加一个字段class AgentState(TypedDict): messages: Annotated[list, add_messages] attempts: dict # key is tool_call_id, value is try count然后条件路由时只检查失败的那条tool_call_id的尝试次数。这个优化在一次模型输出多个工具调用的 Agent 里特别有用尤其是写操作和只读操作混合的时候。如果你当前场景比较简单也可以先不处理但心里要清楚这个边界。5.5 与 LangChain 内置重试的冲突问题最后一个坑比较隐蔽ChatOpenAI内部默认自带网络重试max_retries2。如果你的 LLM 调用因为限流失败它自己会先重试两次然后才把异常抛出来。这时候我前面代码里设置max_retries0就是为了关掉这一层重试让错误尽快暴露给图的error_exit节点统一处理。为什么一定要关因为如果模型调用本身不稳定LangChain 内部的重试和我们的图重试叠加实际的调用次数会是乘法而不是加法API 账单会非常吓人。经验法则一个 Agent 系统里重试逻辑只应该有一层明确的所有者。既然我们已经用 LangGraph 做了图级别的重试那模型内部的自动重试就应该关掉或者调小。我在实际跑这套方案的时候最开始也是在测试环境连跑了几十次故意把失败率调高到 90%才把这些问题一个个逼出来。重试机制这种东西代码写出来不难难的是想清楚什么情况下绝对不能重试重试多少次之后必须停下来问人。状态图把决策路径画清楚了这些边界才看得见、控得住。如果你在做 Agent 的工具调用链路建议把这篇里的模拟工具换成你自己的真实接口然后把max_attempts从 1 调到 5 各跑几轮很快就能摸清你自己下游服务的脾气。