异常处理与重试

Agent 在生产环境中会面临各种异常：LLM API 错误、工具调用失败、网络超时等。LangGraph 提供了多层级的异常处理机制。

异常类型

类型	示例	处理方式
LLM 调用失败	API 超时、速率限制	重试、降级
工具执行错误	文件不存在、API 返回错误	捕获并返回错误消息
State 异常	类型错误、缺少必要字段	校验和默认值
图结构错误	节点不存在、循环检测	编译期检查
资源耗尽	Token 超限、内存不足	限制和熔断

节点级错误处理

使用 try-except 捕获

def safe_node(state: State):
    try:
        result = risky_operation(state["input"])
        return {"output": result, "status": "success"}
    except APIError as e:
        return {
            "output": f"API 错误: {e}",
            "status": "error",
            "retry_count": state.get("retry_count", 0) + 1
        }
    except Exception as e:
        return {
            "output": f"未知错误: {e}",
            "status": "error"
        }

路由到错误处理节点

from typing import Literal

def try_operation(state: State) -> dict | Command:
    try:
        result = risky_operation(state["input"])
        return {"output": result}
    except Exception as e:
        # 返回特殊标记，让条件边路由到错误处理
        return {"error": str(e), "error_occurred": True}

def error_router(state: State) -> Literal["error_handler", "next"]:
    if state.get("error_occurred"):
        return "error_handler"
    return "next"

def error_handler(state: State):
    """处理错误，记录日志，尝试恢复"""
    log_error(state["error"])
    # 尝试用默认值恢复
    return {"output": "fallback_result", "error": None}

# 图结构
builder.add_edge(START, "try_operation")
builder.add_conditional_edges("try_operation", error_router, ...)

重试机制

手动重试

def resilient_node(state: State):
    max_retries = 3
    retry_count = state.get("retry_count", 0)

    for attempt in range(max_retries - retry_count):
        try:
            result = api_call(state["input"])
            return {"output": result, "retry_count": 0}  # 重置
        except Exception as e:
            if attempt < max_retries - retry_count - 1:
                time.sleep(2 ** attempt)  # 指数退避
                continue
            return {
                "error": str(e),
                "retry_count": retry_count + attempt + 1
            }

    return {"error": "重试耗尽", "retry_count": retry_count + max_retries}

条件边重试

def should_retry(state: State) -> Literal["retry_node", "fallback", END]:
    if state.get("error") and state.get("retry_count", 0) < 3:
        return "retry_node"
    elif state.get("error"):
        return "fallback"
    return END

builder.add_conditional_edges(
    "api_node",
    should_retry,
    ["api_node", "fallback_node", END]
)

使用图循环做重试

from langgraph.graph import StateGraph, START, END, Command

def api_call_node(state: State):
    try:
        result = api_service.call(state["input"])
        return Command(goto=END, update={"result": result, "status": "ok"})
    except Exception as e:
        retries = state.get("retries", 0) + 1
        if retries < 4:
            return Command(
                goto="api_call_node",  # 回到自己，形成循环
                update={"retries": retries, "last_error": str(e)}
            )
        return Command(
            goto="fallback",
            update={"status": "failed", "last_error": str(e)}
        )

容错超时

节点级别超时

import asyncio

async def timeout_node(state: State):
    try:
        result = await asyncio.wait_for(
            async_api_call(state["input"]),
            timeout=30.0
        )
        return {"output": result}
    except asyncio.TimeoutError:
        return {"output": "DEFAULT", "timeout": True}

优雅降级

当主要路径失败时，使用备选方案：

def search_with_fallback(state: State):
    # 尝试多种搜索后端
    backends = [
        search_tool_a,
        search_tool_b,
        search_tool_c,
    ]

    for backend in backends:
        try:
            result = backend.invoke(state["query"])
            if result:
                return {"search_results": result, "backend": backend.__name__}
        except Exception as e:
            print(f"Backend {backend.__name__} failed: {e}")
            continue

    # 全部失败，返回空结果
    return {"search_results": [], "backend": "none"}

全局容错策略

在生产部署中使用 Checkpointer

from langgraph.checkpoint.postgres import PostgresSaver

# 使用持久化 checkpoint，节点崩溃后可以从最后一步恢复
checkpointer = PostgresSaver.from_conn_string(DB_URL)
graph = builder.compile(checkpointer=checkpointer)

# 如果执行中途崩溃，可以从最近的 checkpoint 恢复
config = {"configurable": {"thread_id": "task_1"}}
try:
    result = graph.invoke(inputs, config)
except Exception:
    # 检查最后的状态
    state = graph.get_state(config)
    print(f"执行在第 {state.metadata['step']} 步失败")
    # 可以选择恢复
    result = graph.invoke(None, config)

日志和监控

def logged_node(state: State, config: RunnableConfig):
    import logging
    logger = logging.getLogger(__name__)

    thread_id = config["configurable"]["thread_id"]
    logger.info(f"Thread {thread_id}: 开始执行 node: {__name__}")

    try:
        result = process(state)
        logger.info(f"Thread {thread_id}: 执行成功")
        return result
    except Exception as e:
        logger.error(f"Thread {thread_id}: 执行失败: {e}", exc_info=True)
        # 发送告警
        alert_system.send(f"Node failed: {e}")
        raise

最佳实践

1. 所有节点都要有错误处理，不要相信任何外部调用
2. 使用指数退避重试，避免 API 限流
3. 设置合理的重试次数上限，防止无限循环
4. 关键路径使用 @task 异步+超时，避免阻塞
5. 生产环境一定要开 tracing，错误排查全靠它
6. 使用 Checkpointer 实现容错恢复，从失败的 step 继续

参考

• 官方文档：Fault tolerance
• 官方文档：Persistence