LangGraph 错误处理与容错机制

生产环境中，Agent 的失败不是"如果"的问题，而是"什么时候"的问题。网络超时、LLM API 限流、工具调用返回异常数据、下游服务不可用——这些都是日常。没有容错机制的 Agent，在压测时会崩，在凌晨偶发流量高峰时会崩，在 OpenAI API 降级时会崩。

LangGraph 提供了从节点级到图级的完整容错工具链。

1.1 LangGraph 中错误的类型

LangGraph 四种错误处理策略——重试、回退、降级、人工介入，覆盖不同类型的故障场景

按照来源分类，Agent 运行时的错误主要有四类：

错误类型	具体表现	发生位置
节点内部异常	Python 代码抛出异常（TypeError、ValueError 等）	任意节点函数
工具调用失败	搜索 API 返回 404、数据库连接超时、JSON 解析错误	工具节点
LLM 调用异常	API 限流（429，HTTP 状态码，表示请求过于频繁被服务端拒绝）、超时、内容过滤拒绝	调用 LLM 的节点
业务逻辑错误	输出格式不符合预期、关键字段缺失	数据处理节点

这四类错误的处理策略不同：节点内部异常通常需要立即修复或记录；工具失败适合重试；LLM 失败适合切换备用模型；业务逻辑错误适合走备用路径。

1.2 错误状态字段设计

在 State 里专门设计错误相关字段，是容错架构的基础。比把错误信息打到日志里强得多——State 里的错误信息可以被后续节点读取，从而触发不同的处理路径。

python

# error_state_design.py
# 演示在 State 中设计错误相关字段

from typing import TypedDict, List, Optional, Annotated
import operator


class RobustAgentState(TypedDict):
    # 正常业务字段
    user_query: str
    search_results: List[str]
    answer: str

    # 错误处理字段
    # Optional 表示正常情况下为 None，出错时才有值
    error_message: Optional[str]

    # 使用 Annotated + operator.add 追加错误历史，不覆盖
    # 这样可以看到整个错误链路，便于排查
    error_history: Annotated[List[str], operator.add]

    # 重试计数器：每次重试递增，超过阈值时放弃
    retry_count: int

    # 当前节点标识：方便错误日志定位是哪个节点出了问题
    current_node: str

设计原则：

error_message 用 Optional[str]，正常时为 None，出错时填充，条件边根据是否为 None 决定路由
error_history 用追加列表，不要覆盖，方便事后排查错误链路
retry_count 作为重试限制的依据

1.3 try-except 在节点函数中的处理

节点函数里的 try-except 不应该只是静默吞掉异常，而应该把错误信息写回 State，让后续节点和条件边知道"这里出了问题"。

python

# node_error_handling.py
# 演示节点函数内的错误处理模式

import time
import random
from typing import TypedDict, List, Optional, Annotated
import operator
from langgraph.graph import StateGraph, START, END


class AgentState(TypedDict):
    query: str
    search_results: List[str]
    answer: str
    error_message: Optional[str]
    error_history: Annotated[List[str], operator.add]
    retry_count: int


# ============================================================
# 模拟一个不稳定的工具（30% 概率失败）
# ============================================================

def unstable_search_api(query: str) -> List[str]:
    """模拟不稳定的外部搜索 API。"""
    # 模拟 30% 的失败率
    if random.random() < 0.3:
        raise ConnectionError(f"搜索 API 连接超时：无法完成对「{query}」的搜索")
    return [
        f"搜索结果1：关于「{query}」的基础介绍",
        f"搜索结果2：「{query}」的应用案例",
    ]


# ============================================================
# 错误处理模式一：捕获异常，写回 State，让图决定下一步
# ============================================================

def search_node_with_error_handling(state: AgentState) -> dict:
    """
    搜索节点：带完整错误处理。
    核心原则：不要在节点内部决定"出错后怎么办"，
    而是把错误状态写回 State，让条件边做路由决策。
    这样错误处理逻辑和业务逻辑分离，图结构更清晰。
    """
    try:
        results = unstable_search_api(state["query"])
        return {
            "search_results": results,
            "error_message": None,  # 成功时显式清空错误状态
            "error_history": [],    # 没有新错误
        }
    except ConnectionError as e:
        # 工具调用失败：记录错误信息，不重新抛出
        # 返回后由条件边决定是重试、走备用路径还是直接失败
        return {
            "search_results": [],
            "error_message": str(e),
            "error_history": [f"[search_node] {str(e)}"],
        }
    except Exception as e:
        # 未预期的异常：同样记录，但标记为 UNEXPECTED 方便区分
        return {
            "search_results": [],
            "error_message": f"UNEXPECTED: {str(e)}",
            "error_history": [f"[search_node] 未预期异常: {str(e)}"],
        }


# ============================================================
# 错误处理模式二：节点内部自动重试（适合简单场景）
# ============================================================

def search_node_with_retry(state: AgentState) -> dict:
    """
    节点内部重试版本。
    适合：重试逻辑简单、不需要在图层面感知重试过程的场景。
    不适合：需要在重试前做其他操作（比如切换模型、清理缓存）的场景。
    """
    MAX_RETRIES = 3
    RETRY_DELAY = 1  # 秒

    last_error = None
    for attempt in range(MAX_RETRIES):
        try:
            results = unstable_search_api(state["query"])
            return {
                "search_results": results,
                "error_message": None,
                "error_history": [f"[search_node] 第 {attempt + 1} 次尝试成功"] if attempt > 0 else [],
            }
        except Exception as e:
            last_error = e
            if attempt < MAX_RETRIES - 1:
                # 指数退避（Exponential Backoff）：每次重试等待时间翻倍，避免在服务恢复期间频繁冲击
                wait_time = RETRY_DELAY * (2 ** attempt)
                time.sleep(wait_time)

    # 所有重试都失败
    return {
        "search_results": [],
        "error_message": f"重试 {MAX_RETRIES} 次后仍失败：{last_error}",
        "error_history": [f"[search_node] 重试 {MAX_RETRIES} 次全部失败"],
    }


def answer_node(state: AgentState) -> dict:
    """生成答案节点：基于搜索结果生成回复。"""
    results = state.get("search_results", [])
    if not results:
        return {
            "answer": "抱歉，由于搜索服务不可用，无法为您提供准确答案。",
        }

    answer = f"基于搜索结果：\n" + "\n".join(f"- {r}" for r in results)
    return {"answer": answer}


def fallback_node(state: AgentState) -> dict:
    """
    备用节点：当主流程失败时执行。
    可以用缓存数据、静态回复、或降级的简单逻辑来响应。
    备用路径要保证不会再次失败——它是最后的防线。
    """
    query = state["query"]
    error = state.get("error_message", "未知错误")

    return {
        "answer": (
            f"搜索服务暂时不可用（原因：{error}）。\n"
            f"针对您的问题「{query}」，建议您稍后重试，"
            f"或访问帮助中心获取相关信息。"
        ),
        "error_history": ["[fallback_node] 使用备用响应"],
    }

1.4 条件边实现错误路由

错误写回 State 之后，用条件边来决定走哪条路径：

python

# error_routing.py
# 演示用条件边实现错误路由

from typing import Literal


def route_after_search(state: AgentState) -> Literal["answer", "fallback"]:
    """
    搜索节点完成后的路由函数。
    有错误 → 走备用路径
    正常 → 走答案生成路径
    """
    if state.get("error_message"):
        return "fallback"
    return "answer"


def route_with_retry(state: AgentState) -> Literal["search", "answer", "fallback"]:
    """
    支持重试的路由函数。
    有错误且重试次数未超限 → 重试搜索
    有错误且重试次数超限 → 走备用路径
    正常 → 走答案生成路径
    """
    if not state.get("error_message"):
        return "answer"

    retry_count = state.get("retry_count", 0)
    MAX_RETRIES = 3

    if retry_count < MAX_RETRIES:
        return "search"  # 路由回搜索节点，形成重试循环
    return "fallback"


# 构建带错误路由的图
def build_robust_graph():
    graph = StateGraph(AgentState)

    graph.add_node("search", search_node_with_error_handling)
    graph.add_node("answer", answer_node)
    graph.add_node("fallback", fallback_node)

    graph.add_edge(START, "search")

    # 条件边：根据 error_message 字段决定下一步
    graph.add_conditional_edges(
        "search",
        route_after_search,
        {
            "answer": "answer",
            "fallback": "fallback",
        }
    )

    graph.add_edge("answer", END)
    graph.add_edge("fallback", END)

    return graph.compile()

1.5 完整示例：带错误处理的 Agent

python

# robust_agent_complete.py
# 完整的带容错机制的 Agent：包含重试、备用路径、错误记录

import time
import random
from typing import TypedDict, List, Optional, Annotated, Literal
import operator
from langgraph.graph import StateGraph, START, END


class AgentState(TypedDict):
    query: str
    search_results: List[str]
    answer: str
    error_message: Optional[str]
    error_history: Annotated[List[str], operator.add]
    retry_count: int


def search_node(state: AgentState) -> dict:
    """搜索节点：30% 概率失败，模拟真实的不稳定外部服务。"""
    if random.random() < 0.3:
        error_msg = "搜索 API 超时（模拟）"
        return {
            "search_results": [],
            "error_message": error_msg,
            # retry_count 在路由函数里递增，这里只记录错误
            "error_history": [f"[search] 失败（第 {state.get('retry_count', 0) + 1} 次尝试）：{error_msg}"],
        }

    results = [
        f"结果1：{state['query']} 的相关文档",
        f"结果2：{state['query']} 的案例分析",
    ]
    return {
        "search_results": results,
        "error_message": None,
        "error_history": [f"[search] 成功，找到 {len(results)} 条结果"],
    }


def answer_node(state: AgentState) -> dict:
    """生成答案：只在搜索成功后才被调用。"""
    results = state["search_results"]
    answer = "基于搜索结果的回答：\n" + "\n".join(f"• {r}" for r in results)
    return {
        "answer": answer,
        "error_history": ["[answer] 答案生成完成"],
    }


def fallback_node(state: AgentState) -> dict:
    """备用节点：搜索彻底失败后的最终保障。"""
    return {
        "answer": f"搜索服务暂时不可用，请稍后重试。您的问题：「{state['query']}」",
        "error_history": ["[fallback] 使用降级响应"],
    }


def route_after_search(state: AgentState) -> Literal["answer", "search", "fallback"]:
    """
    路由逻辑：
    - 成功 → answer
    - 失败且重试次数 < 3 → 重试（回到 search）
    - 失败且重试次数 >= 3 → fallback
    """
    if not state.get("error_message"):
        return "answer"

    # 注意：retry_count 的递增在这里而不是在节点里
    # 这样路由函数同时负责计数，避免节点和路由逻辑耦合
    current_retry = state.get("retry_count", 0)
    MAX_RETRIES = 3

    if current_retry < MAX_RETRIES:
        return "search"
    return "fallback"


def increment_retry_node(state: AgentState) -> dict:
    """
    重试前的中间节点：递增重试计数。
    单独做一个节点，职责清晰，避免在 search_node 里混入重试逻辑。
    """
    return {
        "retry_count": state.get("retry_count", 0) + 1,
        "error_history": [f"[retry] 准备第 {state.get('retry_count', 0) + 1} 次重试"],
    }


def build_full_robust_graph():
    graph = StateGraph(AgentState)

    graph.add_node("search", search_node)
    graph.add_node("answer", answer_node)
    graph.add_node("fallback", fallback_node)
    graph.add_node("retry_prep", increment_retry_node)  # 重试计数节点

    graph.add_edge(START, "search")
    graph.add_conditional_edges(
        "search",
        route_after_search,
        {
            "answer": "answer",
            "search": "retry_prep",   # 失败时先走 retry_prep 递增计数
            "fallback": "fallback",
        }
    )
    graph.add_edge("retry_prep", "search")  # 计数后重新执行搜索
    graph.add_edge("answer", END)
    graph.add_edge("fallback", END)

    return graph.compile()


def run_robust_demo():
    app = build_full_robust_graph()

    print("=" * 60)
    print("运行带容错机制的 Agent（多次运行观察重试效果）")
    print("=" * 60)

    for run_id in range(3):
        print(f"\n--- 第 {run_id + 1} 次运行 ---")

        initial_state = {
            "query": "LangGraph 错误处理最佳实践",
            "search_results": [],
            "answer": "",
            "error_message": None,
            "error_history": [],
            "retry_count": 0,
        }

        final_state = app.invoke(
            initial_state,
            config={"recursion_limit": 20}  # 双重保险
        )

        print(f"最终答案：{final_state['answer'][:80]}...")
        print("执行日志：")
        for log in final_state["error_history"]:
            print(f"  {log}")


if __name__ == "__main__":
    run_robust_demo()

1.6 LangGraph 内置的 retry 配置

LangGraph 支持在节点层面配置重试策略，不需要在节点函数里手动写重试逻辑：

python

from langgraph.graph import StateGraph
from langgraph.pregel import RetryPolicy

# 为节点配置重试策略
graph.add_node(
    "search",
    search_node,
    retry=RetryPolicy(
        max_attempts=3,           # 最多重试 3 次
        initial_interval=1.0,     # 第一次重试等待 1 秒
        backoff_factor=2.0,       # 指数退避系数
        max_interval=10.0,        # 最长等待 10 秒
        # 只对特定异常类型重试，其他异常直接报错
        retry_on=(ConnectionError, TimeoutError),
    )
)

内置 RetryPolicy 适合处理可重试的基础设施错误（网络超时、临时限流），对于业务逻辑错误（输出格式不对）不应该重试，重试也没用。

1.7 回退策略：主模型失败时切换备用模型

LLM API 偶发失败时，切换到备用模型是比重试更实用的策略：

python

# model_fallback.py
# 主模型失败时自动切换备用模型

from typing import Optional

def llm_node_with_fallback(state: AgentState) -> dict:
    """
    带模型回退的 LLM 节点。
    主模型：GPT-4o（效果好，但偶发超时或限流）
    备用模型：GPT-4o-mini（更便宜、更稳定）
    最后备用：静态模板回答（保证服务不中断）
    """
    query = state["query"]
    context = "\n".join(state.get("search_results", []))

    # 尝试主模型
    primary_response = _try_llm_call(
        model="gpt-4o",
        query=query,
        context=context
    )
    if primary_response is not None:
        return {
            "answer": primary_response,
            "error_history": ["[llm] 主模型 gpt-4o 调用成功"],
        }

    # 主模型失败，尝试备用模型
    fallback_response = _try_llm_call(
        model="gpt-4o-mini",
        query=query,
        context=context
    )
    if fallback_response is not None:
        return {
            "answer": fallback_response,
            "error_history": ["[llm] 主模型失败，使用备用模型 gpt-4o-mini"],
        }

    # 所有模型都失败，返回静态兜底
    return {
        "answer": f"AI 服务暂时不可用，请稍后重试。您的问题：{query}",
        "error_message": "所有 LLM 调用均失败",
        "error_history": ["[llm] 主模型和备用模型均失败，使用静态回复"],
    }


def _try_llm_call(model: str, query: str, context: str) -> Optional[str]:
    """
    封装单次 LLM 调用，失败时返回 None 而非抛出异常。
    把异常转换为 None 值，让调用者通过返回值判断成功与否，
    比让调用者处理异常更简洁。
    """
    try:
        # 实际使用时替换为真实的 LLM 调用
        # from langchain_openai import ChatOpenAI
        # llm = ChatOpenAI(model=model, timeout=10)
        # response = llm.invoke(f"基于以下信息回答：{context}\n\n问题：{query}")
        # return response.content

        # 演示用模拟
        if model == "gpt-4o" and random.random() < 0.4:
            raise TimeoutError("模型调用超时（模拟）")
        return f"[{model}] 针对「{query}」的回答：基于上下文的综合分析..."

    except Exception as e:
        print(f"模型 {model} 调用失败：{e}")
        return None

1.8 错误处理流程图

1.9 小结

LangGraph 的容错体系分三层：

节点层：try-except 捕获异常，把错误状态写回 State，不要吞掉或再次抛出
图层：条件边读取 State 里的错误字段，决定走重试路径还是备用路径
基础设施层：RetryPolicy 处理可重试的基础设施错误（超时、限流）

设计要点：错误处理逻辑和业务逻辑分离——节点负责执行，条件边负责路由决策，备用节点负责降级响应。不要在节点函数内部做复杂的错误路由判断，那是条件边的职责。

下一篇是 LangGraph 实战：构建一个完整的研究报告生成 Agent，会把 Multi-Agent 协作和错误处理的知识综合应用到一个真实项目中。