使用langgraph创建基于AI代理的系统:添加持久性和流(逐步指南)
2025-02-02 06:55:13
将持久性和流式功能添加到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代理的功能。持久性确保对话在多个交互之间进行维持,而流媒体允许与代理商的思维过程进行实时互动。这种设置对于构建强大的对话系统至关重要,在该系统中维护上下文和观察实时行为是必不可少的特征。