错误处理与重试

现实里网络会断、API 会限流、工具会传错参数。一个鲁棒的工作流必须能优雅地处理失败，并在合理范围内重试。本篇讲清楚 LangGraph 里的几套错误处理套路。

一、错误来源

LangGraph 工作流里的失败大致分三类：

来源	典型表现	推荐处理
节点业务逻辑	抛异常、返回非法值	节点内 try/except
工具执行失败	ToolNode 抛异常	返回错误 ToolMessage，让 LLM 自我纠正
外部依赖（LLM/网络）	超时、限流 429	用 tenacity 重试

二、节点内 try/except

最朴素的办法：在节点函数里捕获异常，把错误信息写进状态，而不是让异常向上传播崩溃整个图。

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    query: str
    result: str
    error: str

def risky_call(state: State) -> dict:
    try:
        # 模拟一个可能失败的调用
        if "bad" in state["query"]:
            raise ValueError("触发了坏输入")
        return {"result": f"处理完成：{state['query']}", "error": ""}
    except Exception as e:
        # 捕获并把错误写进状态，不向上抛
        return {"result": "", "error": f"调用失败：{e}"}

graph = StateGraph(State)
graph.add_node("call", risky_call)
graph.add_edge(START, "call")
graph.add_edge("call", END)
app = graph.compile()

print(app.invoke({"query": "正常请求"}))   # result 有值
print(app.invoke({"query": "bad query"}))   # error 有值，没崩

三、工具失败：让 LLM 自我纠正

工具执行失败时，更好的做法不是直接报错，而是把错误信息作为 ToolMessage 返回给 LLM，让它在下一轮自己换参数重试。ToolNode 支持这个：

from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage
from langgraph.prebuilt import ToolNode

@tool
def divide(a: int, b: int) -> float:
    """两数相除。"""
    if b == 0:
        raise ZeroDivisionError("除数不能为 0")
    return a / b

# 关键：handle_tool_errors=True，工具抛错会自动转成 ToolMessage 而不是崩溃
tool_node = ToolNode([divide], handle_tool_errors=True)

当 LLM 调 divide(1, 0)，ToolNode 不会崩溃，而是返回一条内容为"除数不能为 0"的 ToolMessage。LLM 看到后会自己改成非零除数再试。这就是 ReAct 的"自我纠正"循环。

四、用 tenacity 做重试

外部调用（LLM API、网络请求）常因限流/超时失败，这类失败重试几次通常就好。Python 生态里 tenacity 是事实标准。

先装：

pip install tenacity -i https://pypi.tuna.tsinghua.edu.cn/simple

Windows 用户：PowerShell 里直接用上面命令。macOS/Linux 同命令。

from tenacity import retry, stop_after_attempt, wait_exponential
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=8))
def call_llm(prompt: str) -> str:
    # 如果限流 429，会自动重试，最多 3 次，指数退避
    return llm.invoke(prompt).content

def llm_node(state):
    try:
        ans = call_llm(state["query"])
        return {"result": ans, "error": ""}
    except Exception as e:
        # 重试用尽还是失败，降级处理
        return {"result": "", "error": f"模型不可用：{e}"}

wait_exponential 做指数退避（1s、2s、4s），避免短时间内反复打爆 API。

五、用条件边做"失败重试"循环

有时失败不该在节点内静默重试，而是要让图层面看到"失败了→走回头路重试"。用条件边+计数器实现。

flowchart LR
    START([START]) --> R[调用节点]
    R -->|成功| END([END])
    R -->|失败, count<3| R
    R -->|失败, count>=3| F[兜底节点]
    F --> END

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
import random

class State(TypedDict):
    query: str
    result: str
    fail_count: int
    done: bool

def call_node(state: State) -> dict:
    try:
        # 模拟 50% 概率失败
        if random.random() < 0.5:
            raise RuntimeError("随机失败")
        return {"result": f"成功处理 {state['query']}", "done": True}
    except Exception as e:
        return {"fail_count": state.get("fail_count", 0) + 1,
                "result": "", "done": False}

def fallback(state: State) -> dict:
    return {"result": "重试用尽，返回兜底答案"}

def route(state: State) -> str:
    if state.get("done"):
        return "end"
    if state.get("fail_count", 0) >= 3:
        return "fallback"
    return "retry"   # 回到 call_node

graph = StateGraph(State)
graph.add_node("call", call_node)
graph.add_node("fallback", fallback)
graph.add_edge(START, "call")
# 条件边决定：成功→END，失败次数够→fallback，否则→重试
graph.add_conditional_edges("call", route, ["end", "fallback", "retry"])
# retry 这个名字会回到 call 节点
# 注意：add_conditional_edges 里 "retry" 需要映射到真实节点

小技巧：条件边返回的字符串如果和某个节点名相同，就会去那个节点。所以直接返回 "call" 即可让它重试。改成：

def route(state: State) -> str:
    if state.get("done"):
        return "end"
    if state.get("fail_count", 0) >= 3:
        return "fallback"
    return "call"   # 返回节点名，回到 call 节点重试

graph.add_conditional_edges("call", route, ["fallback", "call"])

"end" 是特殊值，会映射到 END。

六、完整示例：会重试的工具调用工作流

把上面几样组合：一个工具可能失败，失败后转成 ToolMessage 让假 LLM 重试，最多 3 轮。

from typing import TypedDict, Annotated
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode, tools_condition

call_log = []  # 记录调用次数，演示重试

@tool
def flaky_lookup(key: str) -> str:
    """查询某个键的值（前两次会失败，第三次成功）。"""
    call_log.append(key)
    if len(call_log) < 3:
        raise RuntimeError("服务暂时不可用")
    return f"「{key}」的值是 42"

tools = [flaky_lookup]
tool_node = ToolNode(tools, handle_tool_errors=True)

# 假 LLM：第一轮调工具，拿到错误后继续调，拿到结果后回答
def fake_llm(state):
    msgs = state["messages"]
    tool_msgs = [m for m in msgs if isinstance(m, ToolMessage)]
    if not tool_msgs:
        # 第一轮：发起工具调用
        return {"messages": [AIMessage(content="", tool_calls=[
            {"name": "flaky_lookup", "args": {"key": "answer"}, "id": "c1"}])]}
    last_tool = tool_msgs[-1]
    if "失败" in last_tool.content or "不可用" in last_tool.content:
        # 看到错误，换个 key 重试（模拟 LLM 自我纠正）
        return {"messages": [AIMessage(content="", tool_calls=[
            {"name": "flaky_lookup", "args": {"key": f"retry-{len(call_log)}"}, "id": f"c{len(call_log)}"})]}
    # 成功，给出最终回答
    return {"messages": [AIMessage(content=f"最终答案：{last_tool.content}")]}

graph = StateGraph(MessagesState)
graph.add_node("llm", fake_llm)
graph.add_node("tools", tool_node)
graph.add_edge(START, "llm")
graph.add_conditional_edges("llm", tools_condition)
graph.add_edge("tools", "llm")
app = graph.compile()

result = app.invoke({"messages": [HumanMessage(content="查一下 answer")]})
print("最终回复:", result["messages"][-1].content)
print("工具调用次数:", len(call_log))

运行后会看到工具被调用了 3 次（前两次失败、第三次成功），最终给出答案。整个过程没有崩溃。

七、异常向上传播 vs 捕获恢复

两种策略要分清：

• 捕获恢复：节点内 try/except，把错误转成状态字段或 ToolMessage，图继续跑。适合"可恢复、可重试"的失败。
• 向上传播：不捕获，异常直接抛出，图执行中断，状态停在失败前那步的检查点（如果有 checkpointer）。适合"必须人工介入"的致命错误，配合 中断与恢复 做断点续跑。

选择原则：能自动恢复就恢复，不能就让它崩，但要配 checkpointer 留好现场。

八、常见踩坑

踩坑 1：无限重试死循环

重试没有次数上限，遇到一直失败的依赖就死循环。任何重试逻辑都要配 stop_after_attempt(N) 或计数器兜底。

踩坑 2：吞掉异常难排查

except Exception:
    pass   # ❌ 静默吞掉，出问题完全不知道

至少要把异常信息写进状态字段或日志，方便排查。生产环境接上日志框架。

踩坑 3：重试时不退避，被限流封 IP

对 LLM API 重试一定要加 wait_exponential 指数退避，否则短时间内高频重试会被判定为攻击性请求。

踩坑 4：条件边重试忘了清状态

重试路径如果不清掉上次的失败标记（如 done=False、error 残留），路由判断可能错乱。重试前显式重置相关字段。

踩坑 5：handle_tool_errors 没开，工具异常直接崩图

默认 ToolNode 不开错误处理时，工具抛异常会让整个图执行中断。需要自我纠正循环就显式传 handle_tool_errors=True。

九、小结

• 节点内 try/except 处理业务异常，把错误写进状态而非崩溃。
• 工具失败用 ToolNode(handle_tool_errors=True) 转成 ToolMessage，让 LLM 自我纠正。
• 外部依赖用 tenacity 做带指数退避的重试。
• 条件边+计数器实现图层面的"失败重试"循环，必须有次数兜底。
• 致命错误让它传播，配 checkpointer 留现场，便于人工恢复。