做Agent这类大模型应用最让人头疼的往往不是模型本身答得不好而是模型在调用外部工具时莫名其妙就失败。你以为让它查个天气、调个数据库结果工具抛个异常、返回个错误码整个流程就断在那里用户那边只能看到一句“抱歉我暂时无法完成这个操作”。说实话这种体验做一次demo还能忍放到生产环境里基本就是事故。这时候给工具调用加一套可靠的重试机制就成了从demo走向生产的必修课。这篇文章我会用LangGraph完整演示一个场景让大模型决定是否调用天气查询工具而这个工具有时会失败我们需要通过LangGraph的条件路由让流程自动重试直到工具调用成功或者达到设定的上限最后给用户一个可解释的结果。文章包含完整代码、路由设计思路和我在实际调试中踩过的坑适合正在用langchain/langgraph做Agent、被工具调用稳定性折磨过的朋友参考。1. 工具调用为什么会失败重试机制到底要解决什么1.1 大模型调用工具失败的常见原因先别急着写重试逻辑我们把“失败”这件事拆开看。大模型调用工具整个过程其实分两段第一段是模型本身根据用户问题决定“该调用哪个工具、传什么参数”第二段是执行层真正去调用那个工具拿到结果或异常。第一段出问题常见表现是模型生成了不存在的工具名、参数格式不对、必填参数缺失。这种情况不是重试能解决的即使重试一百遍模型大概率还会犯同样的错需要做的是把错误信息反馈给模型让它修正调用方式或者干脆用验证层兜住。第二段出问题才是重试机制真正要管的范畴。典型场景包括第三方API超时、上游服务返回5xx、数据库连接池满了、网络闪断、被限流。这些失败具备一个共同特征它们不是固定的、可预期的逻辑错误而往往是临时性的、概率性的故障。对付这类故障最简单也最有效的策略就是重试一次不行两次两次不行三次直到服务恢复。我在标题里专门强调了“模拟”两个字原因就在这里真实项目里你没法控制第三方服务什么时候挂但做演示和测试时我们需要一个能人为控制失败概率的工具这样才能验证重试逻辑在各条分支下是否都走得对。后面我实现的天气工具会以40%的概率随机抛异常模拟一个不稳定的上游服务。1.2 重试机制的设计目标与边界重试不等于无脑循环一个合格的重试机制至少要回答四个问题重试多少次每次间隔多久什么时候彻底放弃放弃之后给用户什么反馈第一个问题重试次数不能拍脑袋定。很多老手喜欢设3次因为经验值。更稳的做法是根据失败类型和业务容忍度来定读操作、查询类工具重试5到10次都无所谓写操作、下单支付类工具重复执行可能产生脏数据那宁可少重试甚至必须加幂等键。第二个问题间隔多久。如果上游已经限流了你还用固定1秒间隔疯狂重试只会把限流拖得更久。平时我会用指数退避加抖动比如第1次失败后等0.5秒第2次等1秒第3次等2秒加上一点随机偏移避免多个请求同时撞上服务恢复窗口。第三个问题什么时候放弃。重试次数达到上限就该停但不能停下来之后装作无事发生。最后一定要把“工具连续失败N次、最后一次错误是什么”这段信息作为工具结果返回给大模型让模型基于这个错误生成一个诚实的兜底回答而不是编造一个假数据或者直接卡死。第四个问题在LangGraph里其实有很好的实现方式就是利用条件边把“重试”和“放弃”都规划成图里的路径。下面我会用一个完整例子把这套逻辑跑通。2. 方案选型为什么用LangGraph而不是普通循环2.1 LangGraph核心概念速览如果你还没用过LangGraph先花30秒建立一下基本认知。LangGraph是LangChain团队推出的有状态Agent编排框架核心抽象就是三样东西State、Node、Edge。State是贯穿整个执行流程的共享状态你可以把它理解成一张“公用的纸”每个节点都能在上面读写内容。在代码里通常用TypedDict定义比如{messages: [...], attempts: 0}。Node是图里的执行单元每个节点是一个函数输入State返回更新后的State。Edge负责把节点串起来普通边就是“A走完必然去B”条件边则是“A走完根据State的内容决定去B还是去C”。这三样组合起来就能把Agent的高层控制流画成一张有向图。尤其是条件边这正是实现“失败后重试直到成功”的关键设施工具调用结果不理想时条件边让我们有能力让流程“绕回去”再试一次而不是只能继续往下走。2.2 两种重试实现思路的对比不用LangGraph也能实现重试最常见的做法是写一个while循环包住工具调用失败就sleep然后再来一发。代码量很少效果立竿见影比如下面这个简陋版本for attempt in range(5): try: result get_weather(city) break except Exception as e: time.sleep(2 ** attempt) print(f第{attempt 1}次失败: {e})这段代码在自己的脚本里没有任何问题但一旦放进Agent完整流程短板就暴露了它没法和大模型的决策、工具结果反馈、对话历史管理形成整体。重试逻辑散落在业务代码里模型前一轮已经生成了 tool_call你重试时是重新让模型生成一次还是延用原来的 tool_call失败信息要不要告诉模型重试超过上限后怎么让模型知道“这个工具不行了请换个方案”这些用while循环处理会很别扭。换成LangGraph的条件边方案就不一样了。重试不再是藏在函数内部的循环而是图结构里的环节点执行的每一步都有记录、可观测、可持久化还能在重试耗尽后优雅地让LLM接手生成兜底语言。从工程可维护性角度讲这套图结构的成本控制得比手写状态机好太多尤其当Agent节点变多时优势非常明显。另外说明一点LangGraph本身也提供节点级的RetryPolicy它会在节点抛异常时自动重跑整个节点。但RetryPolicy重试的是“节点执行”这个动作它不知道工具返回的是业务错误还是系统异常也不负责把失败信息转达给LLM。如果工具调用是抛异常RetryPolicy能用如果工具只是返回一段错误文本RetryPolicy就无能为力了。所以我更倾向于自己用条件边实现重试这样语义完全可控。下面的核心代码就是把这套可控性落到实处的过程。3. 从零搭建完整代码实现LangGraph工具调用重试3.1 环境准备与依赖安装开始前先把依赖装好。我用的Python版本是3.10以上LangGraph的0.2.x版本API相对稳定建议保持畅通更新。pip install -U langgraph langchain langchain-openai然后配置大模型。标题场景用OpenAI兼容接口最方便国内大模型比如DeepSeek、通义、智谱等也大多提供OpenAI兼容的base_url照着填就行from langchain_openai import ChatOpenAI llm ChatOpenAI( modelqwen-plus, # 换成你有的模型名 temperature0, # 工具调用建议低温避免随机行为 openai_api_key填你的密钥, openai_api_base填你的兼容接口地址, # 可选 )temperature设成0是我踩出来的习惯。工具调用场景里模型的参数生成如果引入过多随机性经常会出现同样的调用参数每次生成都不一样这会让重试逻辑的输入不稳定。3.2 定义模拟工具与系统状态先定义一个模拟的天气查询工具模拟“上游不稳定”这个核心前提。我故意让它以40%概率抛异常方便我们反复验证重试路径import random from langchain_core.tools import tool tool def get_city_weather(city: str) - str: 根据中文城市名查询该城市当前天气情况。 if random.random() 0.4: raise RuntimeError(upstream weather service unavailable, please retry later) return f{city}现在天气晴28℃湿度60%适合出行。真实项目里这个函数内部往往是一个HTTP调用比如通过requests或httpx请求第三方API。异常类型也不一定是RuntimeError可能是TimeoutException、HTTPStatusError或自定义的业务异常但处理的思路完全一样。接下来定义整个Agent执行过程中共享的State。这里有一个设计点要特别说明重试计数和错误信息我单独用字段管理不塞进messages列表。from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] attempts: int max_attempts: int last_error: str tool_success: bool为什么不把每次错误都写进messages如果你把每次失败都作为ToolMessage往回塞模型会在token里看到一堆失败历史。重试成功时还好一旦重试耗尽模型被迫看几页错误日志回答质量会明显下降还白白浪费token。让错误只留在State字段里只有最终需要LLM接收时再写入这是更体面的做法。3.3 构建LLM决策节点与工具执行节点先看LLM节点它负责根据当前对话历史决定是否要调用工具、调用哪个工具llm_with_tools llm.bind_tools([get_city_weather]) def call_llm(state: AgentState): response llm_with_tools.invoke(state[messages]) return {messages: [response]}这个节点返回的AIMessage通过add_messages自动追加到状态里。如果模型认为需要调用工具这条消息里会带上tool_calls字段里面包含工具名、参数、以及一个唯一的tool_call_id。再看工具执行节点这是整个重试逻辑的中心。工具失败时我在这里做计数、记录错误并在达到上限时把最终错误作为ToolMessage吐给模型from langchain_core.messages import ToolMessage def call_tool(state: AgentState): last_message state[messages][-1] tool_calls last_message.tool_calls # 单工具场景先取第一个调用验证逻辑多工具场景见第4.2节 tool_call tool_calls[0] try: result get_city_weather.invoke(tool_call[args]) return { messages: [ ToolMessage(contentresult, tool_call_idtool_call[id]) ], attempts: 0, # 成功则清零计数 tool_success: True, last_error: , } except Exception as exc: new_attempts state[attempts] 1 if new_attempts state[max_attempts]: # 重试耗尽把错误结论转告LLM让它生成兜底回答 return { messages: [ ToolMessage( contentf工具调用已失败{new_attempts}次最后一次错误{exc}。 f建议放弃该工具直接基于已有信息回答用户。, tool_call_idtool_call[id], ) ], attempts: new_attempts, tool_success: False, last_error: str(exc), } print(f第{new_attempts}次失败{exc}准备重试) return { attempts: new_attempts, tool_success: False, last_error: str(exc), }这段代码有两个我认为最能体现设计质量的细节。一是失败时不写消息只更新State字段二是达到上限的那一次失败用ToolMessage把“最终放弃”这个结论传给LLM。这样LLM能看到自己发出的tool_call有对应的返回内容不会觉得自己在跟空气对话从而可以自然地生成兜底回答。3.4 实现重试路由逻辑路由逻辑是LangGraph方案相对于普通while循环最有“图味道”的部分。两个路由函数一个决定LLM之后去哪一个决定工具执行之后去哪def route_after_llm(state: AgentState): last_message state[messages][-1] if last_message.tool_calls: return call_tool return end def route_after_tool(state: AgentState): if state[tool_success]: return call_llm # 工具成功回到LLM生成最终回答 if state[attempts] state[max_attempts]: return call_llm # 重试耗尽也要回到LLM但此时带了错误信息 return call_tool # 还没到上限继续重试同一个工具核心逻辑在route_after_tool里只要没成功、没超限条件边就指回call_tool节点。注意这时候messages列表最后一条仍然是之前那条带tool_calls的AIMessage所以call_tool节点每次执行时会读到相同的tool_call参数重试的就是同一次调用而不是让模型重新生成。这一点必须理解否则你会陷入“每次重试参数都不一样”的困惑。3.5 组装编译并验证执行效果最后把节点和边组装成图。结构是这样START进入call_llm如果模型生成tool_calls就去call_tool否则直接结束call_tool执行完通过条件边判断是继续重试、回LLM还是结束。from langgraph.graph import StateGraph, START, END builder StateGraph(AgentState) builder.add_node(call_llm, call_llm) builder.add_node(call_tool, call_tool) builder.add_edge(START, call_llm) builder.add_conditional_edges( call_llm, route_after_llm, { call_tool: call_tool, end: END, }, ) builder.add_conditional_edges( call_tool, route_after_tool, { call_tool: call_tool, call_llm: call_llm, }, ) graph builder.compile()执行的时候注意两点一是初始状态里必须把max_attempts带上二是官网文档没强调但实际很容易踩的recursion_limit参数。LangGraph默认递归上限是25如果我们的max_attempts设得比较大或者工具一直失败导致循环很多轮可能要显式调高result graph.invoke( { messages: [{role: user, content: 查询一下上海的天气}], attempts: 0, max_attempts: 5, last_error: , tool_success: False, }, {recursion_limit: 50}, ) print(result[messages][-1].content)我实测过多次输出可能有两种走向。工具在某次重试中成功流程会输出类似“上海现在天气晴28℃湿度60%适合出行”的正常结果如果5次全部失败打印会连续输出失败日志最后模型会给出类似“天气服务暂时不可用建议稍后再试”的兜底回答。这两种路径一次性都验证通过说明重试机制在逻辑上是完备的。4. 深入细节重试策略的进阶设计4.1 指数退避与抖动上面代码里我没有加任何sleep真实场景是不行的失败后立即重试有时候上游恰恰是因为瞬时高负载才失败立刻再打一次反而容易加重拥堵。重试之间必须等待而且等待时间要有策略。我习惯用指数退避加抖动。指数退避就是等待时间随重试次数指数增长抖动是在退避基础上加一个随机数避免所有客户端在同一时间点“准时”重试形成惊群效应。同步节点直接time.sleep即可import time import random def _backoff_delay(attempt: int) - None: base_delay 2 ** (attempt - 1) # 第1次失败后等1秒第2次等2秒…… jitter random.uniform(0, 0.5) time.sleep(base_delay jitter)注意这段代码不能照搬到异步节点。如果你把call_tool定义成async函数sleep必须改成await asyncio.sleep(...)否则会阻塞事件循环。用LangGraph写异步Agent时这是一个高频踩坑点很多人同步代码跑得好好的一换异步节点就出现性能问题检查了很久才发现是time.sleep卡住了整个事件循环。4.2 多工具并行时的重试语义上面实现只处理了一个tool_call。但真实场景里模型经常一次生成多个tool_calls比如同时查天气、查火车票、查酒店。如果其中两个成功、一个失败怎么重试我的建议是工具执行节点里不要囫囵吞枣整体重试。正确做法是按tool_call_id区分每个tool_call单独执行、单独记录失败次数、单独决定是否重试。一旦某个工具失败且未超限就只重试那一个不要把已经成功的工具调用结果丢掉重来一遍。实现上可以给状态加一个类似tool_attempts: dict[str, int]的字段用tool_call_id做key来记账。这个改造并不复杂但语义清晰很多避免了你批量重试时把成功结果也重新执行、引发重复副作用的问题。如果工具本身有副作用这个问题会非常严重。4.3 幂等性检查与重试安全聊到副作用就绕不开幂等性这个关键话题。工具重试最大的安全隐患就是工具本身不幂等。查询天气重试一万次都没问题但“创建订单”“发送短信”“扣减库存”这类写操作如果第一次调用其实已经成功了只是响应在网络传输中丢了客户端一重试就会重复下单、重复扣款。处理办法有三种思路我按推荐的优先级来说。第一种是在工具调用前做幂等性评估明确哪些工具允许重试、哪些不允许这个信息甚至可以前置配置在工具元数据里。第二种是给每次工具调用生成一个幂等键放在参数里传给上游服务端保证相同幂等键只执行一次。第三种是业务层面兜底工具执行前先查一下“这件事是不是已经发生过”比如查订单号是否已存在。在你做Agent框架设计时我强烈建议把“是否允许重试”提升为工具的一个一等公民属性而不要依赖开发人员的自觉。我的经验是凡是不标注重试策略的工具默认按“不允许重试”处理反而安全很多。4.4 重试上限耗尽后的全局兜底重试耗尽不等于流程失败恰恰是流程进入“兜底阶段”的开始。我最初实现时重试耗尽直接return报错整个graph抛异常用户什么结果都拿不到。后来改成把错误总结成ToolMessage交给LLM处理效果立竿见影。兜底阶段的策略可以再细化。最简单的是已经写好的那种让LLM基于错误信息给出礼貌的抱歉文案。进阶一点可以设计成让模型尝试调用替代工具比如查天气失败改查一个备用的天气数据源。再复杂一点可以引入human-in-the-loop在重试耗尽时用LangGraph的interrupt()暂停执行把“是否放弃当前工具”的决策交给人来处理用户确认后再继续。生产级的Agent系统里这个环节非常值得投入。5. 常见问题与排查实录5.1 无限循环条件边漏了某个分支我在早期版本里犯过一个低级错误route_after_tool忘记判断tool_successTrue的情况导致工具成功后仍然走回call_tool节点又去读messages里最后一条AIMessage的tool_calls然后执行同一个工具陷入死循环。LangGraph的recursion_limit会在25步时强制报错这才把问题暴露出来。排查经验是凡涉及条件边的图先把所有可能的状态组合列出来逐一核对路由目标。工具执行之后无非四类结果成功、失败但可重试、失败且耗尽、没有tool_call。我的路由里前两个走call_tool或call_llm后两个在route_after_llm里兜住返回end这样才算是闭环。5.2 状态里messages无限膨胀如果你在失败时不停往messages里写ToolMessage重试20次就会多出20条工具错误消息token消耗剧烈增长模型还会被一堆错误信息干扰判断。我在第3.2节已经强调过失败日志不要写进messages等重试耗尽时一次性把总结写进去。如果确实需要把每次尝试都记录下来供审计更好的是在State里加一个attempt_log: list[str]字段用普通列表存不经add_messagesreducer这样不占用LLM的上下文窗口。配合LangSmith追踪调试时也能看到完整的尝试历史。5.3 tool_call_id对不上的报错ToolMessage必须带tool_call_id而且要和LLM生成的tool_call里的id严格一致。我见过不少同学手写死一个字符串或者从别的地方复制了一个id结果LangGraph或模型的校验直接报错整个流程崩掉。规范写法永远是从last_message.tool_calls[0][id]里取值不要自己组装。如果一次处理多个tool_calls循环里逐个取出对应的id不要搞混。5.4 异步场景下的超时处理真实工具调用几乎都是网络IO异步写法更合理。但异步节点有几个坑sleep必须用asyncio.sleep每个工具调用要自己控制超时比如用inasyncio.wait_for包住否则上游服务挂死的话你的重试循环也会挂死在那里重试次数设多少都没意义。为每个工具调用设置明确的超时时间比如5秒或10秒超时视作一次失败进入重试逻辑这比让整个节点无限等待要理智得多。5.5 与LangGraph自带RetryPolicy的关系最后说回LangGraph自带的RetryPolicy。它确实能对节点异常做自动重试配置方式如下from langgraph.pregel.retry import RetryPolicy node_with_retry RetryPolicy( max_attempts3, retry_onlambda exc: isinstance(exc, RuntimeError), backoff_factor2.0, )但它的定位是“节点执行层面的韧性”和“工具调用业务层面的重试”是两码事。RetryPolicy会在整个节点抛异常时重跑节点可能出现重复执行已有副作用它也不会把重试的中间过程传达给LLM。我的实践结论是简单的、无副作用的工具用RetryPolicy省事需要精细控制、需要LLM参与兜底的场景用条件边方案。两者可以共存但不要混在一起用否则同一个失败会被两层重试策略各处理一次行为会很诡异。6. 我在实际项目里最后落地的几个习惯这套方案我在几个真实项目里跑过之后沉淀下来一些固定习惯分享出来供你参考。第一所有工具在接入Agent之前先打标标明是否允许重试、允许几次、失败后是否可以换备选工具。这个元数据描述从工具定义阶段就写清楚比等到调试时再临时决定靠谱得多。第二重试次数和退避参数不要写死在代码里放到配置中心或者环境变量因为线上出问题时你往往需要改参数来止血而改代码发版太慢。第三每个工具节点的执行时间、成功次数、失败次数、重试次数全部埋点上报否则你根本不知道自己重试了这么多次到底有没有意义。第四重试日志里一定要带上tool_call_id和具体的参数快照方便在LangSmith或日志平台里把一次异常的前因后果完整还原。我最初做这套东西的时候天真地以为重试就是个do-while循环后来发现真正难的不是重试这个动作而是整个控制流和大模型、工具结果、用户反馈的闭环。当你用LangGraph把图结构搭起来之后重试只是条件边上一个分支而已整个系统的可维护性会高出很多。如果你正在做的Agent也频繁被工具调用稳定性困扰不妨照着这套思路把你自己的工具节点接进去跑一遍相信你能比我更快地踩完这些坑。