Appearance
第16章 预构建 Agent 组件
16.1 引言
前面的章节深入剖析了 LangGraph 的底层基础设施——StateGraph、Channel、Pregel 调度、Checkpoint、Send、Runtime、Store。这些原语提供了极大的灵活性，但直接使用它们构建一个完整的 Agent 需要编写大量的样板代码：定义状态 schema、创建 ToolNode、编写条件边路由、处理错误和重试。
langgraph.prebuilt 模块正是为了解决这个问题。它在底层原语之上提供了一组经过实战验证的高层组件：create_react_agent 工厂函数可以一行代码创建完整的 ReAct Agent；ToolNode 封装了工具执行的并行化、错误处理和状态注入；tools_condition 提供了标准的条件路由；ValidationNode 支持工具调用的 schema 验证；InjectedState 和 InjectedStore 让工具可以直接访问图状态和持久化存储。
本章将从这些组件的源码出发，分析它们如何将底层能力组合成开发者友好的高层 API，同时保持完整的可扩展性。
本章要点
create_react_agent工厂函数——从参数到编译图的完整构建流程ToolNode实现——并行执行、错误处理、状态注入、Command 支持tools_condition路由——标准的 Agent 循环条件判断ValidationNode——工具调用的 Pydantic schema 验证InjectedState、InjectedStore、ToolRuntime——工具级别的依赖注入
16.2 create_react_agent 工厂函数
16.2.1 签名概览
create_react_agent 定义在 langgraph/prebuilt/chat_agent_executor.py 中，是构建 ReAct Agent 的一站式入口：
python
def create_react_agent(
model: str | LanguageModelLike | Callable,
tools: Sequence[BaseTool | Callable | dict] | ToolNode,
*,
prompt: Prompt | None = None,
response_format: StructuredResponseSchema | None = None,
pre_model_hook: RunnableLike | None = None,
post_model_hook: RunnableLike | None = None,
state_schema: StateSchemaType | None = None,
context_schema: type[Any] | None = None,
checkpointer: Checkpointer | None = None,
store: BaseStore | None = None,
interrupt_before: list[str] | None = None,
interrupt_after: list[str] | None = None,
debug: bool = False,
version: Literal["v1", "v2"] = "v2",
name: str | None = None,
) -> CompiledStateGraph:16.2.2 构建流程
16.2.3 模型处理
create_react_agent 支持三种模型传入方式：
python
# 1. 字符串标识符（需要 langchain 包）
graph = create_react_agent("openai:gpt-4", tools)
# 2. LangChain ChatModel 实例
from langchain_openai import ChatOpenAI
graph = create_react_agent(ChatOpenAI(model="gpt-4"), tools)
# 3. 动态模型选择函数
def select_model(state, runtime: Runtime[ModelContext]):
if runtime.context.use_premium:
return ChatOpenAI(model="gpt-4").bind_tools(tools)
return ChatOpenAI(model="gpt-3.5-turbo").bind_tools(tools)
graph = create_react_agent(select_model, tools, context_schema=ModelContext)对于静态模型，框架自动调用 bind_tools 绑定工具。如果模型已经通过 model.bind_tools() 绑定了工具，框架会检查绑定的工具是否与传入的 tools 参数匹配。
16.2.4 v1 vs v2 版本差异
version 参数控制工具节点的执行策略：
v2 版本使用 Send API 将每个 tool_call 分发为独立的 ToolNode 实例。这种设计的优势：
- 中断粒度更细：可以单独中断/恢复某个工具调用
- 超时隔离：一个工具超时不影响其他工具
- Checkpoint 更精确：每个工具调用有独立的 checkpoint 状态
16.2.5 remaining_steps 安全机制
create_react_agent 使用 RemainingSteps managed value 来防止无限循环：
python
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], add_messages]
remaining_steps: NotRequired[RemainingSteps]当 remaining_steps 降至 2 以下且 LLM 仍在请求工具调用时，Agent 会返回一条友好的终止消息，而不是抛出 GraphRecursionError。
16.3 ToolNode 实现
16.3.1 核心职责
ToolNode 是工具执行的中枢，负责：
- 从最后一条 AIMessage 中提取 tool_calls
- 查找对应的工具实现
- 注入状态、Store 等依赖
- 并行执行工具
- 处理错误和返回结果
python
class ToolNode(RunnableCallable):
"""A node that runs tools requested by an AI model."""
def __init__(
self,
tools: Sequence[BaseTool | Callable],
*,
name: str = "tools",
tags: list[str] | None = None,
handle_tool_errors: bool | str | Callable | tuple = True,
):
self.tools_by_name = {tool.name: tool for tool in resolved_tools}
self.handle_tool_errors = handle_tool_errors