【LangGraph】langgraph.prebuilt.ValidationNode 类:验证工具调用的参数是否符合预定义的模式(schemas)
langgraph.prebuilt.ValidationNode 是一个用于验证工具调用的节点类,广泛应用于 LangGraph 的状态图中。它并不实际执行工具,而是专注于验证工具调用的参数是否符合预定义的模式(schemas),并返回验证结果或错误信息。这种功能在需要生成符合复杂模式的结构化输出时尤为有用,尤其是在多轮对话场景中,可以保留原始消息和工具 ID 不丢失。
以下是对 ValidationNode 类的详细介绍,包括其定义、功能、使用场景、实现细节以及示例。
类的定义和参数
ValidationNode 继承自 RunnableCallable,是一个可运行的节点类。它的构造函数接受以下参数:
-
schemas
类型:Sequence[Union[BaseTool, Type[BaseModel], Callable]]
描述:一个列表,包含用于验证工具调用的模式。可以是以下类型之一:- Pydantic 的
BaseModel类(如 Pydantic v1 或 v2 的模型)。 BaseTool实例(将使用其args_schema属性)。- 函数(将根据函数签名自动生成模式)。
作用:定义工具调用参数需要遵守的结构和规则。
- Pydantic 的
-
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 提取工具调用请求,并验证其参数是否符合指定的模式。具体步骤如下:
-
提取消息
从输入中提取最后一条消息,要求其为AIMessage类型。如果不符合,则抛出异常。 -
验证工具调用
对于消息中的每个工具调用,使用schemas中对应的模式验证其参数:- 如果验证成功,将验证后的参数封装成
ToolMessage,内容为序列化的 JSON 字符串。 - 如果验证失败,捕获
ValidationError(Pydantic 的验证异常),调用format_error生成错误消息,并返回带有错误标记的ToolMessage(is_error=True)。
- 如果验证成功,将验证后的参数封装成
-
返回结果
根据输入类型(列表或字典),返回相应的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}
示例解析
- 用户输入“选择一个数字”,模型生成工具调用
SelectNumber(a=42)。 ValidationNode验证发现a != 37,返回错误信息。- 状态图根据条件边返回 “model” 节点,模型重新生成
SelectNumber(a=37)。 - 第二次验证通过,返回正确结果。
实现细节
模式处理
- 在
__init__方法中,schemas被转换为一个字典schemas_by_name,键为工具名称,值为对应的 Pydantic 模型。 - 支持三种输入类型:
BaseTool:使用其args_schema。BaseModel:直接使用。- 函数:通过
create_schema_from_function生成模式。
消息提取
_get_message方法从输入中提取最后一条AIMessage,并确定输出类型(列表或字典)。
验证逻辑
_func方法是核心逻辑,负责验证工具调用:- 对每个工具调用,使用对应的模式(Pydantic
BaseModel)验证参数。 - 使用线程池(
executor)并行处理多个工具调用,提高效率。 - 验证成功时,返回序列化的 JSON 内容;失败时,返回格式化的错误消息。
- 对每个工具调用,使用对应的模式(Pydantic
错误处理
- 捕获 Pydantic 的
ValidationError(支持 v1 和 v2),通过format_error生成错误信息。 - 错误消息的
ToolMessage会附加is_error=True的标记。
注意事项
-
模式要求
- 工具必须定义
args_schema,且为 Pydantic 的BaseModel类型。 - 不支持无模式或非 Pydantic 模式的工具。
- 工具必须定义
-
输入格式
- 输入必须包含消息列表,且最后一条消息为
AIMessage。 - 支持列表或字典(含 “messages” 键)两种格式。
- 输入必须包含消息列表,且最后一条消息为
-
版本兼容性
- 支持 Pydantic v1 和 v2 的
BaseModel,通过条件判断处理差异。
- 支持 Pydantic v1 和 v2 的
-
性能优化
- 对于多个工具调用,使用线程池并行验证,提升处理速度。
总结
langgraph.prebuilt.ValidationNode 是一个功能强大的工具验证节点,适用于需要严格参数格式的场景。它通过与 Pydantic 模型的集成,确保工具调用符合预定义规则,并在验证失败时提供反馈。通过与状态图的条件边结合,可以实现自动重试和错误修正机制,非常适合构建鲁棒的交互式应用,如多轮对话中的数据提取和结构化输出生成。
更多推荐


所有评论(0)