AI创想

标题: 【LangGraph】langgraph.prebuilt 模块：封装了常见的代理逻辑、工具执行和状态管理功能 [打印本页]

---

作者: AI小编    时间: 2025-12-24 08:32
标题: 【LangGraph】langgraph.prebuilt 模块：封装了常见的代理逻辑、工具执行和状态管理功能
作者：彬彬侠
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 调用。用途：快速构建支持推理和工具调用的代理，适合多轮对话或任务分解。
  
• 示例：
  1. from langgraph.prebuilt import create_react_agent
  2. from langchain_openai import ChatOpenAI
  3. from langchain.tools import DuckDuckGoSearchRun
  5. model = ChatOpenAI(model="gpt-4")
  6. tools =[DuckDuckGoSearchRun()]
  7. agent = create_react_agent(model, tools, prompt="你是一个有用的助手")
  8. 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__"：如果无工具调用，结束流程。
  
  用途：用于状态图的条件边，动态决定是否执行工具。
  
• 示例：
  1. from langgraph.prebuilt import tools_condition
  2. from langgraph.graph import StateGraph, END
  4. workflow = StateGraph(State)
  5. workflow.add_node("agent", agent_function)
  6. workflow.add_node("tools", tool_node)
  7. 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"。
  
  用途：在状态图中处理工具调用，适合自动化工作流和代理逻辑。
  
• 示例：
  1. from langgraph.prebuilt import ToolNode
  2. from langchain.tools import DuckDuckGoSearchRun
  4. 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：标签列表。
  
  用途：确保工具调用符合复杂模式，适合多轮对话中的数据完整性验证。
  
• 示例：
  1. from langgraph.prebuilt import ValidationNode
  2. from pydantic import BaseModel
  4. classWeatherSchema(BaseModel):
  5.     city:str
  7. 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，注入整个状态。
  
  用途：工具访问状态（如对话历史），适合上下文相关操作。
  
• 示例：
  1. from langgraph.prebuilt import InjectedState
  2. from typing import Annotated
  4. @tooldefstate_tool(input:str, state: Annotated[dict, InjectedState]):
  5.     user_id = state.get("user_id","default")returnf"User {user_id} processed: {input}"

  复制代码

2.3.2 InjectedStore

类型：注解类描述：用于工具参数，注入 LangGraph 存储对象，隐藏于工具模式。

• 参数：
  
  field: Optional[str] = None：指定存储中的键，若为 None，注入整个存储。
  
  用途：工具访问持久化存储（如用户数据），适合长时记忆。
  
• 示例：
  1. from langgraph.prebuilt import InjectedStore
  2. from typing import Annotated
  4. @tooldefstore_tool(input:str, store: Annotated[Any, InjectedStore]):
  5.     data = store.get(("users",),"profile")returnf"Stored data: {data}"

  复制代码
  注意：需要 langchain-core >= 0.3.8。
  

---

3. 使用方法与示例

3.1 基本使用：创建 ReAct 代理

1. from langgraph.prebuilt import create_react_agent
2. from langchain_openai import ChatOpenAI
3. from langchain.tools import DuckDuckGoSearchRun
5. # 初始化模型和工具
6. model = ChatOpenAI(model="gpt-3.5-turbo")
7. tools =[DuckDuckGoSearchRun()]# 创建代理
8. agent = create_react_agent(model, tools, prompt="你是一个有用的助手")# 运行代理
9. result = agent.invoke({"input":"搜索今日天气"})print(result["messages"][-1].content)

复制代码

解析

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

3.2 工具节点与条件路由

1. from typing import List, TypedDict
2. from langgraph.graph import StateGraph, END
3. from langgraph.prebuilt import ToolNode, tools_condition
4. from langchain_core.messages import BaseMessage, HumanMessage
5. from langchain_openai import ChatOpenAI
6. from langchain.tools import DuckDuckGoSearchRun
8. # 定义状态classState(TypedDict):
9.     messages: List[BaseMessage]# 定义代理节点defagent(state: State)-> State:
10.     model = ChatOpenAI(model="gpt-3.5-turbo").bind_tools(tools)
11.     message = model.invoke(state["messages"])
12.     state["messages"].append(message)return state
14. # 创建工具节点
15. tools =[DuckDuckGoSearchRun()]
16. tool_node = ToolNode(tools)# 构建状态图
17. workflow = StateGraph(State)
18. workflow.add_node("agent", agent)
19. workflow.add_node("tools", tool_node)
20. workflow.add_edge("tools","agent")
21. workflow.add_conditional_edges("agent", tools_condition,{"tools":"tools","__end__": END})
22. workflow.set_entry_point("agent")# 编译并运行
23. graph = workflow.compile()
24. result = graph.invoke({"messages":[HumanMessage(content="搜索今日天气")]})print(result["messages"][-1].content)

复制代码

解析

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

3.3 验证工具调用

1. from langgraph.prebuilt import ValidationNode
2. from pydantic import BaseModel, field_validator
3. from langgraph.graph import StateGraph, END
5. # 定义模式classNumberSchema(BaseModel):
6.     value:int@field_validator("value")defvalidate_value(cls, v):if v <0:raise ValueError("值必须非负")return v
8. # 创建验证节点
9. validation_node = ValidationNode(schemas=[NumberSchema])# 构建状态图
10. workflow = StateGraph(list)
11. workflow.add_node("validation", validation_node)
12. workflow.set_entry_point("validation")
13. workflow.set_finish_point("validation")
14. graph = workflow.compile()# 测试验证
15. messages =[AIMessage(content="", tool_calls=[{"name":"NumberSchema","args":{"value":-1},"id":"1"}])]
16. result = graph.invoke(messages)print(result[-1].content)# 输出: 验证错误信息

复制代码

解析

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

3.4 状态注入

1. from typing import List
2. from langgraph.prebuilt import ToolNode, InjectedState
3. from langgraph.graph import StateGraph, END
4. from typing_extensions import TypedDict, Annotated
5. from langchain_core.tools import tool
6. from langchain_core.messages import BaseMessage
8. # 定义状态classState(TypedDict):
9.     messages: List[BaseMessage]
10.     user_id:str# 定义工具@tooldefstate_tool(input:str, state: Annotated[dict, InjectedState])->str:returnf"User {state['user_id']} processed: {input}"# 创建工具节点
11. tool_node = ToolNode(tools=[state_tool])# 构建状态图
12. workflow = StateGraph(State)
13. workflow.add_node("tools", tool_node)
14. workflow.set_entry_point("tools")
15. workflow.set_finish_point("tools")
16. graph = workflow.compile()# 运行
17. state ={"messages":[HumanMessage(content="test")],"user_id":"user1"}
18. result = graph.invoke(state)print(result["messages"][-1].content)# 输出: User user1 processed: test

复制代码

解析

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

3.5 人类交互

1. from langgraph.prebuilt.interrupt import HumanInterruptConfig, HumanInterrupt, ActionRequest
3. # 配置人类交互
4. config = HumanInterruptConfig(allow_respond=True, allow_edit=True)# 创建中断
5. interrupt = HumanInterrupt(
6.     action_request=ActionRequest(action="respond", args={"response":"继续"}),
7.     config=config,
8.     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 是否正确安装：
  1. 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 系统。

---

参考资料：

LangGraph 官方文档LangGraph 参考：预构建组件LangGraph GitHub 仓库LangGraph PyPI 页面Stack Overflow: No module named ‘langgraph.prebuilt’Medium: Building a ReAct Agent with LangGraph

原文地址：https://blog.csdn.net/u013172930/article/details/148098428

---

欢迎光临 AI创想 (https://www.llms-ai.com/)

Powered by Discuz! X3.4