开启左侧

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

[复制链接]
AI小编 发表于 9 小时前 | 显示全部楼层 |阅读模式 打印 上一主题 下一主题
作者:彬彬侠
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
    4. model = ChatOpenAI(model="gpt-4")
    5. tools =[DuckDuckGoSearchRun()]
    6. agent = create_react_agent(model, tools, prompt="你是一个有用的助手")
    7. 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
    3. workflow = StateGraph(State)
    4. workflow.add_node("agent", agent_function)
    5. workflow.add_node("tools", tool_node)
    6. 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
    3. 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
    3. classWeatherSchema(BaseModel):
    4.     city:str
    5. 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
    3. @tooldefstate_tool(input:str, state: Annotated[dict, InjectedState]):
    4.     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
    3. @tooldefstore_tool(input:str, store: Annotated[Any, InjectedStore]):
    4.     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
  4. # 初始化模型和工具
  5. model = ChatOpenAI(model="gpt-3.5-turbo")
  6. tools =[DuckDuckGoSearchRun()]# 创建代理
  7. agent = create_react_agent(model, tools, prompt="你是一个有用的助手")# 运行代理
  8. 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
  7. # 定义状态classState(TypedDict):
  8.     messages: List[BaseMessage]# 定义代理节点defagent(state: State)-> State:
  9.     model = ChatOpenAI(model="gpt-3.5-turbo").bind_tools(tools)
  10.     message = model.invoke(state["messages"])
  11.     state["messages"].append(message)return state
  12. # 创建工具节点
  13. tools =[DuckDuckGoSearchRun()]
  14. tool_node = ToolNode(tools)# 构建状态图
  15. workflow = StateGraph(State)
  16. workflow.add_node("agent", agent)
  17. workflow.add_node("tools", tool_node)
  18. workflow.add_edge("tools","agent")
  19. workflow.add_conditional_edges("agent", tools_condition,{"tools":"tools","__end__": END})
  20. workflow.set_entry_point("agent")# 编译并运行
  21. graph = workflow.compile()
  22. 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
  4. # 定义模式classNumberSchema(BaseModel):
  5.     value:int@field_validator("value")defvalidate_value(cls, v):if v <0:raise ValueError("值必须非负")return v
  6. # 创建验证节点
  7. validation_node = ValidationNode(schemas=[NumberSchema])# 构建状态图
  8. workflow = StateGraph(list)
  9. workflow.add_node("validation", validation_node)
  10. workflow.set_entry_point("validation")
  11. workflow.set_finish_point("validation")
  12. graph = workflow.compile()# 测试验证
  13. messages =[AIMessage(content="", tool_calls=[{"name":"NumberSchema","args":{"value":-1},"id":"1"}])]
  14. 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
  7. # 定义状态classState(TypedDict):
  8.     messages: List[BaseMessage]
  9.     user_id:str# 定义工具@tooldefstate_tool(input:str, state: Annotated[dict, InjectedState])->str:returnf"User {state['user_id']} processed: {input}"# 创建工具节点
  10. tool_node = ToolNode(tools=[state_tool])# 构建状态图
  11. workflow = StateGraph(State)
  12. workflow.add_node("tools", tool_node)
  13. workflow.set_entry_point("tools")
  14. workflow.set_finish_point("tools")
  15. graph = workflow.compile()# 运行
  16. state ={"messages":[HumanMessage(content="test")],"user_id":"user1"}
  17. 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
  2. # 配置人类交互
  3. config = HumanInterruptConfig(allow_respond=True, allow_edit=True)# 创建中断
  4. interrupt = HumanInterrupt(
  5.     action_request=ActionRequest(action="respond", args={"response":"继续"}),
  6.     config=config,
  7.     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
回复

使用道具 举报

您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发布主题
阅读排行更多+

Powered by Discuz! X3.4© 2001-2013 Discuz Team.( 京ICP备17022993号-3 )