langgraph.prebuilt.ValidationNode 是一个用于验证工具调用的节点类,广泛应用于 LangGraph 的状态图中。它并不实际执行工具,而是专注于验证工具调用的参数是否符合预定义的模式(schemas),并返回验证结果或错误信息。这种功能在需要生成符合复杂模式的结构化输出时尤为有用,尤其是在多轮对话场景中,可以保留原始消息和工具 ID 不丢失。

以下是对 ValidationNode 类的详细介绍,包括其定义、功能、使用场景、实现细节以及示例。


类的定义和参数

ValidationNode 继承自 RunnableCallable,是一个可运行的节点类。它的构造函数接受以下参数:

  • schemas
    类型:Sequence[Union[BaseTool, Type[BaseModel], Callable]]
    描述:一个列表,包含用于验证工具调用的模式。可以是以下类型之一:

    • Pydantic 的 BaseModel 类(如 Pydantic v1 或 v2 的模型)。
    • BaseTool 实例(将使用其 args_schema 属性)。
    • 函数(将根据函数签名自动生成模式)。
      作用:定义工具调用参数需要遵守的结构和规则。
  • format_error
    类型:Optional[Callable[[BaseException, ToolCall, Type[BaseModel]], str]]
    默认值:None(使用默认的 _default_format_error 函数)
    描述:一个可选的函数,用于格式化验证错误。输入包括异常对象、工具调用信息和模式,返回一个格式化的错误字符串。默认实现会返回异常的 repr 表示以及一条提示消息(“在修复所有验证错误后回复”)。

  • name
    类型:str
    默认值:"validation"
    描述:节点的名称,用于标识该节点。

  • tags
    类型:Optional[list[str]]
    默认值:None
    描述:一个可选的标签列表,用于标记节点,便于调试或分类。

返回值:

  • 根据输入类型,返回 Union[Dict[str, List[ToolMessage]], Sequence[ToolMessage]]:
    • 如果输入是消息列表,返回 ToolMessage 列表。
    • 如果输入是字典(包含 “messages” 键),返回包含 ToolMessage 列表的字典。

功能和用途

ValidationNode 的核心功能是从输入中的最后一条 AIMessage 提取工具调用请求,并验证其参数是否符合指定的模式。具体步骤如下:

  1. 提取消息
    从输入中提取最后一条消息,要求其为 AIMessage 类型。如果不符合,则抛出异常。

  2. 验证工具调用
    对于消息中的每个工具调用,使用 schemas 中对应的模式验证其参数:

    • 如果验证成功,将验证后的参数封装成 ToolMessage,内容为序列化的 JSON 字符串。
    • 如果验证失败,捕获 ValidationError(Pydantic 的验证异常),调用 format_error 生成错误消息,并返回带有错误标记的 ToolMessage(is_error=True)。
  3. 返回结果
    根据输入类型(列表或字典),返回相应的 ToolMessage 列表。

使用场景

  • 结构化输出生成:确保工具调用参数符合复杂格式要求,例如数据提取或 API 调用。
  • 多轮对话:验证工具调用并保留原始消息和工具 ID,便于后续对话引用。
  • 错误反馈与修正:在验证失败时提供详细错误信息,提示模型重新生成符合要求的响应。

使用示例

以下是一个使用 ValidationNode 的示例,展示如何在状态图中集成该节点,并通过条件边实现验证失败时的重试逻辑。

示例代码

from typing import Literal, Annotated
from typing_extensions import TypedDict
from langchain_anthropic import ChatAnthropic
from pydantic import BaseModel, field_validator
from langgraph.graph import END, START, StateGraph
from langgraph.prebuilt import ValidationNode
from langgraph.graph.message import add_messages

# 定义一个 Pydantic 模型,要求字段 a 必须为 37
class SelectNumber(BaseModel):
    a: int

    @field_validator("a")
    def a_must_be_meaningful(cls, v):
        if v != 37:
            raise ValueError("Only 37 is allowed")
        return v

# 创建状态图,状态为消息列表
builder = StateGraph(Annotated[list, add_messages])

# 初始化语言模型并绑定工具
llm = ChatAnthropic(model="claude-3-5-haiku-latest").bind_tools([SelectNumber])
builder.add_node("model", llm)

# 添加 ValidationNode 节点
builder.add_node("validation", ValidationNode([SelectNumber]))

# 定义边:从 START 到 model 节点
builder.add_edge(START, "model")

# 条件边:根据 model 输出决定是否验证
def should_validate(state: list) -> Literal["validation", "__end__"]:
    if state[-1].tool_calls:  # 如果有工具调用,转到 validation
        return "validation"
    return END

builder.add_conditional_edges("model", should_validate)

# 条件边:根据验证结果决定是否重试
def should_reprompt(state: list) -> Literal["model", "__end__"]:
    for msg in state[::-1]:
        if msg.type == "ai":  # 找到最近的 AI 消息
            return END
        if msg.additional_kwargs.get("is_error"):  # 如果有错误,返回 model 重试
            return "model"
    return END

builder.add_conditional_edges("validation", should_reprompt)

# 编译并运行状态图
graph = builder.compile()
res = graph.invoke(("user", "Select a number, any number"))

# 输出结果
for msg in res:
    msg.pretty_print()

输出示例

================================ Human Message ================================
Select a number, any number
================================= Ai Message ==================================
[{'id': 'toolu_01JSjT9Pq8hGmTgmMPc6KnvM', 'input': {'a': 42}, 'name': 'SelectNumber', 'type': 'tool_use'}]
Tool Calls:
SelectNumber (toolu_01JSjT9Pq8hGmTgmMPc6KnvM)
Args:
    a: 42
================================ Tool Message =================================
Name: SelectNumber
ValidationError(model='SelectNumber', errors=[{'loc': ('a',), 'msg': 'Only 37 is allowed', 'type': 'value_error'}])
Respond after fixing all validation errors.
================================= Ai Message ==================================
[{'id': 'toolu_01PkxSVxNxc5wqwCPW1FiSmV', 'input': {'a': 37}, 'name': 'SelectNumber', 'type': 'tool_use'}]
Tool Calls:
SelectNumber (toolu_01PkxSVxNxc5wqwCPW1FiSmV)
Args:
    a: 37
================================ Tool Message =================================
Name: SelectNumber
{"a": 37}

示例解析

  1. 用户输入“选择一个数字”,模型生成工具调用 SelectNumber(a=42)。
  2. ValidationNode 验证发现 a != 37,返回错误信息。
  3. 状态图根据条件边返回 “model” 节点,模型重新生成 SelectNumber(a=37)。
  4. 第二次验证通过,返回正确结果。

实现细节

模式处理

  • 在 __init__ 方法中,schemas 被转换为一个字典 schemas_by_name,键为工具名称,值为对应的 Pydantic 模型。
  • 支持三种输入类型:
    • BaseTool:使用其 args_schema。
    • BaseModel:直接使用。
    • 函数:通过 create_schema_from_function 生成模式。

消息提取

  • _get_message 方法从输入中提取最后一条 AIMessage,并确定输出类型(列表或字典)。

验证逻辑

  • _func 方法是核心逻辑,负责验证工具调用:
    • 对每个工具调用,使用对应的模式(Pydantic BaseModel)验证参数。
    • 使用线程池(executor)并行处理多个工具调用,提高效率。
    • 验证成功时,返回序列化的 JSON 内容;失败时,返回格式化的错误消息。

错误处理

  • 捕获 Pydantic 的 ValidationError(支持 v1 和 v2),通过 format_error 生成错误信息。
  • 错误消息的 ToolMessage 会附加 is_error=True 的标记。

注意事项

  1. 模式要求

    • 工具必须定义 args_schema,且为 Pydantic 的 BaseModel 类型。
    • 不支持无模式或非 Pydantic 模式的工具。
  2. 输入格式

    • 输入必须包含消息列表,且最后一条消息为 AIMessage。
    • 支持列表或字典(含 “messages” 键)两种格式。
  3. 版本兼容性

    • 支持 Pydantic v1 和 v2 的 BaseModel,通过条件判断处理差异。
  4. 性能优化

    • 对于多个工具调用,使用线程池并行验证,提升处理速度。

总结

langgraph.prebuilt.ValidationNode 是一个功能强大的工具验证节点,适用于需要严格参数格式的场景。它通过与 Pydantic 模型的集成,确保工具调用符合预定义规则,并在验证失败时提供反馈。通过与状态图的条件边结合,可以实现自动重试和错误修正机制,非常适合构建鲁棒的交互式应用,如多轮对话中的数据提取和结构化输出生成。

更多推荐