作者:彬彬侠
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
- classWeatherSchema(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
- @tooldefstate_tool(input:str, state: Annotated[dict, InjectedState]):
- user_id = state.get("user_id","default")returnf"User {user_id} processed: {input}"
复制代码 2.3.2 InjectedStore
类型:注解类描述:用于工具参数,注入 LangGraph 存储对象,隐藏于工具模式。
- 参数:
field: Optional[str] = None:指定存储中的键,若为 None,注入整个存储。
用途:工具访问持久化存储(如用户数据),适合长时记忆。
- 示例:
- from langgraph.prebuilt import InjectedStore
- from typing import Annotated
- @tooldefstore_tool(input:str, store: Annotated[Any, InjectedStore]):
- data = store.get(("users",),"profile")returnf"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
- # 定义状态classState(TypedDict):
- messages: List[BaseMessage]# 定义代理节点defagent(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
- # 定义模式classNumberSchema(BaseModel):
- value:int@field_validator("value")defvalidate_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
- # 定义状态classState(TypedDict):
- messages: List[BaseMessage]
- user_id:str# 定义工具@tooldefstate_tool(input:str, state: Annotated[dict, InjectedState])->str:returnf"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 系统。
参考资料:
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 |