【LangGraph】langgraph.prebuilt 模块：封装了常见的代理逻辑、工具执行和状态管理功能

langgraph.prebuilt 模块是 LangGraph 的核心模块，提供了 create_react_agent、ToolNode、tools_condition、ValidationNode、InjectedState 等预构建组件，简化了 ReAct 代理、工具执行和状态管理的开发。它通过封装常见逻辑，支持快速构建对话代理、自动化工作流和人类交互场景。模块设计注重易用性和灵活性，适合

正义的彬彬侠

1558人浏览 · 2025-05-20 22:24:35

正义的彬彬侠 · 2025-05-20 22:24:35 发布

langgraph.prebuilt 模块是 LangGraph 库的核心组成部分，提供了一系列预构建的组件和工具，旨在简化复杂 AI 代理和工作流的开发过程。LangGraph 是 LangChain 生态的扩展框架，专注于构建有状态、多步骤的 AI 系统，通过状态图（StateGraph）管理节点和边，支持动态路由、循环和状态管理。该模块通过封装常见的代理逻辑、工具执行和状态管理功能，显著降低了开发者的编码负担，适合快速原型化和生产级应用。

1. 模块背景与作用

1.1 LangGraph 概述

定义：LangGraph 是一个用于构建有状态、多步骤 AI 应用的框架，特别适用于语言模型（LLM）驱动的场景。
功能：通过状态图组织节点（操作）和边（流程），支持动态路由、循环和状态管理。
应用场景：对话式 AI 代理（如聊天机器人）、自动化工作流、复杂任务分解等。
检查点机制：支持状态持久化，确保多轮对话和状态恢复。

1.2 模块作用

核心功能：提供预构建的类、函数和注解，简化代理创建、工具执行、状态管理和人类交互。
设计目标：
- 简化开发：通过开箱即用的组件减少手动编码量。
- 灵活性：支持标准 ReAct（Reasoning and Acting）代理和其他复杂工作流。
- 可扩展性：与 LangChain 生态无缝集成，允许自定义扩展。
地位：作为 LangGraph 的便捷入口，降低开发门槛，适合从初学者到高级开发者的广泛需求。

1.3 适用场景

快速原型：快速构建对话式代理或工具驱动的工作流。
生产级应用：通过检查点和存储支持，构建可靠的 AI 系统。
人类交互：支持人类干预，如审批或编辑，增强流程控制。
复杂逻辑：结合条件路由和状态注入，处理动态、多步骤任务。

2. 主要组件

langgraph.prebuilt 模块包含以下核心组件，基于上文和其他文档分析，分为函数、类和注解三大类：

2.1 函数

2.1.1 create_react_agent

类型：函数
描述：快速创建 ReAct 风格的代理图，自动构建状态图，迭代执行工具调用直到满足停止条件。ReAct（Reasoning and Acting）是一种结合推理和行动的代理架构。
参数：
- model: Union[str, LanguageModelLike]：语言模型，如 ChatOpenAI(model="gpt-4")。
- tools: Union[Sequence[Union[BaseTool, Callable]], ToolNode]：代理可调用的工具列表。
- checkpointer: Optional[BaseCheckpointSaver] = None：检查点保存器，支持状态持久化。
- state_modifier: Optional[Union[str, BasePromptTemplate]] = None：自定义提示模板。
- interrupt_before: Optional[List[str]] = None：指定节点前中断。
- interrupt_after: Optional[List[str]] = None：指定节点后中断。
- debug: bool = False：启用调试模式。
- stream_mode: Literal["values", "messages"] = "values"：流式输出模式。
返回：CompiledGraph，编译后的状态图，可通过 invoke 或 stream 调用。
用途：快速构建支持推理和工具调用的代理，适合多轮对话或任务分解。

示例：

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain.tools import DuckDuckGoSearchRun

model = ChatOpenAI(model="gpt-4")
tools = [DuckDuckGoSearchRun()]
agent = create_react_agent(model, tools, prompt="你是一个有用的助手")
result = agent.invoke({"input": "搜索今日天气"})
print(result)

2.1.2 tools_condition

类型：函数
描述：根据状态中最后一条消息是否包含工具调用，决定路由方向。
参数：
- state: Union[list[AnyMessage], dict[str, Any], BaseModel]：当前状态，包含消息列表。
- messages_key: str = "messages"：状态中存储消息的键，默认 "messages"。
返回：Literal["tools", "__end__"]：
- "tools"：如果最后一条消息包含工具调用，跳转到工具节点。
- "__end__"：如果无工具调用，结束流程。
用途：用于状态图的条件边，动态决定是否执行工具。

示例：

from langgraph.prebuilt import tools_condition
from langgraph.graph import StateGraph, END

workflow = StateGraph(State)
workflow.add_node("agent", agent_function)
workflow.add_node("tools", tool_node)
workflow.add_conditional_edges("agent", tools_condition, {"tools": "tools", "__end__": END})

2.2 类

2.2.1 ToolNode

类型：类
描述：状态图中的节点，用于执行指定的工具调用，从最后一条消息提取工具请求并返回结果。
参数：
- tools: Sequence[Union[BaseTool, Callable]]：工具列表，必需。
- name: str = "tools"：节点名称，默认 "tools"。
- tags: Optional[list[str]] = None：标签列表，用于调试或分类。
- handle_tool_errors: Union[bool, str, Callable[..., str], tuple[type[Exception], ...]] = True：错误处理配置，默认捕获错误。
- messages_key: str = "messages"：消息键，默认 "messages"。
用途：在状态图中处理工具调用，适合自动化工作流和代理逻辑。

示例：

from langgraph.prebuilt import ToolNode
from langchain.tools import DuckDuckGoSearchRun

tool_node = ToolNode(tools=[DuckDuckGoSearchRun()])

2.2.2 ValidationNode

类型：类
描述：验证最后一条 AIMessage 中的工具调用请求，确保符合指定模式，但不执行工具。
参数：
- schemas: Sequence[Union[BaseTool, Type[BaseModel], Callable]]：验证模式列表，支持 Pydantic 模型、工具或函数。
- format_error: Optional[Callable[[BaseException, ToolCall, Type[BaseModel]], str]] = None：格式化错误函数，默认返回异常信息。
- name: str = "validation"：节点名称。
- tags: Optional[list[str]] = None：标签列表。
用途：确保工具调用符合复杂模式，适合多轮对话中的数据完整性验证。

示例：

from langgraph.prebuilt import ValidationNode
from pydantic import BaseModel

class WeatherSchema(BaseModel):
    city: str

validation_node = ValidationNode(schemas=[WeatherSchema])

2.2.3 Human Interaction Components

HumanInterruptConfig：
- 描述：配置人类交互的中断选项。
- 属性：
  - allow_ignore: bool：是否允许忽略。
  - allow_respond: bool：是否允许响应。
  - allow_edit: bool：是否允许编辑。
  - allow_accept: bool：是否允许接受。
- 用途：定义人类干预的权限。
ActionRequest：
- 描述：定义人类交互的动作请求。
- 属性：
  - action: str：动作类型（如 “respond”）。
  - args: dict：动作参数。
HumanInterrupt：
- 描述：配置人类中断。
- 属性：
  - action_request: ActionRequest：动作请求。
  - config: HumanInterruptConfig：中断配置。
  - description: Optional[str]：描述。
HumanResponse：
- 描述：人类响应的类型。
- 属性：
  - type: Literal['accept', 'ignore', 'response', 'edit']：响应类型。
  - arg: Literal['accept', 'ignore', 'response', 'edit']：响应参数。
用途：支持人类在代理流程中的干预，如审批或编辑，增强控制和可靠性。

2.3 注解类

2.3.1 InjectedState

类型：注解类
描述：用于工具参数，注入状态图的当前状态，隐藏于工具模式，防止 LLM 生成。
参数：
- field: Optional[str] = None：指定状态中的键，若为 None，注入整个状态。
用途：工具访问状态（如对话历史），适合上下文相关操作。

示例：

from langgraph.prebuilt import InjectedState
from typing import Annotated

@tool
def state_tool(input: str, state: Annotated[dict, InjectedState]):
    user_id = state.get("user_id", "default")
    return f"User {user_id} processed: {input}"

2.3.2 InjectedStore

类型：注解类
描述：用于工具参数，注入 LangGraph 存储对象，隐藏于工具模式。
参数：
- field: Optional[str] = None：指定存储中的键，若为 None，注入整个存储。
用途：工具访问持久化存储（如用户数据），适合长时记忆。

示例：

from langgraph.prebuilt import InjectedStore
from typing import Annotated

@tool
def store_tool(input: str, store: Annotated[Any, InjectedStore]):
    data = store.get(("users",), "profile")
    return f"Stored data: {data}"

注意：需要 langchain-core >= 0.3.8。

3. 使用方法与示例

3.1 基本使用：创建 ReAct 代理

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain.tools import DuckDuckGoSearchRun

# 初始化模型和工具
model = ChatOpenAI(model="gpt-3.5-turbo")
tools = [DuckDuckGoSearchRun()]

# 创建代理
agent = create_react_agent(model, tools, prompt="你是一个有用的助手")

# 运行代理
result = agent.invoke({"input": "搜索今日天气"})
print(result["messages"][-1].content)

解析

使用 create_react_agent 创建 ReAct 代理，自动处理工具调用和状态管理。
适合快速构建支持搜索或计算的对话代理。

3.2 工具节点与条件路由

from typing import List, TypedDict
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langchain.tools import DuckDuckGoSearchRun

# 定义状态
class State(TypedDict):
    messages: List[BaseMessage]

# 定义代理节点
def agent(state: State) -> State:
    model = ChatOpenAI(model="gpt-3.5-turbo").bind_tools(tools)
    message = model.invoke(state["messages"])
    state["messages"].append(message)
    return state

# 创建工具节点
tools = [DuckDuckGoSearchRun()]
tool_node = ToolNode(tools)

# 构建状态图
workflow = StateGraph(State)
workflow.add_node("agent", agent)
workflow.add_node("tools", tool_node)
workflow.add_edge("tools", "agent")
workflow.add_conditional_edges("agent", tools_condition, {"tools": "tools", "__end__": END})
workflow.set_entry_point("agent")

# 编译并运行
graph = workflow.compile()
result = graph.invoke({"messages": [HumanMessage(content="搜索今日天气")]})
print(result["messages"][-1].content)

解析

使用 ToolNode 执行工具调用，tools_condition 动态路由。
实现了一个循环执行工具的代理，直到无工具调用。

3.3 验证工具调用

from langgraph.prebuilt import ValidationNode
from pydantic import BaseModel, field_validator
from langgraph.graph import StateGraph, END

# 定义模式
class NumberSchema(BaseModel):
    value: int

    @field_validator("value")
    def validate_value(cls, v):
        if v < 0:
            raise ValueError("值必须非负")
        return v

# 创建验证节点
validation_node = ValidationNode(schemas=[NumberSchema])

# 构建状态图
workflow = StateGraph(list)
workflow.add_node("validation", validation_node)
workflow.set_entry_point("validation")
workflow.set_finish_point("validation")
graph = workflow.compile()

# 测试验证
messages = [AIMessage(content="", tool_calls=[{"name": "NumberSchema", "args": {"value": -1}, "id": "1"}])]
result = graph.invoke(messages)
print(result[-1].content)  # 输出: 验证错误信息

解析

使用 ValidationNode 验证工具调用，确保参数符合 NumberSchema。
适合确保多轮对话中的数据完整性。

3.4 状态注入

from typing import List
from langgraph.prebuilt import ToolNode, InjectedState
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict, Annotated
from langchain_core.tools import tool
from langchain_core.messages import BaseMessage

# 定义状态
class State(TypedDict):
    messages: List[BaseMessage]
    user_id: str

# 定义工具
@tool
def state_tool(input: str, state: Annotated[dict, InjectedState]) -> str:
    return f"User {state['user_id']} processed: {input}"

# 创建工具节点
tool_node = ToolNode(tools=[state_tool])

# 构建状态图
workflow = StateGraph(State)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("tools")
workflow.set_finish_point("tools")
graph = workflow.compile()

# 运行
state = {"messages": [HumanMessage(content="test")], "user_id": "user1"}
result = graph.invoke(state)
print(result["messages"][-1].content)  # 输出: User user1 processed: test

解析

使用 InjectedState 注入状态到工具，隐藏于 LLM。
适合工具需要访问上下文（如用户 ID）的场景。

3.5 人类交互

from langgraph.prebuilt.interrupt import HumanInterruptConfig, HumanInterrupt, ActionRequest

# 配置人类交互
config = HumanInterruptConfig(allow_respond=True, allow_edit=True)

# 创建中断
interrupt = HumanInterrupt(
    action_request=ActionRequest(action="respond", args={"response": "继续"}),
    config=config,
    description="需要人类确认"
)

解析

使用人类交互组件配置中断，允许人类响应或编辑。
适合需要人工干预的工作流。

4. 实现原理

4.1 代理逻辑

create_react_agent 内部构建状态图，包含代理节点（调用 LLM）和工具节点（执行工具）。
使用 tools_condition 实现动态路由，基于工具调用决定流程。

4.2 工具执行

ToolNode 从消息中提取工具调用，执行并返回 ToolMessage。
支持错误处理（handle_tool_errors）和并行执行。

4.3 验证与状态

ValidationNode 使用 Pydantic 模型验证工具调用参数，确保数据完整性。
InjectedState 和 InjectedStore 通过运行时注入状态或存储，隐藏于工具模式。

4.4 人类交互

通过 HumanInterruptConfig 和 HumanInterrupt 配置中断逻辑，支持人类干预。

5. 适用场景与限制

5.1 适用场景

快速原型：通过 create_react_agent 快速构建对话代理。
工具驱动工作流：使用 ToolNode 和 tools_condition 实现自动化任务。
数据验证：通过 ValidationNode 确保工具调用符合模式。
上下文管理：使用 InjectedState 和 InjectedStore 实现状态驱动的工具。
人类交互：支持审批或编辑，适合需要人工干预的场景。

5.2 限制

版本依赖：部分功能（如 InjectedStore）需要 langchain-core >= 0.3.8。
功能局限：预构建组件适合标准场景，复杂定制需求需手动实现。
文档更新：功能可能随版本变化，需参考最新 LangGraph 文档.
复杂性：人类交互组件需额外配置，可能增加开发成本。

6. 对比其他方法

特性	langgraph.prebuilt	手动实现
开发速度	快，开箱即用	慢，需自定义逻辑
功能完整性	高，包含 ReAct、工具验证等	低，需手动实现
灵活性	中，适合标准场景	高，完全可定制
维护成本	低，依赖库更新	高，自定义代码需维护
适用场景	标准代理、工具调用	高度复杂工作流

选择建议：

标准场景：使用 langgraph.prebuilt，如 create_react_agent 和 ToolNode。
定制需求：结合手动实现，扩展预构建组件。

7. 注意事项

版本兼容性：
- 确保 LangGraph 版本 >= 0.2，检查依赖版本（如 langchain-core）。
- 参考 PyPI 页面和 GitHub 仓库。
调试技巧：
- 启用 debug=True 查看日志，分析代理行为。
- 使用 graph.get_graph().to_dot() 可视化状态图。
性能优化：
- 对于高并发，使用异步工具调用（如 ToolNode 的异步支持）。
- 配置 checkpointer 和 store 优化 I/O 性能。
安全考虑：
- 使用 InjectedState 和 InjectedStore 时，避免注入敏感数据。
- 验证工具调用时，确保模式定义严格。
模块导入：
- 如果遇到 No module named 'langgraph.prebuilt'，检查 LangGraph 是否正确安装：
```
pip install langgraph-prebuilt
```

8. 学习建议

基础知识：
- 掌握 LangGraph 的状态图、代理和工具概念。
- 了解 LangChain 的 LLM 和工具接口。
文档：
- 阅读 LangGraph 官方文档，重点关注预构建组件。
- 查看 LangGraph 参考。
实践：
- 从 create_react_agent 开始，构建简单对话代理。
- 逐步添加 ToolNode、tools_condition 和 ValidationNode，实现复杂工作流。
- 实验 InjectedState 和人类交互组件，探索上下文管理和干预。
社区：
- 加入 LangChain Discord，查看 GitHub 示例。
- 参考 Stack Overflow。
调试：
- 使用日志和可视化工具（如 graph.get_graph().to_dot()）分析代理行为。

9. 总结

langgraph.prebuilt 模块是 LangGraph 的核心模块，提供了 create_react_agent、ToolNode、tools_condition、ValidationNode、InjectedState 等预构建组件，简化了 ReAct 代理、工具执行和状态管理的开发。它通过封装常见逻辑，支持快速构建对话代理、自动化工作流和人类交互场景。模块设计注重易用性和灵活性，适合从原型到生产级的多种应用场景。通过示例和实践，开发者可以快速上手，构建高效、可控的 AI 系统。

参考资料：

MCP技术社区

欢迎加入 MCP 技术社区！与志同道合者携手前行，一同解锁 MCP 技术的无限可能！

更多推荐

8种封装的1700V国产碳化硅(SiC)功率模块产品介绍及应用

MCP技术社区

（一篇入门）汽车电子电器之电机MCU控制器四

MCP技术社区

OpenCV C++ 入门实战：从基础操作到类封装全解析

Mat数据结构、图像读写、颜色空间转换、像素操作、算术运算和轨迹栏交互，以及面向对象封装思想。图像处理：学习滤波（高斯滤波、中值滤波）、边缘检测（Canny、Sobel）、形态学操作（腐蚀、膨胀）。目标检测：掌握轮廓提取（）、特征匹配（）、Haar 级联分类器。视频处理：使用读取视频，对帧进行实时处理。性能优化：学习多线程、GPU 加速（cv::cuda模块），提升处理速度。通过持续实践，可逐步掌