Store与长期记忆

checkpoint 绑在单次会话的 thread_id 上；用户偏好「默认语言=中文」要跨会话还在——必须另开 Store：按 namespace 元组 + key 存业务 KV，和 Saver 互补，互不替代。

段末注释：Store = 跨 thread 的长期 KV；checkpointer = 单 thread 的图 State 快照链。Saver 在每拍自动 get_tuple/put；Store 不会自动读写，节点里要显式 get/put。

社区方案：官方 InMemoryStore + compile(store=...)，节点用 get_store() 或 runtime.store。不要自写全局 dict 当记忆，也不要把偏好塞进 checkpoint。风险：InMemoryStore 重启即空；search(ns) 是前缀匹配且默认 limit=10，会漏项或串到子命名空间。

---

1. 定位

维度	内容
角色	跨 thread 的业务 KV / 长期记忆
输入 → 输出	store.get/put(ns, key) → 节点写入 State 字段
核心 API	InMemoryStore、compile(store=)、get_store()
依赖 LangChain	无；语义检索才需要 embeddings

出现背景：换 thread_id 后 checkpoint 是空树，但同一 user_id 的偏好还应在。

---

2. 图拓扑

节点表

节点名	职责	读	写
load_mem	get_store().get 读偏好	Store + user_id	profile
reply	用 profile 拼回复	profile	answer
remember	put 记下本轮 answer	user_id, answer	Store（不改 State）

边表

源	目标	类型
START	load_mem	固定
load_mem	reply	固定
reply	remember	固定
remember	END	固定

---

3. invoke 生命周期

compile(store=store, checkpointer=cp)
  → Pregel 挂上两本账：Saver 自动、Store 待节点来取

第一次 invoke(thread=t1, user_id=u1):
  1. Saver.get_tuple(t1) → 无历史
  2. load_mem: get_store().get(("users","u1"), "lang") → profile
  3. reply 写 answer
  4. remember: store.put(("users","u1"), "last_answer", {...})
  5. Saver.put → 只写入 t1 的 State（不含 Store 内容）

第二次 invoke(thread=t2, 同一 u1):
  1. get_tuple(t2) → 空（新会话）
  2. get(("users","u1"), "lang") 仍是 zh；last_answer 也在

---

4. 原理

4.1 compile 如何把 Store 交给图

InMemoryStore 是进程内 dict（可选向量索引）。builder.compile(store=store) 把它赋给 Pregel；调度器不会在 superstep 边界自动 search/put。

节点里取实例的两条官方路（等价）：

from langgraph.config import get_store
# 或 from langgraph.runtime import Runtime

def load_mem(state):
    store = get_store()          # compile 时没传 store → None
    item = store.get(("users", state["user_id"]), "lang")

def load_mem(state, runtime: Runtime):
    item = runtime.store.get(("users", state["user_id"]), "lang")

闭包捕获外层 store 也能跑，但测不到「忘了 compile(store=)」；排障用 get_store()。

和 Saver 对照：

	Checkpointer	Store
compile	checkpointer=cp	store=store
谁发起读写	框架每拍 get_tuple / put	节点显式 get / put / search
主键	(thread_id, checkpoint_ns, checkpoint_id)	(namespace元组, key)
存什么	全量 State 快照	业务 dict
换 thread	新会话、空历史	只要 ns+key 相同就还在

checkpoint_ns 是子图抽屉，不是 Store 的 namespace。

4.2 一次 invoke 里 Store 怎么参与

1. compile(store=store, checkpointer=cp)。
2. invoke(input, {thread_id: t1})：先 get_tuple 灌 checkpoint（与 Store 无关）。
3. load_mem 调 get_store().get(("users", uid), "lang")。命中则 item.value 是 dict；未命中 None。返回 {"profile": ...} 合并进 channels。
4. reply 只读 State，不碰 Store。
5. remember 调 put(ns, "last_answer", {"value": answer})：覆盖同 key，写入进程内账本（或 Postgres 表）。
6. superstep 结束 Saver 只 put State；profile/answer 进 checkpoint，Store 条目不会被复制进 values。

少了第 3 步的 get → 节点看不见长期记忆。
少了 compile(store=) → get_store() 为 None。
第 6 步误以为换 thread 会丢 lang → 那是 Saver 空了，Store 还在。

4.3 namespace / Item / search

主键：namespace: tuple[str, ...] + key: str。典型 ("users", user_id) 或 (org_id, user_id, "prefs")。

put 之后读到的是 Item：value / key / namespace / created_at / updated_at。value 必须是 JSON 可序列化的 dict。

方法	作用
put(ns, key, value)	写入或覆盖
get(ns, key)	精确读，没有则 None
delete(ns, key)	删一条
search(ns_prefix, *, query, filter, limit=10, offset=0)	前缀枚举；("users",) 会扫到 ("users","u1") 下的键
list_namespaces(...)	发现有哪些 ns

search 默认 limit=10，超了静默截断。InMemoryStore 按插入序；Postgres 常按 updated_at 降序——跨后端不要依赖顺序。语义检索要在构造 Store 时配 index={embed, dims, fields}，再 search(..., query="...")。

生产换 PostgresStore 等，接口不变，需 setup()。

---

5. 最小可运行示例

from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.config import get_store
from langgraph.graph import StateGraph, START, END
from langgraph.store.memory import InMemoryStore


class State(TypedDict):
    user_id: str
    profile: str
    answer: str


store = InMemoryStore()
store.put(("users", "u1"), "lang", {"value": "zh"})  # 图外预置，模拟已有偏好


def load_mem(state: State) -> dict:
    item = get_store().get(("users", state["user_id"]), "lang")  # 拓扑 §2 load_mem
    lang = item.value["value"] if item else "en"
    return {"profile": f"lang={lang}"}


def reply(state: State) -> dict:
    return {"answer": f"hello ({state['profile']})"}


def remember(state: State) -> dict:
    get_store().put(
        ("users", state["user_id"]),
        "last_answer",
        {"value": state["answer"]},
    )
    return {}


builder = StateGraph(State)
builder.add_node("load_mem", load_mem)
builder.add_node("reply", reply)
builder.add_node("remember", remember)
builder.add_edge(START, "load_mem")
builder.add_edge("load_mem", "reply")
builder.add_edge("reply", "remember")
builder.add_edge("remember", END)

graph = builder.compile(checkpointer=InMemorySaver(), store=store)

t1 = {"configurable": {"thread_id": "t1"}}
print(graph.invoke({"user_id": "u1", "profile": "", "answer": ""}, t1))

t2 = {"configurable": {"thread_id": "t2"}}  # 新会话，同一用户
out2 = graph.invoke({"user_id": "u1", "profile": "", "answer": ""}, t2)
print(out2["profile"])  # 仍 lang=zh
print(store.get(("users", "u1"), "last_answer").value)

---

6. 执行追踪

调用	Saver thread	Store ("users","u1")	State profile / answer
预置	—	lang=zh	—
invoke t1 后	t1 有完整 State	+ last_answer	lang=zh / hello (lang=zh)
invoke t2 后	t2 新树（无 t1 messages）	lang 与 last_answer 仍在	仍 lang=zh

---

重要配置参数

参数（API 名）	类型 / 默认值	功能说明	作用与影响	参考起点 / 常用范围	配置指导
compile(store=...)	BaseStore / None	把 Store 注入 Pregel	不传则 get_store() 为 None	InMemoryStore()	生产换 Postgres 等
get_store() / runtime.store	节点内取实例	图调用 Store 的入口	闭包捕获看不到注入失败	每个读写节点	不要用全局单例凑合
put(ns, key, value)	tuple + str + dict	写入或覆盖	同 key 覆盖；value 须 JSON 化	偏好、事实	ns 含 tenant/user
get(ns, key)	→ Item / None	精确读	没有就是 None	已知 key	优先于盲 search
search(ns_prefix, limit=10)	前缀枚举	漏 limit 会截断；前缀会扫子 ns	列表记忆	先 get 能 get 就 get	分页用 offset
compile(checkpointer=)	可选，常一起传	会话 State 仍走 Saver	与 Store 正交	InMemorySaver	多轮对话两者都要
index=（构造 Store）	embeddings 配置	打开语义 query=	不配则 query 无效	文档检索	开发可关

---

7. 易踩坑

1. 用 checkpoint 存用户偏好：换 thread_id 即丢。
2. 把 Store namespace 写成 checkpoint_ns：两套键，互不可见。
3. search(("users",)) 当精确列表：前缀会扫到所有用户；默认只 10 条。
4. 节点闭包 store、compile 忘了传：换实例后节点仍打旧对象，或测试以为注入成功。
5. 生产用 InMemoryStore：重启 / 多副本各持一份内存；上线换 PostgresStore / RedisStore 等，接口不变。
6. value 塞不可 JSON 的对象：落盘后端会炸。
7. Python 低于 3.11 的 async 节点用 get_store()：contextvar 传不进子任务；改 runtime.store 或升到 3.11+。

---

小结

• Store = 跨 thread 的长期 KV；Saver = 单 thread 的 State 快照。
• 图只在节点里 get_store().get/put；不会每拍自动读写。
• namespace 用业务元组隔离租户；不要和 checkpoint_ns 混用。

---

参考链接

• Stores
• get_store
• Persistence — checkpointer vs store