https://academy.langchain.com/courses/intro-to-langgraphhttps://github.com/shangxiang0907/langchain-academy文章目录Chatbot with message summarization external DB memory 具备消息摘要与外部数据库记忆的聊天机器人Review 回顾Goals 目标SqlitePersisting state 持久化状态Studio主要讲了什么能解决什么问题实际开发有什么用一、整体运行流程二、代码里的核心部分1. 使用 SQLite 保存状态2. 使用 thread_id 区分会话3. 把旧消息压缩成摘要4. 删除旧消息三、它解决了什么问题1. 程序重启后记忆丢失2. 多轮对话越来越长3. 支持多个独立会话4. Graph 执行被中断后可以恢复四、实际开发有什么用客服机器人AI 助手需要人工审批的 Agent长时间运行的工作流五、它和真正的“长期记忆”有什么区别六、实际项目中的选择Chatbot with message summarization external DB memory 具备消息摘要与外部数据库记忆的聊天机器人Review 回顾We’ve covered how to customize graph state schema and reducer.我们已介绍如何自定义图状态模式state schema与规约器reducer。We’ve also shown a number of tricks for trimming or filtering messages in graph state.我们还展示了多种用于修剪或过滤图状态中消息的技巧。We’ve used these concepts in a Chatbot with memory that produces a running summary of the conversation.我们已在具备记忆功能的聊天机器人中应用了这些概念该机器人可生成对话的持续摘要。Goals 目标But, what if we want our Chatbot to have memory that persists indefinitely?但如果我们希望聊天机器人拥有永久持久化的记忆该怎么办Now, we’ll introduce some more advanced checkpointers that support external databases.接下来我们将介绍一些更高级的支持外部数据库的检查点器checkpointer。Here, we’ll show how to use Sqlite as a checkpointer, but other checkpointers, such as Postgres are available!此处我们将演示如何使用 Sqlite 作为检查点器但其他检查点器例如 Postgres也可用%%capture--no-stderr%pip install--quiet-U langgraph-checkpoint-sqlite langchain_core langgraph langchain_openaiimportos,getpassdef_set_env(var:str):ifnotos.environ.get(var):os.environ[var]getpass.getpass(f{var}: )fromdotenvimportfind_dotenv,load_dotenv load_dotenv(find_dotenv(usecwdTrue))_set_env(OPENAI_API_KEY)SqliteA good starting point here is the SqliteSaver checkpointer.一个良好的起点是 SqliteSaver 检查点器。Sqlite is a small, fast, highly popular SQL database.Sqlite 是一种 轻量、快速且广受欢迎 的 SQL 数据库。If we supply:memory:it creates an in-memory Sqlite database.若传入:memory:它将创建一个内存中的 Sqlite 数据库。importsqlite3# In memoryconnsqlite3.connect(:memory:,check_same_threadFalse)But, if we supply a db path, then it will create a database for us!但若提供数据库路径则会自动为我们创建一个数据库# pull file if it doesnt exist and connect to local db!mkdir-p state_db[!-f state_db/example.db]wget-P state_db https://github.com/langchain-ai/langchain-academy/raw/main/module-2/state_db/example.db db_pathstate_db/example.dbconnsqlite3.connect(db_path,check_same_threadFalse)# Here is our checkpointerfromlanggraph.checkpoint.sqliteimportSqliteSaver memorySqliteSaver(conn)Let’s re-define our chatbot.让我们重新定义聊天机器人。fromtyping_extensionsimportLiteralimportosfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportSystemMessage,HumanMessage,RemoveMessagefromlanggraph.graphimportENDfromlanggraph.graphimportMessagesState modelChatOpenAI(modelos.getenv(OPENAI_MODEL,qwen-plus),base_urlos.getenv(OPENAI_BASE_URL,https://dashscope.aliyuncs.com/compatible-mode/v1),temperature0)classState(MessagesState):summary:str# Define the logic to call the modeldefcall_model(state:State):# Get summary if it existssummarystate.get(summary,)# If there is summary, then we add itifsummary:# Add summary to system messagesystem_messagefSummary of conversation earlier:{summary}# Append summary to any newer messagesmessages[SystemMessage(contentsystem_message)]state[messages]else:messagesstate[messages]responsemodel.invoke(messages)return{messages:response}defsummarize_conversation(state:State):# First, we get any existing summarysummarystate.get(summary,)# Create our summarization promptifsummary:# A summary already existssummary_message(fThis is summary of the conversation to date:{summary}\n\nExtend the summary by taking into account the new messages above:)else:summary_messageCreate a summary of the conversation above:# Add prompt to our historymessagesstate[messages][HumanMessage(contentsummary_message)]responsemodel.invoke(messages)# Delete all but the 2 most recent messagesdelete_messages[RemoveMessage(idm.id)forminstate[messages][:-2]]return{summary:response.content,messages:delete_messages}# Determine whether to end or summarize the conversationdefshould_continue(state:State)-Literal[summarize_conversation,END]:Return the next node to execute.messagesstate[messages]# If there are more than six messages, then we summarize the conversationiflen(messages)6:returnsummarize_conversation# Otherwise we can just endreturnENDNow, we just re-compile with our sqlite checkpointer.现在只需使用我们的 sqlite 检查点器重新编译即可。fromIPython.displayimportImage,displayfromlanggraph.graphimportStateGraph,START# Define a new graphworkflowStateGraph(State)workflow.add_node(conversation,call_model)workflow.add_node(summarize_conversation)# Set the entrypoint as conversationworkflow.add_edge(START,conversation)workflow.add_conditional_edges(conversation,should_continue)workflow.add_edge(summarize_conversation,END)# Compilegraphworkflow.compile(checkpointermemory)display(Image(graph.get_graph().draw_mermaid_png()))Now, we can invoke the graph several times.现在我们可以多次调用该图。# Create a threadconfig{configurable:{thread_id:1}}# Start conversationinput_messageHumanMessage(contenthi! Im Lance)outputgraph.invoke({messages:[input_message]},config)forminoutput[messages][-1:]:m.pretty_print()input_messageHumanMessage(contentwhats my name?)outputgraph.invoke({messages:[input_message]},config)forminoutput[messages][-1:]:m.pretty_print()input_messageHumanMessage(contenti like the 49ers!)outputgraph.invoke({messages:[input_message]},config)forminoutput[messages][-1:]:m.pretty_print()Let’s confirm that our state is saved locally.让我们确认状态已本地保存。config{configurable:{thread_id:1}}graph_stategraph.get_state(config)graph_statePersisting state 持久化状态Using database like Sqlite means state is persisted!使用 Sqlite 等数据库意味着状态将被持久化For example, we can re-start the notebook kernel and see that we can still load from Sqlite DB on disk.例如我们可以重启笔记本内核并仍能从磁盘上的 Sqlite 数据库加载状态。# Create a threadconfig{configurable:{thread_id:1}}graph_stategraph.get_state(config)graph_stateStudio⚠️ Notice⚠️ 注意Since filming these videos, we’ve updated Studio so that it can now be run locally and accessed through your browser.自录制这些视频以来我们已更新 Studio使其现在可本地运行并通过浏览器访问。This is the preferred way to run Studio instead of using the Desktop App shown in the video.这是运行 Studio 的首选方式而非视频中展示的桌面应用程序。It is now calledLangSmith Studioinstead ofLangGraph Studio.它现在被称为LangSmith Studio而非LangGraph Studio。Detailed setup instructions are available in the “Getting Setup” guide at the start of the course.详细的安装说明请参阅本课程开头的“入门设置”指南。You can find a description of Studio here, and specific details for local deployment here.您可在此处查看 Studio 的说明链接以及本地部署的具体细节链接。To start the local development server, run the following command in your terminal in the/studiodirectory in this module:要在本地启动开发服务器请在本模块的/studio目录下于终端中运行以下命令langgraph devYou should see the following output:您应看到如下输出- API: http://127.0.0.1:2024 - Studio UI: https://smith.langchain.com/studio/?baseUrlhttp://127.0.0.1:2024 - API Docs: http://127.0.0.1:2024/docsOpen your browser and navigate to theStudio UIURL shown above.打开您的浏览器并导航至上方显示的Studio UIURL。Load thechatbotin Studio, which usesmodule-2/studio/chatbot.pyset inmodule-2/studio/langgraph.json.在 Studio 中加载chatbot其对应module-2/studio/chatbot.py文件并由module-2/studio/langgraph.json中指定。主要讲了什么能解决什么问题实际开发有什么用这篇主要讲的是如何让 LangGraph 聊天机器人把对话状态保存到 SQLite 数据库并通过“历史摘要”控制上下文长度。它把两种机制结合起来Checkpointer 持久化把消息、摘要等 Graph State 保存到外部数据库。消息摘要与删除消息太多时将旧对话压缩成摘要只保留最近几条原始消息。一、整体运行流程否是用户发送消息根据 thread_id 加载历史状态摘要 最近消息交给 LLM生成回复消息是否超过 6 条保存状态并结束更新历史摘要删除旧消息只保留最近 2 条每次调用graph.invoke()时LangGraph 都会根据thread_id找到对应会话从 SQLite 中恢复之前的messages和summary执行本轮对话将更新后的状态重新写入 SQLite。因此即使程序或 Notebook 重启只要数据库文件还在对话状态就能恢复。二、代码里的核心部分1. 使用 SQLite 保存状态connsqlite3.connect(db_path,check_same_threadFalse)memorySqliteSaver(conn)graphworkflow.compile(checkpointermemory)这里的SqliteSaver是 LangGraph 的检查点器。它会自动保存 Graph State包括classState(MessagesState):summary:str也就是messages近期对话消息summary较早对话的摘要Graph 当前执行到的位置每个thread_id对应的状态版本。需要注意sqlite3.connect(:memory:)只存在于当前进程内程序退出后就消失适合测试。而sqlite3.connect(state_db/example.db)会写入磁盘重启程序后仍然存在才是真正的本地持久化。2. 使用thread_id区分会话config{configurable:{thread_id:1}}thread_id类似聊天系统中的conversation_id。例如thread_id user_001_chat_001 thread_id user_001_chat_002 thread_id user_002_chat_001不同thread_id拥有不同的对话状态。相同thread_id再次调用时会接着之前的状态继续运行。3. 把旧消息压缩成摘要当消息超过 6 条iflen(messages)6:returnsummarize_conversation系统会让模型总结当前对话responsemodel.invoke(messages)如果之前已经有摘要就不是重新总结全部内容而是在旧摘要基础上追加新信息Extend the summary by taking into account the new messages above:这叫“滚动摘要”或“持续摘要”。4. 删除旧消息摘要生成以后delete_messages[RemoveMessage(idm.id)forminstate[messages][:-2]]它会删除除最近两条以外的旧消息。最终状态大致是summary: 用户名叫 Lance喜欢 49ers之前询问过…… messages: 最近一条用户消息 最近一条 AI 回复下次调用模型时再把二者组合messages[SystemMessage(contentfSummary of conversation earlier:{summary})]state[messages]这样模型既知道较早发生过什么又不需要读取全部历史消息。三、它解决了什么问题1. 程序重启后记忆丢失如果只把状态放在 Python 变量或内存中sqlite3.connect(:memory:)进程退出以后对话就会消失。换成磁盘 SQLite 后程序重启仍然可以通过同一个thread_id找回状态。2. 多轮对话越来越长如果每次都把所有历史消息发送给模型会产生Token 消耗越来越大请求越来越慢成本越来越高最终超过模型上下文窗口大量无关历史可能干扰模型回答。通过“摘要 最近消息”上下文长度可以保持在相对稳定的范围。3. 支持多个独立会话借助thread_id同一套 Graph 可以同时管理很多对话用户 A 的会话 → thread_A 用户 B 的会话 → thread_B 用户 A 的另一个窗口 → thread_A_2不会把不同用户或不同聊天窗口的消息混在一起。4. Graph 执行被中断后可以恢复Checkpointer 保存的不只是聊天内容还保存 LangGraph 的执行状态。这对复杂 Agent 很重要。例如 Agent 执行到“等待人工批准”时暂停之后可以从检查点继续而不是重新执行整个流程。四、实际开发有什么用客服机器人机器人可以记住当前工单的历史信息用户报告过什么问题已经尝试过哪些处理方式客服承诺了什么当前工单进展到哪里。即使服务重启也可以恢复这个会话。AI 助手例如你的 JobCopilot可以按用户和任务保存状态thread_id 用户ID 求职任务ID它可以记住用户的目标岗位不考虑哪些职位简历已经修改到哪个版本哪些职位已经分析或投递当前求职工作流执行到哪一步。历史内容过长时可以把旧过程压缩为用户目标是新加坡 AI Platform 岗位需要 EP 不考虑自动驾驶感知岗位 已经分析 15 个职位其中 4 个适合申请……需要人工审批的 Agent例如自动投递职位搜索职位分析匹配度生成申请材料暂停并等待用户批准恢复状态执行投递状态保存到数据库后即使审批发生在几小时或几天后也能继续原来的工作流。长时间运行的工作流适合深度研究 Agent代码生成与测试 Agent数据分析工作流多步骤审批流程需要失败重试的任务多 Agent 协作流程。任务失败或服务重启后可以从检查点恢复减少重复调用模型和外部 API。五、它和真正的“长期记忆”有什么区别这篇虽然称为“外部数据库记忆”但严格来说主要讲的是把某个会话的 Graph State 持久化。它更接近“会话记忆”不完全等于跨会话的用户长期记忆。能力Checkpointer真正的长期记忆保存当前会话状态是可以根据thread_id恢复聊天是可以记住用户长期偏好需要额外设计是跨多个聊天共享记忆默认不行是按语义搜索相关记忆默认不行通常支持典型存储SQLite/PostgresStore、Postgres、向量数据库例如用户在thread_id1中说“我喜欢 Python”新建thread_id2后默认不会自动知道这件事。如果希望跨会话记住用户偏好需要额外设计user_id → 长期记忆 thread_id → 当前会话状态所以实际项目通常会同时使用Checkpointer保存某个工作流或会话的执行状态长期记忆 Store保存用户偏好、事实和历史经验业务数据库保存用户、职位、简历、申请记录等正式数据。六、实际项目中的选择SQLite 适合学习和实验单机应用本地开发个人项目并发量比较小的服务。生产环境、多实例部署或高并发应用一般更适合 PostgreSQL因为多个服务实例需要共享同一份状态而且 SQLite 的并发写入能力有限。一句话总结这篇教你用thread_id SQLite Checkpointer保存每个会话的 LangGraph 状态并通过“滚动摘要 删除旧消息”控制上下文长度从而实现可重启、可恢复、成本可控的多轮聊天或 Agent 工作流。
https://academy.langchain.com/courses/intro-to-langgraphhttps://github.com/shangxiang0907/langchain-academy文章目录Chatbot with message summarization external DB memory 具备消息摘要与外部数据库记忆的聊天机器人Review 回顾Goals 目标SqlitePersisting state 持久化状态Studio主要讲了什么能解决什么问题实际开发有什么用一、整体运行流程二、代码里的核心部分1. 使用 SQLite 保存状态2. 使用 thread_id 区分会话3. 把旧消息压缩成摘要4. 删除旧消息三、它解决了什么问题1. 程序重启后记忆丢失2. 多轮对话越来越长3. 支持多个独立会话4. Graph 执行被中断后可以恢复四、实际开发有什么用客服机器人AI 助手需要人工审批的 Agent长时间运行的工作流五、它和真正的“长期记忆”有什么区别六、实际项目中的选择Chatbot with message summarization external DB memory 具备消息摘要与外部数据库记忆的聊天机器人Review 回顾We’ve covered how to customize graph state schema and reducer.我们已介绍如何自定义图状态模式state schema与规约器reducer。We’ve also shown a number of tricks for trimming or filtering messages in graph state.我们还展示了多种用于修剪或过滤图状态中消息的技巧。We’ve used these concepts in a Chatbot with memory that produces a running summary of the conversation.我们已在具备记忆功能的聊天机器人中应用了这些概念该机器人可生成对话的持续摘要。Goals 目标But, what if we want our Chatbot to have memory that persists indefinitely?但如果我们希望聊天机器人拥有永久持久化的记忆该怎么办Now, we’ll introduce some more advanced checkpointers that support external databases.接下来我们将介绍一些更高级的支持外部数据库的检查点器checkpointer。Here, we’ll show how to use Sqlite as a checkpointer, but other checkpointers, such as Postgres are available!此处我们将演示如何使用 Sqlite 作为检查点器但其他检查点器例如 Postgres也可用%%capture--no-stderr%pip install--quiet-U langgraph-checkpoint-sqlite langchain_core langgraph langchain_openaiimportos,getpassdef_set_env(var:str):ifnotos.environ.get(var):os.environ[var]getpass.getpass(f{var}: )fromdotenvimportfind_dotenv,load_dotenv load_dotenv(find_dotenv(usecwdTrue))_set_env(OPENAI_API_KEY)SqliteA good starting point here is the SqliteSaver checkpointer.一个良好的起点是 SqliteSaver 检查点器。Sqlite is a small, fast, highly popular SQL database.Sqlite 是一种 轻量、快速且广受欢迎 的 SQL 数据库。If we supply:memory:it creates an in-memory Sqlite database.若传入:memory:它将创建一个内存中的 Sqlite 数据库。importsqlite3# In memoryconnsqlite3.connect(:memory:,check_same_threadFalse)But, if we supply a db path, then it will create a database for us!但若提供数据库路径则会自动为我们创建一个数据库# pull file if it doesnt exist and connect to local db!mkdir-p state_db[!-f state_db/example.db]wget-P state_db https://github.com/langchain-ai/langchain-academy/raw/main/module-2/state_db/example.db db_pathstate_db/example.dbconnsqlite3.connect(db_path,check_same_threadFalse)# Here is our checkpointerfromlanggraph.checkpoint.sqliteimportSqliteSaver memorySqliteSaver(conn)Let’s re-define our chatbot.让我们重新定义聊天机器人。fromtyping_extensionsimportLiteralimportosfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportSystemMessage,HumanMessage,RemoveMessagefromlanggraph.graphimportENDfromlanggraph.graphimportMessagesState modelChatOpenAI(modelos.getenv(OPENAI_MODEL,qwen-plus),base_urlos.getenv(OPENAI_BASE_URL,https://dashscope.aliyuncs.com/compatible-mode/v1),temperature0)classState(MessagesState):summary:str# Define the logic to call the modeldefcall_model(state:State):# Get summary if it existssummarystate.get(summary,)# If there is summary, then we add itifsummary:# Add summary to system messagesystem_messagefSummary of conversation earlier:{summary}# Append summary to any newer messagesmessages[SystemMessage(contentsystem_message)]state[messages]else:messagesstate[messages]responsemodel.invoke(messages)return{messages:response}defsummarize_conversation(state:State):# First, we get any existing summarysummarystate.get(summary,)# Create our summarization promptifsummary:# A summary already existssummary_message(fThis is summary of the conversation to date:{summary}\n\nExtend the summary by taking into account the new messages above:)else:summary_messageCreate a summary of the conversation above:# Add prompt to our historymessagesstate[messages][HumanMessage(contentsummary_message)]responsemodel.invoke(messages)# Delete all but the 2 most recent messagesdelete_messages[RemoveMessage(idm.id)forminstate[messages][:-2]]return{summary:response.content,messages:delete_messages}# Determine whether to end or summarize the conversationdefshould_continue(state:State)-Literal[summarize_conversation,END]:Return the next node to execute.messagesstate[messages]# If there are more than six messages, then we summarize the conversationiflen(messages)6:returnsummarize_conversation# Otherwise we can just endreturnENDNow, we just re-compile with our sqlite checkpointer.现在只需使用我们的 sqlite 检查点器重新编译即可。fromIPython.displayimportImage,displayfromlanggraph.graphimportStateGraph,START# Define a new graphworkflowStateGraph(State)workflow.add_node(conversation,call_model)workflow.add_node(summarize_conversation)# Set the entrypoint as conversationworkflow.add_edge(START,conversation)workflow.add_conditional_edges(conversation,should_continue)workflow.add_edge(summarize_conversation,END)# Compilegraphworkflow.compile(checkpointermemory)display(Image(graph.get_graph().draw_mermaid_png()))Now, we can invoke the graph several times.现在我们可以多次调用该图。# Create a threadconfig{configurable:{thread_id:1}}# Start conversationinput_messageHumanMessage(contenthi! Im Lance)outputgraph.invoke({messages:[input_message]},config)forminoutput[messages][-1:]:m.pretty_print()input_messageHumanMessage(contentwhats my name?)outputgraph.invoke({messages:[input_message]},config)forminoutput[messages][-1:]:m.pretty_print()input_messageHumanMessage(contenti like the 49ers!)outputgraph.invoke({messages:[input_message]},config)forminoutput[messages][-1:]:m.pretty_print()Let’s confirm that our state is saved locally.让我们确认状态已本地保存。config{configurable:{thread_id:1}}graph_stategraph.get_state(config)graph_statePersisting state 持久化状态Using database like Sqlite means state is persisted!使用 Sqlite 等数据库意味着状态将被持久化For example, we can re-start the notebook kernel and see that we can still load from Sqlite DB on disk.例如我们可以重启笔记本内核并仍能从磁盘上的 Sqlite 数据库加载状态。# Create a threadconfig{configurable:{thread_id:1}}graph_stategraph.get_state(config)graph_stateStudio⚠️ Notice⚠️ 注意Since filming these videos, we’ve updated Studio so that it can now be run locally and accessed through your browser.自录制这些视频以来我们已更新 Studio使其现在可本地运行并通过浏览器访问。This is the preferred way to run Studio instead of using the Desktop App shown in the video.这是运行 Studio 的首选方式而非视频中展示的桌面应用程序。It is now calledLangSmith Studioinstead ofLangGraph Studio.它现在被称为LangSmith Studio而非LangGraph Studio。Detailed setup instructions are available in the “Getting Setup” guide at the start of the course.详细的安装说明请参阅本课程开头的“入门设置”指南。You can find a description of Studio here, and specific details for local deployment here.您可在此处查看 Studio 的说明链接以及本地部署的具体细节链接。To start the local development server, run the following command in your terminal in the/studiodirectory in this module:要在本地启动开发服务器请在本模块的/studio目录下于终端中运行以下命令langgraph devYou should see the following output:您应看到如下输出- API: http://127.0.0.1:2024 - Studio UI: https://smith.langchain.com/studio/?baseUrlhttp://127.0.0.1:2024 - API Docs: http://127.0.0.1:2024/docsOpen your browser and navigate to theStudio UIURL shown above.打开您的浏览器并导航至上方显示的Studio UIURL。Load thechatbotin Studio, which usesmodule-2/studio/chatbot.pyset inmodule-2/studio/langgraph.json.在 Studio 中加载chatbot其对应module-2/studio/chatbot.py文件并由module-2/studio/langgraph.json中指定。主要讲了什么能解决什么问题实际开发有什么用这篇主要讲的是如何让 LangGraph 聊天机器人把对话状态保存到 SQLite 数据库并通过“历史摘要”控制上下文长度。它把两种机制结合起来Checkpointer 持久化把消息、摘要等 Graph State 保存到外部数据库。消息摘要与删除消息太多时将旧对话压缩成摘要只保留最近几条原始消息。一、整体运行流程否是用户发送消息根据 thread_id 加载历史状态摘要 最近消息交给 LLM生成回复消息是否超过 6 条保存状态并结束更新历史摘要删除旧消息只保留最近 2 条每次调用graph.invoke()时LangGraph 都会根据thread_id找到对应会话从 SQLite 中恢复之前的messages和summary执行本轮对话将更新后的状态重新写入 SQLite。因此即使程序或 Notebook 重启只要数据库文件还在对话状态就能恢复。二、代码里的核心部分1. 使用 SQLite 保存状态connsqlite3.connect(db_path,check_same_threadFalse)memorySqliteSaver(conn)graphworkflow.compile(checkpointermemory)这里的SqliteSaver是 LangGraph 的检查点器。它会自动保存 Graph State包括classState(MessagesState):summary:str也就是messages近期对话消息summary较早对话的摘要Graph 当前执行到的位置每个thread_id对应的状态版本。需要注意sqlite3.connect(:memory:)只存在于当前进程内程序退出后就消失适合测试。而sqlite3.connect(state_db/example.db)会写入磁盘重启程序后仍然存在才是真正的本地持久化。2. 使用thread_id区分会话config{configurable:{thread_id:1}}thread_id类似聊天系统中的conversation_id。例如thread_id user_001_chat_001 thread_id user_001_chat_002 thread_id user_002_chat_001不同thread_id拥有不同的对话状态。相同thread_id再次调用时会接着之前的状态继续运行。3. 把旧消息压缩成摘要当消息超过 6 条iflen(messages)6:returnsummarize_conversation系统会让模型总结当前对话responsemodel.invoke(messages)如果之前已经有摘要就不是重新总结全部内容而是在旧摘要基础上追加新信息Extend the summary by taking into account the new messages above:这叫“滚动摘要”或“持续摘要”。4. 删除旧消息摘要生成以后delete_messages[RemoveMessage(idm.id)forminstate[messages][:-2]]它会删除除最近两条以外的旧消息。最终状态大致是summary: 用户名叫 Lance喜欢 49ers之前询问过…… messages: 最近一条用户消息 最近一条 AI 回复下次调用模型时再把二者组合messages[SystemMessage(contentfSummary of conversation earlier:{summary})]state[messages]这样模型既知道较早发生过什么又不需要读取全部历史消息。三、它解决了什么问题1. 程序重启后记忆丢失如果只把状态放在 Python 变量或内存中sqlite3.connect(:memory:)进程退出以后对话就会消失。换成磁盘 SQLite 后程序重启仍然可以通过同一个thread_id找回状态。2. 多轮对话越来越长如果每次都把所有历史消息发送给模型会产生Token 消耗越来越大请求越来越慢成本越来越高最终超过模型上下文窗口大量无关历史可能干扰模型回答。通过“摘要 最近消息”上下文长度可以保持在相对稳定的范围。3. 支持多个独立会话借助thread_id同一套 Graph 可以同时管理很多对话用户 A 的会话 → thread_A 用户 B 的会话 → thread_B 用户 A 的另一个窗口 → thread_A_2不会把不同用户或不同聊天窗口的消息混在一起。4. Graph 执行被中断后可以恢复Checkpointer 保存的不只是聊天内容还保存 LangGraph 的执行状态。这对复杂 Agent 很重要。例如 Agent 执行到“等待人工批准”时暂停之后可以从检查点继续而不是重新执行整个流程。四、实际开发有什么用客服机器人机器人可以记住当前工单的历史信息用户报告过什么问题已经尝试过哪些处理方式客服承诺了什么当前工单进展到哪里。即使服务重启也可以恢复这个会话。AI 助手例如你的 JobCopilot可以按用户和任务保存状态thread_id 用户ID 求职任务ID它可以记住用户的目标岗位不考虑哪些职位简历已经修改到哪个版本哪些职位已经分析或投递当前求职工作流执行到哪一步。历史内容过长时可以把旧过程压缩为用户目标是新加坡 AI Platform 岗位需要 EP 不考虑自动驾驶感知岗位 已经分析 15 个职位其中 4 个适合申请……需要人工审批的 Agent例如自动投递职位搜索职位分析匹配度生成申请材料暂停并等待用户批准恢复状态执行投递状态保存到数据库后即使审批发生在几小时或几天后也能继续原来的工作流。长时间运行的工作流适合深度研究 Agent代码生成与测试 Agent数据分析工作流多步骤审批流程需要失败重试的任务多 Agent 协作流程。任务失败或服务重启后可以从检查点恢复减少重复调用模型和外部 API。五、它和真正的“长期记忆”有什么区别这篇虽然称为“外部数据库记忆”但严格来说主要讲的是把某个会话的 Graph State 持久化。它更接近“会话记忆”不完全等于跨会话的用户长期记忆。能力Checkpointer真正的长期记忆保存当前会话状态是可以根据thread_id恢复聊天是可以记住用户长期偏好需要额外设计是跨多个聊天共享记忆默认不行是按语义搜索相关记忆默认不行通常支持典型存储SQLite/PostgresStore、Postgres、向量数据库例如用户在thread_id1中说“我喜欢 Python”新建thread_id2后默认不会自动知道这件事。如果希望跨会话记住用户偏好需要额外设计user_id → 长期记忆 thread_id → 当前会话状态所以实际项目通常会同时使用Checkpointer保存某个工作流或会话的执行状态长期记忆 Store保存用户偏好、事实和历史经验业务数据库保存用户、职位、简历、申请记录等正式数据。六、实际项目中的选择SQLite 适合学习和实验单机应用本地开发个人项目并发量比较小的服务。生产环境、多实例部署或高并发应用一般更适合 PostgreSQL因为多个服务实例需要共享同一份状态而且 SQLite 的并发写入能力有限。一句话总结这篇教你用thread_id SQLite Checkpointer保存每个会话的 LangGraph 状态并通过“滚动摘要 删除旧消息”控制上下文长度从而实现可重启、可恢复、成本可控的多轮聊天或 Agent 工作流。
