企业官网建设流程全解析

工具调用失败这件事，做 AI Agent 的应该都遇到过：模型已经正确生成了工具参数，请求发出去，结果第三方接口超时、限流、返回 500，甚至干脆抛异常。如果代码里没有兜底，整个 Agent 流程直接崩掉，用户只看到一句"抱歉，我暂时无法完成"。这个场景我踩过太多次，所以后来专门用 LangGraph 做了一套"失败自动重试直到成功"的机制，核心思路不算复杂，但细节挺多。这篇文章就把完整方案拆开聊，包括状态图怎么设计、条件路由怎么写、重试次数和退避策略怎么定，以及我实际调试时踩过的一堆坑。

1. 工具调用失败不是异常，是常态

先不说代码，聊聊我为什么要把"工具调用失败"当成 Agent 开发的头号问题。很多人第一次搭 LangGraph 工具调用链路的时候，默认的前提是"模型输出正确、工具执行成功、结果返回给模型生成回答"，整个流程像一个线性管道。但真实环境根本不是这样。

1.1 大模型工具调用链路的真实风险清单

我在生产环境里跑 Agent 服务，遇到过的失败类型大概可以分成下面几类：

这里最容易被忽略的是"参数校验失败"。你以为模型生成的参数必然合法？实际上我见过模型把日期格式传错、把城市名拼错、把枚举值传成中文字符串。这类错误如果只做无脑重试，模型大概率会原样再调一次，所以后面我会讲到错误分类，这是重试策略里非常关键的一环。

另一个容易忽略的是并发问题。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 里写"如果工具返回错误，请重新尝试直到成功"，然后把重试的兜底完全交给模型。我试过这种做法，结论是：偶尔有效，但绝对不能作为主方案。原因有三个：

1. 模型对错误信息的理解不稳定。比如工具返回__ERROR__: RateLimitError: 429，不同模型有的能识别是限流，有的会当成普通结果直接回复用户。
2. 模型可能陷入死循环。如果某种错误是持续性的（比如权限问题），模型会一遍遍尝试，消耗大量 token，最后还是失败。
3. 你失去了可观测性。模型内部怎么想的、尝试了哪几次，你全都看不到，排障无从下手。

所以我把 Control Flow 交给 LangGraph 的图结构，让模型只负责两件事：判断工具结果是否正常（通过明确的__OK__和__ERROR__前缀），以及决定重试时重新生成工具调用参数。至于"最多试几次""要不要退避"，这些规则由代码硬性控制。

3. 实操：搭建"失败自动重试直到成功"的 LangGraph 图

这部分是全文核心，我会给出一份可以直接运行的完整代码，然后逐步解释每一段的作用和设计思路。模拟场景是：模型需要调用一个"查询天气"的外部工具，这个工具有 60% 概率失败，我们需要让 Agent 在失败后自动重试，直到成功或达到最大尝试次数。

3.1 环境准备与依赖安装

我实际测试用的版本组合如下：

langgraph>=0.3.0 langchain-core>=0.3.0 langchain-openai>=0.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(model="gpt-4o-mini", temperature=0, max_retries=0) 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( content=f"__OK__: {result}", tool_call_id=tool_call["id"], name=tool_call["name"] )) except Exception as e: messages.append(ToolMessage( content=f"__ERROR__: {type(e).__name__}: {e}", tool_call_id=tool_call["id"], name=tool_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( content=f"工具调用在 {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(checkpointer=MemorySaver())

这里我启动了MemorySavercheckpointer，它有两个作用：一是让图具备断点续跑能力，二是方便我们事后查看每一步的状态快照。生产环境可以换成 Postgres 或 Redis 的 checkpointer，后面第 4 章详述。

运行模拟：

def run_simulation(): config = {"configurable": {"thread_id": f"retry-demo-{time.time()}"}} result = graph.invoke( { "messages": [ SystemMessage(content=( "你需要调用工具完成用户请求。" "如果工具返回 __ERROR__，你必须再次调用同一个工具继续重试，" "不要向用户道歉，直到成功或达到最大尝试次数。" )), HumanMessage(content="帮我查一下北京今天的天气"), ], "attempts": 0, "max_attempts": 3, }, config=config, ) return result, config

我连续跑了五次，输出记录如下：

第 4 轮是"故意"失败给你看效果的：因为工具失败概率是 60%，连续三次都失败的概率是0.6^3 ≈ 21.6%，不算低。我设置max_attempts=3意味着最多执行 3 次工具调用，第 3 次失败后直接进error_exit，不会再多试。

如果想看每一步的中间状态，LangGraph 提供了get_state：

snapshot = 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_attempts=3，退避间隔0.5s/1s/2s
• 外部第三方 API，可能出现限流窗口：max_attempts=3~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( content=f"__ERROR__: {e}", tool_call_id=tool_call["id"], name=tool_call["name"] )) except FatalError as e: messages.append(ToolMessage( content=f"__FATAL__: {e}", tool_call_id=tool_call["id"], name=tool_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:pass@localhost/db") as saver: graph = builder.compile(checkpointer=saver)

配合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(goto="agent", update={"attempts": 0}) return Command(goto=END, update={ "messages": [AIMessage(content="好的，我停止尝试，需要人工介入处理。")] })

当图执行到这里时，会暂停并等待外部输入。你可以查状态、把情况发给前端展示，让用户点击"重试"或者"放弃"按钮：

snapshot = graph.get_state(config) print(snapshot.next) # 会停留在 error_exit 处 # 用户选择重试 graph.invoke(Command(resume="retry"), config=config)

这种模式比"静默重试 N 次后放弃"体验好很多，尤其适合企业内部的 Agent，关键操作不应该让机器独自决定。

5. 调试经验与常见坑：我实测时踩过的雷

最后这部分是我最想讲的。这套重试逻辑看起来简单，但我在多次实测中遇到过不少诡异问题，很多都是文档里不会明确写的。

5.1 add_messages 累加规则导致的重复消息

我在第一版代码里犯过一个错误：节点函数内部直接修改了传入的state["messages"]，然后又整体返回。结果消息在恢复 checkpoint 时出现了重复。

原因是add_messagesreducer 的机制是"追加"，如果你返回的列表里包含了旧消息和新消息，它会把整个列表都追加进去。正确做法是：

• 复制一份messages = list(state["messages"])
• 往副本里追加新消息
• 返回时只返回这一个列表

好在 LangGraph 对消息有按 ID 去重的逻辑，新版消息对象自带 ID，所以重复问题不严重。但如果你手动构造了不带 ID 的ToolMessage，在加 checkpointer 的场景下就可能出问题。经验就是：所有消息尽量带上 ID，用Message(id=str(uuid.uuid4()))或者让 LangChain 自动生成。

5.2 模型不按预期重试：tool_choice 与 Prompt 控制

这是我在测试中最头疼的问题：错误信息已经返回给模型了，但模型就是不再调用工具，而是回复"抱歉，我暂时无法获取天气信息"。

我排查下来，原因是模型在"道歉"和"继续调用工具"之间做了判断，而大部分模型的默认倾向是停止调用。解决办法有三个层次：

第一层，设置tool_choice="any"，强制模型必须输出工具调用：

llm_with_tools = llm.bind_tools([get_weather], tool_choice="any")

注意这会带来副作用：即使没有工具需求，模型也会强行调用一次。所以只建议在重试路径上使用，或者用"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_retries=2）。如果你的 LLM 调用因为限流失败，它自己会先重试两次，然后才把异常抛出来。这时候我前面代码里设置max_retries=0就是为了关掉这一层重试，让错误尽快暴露给图的error_exit节点统一处理。

为什么一定要关？因为如果模型调用本身不稳定，LangChain 内部的重试和我们的图重试叠加，实际的调用次数会是乘法而不是加法，API 账单会非常吓人。经验法则：一个 Agent 系统里，重试逻辑只应该有一层明确的所有者。既然我们已经用 LangGraph 做了图级别的重试，那模型内部的自动重试就应该关掉，或者调小。

我在实际跑这套方案的时候，最开始也是在测试环境连跑了几十次，故意把失败率调高到 90%，才把这些问题一个个逼出来。重试机制这种东西，代码写出来不难，难的是想清楚"什么情况下绝对不能重试""重试多少次之后必须停下来问人"。状态图把决策路径画清楚了，这些边界才看得见、控得住。如果你在做 Agent 的工具调用链路，建议把这篇里的模拟工具换成你自己的真实接口，然后把max_attempts从 1 调到 5 各跑几轮，很快就能摸清你自己下游服务的脾气。