使用langgraph创建基于AI代理的系统：添加持久性和流（逐步指南）

2025-02-02 06:55:13 英文原文

作者：Vineet Kumar

在上一位教程中，我们建立了一个AI代理，能够通过上网浏览网络来回答查询。但是，当为长期运行任务建造代理时，两个关键概念就会发挥作用：持久性和流。持久性使您能够在任何给定点保存代理的状态，从而使您能够在未来的互动中从该状态恢复。这对于长期运行的应用至关重要。另一方面，流媒体使您可以随时发出有关代理商在做什么的实时信号，从而提供透明度和对其动作的控制。在本教程中，我们通过添加这些强大的功能来增强我们的代理商。

设置代理

让我们首先重新创建我们的代理商。我们将加载必要的环境变量，安装和导入所需的库，设置塔维利搜索工具，定义代理状态，最后构建代理。

pip install langgraph == 0.2.53 langgraph-checkpoint == 2.0.6 langgraph-sdk == 0.1.36 langchain-groq langchain-community langgraph-checkpoint-sqlite == 2.0.1.1.1.1.1

导入操作系统os.environ ['tavily_api_key'] =“ <tavily_api_key>”os.environ ['groq_api_key'] =“ <GROQ_API_KEY>”摘自langgraph.graph Import stategraph，结束通过输入导入打字，注释导入操作员来自langchain_core.message导入任何杂音，系统杂项，人类，工具来自langchain_groq导入changroq来自langchain_community.tools.tavily_search导入tavilySearchResults工具= TavilySearchResults（max_results = 2）班级代理（TypedDict）：消息：注释[list [Anymessage]，operator.Add]班级代理：def __init __（自我，模型，工具，系统=“”）：self.System =系统图形= stategraph（agentState）graph.add_node（“ llm”，self.call_openai）graph.add_node（“ action”，self.take_action）graph.add_conditional_edges（“ llm”，self.exists_action，{true：“ action”，false：end：end}）graph.add_edge（“ action”，“ llm”）graph.set_entry_point（“ llm”）self.graph = graph.compile（）self.tools = {t.name：t in tools} t in tools}self.model = model.bind_tools（工具）def call_openai（自我，状态：代理商）：消息=状态['消息']如果Self.System：消息= [SystemMessage（content = self.system）] +消息消息= self.model.invoke（消息）返回{'消息'：[消息]}def存在_action（self，state：agentState）：结果=状态['消息'] [ -  1]返回len（result.tool_calls）> 0def take_action（自我，状态：代理商）：tool_calls = state ['消息'] [ -  1] .tool_calls结果= []对于tool_calls中的t：打印（f“呼叫：{t}”）结果= self.tools [t ['name']。results.append（toolmessage（tool_call_id = t ['id']，name = t ['name']，content = str（resuly））））））打印（“回到模型！”）返回{'消息'：结果}

增加持久性

为了增加持久性，我们将使用langgraph的Checkpointer特征。一个检查器在每个节点之间和之间保存代理的状态。对于本教程，我们将使用sqliteSaver，一个利用SQLite（内置数据库）的简单检查器。虽然我们将使用内存数据库为简单起见，但您可以轻松地将其连接到外部数据库，或使用其他检查点（例如Redis或Postgres），以实现更稳定的持久性。

来自langgraph.checkpoint.sqlite导入sqlitesaver导入sqlite3sqlite_conn = sqlite3.connect（“ checkpoints.sqlite”，check_same_thread = false）内存= sqlitesaver（sqlite_conn）

接下来，我们将修改我们的代理人接受Checkpointer：

班级代理：def __init __（自我，模型，工具，checkpointer，system =“”）：＃其他一切与以前保持不变self.graph = graph.compile（checkpointer = checkpointer）＃之后的其他一切保持不变

现在，我们可以启用持久性创建代理：

提示=“”您是一名智能研究助手。使用搜索引擎查找信息。您可以拨打多个呼叫（一起或顺序）。\ \仅当您确定想要的内容时才查找信息。\ \如果您需要在提出后续问题之前查找一些信息，则可以这样做！”“”model = chatgroq（model =“ Llama-3.3-70b-specdec”）bot = agent（模型，[工具]，系统=提示，checkpointer =内存）

添加流

流对实时更新至关重要。我们将关注两种类型的流媒体：

1。流消息：发射中间消息，例如AI决策和工具结果。

2。流式令牌：从LLM的响应中流式传输单个令牌。
让我们从流媒体消息开始。我们将创建人类信息并使用溪流实时观察代理动作的方法。消息= [humanMessage（content =“德克萨斯州的天气是什么？”]]]

thread = {“配置”：{“ thread_id”：“ 1”}}对于bot.graph.stream中的事件（{“ messages”：message}，线程）：对于event.values（）的v：打印（v ['消息']）

最终输出：德克萨斯州目前的天气晴天，温度为19.4°C（66.9°F），风速为4.3 mph（6.8 kph）。

运行此操作时，您会看到结果流。首先，一条AI消息指示代理商致电Tavily，然后是带有搜索结果的工具消息，最后是一个AI消息回答问题。

了解线程ID

这thread_id是线程配置的关键部分。它允许代理商与不同的用户或上下文保持单独的对话。通过为每个对话分配一个唯一的线程_ID，代理可以同时跟踪多个交互，而无需混合它们。

例如，让我们通过询问la中继续对话吗？

消息= [humanMessage（content =“ la in La中怎么办？”）]]thread = {“配置”：{“ thread_id”：“ 1”}}对于bot.graph.stream中的事件（{“ messages”：message}，线程）：对于event.values（）的v：打印（v）

最终输出：洛杉矶目前的天气晴天，温度为17.2°C（63.0°F），风速为2.2 mph（3.6 kph）。

由于持久性，代理人会进一步询问天气。验证，让我们问，哪个是温暖的？

消息= [humanMessage（content =“哪一个温暖？”）]]thread = {“配置”：{“ thread_id”：“ 1”}}对于bot.graph.stream中的事件（{“ messages”：message}，线程）：对于event.values（）的v：打印（v）

最终产出：德克萨斯州比洛杉矶温暖。德克萨斯州的当前温度为19.4°C（66.9°F），而洛杉矶的当前温度为17.2°C（63.0°F）

代理商正确比较了德克萨斯州和洛杉矶的天气。要测试持续性是否保持对话是否分开thread_id：

消息= [humanMessage（content =“哪一个温暖？”）]]thread = {“配置”：{“ thread_id”：“ 2”}}对于bot.graph.stream中的事件（{“ messages”：message}，线程）：对于event.values（）的v：打印（v）

输出：我需要更多信息来回答该问题。您能提供更多上下文或指定您要比较哪两件事？

这次，代理人感到困惑，因为它无法访问以前的对话历史。

流式令牌

要流式令牌，我们将使用astream_events方法，这是异步的。我们还将切换到异步检查器。

来自langgraph.checkpoint.sqlite.aio导入asyncsqlitesaverasync with asyncsqlitesaver.from_conn_string（“：memory：”）作为checkpointer：abot = agent（模型，[工具]，系统=提示，checkpointer = checkpointer）消息= [humanMessage（content =“ sf中的天气是什么？”）]]thread = {“配置”：{“ thread_id”：“ 4”}}abot.graph.astream_events中的事件的异步kint = event [“ event”]如果bink ==“ on_chat_model_stream”：content = event [“ data”] [“块”]。内容如果内容：＃在Openai的上下文中的空内容＃该模型要求调用工具。＃因此，我们只打印非空内容打印（内容，end =“ |”）

这将实时流式传输，使您可以实时了解代理商的思维过程。

结论

通过添加持久性和流媒体，我们大大提高了AI代理的能力。持久性使代理可以在跨交互作用上维护上下文，而流提供了对其行为的实时见解。这些功能对于构建适合生产的应用程序至关重要，尤其是涉及多个用户或人类互动的应用程序。

在下一个教程中，我们将潜入人类互动，持久性在实现人类与AI代理之间的无缝合作中起着至关重要的作用。敬请关注！

参考：

（deeplearning.ai）https://learn.deeplearning.ai/courses/ai-agents-in-langgraph

另外，不要忘记跟随我们 叽叽喳喳并加入我们的 电报频道和 LinkedIn GrOUP。不要忘记加入我们的 75K+ ml子雷迪特。

ð -满足Intellagent：一个开源多代理框架，用于评估复杂的对话AI系统 ^（晋升）

Vineet Kumar是Marktechpost的咨询实习生。他目前正在坎普尔印度理工学院（IIT）攻读学士学位。他是一个机器学习爱好者。他对研究，计算机视觉和相关领域的最新进步充满热情。

关于《使用langgraph创建基于AI代理的系统：添加持久性和流（逐步指南）》的评论

暂无评论

发表评论

摘要

将持久性和流式功能添加到AI代理中可以显着提高其功能和可用性。以下是如何实现这些功能的详细细分：###持久性持久性可确保AI通过存储历史数据来维持不同交互的环境。当涉及多个用户或上下文时，这对于保持对话连续性至关重要。####实施步骤：1。**数据库设置**：使用SQLITE，PostgreSQL或任何其他数据库系统来存储交互历史记录。2。** CheckPointer配置**： - 初始化与数据库交互的检查台。 - 确保每个线程（对话）具有唯一的“线程_id”。``python来自langgraph.checkpoint.sqlite.aio导入asyncsqlitesaver＃初始化异步sqlite储蓄器以持续异步DEF INIT_CHECKPOINTER（）：返回等待asyncsqlitesaver.from_conn_string（“：memory：”）checkpointer = asyncio.run（init_checkpointer（））````````3。**存储和检索数据**： - 存储消息，上下文和结果中的数据库。 - 基于`thread_id'的数据检索数据以维护对话历史记录。``python＃存储消息的示例async def store_message（checkpointer，thread_id，消息）：等待checkpointer.store_event（{“事件”：“ on_chat_model”，“ data”：{“ thread_id”：thread_id，“ message”：message}}，，版本=“ V1”）＃检索特定线程消息的示例异步def retrieve_messages（checkpointer，thread_id）：返回等待checkpointer.retrieve_events（{“ filter”：{“ event”：“ on_chat_model”，“ data.thread_id”：thread_id}}）````````###流流媒体允许与AI代理的实时互动，从而提供有关其思维过程和行动发生时的见解。####实施步骤：1。**异步流动事件**： - 使用`satream_events'方法进行事件的异步流。``pythonASYNC DEF stream_chat（checkpointer，消息，thread_id）：Abot.graph.astream_events（{“ Message”：Message}，thread =“ configurable”，thread_id = thread_id）中的事件的异步kint = event [“ event”]如果bink ==“ on_chat_model_stream”：content = event [“ data”] [“块”]。内容打印（内容，end =“ |”）````````2。**令牌流**： - 流式传输令牌以实时观察AI的思维过程。``python＃用于特定对话的流式令牌的示例async def stream_tokens（checkpointer，消息，thread_id）：Abot.graph.astream_events（{“ Message”：Message}，thread =“ configurable”，thread_id = thread_id，版本=“ v1”）中的事件的异步kint = event [“ event”]如果bink ==“ on_chat_model_stream”：content = event [“ data”] [“块”]。内容打印（内容，end =“ |”）````````###示例用法让我们将所有内容放在一个示例中，在该示例中，AI代理保持上下文并流式令牌。``python导入异步来自langgraph.checkpoint.sqlite.aio导入asyncsqlitesaver＃初始化CheckPointercheckpointer = asyncio.run（asyncsqlitesaver.from_conn_string（“：memory：“）））异步def main（）：＃通过启用持久性初始化代理abot = agent（模型，[工具]，系统=提示，checkpointer = checkpointer）＃第一消息消息1 = [humanMessage（content =“ sf中的天气是什么？”）]]等待stream_chat（checkpointer，messages1，thread_id =“ 4”）＃带有不同上下文的第二个消息messages2 = [humanMessage（content =“哪一个温暖？”）]]等待stream_tokens（checkpointer，messages2，thread_id =“ 5”）＃优异运行主要功能asyncio.run（main（））````````＃＃＃结论通过实施持久性和流式功能，您可以显着增强AI代理的功能。持久性确保对话在多个交互之间进行维持，而流媒体允许与代理商的思维过程进行实时互动。这种设置对于构建强大的对话系统至关重要，在该系统中维护上下文和观察实时行为是必不可少的特征。