Agent 崩溃不是"某个报错怎么修"，而是要把瞬态故障自动恢复、永久错误优雅降级、死循环兜底、崩溃可断点续跑这四件事系统化做掉。

一、崩溃的根因分类（先定位再动手）

LangChain/LangGraph Agent 在生产中崩，基本逃不出这五类：

类别	典型表现	处理策略
瞬态故障	网络超时、HTTP 5xx、429 限流、LLM 连接重置	重试 + 指数退避
永久错误	参数校验失败、404、鉴权失败、除零	不重试，返回错误给 LLM 让其自我纠正
死循环 / 递归爆栈	RecursionError、GraphRecursionError	设递归上限 + 路由条件修正
状态/序列化问题	_pickle.PicklingError、状态字段丢失	状态只存可序列化数据 + reducer
进程级崩溃	OOM、机器重启、任务跑了几小时中途挂掉	Checkpointer 断点恢复

💡 关键认知：LangGraph 把 Agent 建模成"节点 + 边"的图，重试、超时、错误处理器都直接挂在节点上，这是它做容错的核心优势。

---

二、详细步骤与代码

步骤 1：工具层——把异常"吞掉"变成错误信息返回给 LLM

工具里直接抛异常会导致整个 Agent 运行中断。正确做法是捕获异常、返回错误字符串，让 LLM 看到错误后自己换策略。

from langchain_core.tools import tool
import httpx

@tool
def fetch_data(url: str) -> str:
    """抓取 URL 数据，带完整错误处理。"""
    try:
        response = httpx.get(url, timeout=10)
        response.raise_for_status()
        return response.text[:2000]
    except httpx.TimeoutException:
        # 返回错误字符串而非抛异常，LLM 可见并可换 URL 重试
        return "ERROR: Request timed out. The server may be slow or unreachable."
    except httpx.HTTPStatusError as e:
        return f"ERROR: HTTP {e.response.status_code}. The resource may not exist."
    except Exception as e:
        return f"ERROR: {type(e).__name__}: {e}"

⚠️ 只在工具内捕获你能处理的异常（网络、超时、校验）。编程错误（bug、OOM）应让它向上冒泡，否则会被静默吞掉。

步骤 2：节点级重试 + 超时 + 错误处理器（LangGraph ≥ 1.2）

这是 LangGraph 官方推荐的三段式容错，三者组合顺序固定：节点抛异常 → 重试策略决定是否重试 → 重试耗尽 → 错误处理器接管。

from langgraph.graph import StateGraph, START, END
from langgraph.types import RetryPolicy, default_retry_on
from langgraph.errors import NodeError
from langgraph.runtime import Runtime

def custom_retry_on(exc: BaseException) -> bool:
    """自定义重试判定：业务错误不重试，其余走默认。"""
    if isinstance(exc, ValueError):   # 参数错误属于永久错误，不重试
        return False
    return default_retry_on(exc)

def call_llm_with_fallback(state, runtime: Runtime):
    """重试到第 2 次时切换到备用模型。"""
    if runtime.execution_info.node_attempt > 1:
        return {"result": call_fallback_model(state)}
    return {"result": call_primary_model(state)}

def handle_model_failure(state, error: NodeError):
    """重试耗尽后的兜底：记录错误 + 返回用户友好文案。"""
    import logging
    logging.error(f"LLM 节点最终失败: {error}")
    return {
        "messages": [AIMessage(
            content="抱歉，我暂时无法处理您的请求，请稍后重试。"
        )]
    }

# 装配到节点
builder = StateGraph(State)
builder.add_node(
    "call_llm",
    call_llm_with_fallback,
    retry_policy=RetryPolicy(
        max_attempts=4,           # 首次 + 3 次重试
        initial_interval=0.5,     # 首次重试前等待 0.5s
        backoff_factor=2.0,       # 指数退避：0.5s → 1s → 2s
        max_interval=128.0,
        jitter=True,              # 随机抖动，避免惊群
        retry_on=custom_retry_on, # 只对瞬态故障重试
    ),
    # 节点级超时（LangGraph >= 1.2）
    timeout=TimeoutPolicy(run_timeout=30, idle_timeout=5),
    error_handler=handle_model_failure,  # 重试耗尽后走这里
)

默认重试行为：default_retry_on 会对 ConnectionError、httpx/requests 的 5xx 重试，不会对 ValueError、TypeError、RuntimeError 重试——因为这些几乎都是代码 bug。

步骤 3：防死循环——递归上限 + 路由条件修正

LangGraph 默认递归上限是 25 步（约 12 次工具调用），超出抛 GraphRecursionError。光调大数字只是拖延崩溃，必须修路由。

from langgraph.errors import GraphRecursionError
from langgraph.prebuilt import tools_condition

# ✅ 正确：用预置的 tools_condition 做路由，模型不再要工具时自动 END
builder.add_conditional_edges("agent", tools_condition, {
    "tools": "tools",
    END: END,
})

# ✅ 工具返回要带明确的结束信号
@tool
def search(query: str):
    if not query.strip():
        return "ERROR — 查询为空，请停止并提示用户提供关键词"
    # ... 正常逻辑
    return "SUCCESS — 找到 3 条结果: ..."

# ✅ 调用时用 try/except 包裹，优雅降级
try:
    result = graph.invoke(
        initial_state,
        {"configurable": {"thread_id": "user-42"}, "recursion_limit": 50}
    )
except GraphRecursionError:
    # 兜底：不让用户看到原始堆栈
    result = {"messages": [AIMessage(
        content="I couldn't complete that within the step budget."
    )]}

📌 死循环的真正根因通常是两个：①路由函数永远不返回 END；②工具结果没给模型"任务完成"的信号。先用 stream_mode="updates" 看节点执行轨迹定位，比盲目调大 recursion_limit 有用得多。

步骤 4：Checkpointer 断点恢复——扛进程级崩溃

没有持久化，服务重启后半截任务全丢。Checkpointer 让崩溃后的任务能从最近的快照继续：

from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from typing import Annotated, TypedDict

class State(TypedDict):
    # 消息字段必须用 add_messages reducer，否则历史会丢失
    messages: Annotated[list, add_messages]
    step_count: int

# 编译时挂上 checkpointer
checkpointer = AsyncSqliteSaver.from_conn_string("checkpoints.db")
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "user-request-001"}}

# 第一次调用，跑到第 3 步崩了
try:
    result = graph.invoke(
        {"messages": [HumanMessage(content="帮我搜 LangGraph 最新 release notes")]},
        config
    )
except Exception as e:
    print(f"🔥 崩溃: {e}")
    # 不需要 panic，用同一个 thread_id 恢复

# 重启后：从最近的 Checkpoint 继续
recovered = graph.get_state(config)
print(f"恢复在节点: {recovered.next}")
print(f"已完成步数: {recovered.values.get('step_count', 0)}")

# 传入 None = 不追加新输入，从断点继续
result = graph.invoke(None, config)

关键细节：

• thread_id 是唯一标识，必须有，否则 ValueError: Checkpointer requires a configurable with a thread_id key
• 状态里只放可序列化数据，数据库连接、模型实例等放全局或连接池，否则 _pickle.PicklingError
• 列表类字段（如 messages）必须用 reducer（add_messages），否则每轮被覆盖

步骤 5：模型调用与工具调用的独立重试/降级

如果是用 Deep Agents 或 JS 版 LangChain，可以用中间件把模型重试、工具重试、模型降级串起来：

from langchain.agents import create_agent
from langchain.agents.middleware import (
    ModelRetryMiddleware,
    ToolRetryMiddleware,
    ModelFallbackMiddleware,
)

agent = create_agent(
    model="gpt-5.5",
    tools=[search_tool, database_tool, api_tool],
    middleware=[
        # 模型调用重试：429/5xx 自动重试 3 次
        ModelRetryMiddleware(max_retries=3, backoff_factor=2.0, initial_delay=1.0),
        # 工具调用重试：只对外部 API 类工具重试，文件读取失败不重试
        ToolRetryMiddleware(
            max_retries=2,
            tools=["api_tool"],
            retry_on=(ConnectionError, TimeoutError),
            on_failure="return_message",  # 重试耗尽→返回错误给 LLM
        ),
        # 主模型完全宕机→切换备用模型
        ModelFallbackMiddleware("claude-sonnet-4-6"),
    ],
)

💡 决策原则：429 限流 → 重试为主；5xx → 重试 + 降级；单一厂商完全宕机 → 跨厂商 fallback；400/401/输入错误 → 都不行，返回错误提示。

步骤 6：可观测性——别让生产环境"盲飞"

import os
# 开发阶段全量开启 LangSmith 跟踪
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "..."

# 本地调试用 stream 看节点轨迹
for chunk in graph.stream(initial_state, config, stream_mode="updates"):
    for node_name, state_update in chunk.items():
        print(f"Node: {node_name} | Update: {state_update}")

生产环境开启采样跟踪即可，避免日志成本爆炸。

---

三、完整容错架构总结

把上面六步串起来，生产级 LangGraph Agent 的容错体系是这样的：

┌─────────────────────────────────────────────┐
│  第一层：工具层                               │
│   try/except → 返回错误字符串给 LLM          │
│   瞬态≠永久错误，后者不重试                    │
├─────────────────────────────────────────────┤
│  第二层：节点层（LangGraph 核心）              │
│   RetryPolicy：指数退避 + 抖动                │
│   TimeoutPolicy：单次尝试墙钟超时               │
│   error_handler：重试耗尽后的兜底                │
│   runtime.execution_info：第 2 次起走降级       │
├─────────────────────────────────────────────┤
│  第三层：图层面                               │
│   recursion_limit：防死循环（25-50）           │
│   tools_condition：正确路由到 END             │
│   Checkpointer：崩溃后断点续跑                 │
├─────────────────────────────────────────────┤
│  第四层：模型/工具中间件层                      │
│   ModelRetryMiddleware / ToolRetryMiddleware  │
│   ModelFallbackMiddleware：跨模型降级           │
├─────────────────────────────────────────────┤
│  第五层：可观测性                             │
│   LangSmith / 日志 / 指标上报                  │
│   顶层 try/except → 用户永远看不到原始 500     │
└─────────────────────────────────────────────┘

核心原则

1. 异常分类是前提：瞬态故障（网络/5xx/429）才重试，永久错误（400/401/参数错）立即返回给 LLM 让其自我纠正，不要无脑重试。
2. 重试必须带退避和抖动：backoff_factor=2.0, jitter=True，避免雪崩。
3. 超时是必需的：节点级 TimeoutPolicy 防单点卡死，配合 recursion_limit 防死循环——两者守卫不同故障模式（步数守工作量，时间守延迟）。
4. Checkpointer 是生产底线：没有持久化，跑了几小时的任务中途崩 = 全部重来。状态只存可序列化数据，列表字段务必加 reducer。
5. 用户永远不该看到原始堆栈：顶层 try/except + 友好文案 + 错误上报监控，是上线的最后一公里。

⚠️ 一个常被忽略的点：LangGraph 没有内置的全局超时，需要在应用层用 asyncio.wait_for() 包住 graph.ainvoke()。但即便全局超时触发，checkpoint 状态依然可用，后续可从断点恢复——这就是 Checkpointer 和超时的协同价值。

按这套思路落地后，Agent 的崩溃从"整个任务报废"变成了"瞬态故障自动恢复、永久错误优雅降级、进程崩溃断点续跑"，生产可用性会有质的提升。