Day 56 / 共 60 天 · 阶段 9 预制件与流式

路由与校验:tools_condition + ValidationNode

D53 里 create_react_agent 用的是内部的 should_continue。但如果你自己搭图,官方给了个更通用的现成路由 tools_condition——一行接上就有 ReAct 走廊。另一头,ValidationNode 解决一个正交问题:只校验模型填的参数合不合法、不真跑工具,专用于结构化抽取和"让模型重填到对为止"的重试循环。今天把这两个"辅助件"看透。

📍 阶段 9 · 预制件与流式(6 天)你在这里
D53 总览 D54 逐行① D55 ToolNode D56 校验 D57 流式5模式 D58 流式底层
💡 用一个类比先兜住今天 tools_condition 像地铁站的指示牌:"手里有票(tool_calls)往闸机(tools)走,没票就出站(END)"——一块牌子,谁都能用。ValidationNode 则像签证官:他不放你入境(不执行工具),只检查你填的表格合不合规,填错了盖个"退回重填"的章,你改好再来——反复直到表格合法。两者一个管"去哪",一个管"填得对不对"。
L01

tools_condition 为什么存在

🤔 痛点D53 的 should_continue 是 create_react_agent 内部闭包,你拿不到、也不该拿。可你想自己搭一个 ReAct 图(要更多定制),"判断该不该去 tools"的逻辑难道要每次手写?
💡 本质tools_condition 就是把"看最后一条 AIMessage 有没有 tool_calls"这个最常见判断抽成一个公开工具函数,配 add_conditional_edges 直接用。它是 should_continue 的"简化公开版":不处理 v2 的 Send、不管 post_model_hook,只做最基础的二选一。

它和 ToolNode 一起,构成"手搓 ReAct 三件套":StateGraph + ToolNode + tools_condition。定义在 tool_node.py:1582,docstring 里就给了标准用法:

# tool_node.py:1582 docstring 示例
graph = StateGraph(State)
graph.add_node("llm", call_model)
graph.add_node("tools", ToolNode([my_tool]))
graph.add_conditional_edges(
    "llm",
    tools_condition,                         # ← 一行接上路由
    {"tools": "tools", "__end__": "__end__"},
)
👶 一句话用 create_react_agent 就用不到它(内部有 should_continue);自己搭图时,它帮你省掉写路由函数。
L02

逐行读 tools_condition

函数体很短(tool_node.py:1648 起,即 docstring 之后):

# tool_node.py:1648
def tools_condition(state, messages_key="messages") -> Literal["tools", "__end__"]:
    if isinstance(state, list):
        ai_message = state[-1]                                # 输入是消息列表
    elif (isinstance(state, dict) and (messages := state.get(messages_key, []))) or (
        messages := getattr(state, messages_key, [])):
        ai_message = messages[-1]                             # 输入是 dict / dataclass 状态
    else:
        raise ValueError(f"No messages found in input state to tool_edge: {state}")
    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
        return "tools"                                        # 有 tool_calls → 去 tools
    return "__end__"                                          # 没有 → 结束
isinstance(state, list)三种输入形态之一:直接是消息列表,取最后一条。
state.get(messages_key) / getattr(...)dict 状态或 dataclass/Pydantic 状态,都能取到 messages。messages_key 可自定义(比如你的状态里叫 chat_history)。
hasattr(ai_message,"tool_calls") and len>0核心判断:最后一条消息有非空 tool_calls → 返回 "tools";否则 "__end__"
返回值是字符串注意它返回的是普通字符串(不像 should_continue 会返回 list[Send])——所以它不做 v2 并行扇出,是最朴素的二选一。
💡 设计取舍①:为什么另造一个 tools_condition,而不复用 should_continue?因为两者面向不同用户should_continue 是 create_react_agent 的内部实现,要处理 v2 Send 扇出、post_model_hook、structured_response、return_direct 一大堆分支——复杂但功能全。tools_condition 是给自己搭图的人用的公开 API,只需最常见的"有票去闸机/没票出站"。把复杂留给框架内部,把简单暴露给用户——如果强行让用户直面 should_continue 的全部分支,反而劝退。两个函数各服务一类人,是刻意的分层。
对比记忆:should_continue(chat_agent_executor.py:831,返回 str | list[Send])功能全;tools_condition(tool_node.py:1582,返回 Literal["tools","__end__"])极简公开。
L03

_validate_tool_call:模型点了不存在的工具怎么办

回到 ToolNode 里一个"校验"环节(D55 L05 里提过)。模型可能幻觉出一个不存在的工具名。tool_node.py:1268

# tool_node.py:1268
def _validate_tool_call(self, call: ToolCall) -> ToolMessage | None:
    requested_tool = call["name"]
    if requested_tool not in self.tools_by_name:                  # 不在工具表里
        all_tool_names = list(self.tools_by_name.keys())
        content = INVALID_TOOL_NAME_ERROR_TEMPLATE.format(
            requested_tool=requested_tool,
            available_tools=", ".join(all_tool_names))            # 列出可用工具
        return ToolMessage(content, name=requested_tool,
                           tool_call_id=call["id"], status="error")
    return None                                                   # 工具存在 → 无需拦截
# INVALID_TOOL_NAME_ERROR_TEMPLATE (:108):
# "Error: {requested_tool} is not a valid tool, try one of [{available_tools}]."
requested_tool not in self.tools_by_name用 D55 建的名字索引查表。查不到就是幻觉工具。
", ".join(all_tool_names)关键:错误消息里列出所有可用工具名,等于给模型一份"正确答案清单"。
return ToolMessage(status="error")返回一条错误消息(不抛异常、不崩)。模型下一轮看到"XX不是有效工具,试试[weather, add]",大概率就改对了。
return None工具存在 → 返回 None,表示"无需拦截,正常执行"。
💡 本质:把"错误提示"写成"引导提示"光说"XX 无效"没用,模型不知道该用啥。源码特意把可用工具清单塞进错误消息——这是"错误消息即 few-shot 引导"的小技巧。模型的自纠能力,很大程度取决于你把错误信息写得多"可操作"。
L04

ValidationNode:只校验参数、不真跑工具

🤔 痛点有些场景你根本不想"执行"工具,只想让模型输出符合复杂 schema 的结构化数据(比如抽取一个 Person(name, age))。模型可能填错(age 填成字符串、漏字段)。你想校验、给模型退回、让它重填,直到合法——但保留原始消息和 tool_call id 以便多轮对话。

这就是 ValidationNodetool_validator.py:47)。它继承 RunnableCallable,也是个图节点,但和 ToolNode 的差别是:它 model_validate 参数,不 invoke 工具

# tool_validator.py:47(类头 + docstring 要点)
class ValidationNode(RunnableCallable):
    """A node that validates all tools requests from the last AIMessage.
    This node does NOT actually run the tools, it only validates the tool calls,
    which is useful for extraction and other use cases where you need to generate
    structured output that conforms to a complex schema ..."""
👶 一句话ToolNode = 执行工具、返回工具结果;ValidationNode = 只校验参数合不合 schema、合法就返回规整后的 JSON、不合法就返回错误让模型重填。一个做事,一个查表格。
源码顶上也有 @deprecated(:43)——官方推荐迁到 langchain.agents 的新错误处理。但作为"校验型节点"的设计范本,它的思路依然经典。
L05

__init__:schema 的三种来源

tool_validator.py:116,把各种输入统一成"名字 → Pydantic 模型":

# tool_validator.py:116
def __init__(self, schemas, *, format_error=None, name="validation", tags=None):
    super().__init__(self._func, None, name=name, tags=tags, trace=False)
    self._format_error = format_error or _default_format_error
    self.schemas_by_name: dict[str, type[BaseModel]] = {}
    for schema in schemas:
        if isinstance(schema, BaseTool):                          # ① 工具 → 用它的 args_schema
            if schema.args_schema is None:
                raise ValueError(f"Tool {schema.name} does not have an args_schema defined.")
            ...
            self.schemas_by_name[schema.name] = schema.args_schema
        elif isinstance(schema, type) and issubclass(schema, (BaseModel, BaseModelV1)):
            self.schemas_by_name[schema.__name__] = schema        # ② 直接是 Pydantic 类
        elif callable(schema):
            base_model = create_schema_from_function("Validation", schema)  # ③ 函数 → 推 schema
            self.schemas_by_name[schema.__name__] = base_model
        else:
            raise ValueError(f"Unsupported input to ValidationNode. ...")
BaseTool → args_schema传工具,就取它的参数 schema。若工具没定义 args_schema 就报错——校验必须有 schema 做依据。
Pydantic 类 → 直接用最常见:你直接传 Person 这样的 Pydantic 模型当校验规则。
callable → create_schema_from_function传普通函数,从它的签名/类型注解推出一个 Pydantic schema。
schemas_by_name和 ToolNode 的 tools_by_name 呼应:都是"名字→XXX"的索引,执行时按 call["name"] 查。
_default_format_error(:34)默认错误格式化:f"{repr(error)}\n\nRespond after fixing all validation errors."——提示模型修完再答。
💡 本质:和 ToolNode 是"同构"的两兄弟你会发现 ValidationNode 的结构几乎复刻 ToolNode:__init__ 建 xxx_by_name 索引、_func 里 get_executor_for_config 并行 run_one、输出按 list/dict 对齐。同一套"节点骨架",换掉核心动作(一个 invoke 工具、一个 model_validate 参数)。看懂 D55,这里就是举一反三。
L06

_func:校验成功出 JSON,失败出错误

核心 tool_validator.py:184

# tool_validator.py:184
def _func(self, input, config):
    output_type, message = self._get_message(input)          # 取最后一条 AIMessage
    def run_one(call: ToolCall) -> ToolMessage:
        schema = self.schemas_by_name[call["name"]]
        try:
            if issubclass(schema, BaseModel):
                output = schema.model_validate(call["args"])  # ★ 校验参数(不执行工具!)
                content = output.model_dump_json()            # 合法 → 规整成 JSON
            elif issubclass(schema, BaseModelV1):
                output = schema.validate(call["args"]); content = output.json()
            ...
            return ToolMessage(content=content, name=call["name"], tool_call_id=call["id"])
        except (ValidationError, ValidationErrorV1) as e:
            return ToolMessage(                               # 不合法 → 错误消息
                content=self._format_error(e, call, schema),
                name=call["name"], tool_call_id=call["id"],
                additional_kwargs={"is_error": True})         # ★ 打上 is_error 标记
    with get_executor_for_config(config) as executor:
        outputs = [*executor.map(run_one, message.tool_calls)]  # 并行校验每个 call
        return outputs if output_type == "list" else {"messages": outputs}
schema.model_validate(call["args"])今天的高潮:用 Pydantic 校验模型填的参数。这里只校验、绝不 invoke 工具——这是它和 ToolNode 的本质区别。
content = model_dump_json()校验通过:把规整后的对象转成 JSON 字符串放进 ToolMessage。相当于"帮你把数据洗干净"。
except ValidationError → ToolMessage校验失败:把 Pydantic 的报错格式化成消息,附 is_error=True 标记,喂回模型。
additional_kwargs={"is_error": True}关键:这个标记让下游路由能判断"这轮校验有没有出错",从而决定是否重试(L07)。
executor.map(run_one, ...)和 ToolNode 一样,多个 tool_call 并行校验。
⚠️ 边界:为什么校验节点要保留 tool_call_id、不直接返回裸数据?朴素做法可能想直接 return Person(...)。但模型 API 要求每个 tool_call 必须有配对的 tool 结果(D54 那个校验!)。如果 ValidationNode 不产出带 tool_call_id 的 ToolMessage,下一轮调模型就会因"孤儿 tool_call"报 400。所以它即使只是校验,也要伪装成一次完整的"工具应答"——用 ToolMessage 包装,带上原 id。这是"结构化抽取"要在多轮对话里工作的隐形约束。
L07

重试环:校验失败→重填→再校验 + 小结

ValidationNode 的威力在于配一个"重试路由"(tool_validator.py:47 docstring 里的完整例子):

# tool_validator.py docstring 示例(re-prompt 循环)
builder.add_node("model", llm)
builder.add_node("validation", ValidationNode([SelectNumber]))
def should_validate(state) -> Literal["validation", "__end__"]:
    if state[-1].tool_calls: return "validation"      # 模型给了 tool_call → 去校验
    return END
builder.add_conditional_edges("model", should_validate)
def should_reprompt(state) -> Literal["model", "__end__"]:
    for msg in state[::-1]:
        if msg.type == "ai": return END               # 回溯到 AIMessage 都没错 → 结束
        if msg.additional_kwargs.get("is_error"): return "model"  # 有错 → 回模型重填
    return END
builder.add_conditional_edges("validation", should_reprompt)
should_validate模型产出 tool_call → 去 validation 校验(而不是去执行)。
should_reprompt 回溯 is_error从后往前看校验结果:碰到 is_error → 回 model 重填;一路没错到 AIMessage → END。这就是靠 L06 那个标记实现的。
model ↔ validation 成环模型填→校验→(错就)回模型重填→再校验……直到合法。一个"自我修正到合规"的循环。
数据结构:ToolNode vs ValidationNode 同骨架不同核 ToolNode _tools_by_name run_one: tool.invoke(args) ← 执行 出:工具真实结果 错误→ToolMessage(error) ValidationNode schemas_by_name run_one: schema.model_validate ← 校验 出:规整 JSON 失败→is_error=True
图注:两个节点结构对称,只有 run_one 里"执行 vs 校验"这一核不同。
控制流:re-prompt 自纠环 model validation END 有tool_call is_error→回 model 重填 全合法
图注:校验失败就把模型拽回来重填,直到参数全合法才 END。

🧠 今天你应该能回答

  • tools_condition 和 should_continue 的区别?(公开极简版 vs 内部全功能版;前者不做 Send 扇出)
  • tools_condition 怎么判断?(最后一条消息有非空 tool_calls→"tools",否则"__end__")
  • 模型幻觉出不存在的工具,ToolNode 怎么处理?(返回列出可用工具的错误 ToolMessage)
  • ValidationNode 和 ToolNode 的本质区别?(model_validate 校验 vs tool.invoke 执行)
  • ValidationNode 的 schema 三来源?(BaseTool 的 args_schema / Pydantic 类 / 函数推断)
  • is_error 标记有什么用?(让下游路由判断该不该把模型拽回重填)
  • 为什么校验节点也要产 ToolMessage 带 tool_call_id?(否则孤儿 tool_call 会让下轮调模型 400)

✋ 10 分钟动手

# 1. 三段核心逐行读
sed -n '1648,1662p' libs/prebuilt/langgraph/prebuilt/tool_node.py       # tools_condition 主体
sed -n '1268,1280p' libs/prebuilt/langgraph/prebuilt/tool_node.py       # 幻觉工具校验
sed -n '184,221p'   libs/prebuilt/langgraph/prebuilt/tool_validator.py  # 校验 _func

# 2. 亲手看 ValidationNode 校验失败
python - <<'PY'
from langgraph.prebuilt import ValidationNode
from langchain_core.messages import AIMessage
from pydantic import BaseModel, field_validator
class SelectNumber(BaseModel):
    a: int
    @field_validator("a")
    def only_37(cls, v):
        if v != 37: raise ValueError("Only 37 is allowed")
        return v
node = ValidationNode([SelectNumber])
ai = AIMessage(content="", tool_calls=[{"name":"SelectNumber","args":{"a":5},"id":"1","type":"tool_call"}])
print(node.invoke([ai])[0].content)   # 看到 Only 37 is allowed + Respond after fixing...
PY
明天预告 · Day 57:换个大主题——流式graph.stream(stream_mode=...) 的五种模式 values/updates/messages/custom/debug 各输出什么、在 pregel 里哪里被 emit、以及"传一个 list 同时要多种模式"怎么实现。
← Day 55 ToolNode Day 57 · 流式 5 模式 →