使用LangGraph在Python中构建代理工作流

使用LangGraph在Python中构建代理工作流

💡 原文英文,约2600词,阅读约需10分钟。
📝

内容提要

本文介绍如何使用LangGraph在Python中构建代理工作流,涵盖状态、节点和边的基本概念,通过MessagesState管理对话历史,在节点中调用模型,注册工具并路由工具调用,以及使用检查点持久化跨会话的对话。文章从基础图构建到工具使用代理,逐步演示完整流程,并强调LangGraph组件的独立性和可扩展性。

🔎

延伸解读

状态、节点和边:理解LangGraph的核心抽象

LangGraph将代理工作流建模为图,其中状态是共享内存,节点是执行单元,边定义执行顺序。这种设计使得每个推理步骤、工具调用和响应都成为图状态的一部分,从而提供完整的可观测性。理解这些原语是构建复杂代理的基础,因为它们决定了数据如何在节点间流动以及如何被更新。

工具调用的循环:ReAct模式与成本考量

当模型决定使用工具时,图会进入一个循环:模型调用、工具执行、再次模型调用。这个循环是ReAct模式的体现,但每次工具使用都意味着两次模型调用,这会增加延迟和成本。在设计代理时,应权衡工具的使用频率和必要性,以优化性能和费用。

持久化与状态管理:从开发到生产

使用检查点(checkpointer)可以跨会话持久化对话状态,但InMemorySaver仅适用于开发测试。生产环境需要基于数据库的持久化方案。此外,检查点保存的是图状态,而Store则用于存储独立于对话的长期数据,如用户偏好。理解两者的区别有助于设计可扩展的代理应用。

Q&A

LangGraph中的状态、节点和边分别是什么?

在LangGraph中,状态是一个TypedDict,作为整个图的共享内存,节点读取和更新它;节点是普通的Python函数,接收当前状态并返回要更新的字段;边定义了执行顺序,add_edge表示顺序执行,add_conditional_edges根据路由函数决定下一步。

如何在LangGraph中管理对话历史?

LangGraph提供了内置的MessagesState,它是一个TypedDict,包含一个messages字段,使用add_messages reducer,每次节点返回新消息时,会追加到现有列表而不是覆盖,从而自动管理对话历史。

如何在LangGraph中调用语言模型?

在节点中调用语言模型,例如使用ChatOpenAI,将系统消息和当前消息列表传递给模型,模型返回AIMessage,然后以字典形式返回{'messages': [response]},这样响应就会添加到状态中。

如何在LangGraph中注册工具并路由工具调用?

使用@tool装饰器定义工具,然后用llm.bind_tools(tools)将工具绑定到模型,使模型知道工具的存在。在图中添加ToolNode来执行工具,并使用tools_condition作为条件边,根据模型是否产生工具调用来路由到工具节点或结束。

LangGraph中的ReAct循环是如何工作的?

ReAct循环是模型调用、工具执行、再模型调用的循环。模型首先决定是否需要工具,如果需要,则生成带有tool_calls的AIMessage,ToolNode执行工具并返回ToolMessage,然后模型再次被调用以解释工具结果并生成最终答案。每次工具使用需要两次模型调用。

如何在LangGraph中持久化对话状态?

在编译图时附加一个检查点器(checkpointer),例如InMemorySaver,然后在每次调用时传递相同的thread_id,这样图的状态会在调用之间保存和恢复。生产环境可以使用持久化检查点器,如数据库支持的存储。

LangGraph中的Store和Checkpointer有什么区别?

Checkpointer用于持久化图的状态,特别是对话历史,按线程隔离。Store用于持久化应用程序级数据,如用户配置文件或长期记忆,可以跨线程共享。两者互补,Store提供图执行期间可访问的持久存储。

🏷️

标签

➡️

继续阅读