LangGraph 记忆分层设计

QuibblerAgentQuibblerAgent 2026-11-04 约 30 分钟 3 次阅读

第 8 篇把 checkpointer 讲透了:thread_id 是档案抽屉号,图每推进一步就给黑板打快照,进程崩了能续跑,想回看能时间旅行。但快照有一条天然边界——它只记"这一次会话"。明天用户带着新的 thread_id 回来,Agent 对他一无所知:上次定下的回复风格、纠正过的口径、练到一半的知识点,全都锁在旧抽屉里。真实产品里"越用越懂你"的那部分能力,checkpointer 给不了。本篇是第 9 篇,把长期记忆 Store 装进图里,与短期层拼成完整的记忆分层设计。

版本口径与全系列一致:langgraph 1.2.x,PyPI 独立发布,接口以官方 Stores 文档与 BaseStore API 参考为准。LangChain 基础系列的第 7 篇从组件视角讲过同一主题,本篇换图编排视角:记忆是图的一等公民——Store 挂在图上,节点按需读写,会话开始时动态注入。全文按"checkpointer 的边界 → Store 编址模型 → 节点读写 → 记忆工具化 → 记忆预热 → 分层总架构 → 学习助手实战 → 总结"展开。


1、checkpointer 的边界

第 8 篇的结论先摆回来:checkpointer 的全部组织维度是 thread。thread_id 划定抽屉,快照只进本抽屉,恢复、续跑、时间旅行都在抽屉内发生。这套机制对"会话内"的诉求是完备的,边界出现在"会话外"——三类需求它天然接不住:

  • 用户画像:称呼偏好、回复风格、输出格式,说一次就该一直记得
  • 事实纠偏:上次纠正过的口径与禁忌,下次不该再犯
  • 跨会话进度:学习到哪一课、订单办到第几步,要跨天累计

直觉方案是"不换 thread_id,永远接着聊",三条毛病立刻冒头。其一,历史只增不减,上下文越拖越长;其二,快照存的是整份黑板,想取"称呼偏好"这一条,只能把整个 thread 重放出来自己翻;其三,多端并发写同一个 thread 会互相踩踏。更根本的问题在于:快照是执行状态,不是知识——它不知道哪一条值得留到明天。

两层记忆的分工对照如下:

维度 短期记忆 checkpointer 长期记忆 Store
组织维度 按 thread 归档 按 namespace 编址
生命周期 会话内 跨会话跨 thread
存储形态 超步快照 一条条记忆条目
检索方式 按 thread_id 整份取 按前缀或语义查

结论一句话:短期与长期不是互相替代,是分工。LangGraph 的答案是把长期记忆也收编为框架的一等公民——独立的 BaseStore 抽象与 checkpointer 平行,专门装"值得带走的东西"。

2、Store 的 namespace 编址

BaseStore 的接口小得可以一口气看完,四个同步方法如下,异步另有 aput、aget、asearch 一套:

方法 作用 说明
put 写入一条记忆 命名空间、键、值三要素
get 点取一条记忆 命名空间加键精确定位
search 前缀检索一批 可加过滤与语义查询
list_namespaces 列出命名空间 按前缀或模式过滤

编址是这一层的关键设计。namespace 是字符串元组,段数不限、顺序即层级:("memories", "alice") 读作"记忆里 alice 的那份",第一段定"记什么",后几段定"归谁"。search 按前缀匹配:检索 ("memories", "alice") 只命中 alice 的条目,检索 ("memories",) 则跨用户全捞——要全用户共享的常识库,第一段用固定值;要按用户隔离,第二段放 user_id。开发测试用官方自带的 InMemoryStore,生产同族有 SQLite、Postgres 的 Store 实现。

直连 Store 的最小用法如下:

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
store.put(("memories", "alice"), "style",         # 命名空间 + 键 + 值
          {"text": "回复保持简短"})
store.put(("memories", "alice"), "level",
          {"text": "LangGraph 学到第 9 篇"})
store.put(("memories", "bob"), "style",
          {"text": "回答附代码示例"})

item = store.get(("memories", "alice"), "style")  # 点取一条
print(item.value["text"])
for it in store.search(("memories", "alice")):    # 前缀检索
    print(it.key, "->", it.value["text"])

输出如下:

回复保持简短
style -> 回复保持简短
level -> LangGraph 学到第 9 篇

解读:三次 put 写进两个命名空间,get 按"命名空间加键"点取;search 只给前缀 ("memories", "alice"),命中 alice 的两条,bob 的那条不出现——隔离与共享全由编址表达。条目的 value 是普通字典,字段自己定。

Store 与各会话的关系画出来是这样:

flowchart LR A[会话一] -->|写入偏好| S[(Store)] B[会话二] -->|写入进度| S C[明天的新会话] -->|预热读取| S S -->|注入画像| C

解读:Store 独立于任何 thread 存活。会话一写入偏好、会话二写入进度,都只是往同一个 Store 放条目;明天的新会话开一个全新 thread_id,checkpointer 里一片空白,Store 里 alice 的画像原样躺着。跨 thread 共享的本质,就是把记忆的编址维度从 thread 换成 namespace——记忆不再随会话生灭。

两个接口细节的现状也交代一下。put 另收 ttl 与 index 两个可选参数:前者给条目定寿命、到期清理,后者可按条目跳过语义索引;官方 put 参考同时注明 TTL 是否生效取决于具体存储实现,上线前要在目标后端的文档里核对。语义检索同理,search 的 query 参数走向量相似度,前提是 Store 实例配置了 embeddings,本文不展开。

3、节点读写 Store

图这一侧的接法两步:编译时把 Store 挂上图,节点签名里声明 store 参数。compile 的 store 参数与第 8 篇的 checkpointer 并列,一次编译可以同时挂两层记忆。节点侧是按需注入:函数签名写 store: BaseStore,框架检测到就把实例递进来;不声明的节点完全不受影响——权限按需领取。

一个"写入加回读"的最小双节点图如下:

from langgraph.store.base import BaseStore
from langgraph.graph import MessagesState, StateGraph, START, END

def remember(state: MessagesState,
             *, store: BaseStore):                # 声明即注入
    text = state["messages"][-1].content          # 用户最新一句
    store.put(("memories", "alice"), "last_topic",
              {"text": text})                     # 写:跨会话可见

def recall(state: MessagesState,
           *, store: BaseStore) -> dict:
    items = store.search(("memories", "alice"))   # 读:前缀检索
    memo = ";".join(i.value["text"] for i in items)
    return {"messages": [{"role": "assistant",
                          "content": f"我记得:{memo}"}]}

builder = StateGraph(MessagesState)
builder.add_node("remember", remember)
builder.add_node("recall", recall)
builder.add_edge(START, "remember")
builder.add_edge("remember", "recall")
builder.add_edge("recall", END)
graph = builder.compile(store=store)              # Store 挂在图上

解读三件事。其一,remember 与 recall 都声明了 store 参数,实例来自 compile 挂载,节点代码里没有任何全局变量。其二,remember 只写 Store、不写黑板——节点返回 None 表示本轮零更新,这是合法返回值。其三,recall 把检索结果拼成一条消息写回黑板,这就是 Store 的内容进入 State 的最短路径。user_id 写死在命名空间里只为演示;真实用法从 config 取——节点再加一个 config: RunnableConfig 参数,config["configurable"]["user_id"] 拿到用户标识,与第 6 篇 interrupt 恢复共用同一条 config 通道。

4、记忆工具化

节点读写是最直接的路,但"什么值得记"常常只有模型自己知道——用户随口一句"以后叫我老王",规则节点未必抓得住。第二条路是把记忆的读写做成工具,交给 Agent 自主决定。save_memory 与 recall 两个工具如下:

from uuid import uuid4
from langchain.tools import tool
from langchain.agents import create_agent
from langgraph.store.base import BaseStore
from langchain_core.runnables import RunnableConfig

@tool
def save_memory(content: str, *, config: RunnableConfig,
                store: BaseStore) -> str:
    """把一条值得长期记住的信息存入记忆"""
    uid = config["configurable"]["user_id"]      # 注入的 config 通道
    store.put(("memories", uid), uuid4().hex[:8],
              {"text": content})                 # 随机键:追加不覆盖
    return "已记住"

@tool
def recall(query: str, *, config: RunnableConfig,
           store: BaseStore) -> str:
    """按关键词回忆此前存下的记忆"""
    uid = config["configurable"]["user_id"]
    hits = store.search(("memories", uid))
    return ";".join(i.value["text"] for i in hits
                     if query in i.value["text"]) or "没有相关记忆"

agent = create_agent(model="openai:gpt-6",
                     tools=[save_memory, recall],
                     store=store)                # 成品代理同收 store

解读:工具定义沿用第 5 篇的 @tool 口径,store 与 config 是注入参数,不会出现在模型可见的工具描述里——模型只看到 content 与 query 两个真参数。create_agent 同样收 store 参数,实例由此传进内部图,工具执行时由工具节点注入,第 3 节手搓的机制在成品代理里原样生效。存取的键用 uuid4().hex[:8] 现场生成——长期记忆是追加型数据,同键覆盖的语义交给上层判断。recall 演示的是子串匹配,Store 配上 embeddings 之后同一个签名直接升级为语义检索。这套"记忆当工具"的模式与 LangChain 基础系列第 7 篇同源,何时存、存什么的策略那篇已经展开,本篇不重复;图编排视角只补一句——记忆操作发生在图内的工具节点上,照样吃 checkpointer 快照与 stream 轨迹,可观测性不打折。

5、记忆预热节点

工具化解决"怎么存",但每轮都指望模型自己想起 recall,费轮次也不可靠。更工程化的姿势是会话开始时预取:START 正后方放一个预热节点,把该用户的记忆一次捞全写进 State,后续节点把它当普通字段用——这就是"记忆预热"模式。

from typing import Annotated, TypedDict
import operator
from langgraph.store.base import BaseStore
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, START, END

class LearnState(TypedDict):
    question: str
    memories: Annotated[list, operator.add]      # 预热区:追加语义

def warm_up(state: LearnState, *, config: RunnableConfig,
            store: BaseStore) -> dict:
    uid = config["configurable"]["user_id"]      # config 取用户标识
    items = store.search(("memories", uid))      # 记忆一次捞全
    return {"memories": [i.value["text"] for i in items]}

def answer(state: LearnState) -> dict:
    profile = ";".join(state["memories"])       # 记忆当普通字段用
    return {"question": f"结合画像作答:{profile}"}

builder = StateGraph(LearnState)
builder.add_node("warm_up", warm_up)             # 永远的第一站
builder.add_node("answer", answer)
builder.add_edge(START, "warm_up")
builder.add_edge("warm_up", "answer")
builder.add_edge("answer", END)
graph = builder.compile(store=store)             # 生产形态再挂 checkpointer

解读:warm_up 是全图第一站,只干一件事——search 全量、写进 memories 字段。answer 节点拿到的是普通黑板字段,它根本不知道 Store 的存在——存储访问被压缩到预热一处,业务节点与记忆后端彻底解耦;调用时 config 里带上 user_id 即可。真实工程里 answer 会把画像拼进 system prompt;记忆一多,全量注入就换成带过滤或语义查询的检索,只捞本次会话相关的几条。再看编译那一行:目前只挂了 store,把第 8 篇的 checkpointer 也加上,就是下一节的分层架构。

6、三层记忆总架构

两件兵器凑齐,还差第三件。长会话里 checkpointer 的快照越滚越大,每轮都拖着完整历史进上下文;治法是在图里加一个摘要压缩节点——消息超过阈值,就把前情折成一段摘要写回黑板,黑板重新变短,对话照常续。三层在同一张图上的分工如下:

flowchart TD U[用户] --> W[预热节点] S[(长期 Store)] -->|会话开始注入| W W --> A[对话节点] A -->|每超步快照| K[(短期 checkpointer)] A -->|历史过长| Z[压缩节点] Z -->|摘要写回| A A -->|沉淀要点| S

解读:数据三层流动。会话开始,预热节点从 Store 注入长期画像;会话之中,checkpointer 每个超步打快照、崩了能续;历史过长时压缩节点折摘要;会话里冒出的长期价值信息,由工具或节点沉淀回 Store,明天的新会话再从预热开始循环。三层不是三套系统,是同一张图上的三种节点职责。

什么需求落在哪一层,选型对照如下:

需求 归属层 落点
会话内多轮上下文 短期层 checkpointer 快照
断点续跑与时间旅行 短期层 thread_id 加快照
长会话历史过载 压缩节点 摘要写回黑板
用户偏好与画像 长期层 Store 命名空间
跨会话任务进度 长期层 Store 加预热节点

分层的判断标准一句话:上下文里用的放短期,值得带走的放长期——问一句"明天还有用吗",有用的进 Store,没用的留在 thread 里随会话沉底。

后端组合顺带交代:短期层落盘换 SQLite、Postgres 的 checkpointer;长期层同族有 SQLite、Postgres 的 Store 实现,MongoDB、DynamoDB 等集成也已就位。语义检索与 TTL 是否可用取决于所选后端,第 2 节已提醒过;这里只补一句——两层后端可以分开选型,不必绑死同一家。

7、学习助手实战

机制凑齐,合体做一个真东西:学习助手——跨会话记住每个知识点的掌握进度,弱点优先加练。LearnState 沿用第 5 节,再加 topic 与 answer 两个 str 字段;出题与判分节点如下:

from langgraph.types import Command, interrupt

TOPICS = ["条件边", "reducer"]                   # 知识点清单
ANSWERS = {"条件边": "路由", "reducer": "合并"}    # 演示用标准答案

def quiz(state: LearnState, *, config: RunnableConfig,
         store: BaseStore) -> dict:
    uid = config["configurable"]["user_id"]
    done = {i.key for i in store.search(("progress", uid))
            if i.value["mastered"]}              # 已掌握集合
    weak = [t for t in TOPICS if t not in done]  # 其余全是弱点
    if not weak:
        return {"topic": "", "question": "全部掌握,自由提问"}
    topic = weak[0]                              # 弱点优先加练
    reply = interrupt(f"请说明「{topic}」的作用")  # 挂起等人作答
    return {"topic": topic, "answer": reply}

def grade(state: LearnState, *, config: RunnableConfig,
          store: BaseStore) -> dict:
    uid = config["configurable"]["user_id"]
    ok = ANSWERS.get(state["topic"], "") in state["answer"]
    store.put(("progress", uid), state["topic"],
              {"mastered": ok})                  # 进度写回长期层
    return {"question": f"「{state['topic']}」"
                        f"{'通过' if ok else '没过,明天再练'}"}

解读:quiz 从 Store 读进度、把没掌握的知识点挑出来——弱点自动加练的全部逻辑就是这一个列表推导;没有弱点时收束为自由问答。出题后 interrupt 挂起等人作答,这是第 6 篇的机制,作答文本随返回值写进黑板。grade 的判分是演示口径——标准答案表里查关键词,真实产品换成模型结构化判分即可,图结构一字不改;判完把 mastered 写回 Store,进度沉淀进长期层,会话结束也不丢。

建图与第一天的运行如下:

from langgraph.checkpoint.memory import MemorySaver
from langgraph.store.memory import InMemoryStore

def warm_up(state: LearnState, *, config: RunnableConfig,
            store: BaseStore) -> dict:
    uid = config["configurable"]["user_id"]
    items = store.search(("progress", uid))      # 进度一次捞全
    return {"memories": [
        f"{i.key}:{'已掌握' if i.value['mastered'] else '未掌握'}"
        for i in items]}

store = InMemoryStore()                          # 全新 Store
builder = StateGraph(LearnState)
builder.add_node("warm_up", warm_up)             # 预热:进度进黑板
builder.add_node("quiz", quiz)                   # 弱点优先出题
builder.add_node("grade", grade)                 # 进度写回 Store
builder.add_edge(START, "warm_up")
builder.add_edge("warm_up", "quiz")
builder.add_edge("quiz", "grade")
builder.add_edge("grade", END)
tutor = builder.compile(checkpointer=MemorySaver(),
                        store=store)             # 双层记忆一次挂齐

day1 = {"configurable": {"user_id": "alice",
                         "thread_id": "day-1"}}
r1 = tutor.invoke({"memories": []}, config=day1) # 首轮在题面处挂起
r2 = tutor.invoke(Command(resume="条件边负责按状态路由"),
                  config=day1)                   # 交卷:判分写回
print(r2["question"])

输出如下:

「条件边」通过

第二天换一个全新会话再问:

day2 = {"configurable": {"user_id": "alice",
                         "thread_id": "day-2"}}
r3 = tutor.invoke({"memories": []}, config=day2) # 全新 thread
print(r3["__interrupt__"][0].value)              # 新会话的加练题

输出如下:

请说明「reducer」的作用

解读:day-2 是全新 thread,短期层一片空白,但 quiz 从 Store 读到的进度里条件边已掌握,加练题自动落到下一个弱点——会话散了记忆不散,正是分层在起作用。也注意 warm_up 与 quiz 都查了 Store:前者把进度摘要注入黑板供模型参考,后者现查现用做路由依据,两条路各有用处。这张演示图消息量小,用不上压缩节点,历史一长就按第 6 节把它挂进 quiz 之前。

8、总结

本篇把长期记忆装进了图。checkpointer 的组织维度是 thread,Store 的组织维度是 namespace 元组——put、get、search 配前缀检索,跨 thread 共享的本质是编址维度换了。图这一侧两步接法:compile 的 store 参数挂图,节点签名声明 store 即注入。用法两条路线:记忆工具化交给模型自主存取,预热节点在会话开始把画像写进 State;再加摘要压缩节点,短期、压缩、长期三层在同一张图上各司其职。学习助手把两层串成闭环:弱点从 Store 来,进度回 Store 去,会话散了记忆不散。

练习两道。练习一:给学习助手加"隔天复测"——grade 通过时除了 mastered 再写一个复习到期日(演示口径用字符串日期),quiz 优先挑"已掌握但到期"的知识点复测,答错就标记回未掌握。练习二:把第 4 节的 save_memory 接进学习助手,让模型判分后自己决定要不要把薄弱原因存进 ("memories", uid),跑完两个会话后遍历 Store,人工审一遍它存下的条目该不该记。对于正在做对话产品的读者,建议先用 InMemoryStore 把 namespace 的层级设计跑通——第一段定记什么、第二段定归谁——再换落盘后端;判断什么进 Store,就用那句话:明天还有用吗。

相关推荐

获取系统SDK版本、判断手机ROM
精选Code

获取系统SDK版本、判断手机ROM

Build获取系统SDK版本Android中部分API的使用,需要在特定的SDK版本之后才能使用,因此在兼容老版本SDK的时候,经常需要判断API的版本。各种Android版本的对应关系参考《Android各版本对应的SDK版本》判断手机ROM有时候需要判断手机系统的ROM,检测ROM是MIUI、EMUI还是Flyme,可以使用getprop命令,去系统build.prop文件(关于build.p

4.2k
AndroidStudio中各种中文乱码问题
精选Code

AndroidStudio中各种中文乱码问题

AndroidStudio中各种中文乱码问题1、编译Java错误信息乱码在出现这个Annotation processors must be explicitly declared now...问题的时候,正好也发现AndroidStudio Build Output错误信息都乱码。经常遇到各种问题,习惯了,现在遇到的问题,都是以后的答案。 1.1、修改项目build.gradle(无效)在整个p

2.8k
PydanticAI 详解:类型安全的 Python AI SDK
精选AI

PydanticAI 详解:类型安全的 Python AI SDK

PydanticAI 详解:类型安全的 Python AI SDK写 LLM 应用最疼的两件事:模型返回的 JSON 不合预期、换一家模型商要改一串代码。pydantic/pydantic-ai(PydanticAI)用 Pydantic 团队的老本行回答了第一件——输出用 BaseModel 定义、运行前先校验,错误从运行时提前到编码时;第二件则用"模型即字符串"回答:换个厂商只改一个参数。如今

75