纲要持久化层与记忆的关系持久化层保存和恢复图执行状态的机制记忆构建在持久化层之上的认知功能核心组件Checkpoint状态的快照MemorySaver短期内存记忆InMemoryStore跨线程共享存储MongoDBSaver长期持久化记忆线程隔离与跨线程调用基于thread_id的会话隔离基于user_id的跨线程数据共享短期记忆与长期记忆的实现内置create_react_agent快速集成自定义StateGraph手动控制记忆优化策略消息过滤限制窗口大小对话摘要自动压缩历史记录完整可运行示例环境准备短期记忆演示长期记忆MongoDB演示记忆优化对话摘要总结与相关度说明持久化层与记忆的关系在构建多轮对话智能体时最基础的需求就是“记住”之前说过的话。LangGraph 通过持久化层提供了这一能力。持久化层是一种保存和恢复图执行状态的机制它基于Checkpoint记录每一步的状态快照包括对话消息、工具调用结果等。记忆则是构建在持久化层之上的认知功能它定义了我们如何组织、检索和使用这些被存储的信息。可以简单理解为持久化层解决“数据存在哪里”的问题记忆解决“如何使用数据”的问题。LangGraph 支持多种持久化后端后端类型适用场景MemorySaver内存开发调试、短期会话InMemoryStore内存跨线程共享、用户维度记忆MongoDBSaver硬盘生产环境、长期持久化RedisSaver硬盘高并发、分布式部署线程隔离与跨线程调用LangGraph 的持久化层天然支持线程隔离——通过不同的thread_id每个对话会话拥有独立的状态空间。但有时我们需要在不同会话之间共享某些用户信息例如用户档案这可以通过user_id和InMemoryStore实现跨线程调用。InMemoryStore (user_iduser_001)线程2 (thread_id2)线程1 (thread_id1)InMemoryStore (user_iduser_001)线程2 (thread_id2)线程1 (thread_id1)存储记忆 名字Tommy查询用户信息返回 名字Tommy生成个性化回复实战短期记忆——一个带记忆的对话机器人以下示例使用MemorySaver激活短期记忆同一thread_id下可记住上下文。环境准备pipinstalllanggraph langchain-openai python-dotenv完整代码importosfromdotenvimportload_dotenvfromlanggraph.graphimportStateGraph,ENDfromlanggraph.checkpoint.memoryimportMemorySaverfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportHumanMessage,AIMessage load_dotenv()# 1. 定义 StateclassChatState(dict):messages:list# 2. 创建模型llmChatOpenAI(modelgpt-3.5-turbo,temperature0)# 3. 定义节点defchat_node(state:ChatState)-ChatState:responsellm.invoke(state[messages])return{messages:[response]}# 4. 构建图并激活 MemorySaverbuilderStateGraph(ChatState)builder.add_node(chat,chat_node)builder.set_entry_point(chat)builder.add_edge(chat,END)memoryMemorySaver()appbuilder.compile(checkpointermemory)# 5. 多轮对话测试config{configurable:{thread_id:session-001}}# 第一轮input1{messages:[HumanMessage(content我叫 Tommy)]}foreventinapp.stream(input1,config):forkey,valueinevent.items():ifmessagesinvalueandvalue[messages]:last_msgvalue[messages][-1]print(f助手:{last_msg.content})# 第二轮input2{messages:[HumanMessage(content我叫什么名字)]}foreventinapp.stream(input2,config):forkey,valueinevent.items():ifmessagesinvalueandvalue[messages]:last_msgvalue[messages][-1]print(f助手:{last_msg.content})运行后第二轮会正确回答“Tommy”。如果修改thread_id则会丢失记忆。实战长期记忆——使用 MongoDB 持久化将MemorySaver替换为MongoDBSaver记忆即可持久化到磁盘重启不丢失。安装 MongoDB 驱动pipinstallpymongo启动 MongoDBDockerdockerrun-d-p27017:27017--namemongodb mongo:7完整代码importosfromdotenvimportload_dotenvfromlanggraph.graphimportStateGraph,ENDfromlanggraph.checkpoint.mongodbimportMongoDBSaverfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportHumanMessage load_dotenv()# 初始化模型llmChatOpenAI(modelgpt-3.5-turbo,temperature0)# 定义 State 和节点classChatState(dict):messages:listdefchat_node(state:ChatState)-ChatState:responsellm.invoke(state[messages])return{messages:[response]}# 构建图使用 MongoDBSaverbuilderStateGraph(ChatState)builder.add_node(chat,chat_node)builder.set_entry_point(chat)builder.add_edge(chat,END)# 连接 MongoDB默认 localhost:27017checkpointerMongoDBSaver.from_conn_string(mongodb://localhost:27017/)appbuilder.compile(checkpointercheckpointer)# 多轮对话config{configurable:{thread_id:persistent-001}}input1{messages:[HumanMessage(content记住我最喜欢的语言是 Python)]}foreventinapp.stream(input1,config):forkey,valueinevent.items():ifmessagesinvalueandvalue[messages]:print(助手:,value[messages][-1].content)# 即使重启程序用同样的 thread_id 仍能回忆input2{messages:[HumanMessage(content我最喜欢的语言是什么)]}foreventinapp.stream(input2,config):forkey,valueinevent.items():ifmessagesinvalueandvalue[messages]:print(助手:,value[messages][-1].content)再次运行程序并传入相同的thread_id智能体仍能回答“Python”。记忆优化用对话摘要压缩上下文直接保存所有历史消息会导致上下文窗口迅速膨胀。一种成熟方案是对话摘要当消息数量超过阈值时自动将旧消息压缩为一段摘要同时删除原始消息在保持关键信息的同时控制 token 开销。实现思路在State中增加summary字段编写一个summarize节点当消息列表长度超过设定值如 6 条时触发。该节点调用 LLM 生成摘要并调用RemoveMessage删除旧消息。fromlanggraph.graphimportStateGraph,END,MessagesStatefromlanggraph.checkpoint.memoryimportMemorySaverfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportSystemMessage,HumanMessage,RemoveMessagefromlanggraph.graph.messageimportadd_messages# 自定义 State增加 summary 字段classState(MessagesState):summary:strllmChatOpenAI(modelgpt-3.5-turbo,temperature0)defcall_model(state:State)-State:messagesstate[messages]# 如果有摘要添加到系统提示中ifstate.get(summary):summary_msgSystemMessage(content历史对话摘要state[summary])messages[summary_msg]messages responsellm.invoke(messages)return{messages:[response]}defsummarize_conversation(state:State)-State:# 仅当消息超过 6 条时执行总结iflen(state[messages])6:returnstate# 提取前两条消息作为总结素材实际可取全部历史old_messagesstate[messages][:-2]summary_prompt请用一段话总结以下对话的关键信息 .join([m.contentforminold_messages])summaryllm.invoke([HumanMessage(contentsummary_prompt)]).content# 删除旧消息保留最后两条delete_messages[RemoveMessage(idm.id)forminold_messages]return{summary:summary,messages:delete_messages}defshould_summarize(state:State)-str:returnsummarizeiflen(state[messages])6elsemodel# 构建图builderStateGraph(State)builder.add_node(model,call_model)builder.add_node(summarize,summarize_conversation)builder.set_entry_point(model)builder.add_conditional_edges(model,should_summarize)builder.add_edge(summarize,model)builder.add_edge(model,END)# 当 should_summarize 返回 model 时直接结束memoryMemorySaver()appbuilder.compile(checkpointermemory)# 多轮对话测试config{configurable:{thread_id:summary-test}}queries[我叫张三,我喜欢编程,我在学习 LangGraph,请记住这些信息,]forqinqueries:input_data{messages:[HumanMessage(contentq)]}foreventinapp.stream(input_data,config):forkey,valueinevent.items():ifmessagesinvalueandvalue[messages]:last_msgvalue[messages][-1]print(f助手:{last_msg.content})# 当消息数超过阈值后再次提问仍能回忆历史input_data{messages:[HumanMessage(content我叫什么名字)]}foreventinapp.stream(input_data,config):forkey,valueinevent.items():ifmessagesinvalueandvalue[messages]:last_msgvalue[messages][-1]print(f助手:{last_msg.content})运行后当对话超过 6 条时旧消息会被压缩成摘要窗口大小始终可控但智能体依然能回答姓名等历史信息。总结LangGraph 的持久化层为智能体提供了稳固的记忆基础通过组合MemorySaver、MongoDBSaver和InMemoryStore可以灵活实现线程隔离、跨线程共享以及长期持久化。而消息过滤和对话摘要等优化策略则让记忆系统在有限窗口下仍能保持高效的上下文利用。这些技术是构建复杂、可靠的多智能体应用的必备技能。本文覆盖了持久化层概念、Checkpoint、MemorySaver、线程隔离、跨线程调用、短期/长期记忆实现、记忆优化过滤、摘要等全部技术要点并给出了代码示例。