🎯 课程主题

详解 LangChain Agent 中 handle_errors 参数的五种错误处理策略，涵盖从默认自动重试到自定义函数的完整方案。

📝 核心知识点

1. handle_errors 参数概述

  • 概念说明：create_agent 中的 handle_errors 参数控制当大模型输出不符合结构化格式要求时的处理策略。受限于模型能力，大模型输出很可能不符合格式要求，需要该参数来兜底。
  • 关键细节：该参数共有五种取值方式——True（默认）、False、自定义字符串、指定异常类型、自定义处理函数。

2. 五种错误处理策略

策略行为适用场景
True（默认）捕获所有异常，使用内置错误消息模板让模型重试，直到输出有效数据或达到重试上限大多数希望自动处理的场景
False关闭自动重试，出现任何异常直接抛出，程序中断需要严格把控输出质量的场景
自定义字符串捕获所有异常后返回指定的友好提示字符串，替代默认错误信息需要给用户友好提示的业务场景
异常类型（元组）仅捕获指定的异常类型，其他异常直接抛出需要对特定异常做精准控制的场景
自定义函数传入一个函数，根据异常类型分别编写处理逻辑，最灵活需要复杂精细化错误处理的场景

3. 常见异常类型

  • MultipleStructuredOutputsError：返回的工具调用请求数量大于1时抛出（Union 类型只能二选一）
  • StructuredOutputValidationError：输出格式不符合结构化要求时抛出（如字段名不匹配）

🏗️ 架构与工作流

大模型输出 → 结构化解析
    ├── 解析成功 → 返回结构化结果
    └── 解析失败 → handle_errors 拦截
          ├── True: 内置模板重试（循环，直到成功或达上限）
          ├── False: 直接抛出异常
          ├── 字符串: 返回友好提示
          ├── 异常类型: 仅匹配指定异常，否则抛出
          └── 函数: 自定义精细化处理

💻 代码实战

# ==================== 环境准备 ====================
from langchain_deepseek import ChatDeepSeek

model = ChatDeepSeek(model="deepseek-v4-flash")

# ==================== 定义结构化输出 ====================
from pydantic import BaseModel, Field
from typing import Union

class ContentInfo(BaseModel):
    name: str = Field(description="用户姓名")
    email: str = Field(description="用户邮箱")

class EventDetails(BaseModel):
    event_name: str = Field(description="活动名称")
    event_date: str = Field(description="活动日期")

# ==================== 示例 Agent ====================
from langgraph.prebuilt import create_react_agent

# 故意构造一个可能触发多个结构化输出的场景（Union 只能二选一）
user_input = "张三的邮箱是 zhangsan@example.com，参加的是AI峰会，活动日期是2025-06-15"

# === 策略一：True（默认，捕获所有异常并重试）===
agent = create_react_agent(
    model=model,
    tools=[],
    response_format=Union[ContentInfo, EventDetails],
    handle_errors=True  # 默认值，可省略
)
result = agent.invoke({"messages": [{"role": "user", "content": user_input}]})
print(result)

# === 策略二：False（不捕获，直接抛出）===
agent = create_react_agent(
    model=model,
    tools=[],
    response_format=Union[ContentInfo, EventDetails],
    handle_errors=False
)
# 运行将抛出 MultipleStructuredOutputsError
# result = agent.invoke({"messages": [{"role": "user", "content": user_input}]})

# === 策略三：自定义字符串 ===
agent = create_react_agent(
    model=model,
    tools=[],
    response_format=Union[ContentInfo, EventDetails],
    handle_errors="请检查输入的数据"
)
result = agent.invoke({"messages": [{"role": "user", "content": user_input}]})
# 异常时 AIMessage.content 将被替换为 "请检查输入的数据"

# === 策略四：指定异常类型 ===
from langgraph.errors import MultipleStructuredOutputsError, StructuredOutputValidationError

agent = create_react_agent(
    model=model,
    tools=[],
    response_format=Union[ContentInfo, EventDetails],
    handle_errors=(MultipleStructuredOutputsError, StructuredOutputValidationError)
)
# 仅捕获元组中的异常类型，其他异常直接抛出
result = agent.invoke({"messages": [{"role": "user", "content": user_input}]})

# === 策略五：自定义错误处理函数 ===
def custom_error_handler(exception: Exception) -> str:
    """自定义错误处理器"""
    print(f"捕获到的错误类型: {type(exception).__name__}")
    print(f"错误详情: {str(exception)}")

    if isinstance(exception, MultipleStructuredOutputsError):
        return "检测到多个响应，请选择最相关的一个进行返回"
    elif isinstance(exception, StructuredOutputValidationError):
        return "数据格式有误，请检查字段是否符合要求"
    else:
        return f"发生未知错误: {str(exception)}"

agent = create_react_agent(
    model=model,
    tools=[],
    response_format=Union[ContentInfo, EventDetails],
    handle_errors=custom_error_handler
)
result = agent.invoke({"messages": [{"role": "user", "content": user_input}]})

⚠️ 常见问题与避坑指南

  • 当 handle_errors=True 且错误反复重试后仍无法得到有效结果时，Agent 会在达到重试上限后结束，注意此时输出的内容可能并非预期的结构化数据。
  • 设置为固定字符串时，重试循环中每次都会用该字符串替换异常输出，可能导致输出日志非常冗长。
  • 使用异常类型元组时，如果实际出现的异常不在元组内，程序仍会中断（等同于 False 行为）。
  • 日常开发推荐使用默认的 handle_errors=True，简单有效；有精细化需求时使用自定义函数。

💡 个人总结与延伸

handle_errors 本质上是 LangChain 在结构化输出场景下的异常兜底机制，类似于 try/except 的声明式封装。默认行为适合大多数场景，自定义函数赋予开发者最大的灵活性。后续可以结合中间件的容错机制（如自动重试中间件）构建更健壮的 Agent 系统。

最近更新:
Contributors: yangwenguang