本文是「LangGraph 教程系列」第 8 篇。写作时基于 langgraph 1.2.10、langchain 1.3.14、langchain-openai 1.4.1、Python 3.12。配套代码仓库 https://github.com/wxj006007/deep-research-assistant 本篇对应 tagv2.2。上一版的研究助手已经会停下来等人审批。审批后恢复得很好只要还是同一个thread_id它就记得这轮检索查过什么、现在该去哪一个节点。但用户换一条新会话呢比如昨天研究过 checkpoint今天继续问 Store又比如他每次都希望中文回答、先给结论再列步骤。checkpoint 不应承担这些跨会话信息——它记录的是某一次执行不是用户档案。这一篇把“记住过去”拆清楚短期记忆仍归 checkpointer长期记忆交给 Store。一、三种东西不要都叫 memory“记忆”在 agent 系统里很容易被说成一个模糊概念。这里至少有三层层次存放内容生命周期本例实现当前 state当前节点需要的 question、docs、round一次图运行ResearchState短期记忆可暂停、恢复、回放的执行快照一个thread_idMemorySaver长期记忆用户偏好、历史研究摘要跨多个 threadInMemoryStore前两层跟着图的执行走第三层以用户为边界独立于某次执行。把它们混在一起常见后果是新会话读不到信息或者所有旧资料不断塞进 statecheckpoint 越来越大。二、v2.2 的记忆分层用户thread_idMemorySaver短期记忆InMemoryStore长期记忆用户偏好历史研究摘要一个用户可以有多条 thread每条 thread 的 checkpoint 相互隔离但它们都可以读取同一个用户 namespace 下的长期 Store。反过来另一个用户即便问了同样的问题也不应读取到前者的资料。三、Store 的最小模型namespace、key 和 value这一篇故意不引入 embedding 或向量库。我们先用可复现的 namespace/key 模型讲清数据归属(users, user_id, profile) - preferences (users, user_id, research) - summaries第一条记录保存稳定的用户偏好第二条记录保存一个摘要列表。它们的 value 都是普通字典store.put((users,alice,profile),preferences,{topics:[LangGraph],style:先给结论再给简洁步骤,language:中文,},indexFalse,)indexFalse明确表示本例不建向量索引、不做语义检索。以后需要从大量历史资料中按自然语言召回时再讨论search()、embedding 和索引策略现在的重点是把跨会话存取边界设计正确。四、把用户身份放进运行时上下文用户身份不是问题本身也不该作为每个节点都写入的 state 字段。LangGraph 的Runtime适合承载这类运行时信息fromdataclassesimportdataclassdataclass(frozenTrue)classResearchContext:user_id:strresearch_id:str编译图时同时传入 checkpointer 与 storegraphbuilder.compile(checkpointerMemorySaver(),storeInMemoryStore(),)节点通过runtime.context知道当前用户通过runtime.store读写长期记忆defload_memory_node(_:MemoryResearchState,runtime:Runtime[ResearchContext])-dict:profileruntime.store.get(profile_namespace(runtime.context.user_id),preferences)summariesruntime.store.get(research_namespace(runtime.context.user_id),summaries)return{memory_context:{preferences:profile.valueifprofileelse{},summaries:summaries.value.get(items,[])ifsummarieselse[],}}memory_context只是这次执行读取到的记忆副本放入 state 供 planner 和 writer 使用真正长期保存的数据仍在 Store 中。这样 checkpoint 保存的是“本轮曾读取了什么上下文”Store 保存的是“下次还能取到什么档案”。五、图结构先读后写v2.2 保留第7篇的人在回路。新增两个节点开始时load_memory成稿后persist_memory。通过拒绝继续完成START读取长期记忆规划查询人工审核搜索取消评估成稿保存研究摘要END读取节点在plan前因此查询规划可以根据用户的语言、风格和已有研究避免重复。保存节点只在成功成稿后执行用户在审批阶段拒绝任务时不应凭空新增一条“研究完成”记忆。六、保存摘要时为什么要有research_id图可能因重试、进程恢复或调用方重复提交而再次执行。若每次都 append一项研究会在长期记忆里留下多份重复摘要。因此 v2.2 的上下文里还有research_id。persist_memory_node写入前会删除同 id 的旧项再保存新项并把历史限制在最近 5 条items[itemforiteminitemsifitem.get(research_id)!research_id]items.append({research_id:research_id,question:state[question],summary:state.get(answer,)[:500],})store.put(namespace,summaries,{items:items[-5:]},indexFalse)这不是分布式事务的完整方案但已经给出一个重要的工程习惯外部写入要有稳定业务 id并且能安全重试。七、跨会话演示先为 Alice 保存偏好再在 thread A 中完成研究context_aResearchContext(user_idalice,research_idcheckpoint-basics)thread_a{configurable:{thread_id:alice-thread-a}}firstgraph.invoke({question:checkpoint 和长期记忆有什么区别},configthread_a,contextcontext_a,)之后 Alice 在一个全新的 thread B 提问context_bResearchContext(user_idalice,research_idmemory-followup)thread_b{configurable:{thread_id:alice-thread-b}}secondgraph.invoke({question:那 Store 应该保存哪些数据},configthread_b,contextcontext_b,)assertsecond[memory_context][preferences][language]中文assertany(item[research_id]checkpoint-basicsforiteminsecond[memory_context][summaries])这里两个thread_id不同所以 checkpoint 不共享但user_id一样所以load_memory读到了同一份长期记忆。把user_id换成 Bob读取结果就是空字典和空摘要列表这才是正确的租户隔离。八、实践边界记忆越多不一定越好长期记忆通常保存的是用户数据因此比普通 state 更需要边界。最小化只保存对后续任务有用的信息本例保存摘要而非全文 docs。显式写入用户偏好应由设置操作或明确同意写入不要从一次回答里偷偷推断永久标签。可删除以用户 namespace 为边界提供清除偏好、清除研究档案的能力。限制容量本例只保留 5 条摘要生产环境还应有 TTL、归档和监控。隔离与授权namespace 必须以可信的用户 id 构造不能直接相信客户端任意传来的 id。InMemoryStore适合教学、调试和单进程 Demo进程重启后数据会消失。真正上线时再替换为持久化 Store读取和写入节点的接口不需要跟着业务代码一起重写。九、跑起来代码在src/v2_2_memory.pypython-msrc.v2_2_memory脚本会验证四件事Alice 的 thread B 能读取 thread A 留下的偏好和摘要同一research_id重试只保留一条摘要Bob 读取不到 Alice 的 Store 数据checkpoint 仍按thread_id隔离Store 则按user_id隔离。十、本篇小结第7篇的 checkpoint 解决的是“这次执行如何暂停和恢复”第8篇的 Store 解决的是“下一次会话还记得什么”。v2.2 在开始时按用户读取偏好和历史摘要在结束时用research_id幂等地保存新的摘要。最关键的区分是thread 管短期执行Store 管长期数据。把这条边界守住记忆才不会变成无限膨胀、难以删除、跨用户泄露的一团 state。有了暂停、恢复和跨会话记忆研究助手的能力已经不只是一口气跑完任务。下一篇我们让它把执行过程以流的形式交给用户什么时候该展示 token什么时候该展示事件什么时候该展示状态更新。