LangGraph 设计与实现
第15章 Store 与长期记忆
15.1 引言
Checkpoint 是 LangGraph 的"短期记忆"——它保存单个线程（thread）内的完整状态历史，支持时间旅行和中断恢复。但在许多 LLM 应用中，我们还需要一种"长期记忆"：跨线程、跨对话的持久化存储。例如，用户的偏好设置在所有对话中都应该可用；一个文档分析 Agent 应该记住之前处理过的文档；多个 Agent 之间需要共享知识库。
LangGraph 通过 BaseStore 接口和 InMemoryStore 实现提供了这种长期记忆能力。Store 是一个层级化的键值存储，支持命名空间（namespace）组织、精确查询和可选的向量语义搜索。它通过 Runtime.store 注入到节点函数中，也可以在工具函数中通过 ToolRuntime.store 访问。
本章将从 BaseStore 的抽象接口出发，分析 Item 数据模型、操作类型（Get/Put/Search/ListNamespaces）的设计，深入 InMemoryStore 的实现细节，并对比 Store 与 Checkpoint 在定位、生命周期和使用场景上的本质区别。
本章要点
BaseStore接口的操作原语——Get、Put、Search、ListNamespaces 四种操作Item数据模型——key/value/namespace/created_at/updated_atInMemoryStore的实现——字典存储 + 可选向量搜索- Store 与 Checkpoint 的区别——跨线程 vs 线程内、键值 vs 快照
runtime.store的访问模式——从节点和工具中使用 Store
15.2 BaseStore 接口
15.2.1 接口定义
BaseStore 定义在 libs/checkpoint/langgraph/store/base/__init__.py:700，是一个抽象基类，定义了 Store 的全部操作原语（下面按 ABC 的形状归纳，两个抽象方法 batch / abatch 分别在 :725 与 :737）：
class BaseStore(ABC):
"""Abstract base class for persistent key-value stores."""
@abstractmethod
def batch(self, ops: Iterable[Op]) -> list[Result]:
"""Execute a batch of operations."""
@abstractmethod
async def abatch(self, ops: Iterable[Op]) -> list[Result]:
"""Async version of batch."""
# 便捷方法（基于 batch 实现）
def get(self, namespace: tuple[str, ...], key: str) -> Item | None: ...
def put(self, namespace: tuple[str, ...], key: str, value: dict) -> None: ...
def delete(self, namespace: tuple[str, ...], key: str) -> None: ...
def search(self, namespace_prefix: tuple[str, ...], **kwargs) -> list[Item]: ...
def list_namespaces(self, **kwargs) -> list[tuple[str, ...]]: ...
# 对应的 async 版本
async def aget(self, ...) -> Item | None: ...
async def aput(self, ...) -> None: ...
async def adelete(self, ...) -> None: ...
async def asearch(self, ...) -> list[Item]: ...
async def alist_namespaces(self, ...) -> list[tuple[str, ...]]: ...所有便捷方法最终都委托给 batch 或 abatch 方法。这种设计的好处是实现类只需要实现两个方法就能获得完整的接口，同时 batch 方法天然支持操作批量化，有利于网络 I/O 优化。
15.2.2 操作类型
Store 定义了四种操作原语，每种都是一个 NamedTuple：
graph TB
subgraph 操作类型
GetOp["GetOp<br/>namespace + key<br/>精确查找单个 Item"]
PutOp["PutOp<br/>namespace + key + value<br/>创建或更新 Item"]
SearchOp["SearchOp<br/>namespace_prefix + filter/query<br/>搜索多个 Item"]
ListNS["ListNamespacesOp<br/>match_conditions<br/>列举命名空间"]
end
GetOp -->|返回| Item["Item | None"]
PutOp -->|返回| None_["None"]
SearchOp -->|返回| Items["list[SearchItem]"]
ListNS -->|返回| NS["list[tuple[str, ...]]"]
15.2.3 GetOp
class GetOp(NamedTuple):
namespace: tuple[str, ...]
key: str
refresh_ttl: bool = TrueGetOp 通过 namespace + key 精确定位一个 Item。refresh_ttl 控制是否刷新该 Item 的 TTL（存活时间）。
15.2.4 PutOp
class PutOp(NamedTuple):
namespace: tuple[str, ...]
key: str
value: dict[str, Any] | None # None 表示删除
index: Literal[False] | list[str] | None = None
ttl: float | None = NonePutOp 用于创建、更新或删除 Item。当 value 为 None 时表示删除。index 字段控制哪些值路径需要被向量索引，ttl 是该条目的存活分钟数（只有 supports_ttl 为 True 的后端才接受；对不支持的后端调用 put(..., ttl=...) 会直接抛 NotImplementedError，见 §15.2.6）。
15.2.5 SearchOp
class SearchOp(NamedTuple):
namespace_prefix: tuple[str, ...]
filter: dict[str, Any] | None = None
limit: int = 10
offset: int = 0
query: str | None = None
refresh_ttl: bool = TrueSearchOp 是最灵活的查询操作。它支持三种过滤方式：
- 命名空间前缀：只搜索指定前缀下的 Item
- 结构化过滤：通过
filter字典做精确匹配或比较操作 - 语义搜索：通过
query字符串做向量相似度搜索
过滤操作支持多种比较运算符：
# 精确匹配
{"status": "active"}
# 比较运算符
{"score": {"$gt": 4.99}}
{"score": {"$gte": 3.0}}
{"priority": {"$lt": 5}}
{"count": {"$lte": 100}}
{"status": {"$ne": "deleted"}}15.2.6 BaseStore 的两个被忽视的类级属性：supports_ttl 与 weakref
打开 libs/checkpoint/langgraph/store/base/__init__.py:700，BaseStore 类头定义里藏着两个教学资料极少提到的类级变量：
class BaseStore(ABC):
supports_ttl: bool = False
ttl_config: TTLConfig | None = None
__slots__ = ("__weakref__",)1、supports_ttl: bool = False 是一个 feature capability flag。子类必须显式覆写为 True 才能"声称支持 TTL"。InMemoryStore 没覆写——所以它的 supports_ttl 读出来是 False。这个标志不只是文档：BaseStore.put/aput 在校验完 namespace 后紧接着检查它（libs/checkpoint/langgraph/store/base/__init__.py:912、:1173），只要传了非 None 的 ttl 而后端不支持，就立刻抛 NotImplementedError。用户代码如果依赖 TTL 语义必须用 Postgres/Sqlite 后端（二者都设了 supports_ttl = True）。Python 里没有 trait/interface 支持特性检查，用类属性模拟出来——使用方法（实跑）：
store = InMemoryStore()
assert store.supports_ttl is False # 静态查询能力
store.put(("a",), "k", {"x": 1}, ttl=5)
# NotImplementedError: TTL is not supported by InMemoryStore. Use a store implementation that supports TTL or set ttl=None.2、__slots__ = ("__weakref__",) 有两层用意：
- 禁止动态添加实例属性。如果子类定义了
__slots__但基类没有，就会破坏 slots 隔离（实例会有__dict__）——所以基类也要声明自己的 slots。 - 显式保留
__weakref__。slot 类默认不支持弱引用（weakref.ref(store)会抛 TypeError），把"__weakref__"放进 slots 恢复这个能力。这个能力在本模块内就有真实消费者：AsyncBatchedBaseStore._ensure_task会用weakref.ref(self)把自己交给后台协程_run（libs/checkpoint/langgraph/store/base/batch.py:80, 328），避免后台 task 持强引用导致 store 永远不被回收——因此 BaseStore 必须支持 weakref。
看似不起眼的一行代码背后是"内存布局约束 + 弱引用生态互通"的双重考虑。这条实践在 Python 库开发里值得记住：写基类用 __slots__，但永远别忘了加 "__weakref__"——否则库在复杂场景下被使用时会出现"cannot create weak reference to 'Xxx' object"之类的诡异错误。
15.3 Item 数据模型
15.3.1 _validate_namespace：写入时强加的四条铁律
Store 接口接受的 namespace 是 tuple[str, ...]——Python 里一个普通元组没有格式要求。但 BaseStore.put/BaseStore.aput 在方法体第一行都会调 _validate_namespace(namespace)（libs/checkpoint/langgraph/store/base/__init__.py:911, 1172；AsyncBatchedBaseStore 覆写的 aput 与同步 put 在 base/batch.py:141, 239 也各调一次）。注意这是运行时每次写入的检查，与图的编译无关；而且只有 put 路径校验——get/search/delete 和直接调 batch 都不校验。这个校验函数（第 1255 行）把四条看似平常的规则写死在了框架里：
def _validate_namespace(namespace: tuple[str, ...]) -> None:
if not namespace:
raise InvalidNamespaceError("Namespace cannot be empty.")
for label in namespace:
if not isinstance(label, str):
raise InvalidNamespaceError(...)
if "." in label:
raise InvalidNamespaceError(
f"Invalid namespace label '{label}' found in {namespace}. Namespace labels cannot contain periods ('.')."
)
elif not label:
raise InvalidNamespaceError(
f"Namespace labels cannot be empty strings. Got {label} in {namespace}"
)
if namespace[0] == "langgraph":
raise InvalidNamespaceError(
f'Root label for namespace cannot be "langgraph". Got: {namespace}'
)四条规则背后的动机，源码没有注释说明，下面结合后端实现来理解：
规则 1：namespace 不能为空。空 tuple () 作为 "根 namespace" 在语义上模糊——到底是"存到顶层"还是"查所有"？拒绝它就消除了歧义。你想存顶层文档应该写 ("documents",) 这种至少有一层的 namespace。
规则 2：label 必须是 str。PostgresStore/SqliteStore 都用 _namespace_to_text 把 namespace ".".join 成一个 text 列 prefix（libs/checkpoint-postgres/langgraph/store/postgres/base.py:1251、libs/checkpoint-sqlite/langgraph/store/sqlite/base.py:96）——str.join 只接受字符串，tuple 里出现 int/UUID 会直接出错。统一要求 str 让所有后端的拼接方式一致。
规则 3：label 不能含 .。点号正是这个拼接的分隔符——如果允许 ("users.alice", "prefs")，拼出来的 users.alice.prefs 与 ("users", "alice", "prefs") 完全相同，两个不同的 namespace 在数据库里会撞成同一个。InMemoryStore 以 tuple 为键，自己不会有这个问题，但为了跨后端可移植性，校验器统一禁止。用户需要在 label 里表达多段信息时应该用下划线或连字符（("users", "alice_corp")），而不是点号。
规则 4：根 label 不能是 "langgraph"。引入这条检查的提交（#1961 "Validate no empty namespace is added"）没有说明原因，pin 树里框架自身也没有往 Store 写 ("langgraph", ...) 前缀的数据——只能理解为给框架预留的根命名空间，提前禁止用户占用。
这四条校验看似冗余、实则决定了 Store 在多后端场景下的一致性。用户实际代码被拒时错误消息非常精确，会点名哪个 label 有问题（实跑：('a.b',) 报 Invalid namespace label 'a.b' found in ('a.b',). Namespace labels cannot contain periods ('.').）。但要记住校验只挂在 put 路径上：store.delete(("a.b",), "k") 不会报错，get/search 传入非法 namespace 也只是查不到东西。
15.3.2 结构定义
# libs/checkpoint/langgraph/store/base/__init__.py:51（docstring 有删节，省略 __eq__/__hash__/dict/__repr__）
class Item:
"""Represents a stored item with metadata."""
__slots__ = ("value", "key", "namespace", "created_at", "updated_at")
def __init__(
self,
*,
value: dict[str, Any],
key: str,
namespace: tuple[str, ...],
created_at: datetime,
updated_at: datetime,
):
self.value = value
self.key = key
self.namespace = tuple(namespace)
self.created_at = (
datetime.fromisoformat(cast(str, created_at))
if isinstance(created_at, str)
else created_at
)
self.updated_at = (
datetime.fromisoformat(cast(str, updated_at))
if isinstance(updated_at, str)
else updated_at
)Item 使用 __slots__ 优化内存，值类型固定为 dict[str, Any]。时间戳字段支持从 ISO 8601 字符串反序列化——这是为了兼容 JSON 存储后端。
15.3.3 命名空间的层级设计
graph TB
Root["()"]
Root --> Users["('users',)"]
Root --> Docs["('docs',)"]
Users --> U1["('users', 'alice')"]
Users --> U2["('users', 'bob')"]
U1 --> Prefs["('users', 'alice', 'prefs')"]
U1 --> History["('users', 'alice', 'history')"]
Docs --> D1["('docs', 'project_a')"]
style Root fill:#f9f9f9,stroke:#333
style Users fill:#e6f3ff,stroke:#333
style Docs fill:#fff3e6,stroke:#333
命名空间是一个字符串元组，形成类似文件系统的层级结构。这种设计的优势：
- 组织清晰：自然地按用户、项目、类型分层
- 搜索高效：通过
namespace_prefix可以搜索某个子树下的所有 Item - 访问控制：可以基于命名空间前缀实现权限隔离
15.3.4 SearchItem
class SearchItem(Item):
"""Represents an item returned from a search operation."""
__slots__ = ("score",)
def __init__(self, ..., score: float | None = None):
super().__init__(...)
self.score = scoreSearchItem 继承自 Item，增加了 score 字段，用于向量搜索时返回相似度分数。
15.4 InMemoryStore 实现
15.4.1 数据结构
# libs/checkpoint/langgraph/store/memory/__init__.py:136（简化：省略 docstring、
# index_config 的 copy()、预先 tokenize fields 写入 "__tokenized_fields"，以及无 index 时的 else 分支）
class InMemoryStore(BaseStore):
__slots__ = ("_data", "_vectors", "index_config", "embeddings")
def __init__(self, *, index: IndexConfig | None = None) -> None:
self._data: dict[tuple[str, ...], dict[str, Item]] = defaultdict(dict)
self._vectors: dict[tuple[str, ...], dict[str, dict[str, list[float]]]] = (
defaultdict(lambda: defaultdict(dict))
)
self.index_config = index
if self.index_config:
self.embeddings = ensure_embeddings(self.index_config.get("embed"))InMemoryStore 使用两层嵌套字典：外层按命名空间分组，内层按 key 索引。向量数据单独存储在 _vectors 字典中，第三层按字段路径（如 "text"、"sections[*].body.0"）区分。
graph LR
subgraph "_data 存储结构"
NS1["('users', 'alice')"] --> K1["'prefs' -> Item"]
NS1 --> K2["'profile' -> Item"]
NS2["('docs',)"] --> K3["'doc1' -> Item"]
end
subgraph "_vectors 向量索引"
VNS1["('docs',)"] --> VK1["'doc1'"]
VK1 --> VP1["'text' -> [0.1, 0.2, ...]"]
end
15.4.2 batch 入口：三型拆分到三种处理路径
InMemoryStore.batch（libs/checkpoint/langgraph/store/memory/init.py:206）的真实实现只有 14 行，但把 Store 所有"批量做什么"的逻辑摊在一个函数里：
def batch(self, ops: Iterable[Op]) -> list[Result]:
# The batch/abatch methods are treated as internal.
# Users should access via put/search/get/list_namespaces/etc.
results, put_ops, search_ops = self._prepare_ops(ops)
if search_ops:
queryinmem_store = self._embed_search_queries(search_ops)
self._batch_search(search_ops, queryinmem_store, results)
to_embed = self._extract_texts(put_ops)
if to_embed and self.index_config and self.embeddings:
embeddings = self.embeddings.embed_documents(list(to_embed))
self._insertinmem_store(to_embed, embeddings)
self._apply_put_ops(put_ops)
return results_prepare_ops 把传入的 ops 按类型分成三路（libs/checkpoint/langgraph/store/memory/init.py:375）：
for i, op in enumerate(ops):
if isinstance(op, GetOp):
item = self._data[op.namespace].get(op.key)
results.append(item) # ← 立即返回
elif isinstance(op, SearchOp):
search_ops[i] = (op, self._filter_items(op))
results.append(None) # ← 占位，稍后填
elif isinstance(op, ListNamespacesOp):
results.append(self._handle_list_namespaces(op)) # ← 立即返回
elif isinstance(op, PutOp):
put_ops[(op.namespace, op.key)] = op
results.append(None) # ← 占位
else:
raise ValueError(f"Unknown operation type: {type(op)}")这套分型有两条工程巧思值得细读：
1、put_ops 用 (namespace, key) 作 dict 的 key——天然去重。如果同一个 batch 里出现了两条针对同一 (ns, key) 的 PutOp，后者覆盖前者，只有一条最终被执行——同批次对同一 key 的多次 put 以最后一条为准。
2、SearchOp 和 PutOp 都先占 None 返回位，Search 稍后回填。SearchOp 要等 query 向量算出来才能打分；PutOp 的返回值本来就是 None，但它要等 Get/Search 都处理完才在最后一步 _apply_put_ops 落盘——所以同一批里的 Get/Search 看到的是这批 put 之前的数据（实跑 store.batch([PutOp(("z",), "k", {"v": 1}), GetOp(("z",), "k")]) 返回 [None, None]）。embedding 方面：同批所有 PutOp 待嵌入的文本由 _extract_texts 汇总，合成一次 embed_documents 调用；SearchOp 的 query 则先按文本去重，再每条 query 各调一次 embed_query（用线程池并发），并不合并成一次调用。实跑同批 3 条带 query 的 SearchOp + 3 条 PutOp，嵌入函数被调用 4 次：3 次各含一条 query，1 次含全部 3 条文本。
从这个设计能反推出 LangGraph Store 的一条思路：API 层暴露细粒度操作、实现层成批优化。用户随便调 store.get()/store.put()——接口简单、不用管批量；但如果你写 store.batch([...]) 传入复杂操作集，实现层会在能合并的地方（如 put 的文本嵌入）做聚合。
15.4.3 基本操作示例
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
# 存储用户偏好
store.put(("users", "alice"), "prefs", {"theme": "dark", "language": "zh"})
# 读取
item = store.get(("users", "alice"), "prefs")
print(item.value) # {'theme': 'dark', 'language': 'zh'}
print(item.key) # prefs
print(item.namespace) # ('users', 'alice')
# 搜索某用户下的所有数据（返回 list[SearchItem]，无 query 时 score 为 None）
results = store.search(("users", "alice"))
# 列举 ("users",) 前缀下的命名空间
namespaces = store.list_namespaces(prefix=("users",))
# [('users', 'alice')]
# 删除
store.delete(("users", "alice"), "prefs") # 内部就是 PutOp(value=None)15.4.4 _apply_put_ops 删除时同时清理 _data 和 _vectors
value=None 表示删除这个约定比看起来要多一层。真实的 _apply_put_ops（libs/checkpoint/langgraph/store/memory/init.py:404）：
def _apply_put_ops(self, put_ops):
for (namespace, key), op in put_ops.items():
if op.value is None:
self._data[namespace].pop(key, None)
self._vectors[namespace].pop(key, None) # ← 同时清向量索引
else:
self._data[namespace][key] = Item(
value=op.value, ... # created_at/updated_at 都取当前时间
)注意删除时同时 pop _data 和 _vectors。搜索的候选来自 _data（_filter_items 遍历的是 _data，再按 key 去 _vectors 取向量），所以残留向量本身不会被单独搜出来；但如果只清 _data，以后同一 key 再被 put 一个没有可索引文本的新值时，旧向量就会被这个新 Item "认领"，让它按旧内容参与打分。
相反的，更新 put 的路径没有清 _vectors。更新 value 后，新值里能取到的字段会由 _extract_texts + _insertinmem_store 重新计算并覆盖同一路径的旧向量——所以这些路径不必先清。但如果你把 value 从 "有 text 字段" 改成 "没 text 字段"，旧的 embedding 会继续存活（因为新 value 里没 text，不触发 re-embed）。实跑：doc1 先存 {"text": "Python 编程入门教程"}，再 put 成 {"summary": "nothing"}，按 "Python 教程" 搜索时 doc1 仍以旧分数排第一。PostgresStore 也有同样的问题：更新走 INSERT ... ON CONFLICT DO UPDATE，只 upsert 新值里取到的字段向量，不删除旧字段的向量（libs/checkpoint-postgres/langgraph/store/postgres/base.py:386-405）。
实际使用这条知识的价值：如果你做了"从 {text: "foo"} 改成 {}"的 put，语义上那个 Item 应该不可搜索了，但向量搜索仍能用旧 embedding 找到它。要避免，应该先 delete(ns, key) 删除、再重新 put 新 value。
15.4.5 向量搜索配置
from langgraph.store.memory import InMemoryStore
# 使用 LangChain embeddings（需另装 langchain 与 langchain-openai，并配置 OpenAI API key）
from langchain.embeddings import init_embeddings
store = InMemoryStore(
index={
"dims": 1536,
"embed": init_embeddings("openai:text-embedding-3-small"),
"fields": ["text"], # 指定要索引的字段
}
)
# 存储文档
store.put(("docs",), "doc1", {"text": "Python 编程入门教程", "author": "Alice"})
store.put(("docs",), "doc2", {"text": "TypeScript 类型系统详解", "author": "Bob"})
# 语义搜索
results = store.search(("docs",), query="编程语言教程")
for item in results:
print(f"{item.key}: {item.value['text']} (score={item.score})")向量搜索也支持使用原生的 OpenAI SDK 或任意嵌入函数：
from openai import OpenAI
client = OpenAI()
def embed_texts(texts: list[str]) -> list[list[float]]:
response = client.embeddings.create(
model="text-embedding-3-small",
input=texts
)
return [e.embedding for e in response.data]
store = InMemoryStore(
index={"dims": 1536, "embed": embed_texts}
)embed 既然可以是任意函数，本地实验时也可以不调任何 API，用一个玩具函数把机制跑通（按词表计数，只为演示打分流程）：
from langgraph.store.memory import InMemoryStore
VOCAB = ["python", "typescript", "编程", "教程", "类型"]
def toy_embed(texts: list[str]) -> list[list[float]]:
return [[float(t.lower().count(w)) + 0.01 * i for i, w in enumerate(VOCAB)] for t in texts]
store = InMemoryStore(index={"dims": 5, "embed": toy_embed, "fields": ["text"]})
store.put(("docs",), "doc1", {"text": "Python 编程入门教程", "author": "Alice"})
store.put(("docs",), "doc2", {"text": "TypeScript 类型系统详解", "author": "Bob"})
for item in store.search(("docs",), query="Python 教程"):
print(f"{item.key}: {item.value['text']} (score={item.score:.4f})")
# doc1: Python 编程入门教程 (score=0.8233)
# doc2: TypeScript 类型系统详解 (score=0.0398)没装 numpy 时，第一次向量搜索还会打出一条 "NumPy not found..." 警告（见 §15.8.6），结果不受影响。
15.4.6 _batch_search 的四段语义：flatten → score → max-pool → 分页
InMemoryStore 的向量搜索核心逻辑在 _batch_search（libs/checkpoint/langgraph/store/memory/init.py:302）。逐段翻译它做什么（节选 op.query 存在且 query 向量已算出的分支；没配 index 或没传 query 时走另一个分支，直接对过滤结果按 offset/limit 切片、score 为 None）：
# 阶段 1：展平 (item, vectors) 为 (item, single_vector) 列表
flat_items, flat_vectors = [], []
scoreless = []
for item, vectors in candidates:
for vector in vectors:
flat_items.append(item)
flat_vectors.append(vector)
if not vectors:
scoreless.append(item) # ← 没有 embedding 的 item 单独记下
# 阶段 2：一次性计算所有 (query, vector) 的余弦相似度
scores = _cosine_similarity(query_embedding, flat_vectors)
# 阶段 3：按分数排序
sorted_results = sorted(
zip(scores, flat_items, strict=False),
key=lambda x: x[0],
reverse=True,
)
# 阶段 4：max pooling——同一个 (namespace, key) 只保留分最高的那份
seen = set()
kept = []
for score, item in sorted_results:
key = (item.namespace, item.key)
if key in seen:
continue
ix = len(seen)
seen.add(key)
if ix >= op.offset + op.limit:
break
if ix < op.offset:
continue
kept.append((score, item))
if scoreless and len(kept) < op.limit:
# Corner case: fill tail with non-scored items
kept.extend((None, item) for item in scoreless[: op.limit - len(kept)])四个设计点值得细读：
1、多字段索引的"每字段一条向量"。如果 index.fields = ["text", "summary"]，一个 Item 会产生 两条向量（text 一条、summary 一条）。flat_items 在 flatten 阶段会把同一 Item 塞进去两遍（每个 vector 对应一行）。
2、max pooling 消重。多字段产生的多个向量在分数上互相独立——同一 Item 可能同时以 text_score=0.85 和 summary_score=0.90 两条身份出现。seen set 按 (namespace, key) 去重——只保留得分最高的那一条。sorted_results 是降序排的，所以遍历时第一次遇到一个 key 就是最高分，后续全 skip。这就是 "max pooling" 的字面含义——把多向量 collapse 到单 Item 时取最大相似度。
3、offset/limit 在去重之后。注意 ix = len(seen) 统计的是去重后的序号——offset=5 表示"跳过前 5 个不同 Item"，而不是"跳过前 5 个向量"。这是正确的语义（用户写 limit=10 期望拿到 10 个不同 Item）。一旦去重后的序号达到 offset + limit 就 break，不必走完整个列表；但在此之前，所有向量都已经算过分、整体排过序——InMemoryStore 是暴力全量打分，没有任何 ANN 索引。
4、scoreless 兜底。没有任何向量的 Item（比如 index.fields=["text"] 但某 Item 没有 text 字段）在 filter 后仍是候选，但没得分。如果用户 limit=10 但有得分的 Item 只有 3 个，就从 scoreless 里最多补 7 个（score=None）。设计理由是"用户显式指定了 limit 就要尽量返回那么多"——哪怕部分 item 没相似度分（实跑：三条文档中一条没有 text 字段，它以 score=None 排在最后返回）。这对"先粗筛再人工 rerank"的流程很友好。
15.4.7 _cosine_similarity：numpy 可选 + 纯 Python 回退
向量相似度计算函数（libs/checkpoint/langgraph/store/memory/init.py:493）同时提供 numpy 优化路径和纯 Python 兜底：
def _cosine_similarity(X, Y):
if not Y:
return []
if _check_numpy():
import numpy as np
X_arr = np.array(X) if not isinstance(X, np.ndarray) else X
Y_arr = np.array(Y) if not isinstance(Y, np.ndarray) else Y
X_norm = np.linalg.norm(X_arr)
Y_norm = np.linalg.norm(Y_arr, axis=1)
# Avoid division by zero
mask = Y_norm != 0
similarities = np.zeros_like(Y_norm)
similarities[mask] = np.dot(Y_arr[mask], X_arr) / (Y_norm[mask] * X_norm)
return similarities.tolist()
# 纯 Python fallback
similarities = []
for y in Y:
dot_product = sum(a * b for a, b in zip(X, y, strict=False))
norm1 = sum(a * a for a in X) ** 0.5
norm2 = sum(a * a for a in y) ** 0.5
similarity = dot_product / (norm1 * norm2) if norm1 > 0 and norm2 > 0 else 0.0
similarities.append(similarity)
return similarities四条教学点：
1、numpy 是可选依赖。LangGraph 的 checkpoint 子包不强依赖 numpy，Store 用懒检测 _check_numpy()（内部是 importlib.util.find_spec("numpy") 探测，不真的 import）。这让最小化部署（比如 serverless 场景）可以不装 numpy——代价是向量搜索走纯 Python 路径、性能明显下降（源码 warning 原文是 "may significantly impact performance"），但对开发/测试 InMemoryStore 可忽略。
2、numpy 路径里的 mask。mask = Y_norm != 0——把"所有维度为 0 的向量"排除。这些向量不能 normalize（除 0），让它们的相似度直接是 0。纯 Python 路径的 if norm1 > 0 and norm2 > 0 else 0.0 是同样的保护，只是语法看起来不像矩阵运算。
3、zip(X, y, strict=False)。Python 3.10+ 新增 strict 参数——True 时要求两侧长度一致、不等就 raise。这里传 False 是显式选择"允许长度不同"（虽然 embedding 场景下应该永远相等）——宁可少算几维也不在库内部抛异常。顺带一提，strict 参数本身要求 Python 3.10+，而 langgraph、langgraph-checkpoint、langgraph-prebuilt 三个包的 pyproject.toml 都写着 requires-python = ">=3.10"，所以这里可以放心用。如果你的生产代码要求严格校验，建议改 strict=True 以便在 embedding 维度不匹配时立刻炸出来（而不是默默计算出错误相似度）。
4、双实现路径的维护成本。一份 numpy、一份纯 Python——两条路径必须语义一致。这份代码很短所以好维护，但如果你给自己的系统写类似的双路径，记得写交叉测试：同一组输入、两条路径结果应该在浮点容差内相等。否则会出现"开发环境没装 numpy 结果 A、生产环境装了 numpy 结果 B"的诡异分叉。
15.5 Store 与 Checkpoint 的区别
15.5.1 核心对比
graph TB
subgraph "Checkpoint（检查点）"
direction TB
CP1["线程级别<br/>每个 thread_id 独立"]
CP2["快照语义<br/>保存完整状态"]
CP3["自动管理<br/>框架在每步保存"]
CP4["时间旅行<br/>可回溯到任意步骤"]
CP5["序列化格式<br/>Channel 值的快照"]
end
subgraph "Store（存储）"
direction TB
ST1["全局级别<br/>跨所有 thread 共享"]
ST2["键值语义<br/>namespace + key 定位"]
ST3["手动管理<br/>节点显式读写"]
ST4["持久化存储<br/>独立于图执行"]
ST5["结构化数据<br/>dict[str, Any] 值"]
end
| 维度 | Checkpoint | Store |
|---|---|---|
| 作用域 | 单个线程（thread_id） | 全局跨线程 |
| 数据模型 | Channel 值的完整快照 | namespace/key/value 键值对 |
| 管理方式 | 框架自动保存 | 开发者显式操作 |
| 时间语义 | 有序历史（可时间旅行） | 最新值（无历史） |
| 访问方式 | get_state() / get_state_history() |
runtime.store.get/put/search |
| 典型用途 | 对话上下文、中断恢复 | 用户偏好、知识库、跨对话记忆 |
15.5.2 互补关系
Checkpoint 和 Store 不是替代关系，而是互补的：
# 示意代码：llm、summarize 需自行定义；runtime.context 需要图声明了带 user_id 的 context_schema
from langchain_core.messages import SystemMessage
from langgraph.runtime import Runtime
def personalized_chat(state: State, runtime: Runtime) -> State:
# 从 Checkpoint 恢复的状态中获取对话历史（线程内）
messages = state["messages"]
# 从 Store 中获取用户偏好（跨线程）
prefs = runtime.store.get(
("users", runtime.context.user_id), "preferences"
)
# 结合两者生成响应
prompt = f"User prefers {prefs.value['style']}" if prefs else ""
response = llm.invoke(messages + [SystemMessage(content=prompt)])
# 更新 Store 中的记忆（跨线程持久化）
runtime.store.put(
("users", runtime.context.user_id, "memory"),
f"conv_{runtime.execution_info.thread_id}",
{"summary": summarize(messages)}
)
return {"messages": [response]}15.5.3 数据流对比
sequenceDiagram
participant User as 用户
participant Graph as 图执行
participant CP as Checkpoint
participant Store as Store
Note over CP: 线程 A 的状态
User->>Graph: 开始对话（线程 A）
Graph->>CP: 保存每步状态
Graph->>Store: 保存用户偏好
Graph-->>User: 响应
Note over CP: 线程 B 的状态
User->>Graph: 新对话（线程 B）
Graph->>CP: 读取线程 B 的状态（空）
Graph->>Store: 读取用户偏好（跨线程！）
Note over Graph: 用户偏好从线程 A 延续到线程 B
Graph-->>User: 个性化响应
15.6 runtime.store 访问模式
15.6.1 在节点中使用
def my_node(state: State, runtime: Runtime) -> State:
if runtime.store is None:
# 没有配置 Store，降级处理
return {"result": "default"}
# 读取
item = runtime.store.get(("config",), "settings")
# 写入
runtime.store.put(
("results", state["session_id"]),
"analysis",
{"score": 0.95, "category": "positive"}
)
# 搜索
similar = runtime.store.search(
("knowledge",),
query=state["question"],
limit=5
)
return {"result": [s.value for s in similar]}除了 runtime: Runtime 参数，节点函数也可以直接声明一个类型注解为 BaseStore（或 BaseStore | None）、名为 store 的参数，框架按签名把同一个 store 注入进来（签名注入表 KWARGS_CONFIG_KEYS 见第 14 章 §14.7.6）：
from langgraph.store.base import BaseStore
def remember(state: State, store: BaseStore) -> State:
store.put(("users", "u1"), "profile", {"seen": True})
return {}15.6.2 在工具中使用
通过 ToolRuntime 或 InjectedStore 注解（下面两个工具放进 ToolNode、用 compile(store=...) 编译后实跑可用）：
from datetime import datetime
from langchain_core.tools import tool
from langgraph.prebuilt import ToolRuntime
@tool
def save_memory(content: str, runtime: ToolRuntime) -> str:
"""保存一段记忆到长期存储"""
runtime.store.put(
("memories",),
f"mem_{hash(content)}",
{"content": content, "timestamp": datetime.now().isoformat()}
)
return f"Memory saved: {content[:50]}..."
@tool
def recall_memories(query: str, runtime: ToolRuntime) -> str:
"""回忆相关记忆"""
results = runtime.store.search(("memories",), query=query, limit=3)
return "\n".join(r.value["content"] for r in results)15.6.3 Store 配置
Store 在图编译时注入：
from langgraph.store.memory import InMemoryStore
store = InMemoryStore(index={"dims": 1536, "embed": my_embed_fn})
graph = (
StateGraph(State)
.add_node("process", process_node)
.set_entry_point("process")
.set_finish_point("process")
.compile(store=store) # 注入 Store
)编译后，每次 invoke/stream 时 Pregel.stream 用这个 store 构造 Runtime 并放进 config（libs/langgraph/langgraph/pregel/main.py:2657，异步路径在 :3030），准备任务时再随 CONFIG_KEY_RUNTIME 注入到每个任务的配置中（见第 14 章 §14.7.3、§14.7.5）。
15.7 跨线程记忆模式
15.7.1 用户画像累积
def update_user_profile(state: State, runtime: Runtime) -> State:
user_ns = ("users", runtime.context.user_id)
# 获取现有画像
profile = runtime.store.get(user_ns, "profile")
existing = profile.value if profile else {}
# 从对话中提取新信息
new_info = extract_user_info(state["messages"])
# 合并更新
merged = {**existing, **new_info}
runtime.store.put(user_ns, "profile", merged)
return state15.7.2 共享知识库
def research_agent(state: State, runtime: Runtime) -> State:
# 搜索已有研究成果
existing = runtime.store.search(
("research", state["topic"]),
query=state["question"],
limit=10
)
if relevant := [r for r in existing if r.score and r.score > 0.8]:
# 复用已有成果
return {"answer": synthesize(relevant)}
# 生成新的研究成果
answer = do_research(state["question"])
# 保存到共享知识库
runtime.store.put(
("research", state["topic"]),
f"q_{hash(state['question'])}",
{"question": state["question"], "answer": answer}
)
return {"answer": answer}15.7.3 命名空间设计模式
graph TB
subgraph 常见命名空间模式
U["('users', user_id)"] --> UP["用户画像"]
U --> UM["('users', user_id, 'memories')"] --> UMI["对话记忆"]
U --> UH["('users', user_id, 'history')"] --> UHI["历史记录"]
D["('docs', project_id)"] --> DI["文档索引"]
K["('knowledge', domain)"] --> KI["知识条目"]
C["('cache',)"] --> CI["缓存数据"]
end
15.8 MatchCondition 与 ListNamespacesOp
15.8.1 命名空间查询
ListNamespacesOp 支持复杂的命名空间匹配：
class MatchCondition(NamedTuple):
match_type: NamespaceMatchType # "prefix" | "suffix"
path: NamespacePath # 可包含 "*" 通配符
class ListNamespacesOp(NamedTuple):
match_conditions: tuple[MatchCondition, ...] | None = None
max_depth: int | None = None
limit: int = 100
offset: int = 0便捷方法 list_namespaces 并不直接接受 match_conditions——它只有 prefix/suffix/max_depth/limit/offset 五个关键字参数，内部再把 prefix、suffix 各包成一个 MatchCondition（libs/checkpoint/langgraph/store/base/__init__.py:938）。写成 store.list_namespaces(match_conditions=...) 会抛 TypeError: BaseStore.list_namespaces() got an unexpected keyword argument 'match_conditions'（实跑）。示例：
from langgraph.store.base import ListNamespacesOp, MatchCondition
# 列举所有用户命名空间
store.list_namespaces(prefix=("users",))
# 列举所有以 "v1" 结尾的命名空间
store.list_namespaces(suffix=("v1",))
# 组合条件：以 "docs" 开头且以 "draft" 结尾
store.list_namespaces(prefix=("docs",), suffix=("draft",))
# 需要手写 MatchCondition 时，直接构造 ListNamespacesOp 走 batch
store.batch([
ListNamespacesOp(
match_conditions=(
MatchCondition("prefix", ("docs",)),
MatchCondition("suffix", ("draft",)),
)
)
])[0]15.8.2 SearchOp：自然语言检索 + MongoDB 风格过滤
ListNamespacesOp 之外、Store 还有另一个查询原语 SearchOp，§15.2.5 只列了它的字段，这里展开。打开 libs/checkpoint/langgraph/store/base/__init__.py:203：
class SearchOp(NamedTuple):
namespace_prefix: tuple[str, ...]
filter: dict[str, Any] | None = None
limit: int = 10
offset: int = 0
query: str | None = None # ← 自然语言查询
...两种独立的检索维度：
1. filter 字段——MongoDB 风格过滤算子（源码 docstring 明确列出 6 种）：
$eq Equal to（同直接值比较）
$ne Not equal to
$gt Greater than
$gte Greater than or equal to
$lt Less than
$lte Less than or equal to用法（注意 search 的第一个参数是仅限位置参数——签名里跟着 /，写成 namespace_prefix=... 会抛 TypeError）：
store.search(
("documents",),
filter={
"status": "active", # ← 直接值 = $eq
"priority": {"$gte": 5}, # ← 算子比较
"type": {"$ne": "draft"},
}
)$ 前缀的算子写法来自 MongoDB 的查询语法，在文档数据库里很常见，用户上手成本低、跨存储后端也能复用同一套 filter 语法。PostgresStore / SqliteStore / InMemoryStore 三种 Store 实现都要能接同样的 filter——这个算子抽象就是那层共用 API。
但"同一套语法"不等于"同一套语义"——这里有个只看 docstring 发现不了的坑。 四个比较算子在三个后端的实现方式并不一致：
- InMemoryStore 的
_apply_operator（libs/checkpoint/langgraph/store/memory/__init__.py:577-592）对$gt/$gte/$lt/$lte走float(value) > float(op_value)——数值比较。注意它的 docstring 自称 "matching PostgreSQL's JSONB behavior"。 - SqliteStore 的
_get_filter_condition（libs/checkpoint-sqlite/langgraph/store/sqlite/base.py:634）对数值显式CAST(json_extract(value, '$.key') AS REAL) > ?，注释还特意写了 "SQLite needs to compare as numbers, not strings"——也是数值比较。 - PostgresStore 的同名方法（
libs/checkpoint-postgres/langgraph/store/postgres/base.py:622-636）却是value->>%s > %s配str(value)——->>取出来的是 text，两边都按字符串字典序比。
于是同一条 filter={"score": {"$gte": 10}}，对 score = 9 的记录，内存与 SQLite 后端判 False，Postgres 后端判 True（因为 '9' >= '10'）。上面那句"自称匹配 PostgreSQL 行为"的 docstring，正是注释是断言不是证据的典型：真正算数的是 ->>。做数值范围过滤前，先在你实际用的后端上验一遍。
2. query: str 字段——自然语言语义检索：
store.search(
("users", "content"),
query="technical documentation about APIs", # ← 向量语义搜索
limit=20,
)但 docstring（line 212）写了一个重要警示：
"Natural language search support depends on your store implementation."
不是所有 Store 都支持——InMemoryStore 必须配了 index（含 embedder）才能做 semantic search；PostgresStore 需要 pgvector 扩展；SqliteStore 需要 sqlite-vec 扩展。更隐蔽的是，没配 index 时 query=... 不报错，而是被静默忽略：InMemoryStore 照常返回按 filter 过滤、按插入顺序切片的结果，score 全是 None（实跑）。用户代码如果依赖 query=...，务必先确认后端真的配了索引——这是 Store 接口设计的一个"软约束"。
filter 和 query 可以组合——先按 filter 过滤出候选，再对候选做向量打分排序（InMemoryStore 在 _filter_items 里先过滤，_batch_search 再打分）：
store.search(
("docs",),
query="machine learning", # 语义相关
filter={"year": {"$gte": 2024}}, # 年份 ≥ 2024
limit=10,
)这是典型的 "hybrid search"——在向量检索和关键字/元数据过滤之间做交叉，也是生产 RAG 系统里常见的做法。LangGraph Store 在 API 层就把这个能力内置了。
15.8.3 MatchCondition.path 的通配符 *
前面例子里 MatchCondition("suffix", ("v1",)) 是纯字符串路径。真实 path 类型支持通配符（libs/checkpoint/langgraph/store/base/__init__.py:351 docstring 示例）：
MatchCondition(match_type="suffix", path=("cache", "*"))在 InMemoryStore 的 _does_match 里，* 匹配任意单段路径。所以 ("cache", "*") 能匹配 ("cache", "v1")、("cache", "v2") 等末尾两段。组合多个 * 和固定段可以表达 docs/*/draft/* 这种结构化匹配。PostgresStore 的实现不同：它把 * 换成 SQL 的 % 再拼进 prefix LIKE ...（_namespace_to_text(..., handle_wildcards=True)），% 可以跨过 . 匹配多段——跨后端时别依赖"恰好一段"的语义。
这让命名空间组织可以采用"段化 pattern" 而不是字符串拼接——比如 ("users", user_id, "preferences") 比 f"users/{user_id}/preferences" 更容易过滤：
# 列所有用户的 preferences namespace
store.list_namespaces(prefix=("users",), suffix=("preferences",))
# 列某一用户的所有 namespace
store.list_namespaces(prefix=("users", user_id))这种 "结构化 path + tuple 段 + 通配符" 的设计让命名空间既有层次性、又便于程序化查询——比纯字符串 key 表达力强得多。
15.8.4 _handle_list_namespaces：max_depth 截断 + sorted + offset/limit 的四段管线
list_namespaces 看起来平平无奇，真实实现（libs/checkpoint/langgraph/store/memory/init.py:460）是把 4 个相关操作串在一起：
def _handle_list_namespaces(self, op: ListNamespacesOp) -> list[tuple[str, ...]]:
# 阶段 1：快照所有 namespace key，避免迭代时被并发修改
all_namespaces = list(self._data.keys())
namespaces = all_namespaces
# 阶段 2：按 match_conditions 过滤
if op.match_conditions:
namespaces = [
ns for ns in namespaces
if all(_does_match(condition, ns) for condition in op.match_conditions)
]
# 阶段 3：max_depth 截断 + 去重
if op.max_depth is not None:
namespaces = sorted({ns[: op.max_depth] for ns in namespaces})
else:
namespaces = sorted(namespaces)
# 阶段 4：offset/limit
return namespaces[op.offset : op.offset + op.limit]两个值得深看的点：
1、"Avoid collection size changing while iterating" 的注释。list(self._data.keys()) 显式把 keys 拷贝成 list——这是为了防止迭代过程中有别的线程 put/delete 导致 dict changed size during iteration 错误（这段同步代码中间没有 await，同一事件循环里的其他协程插不进来；能并发改 _data 的是线程，比如同一超步里在线程池并行执行的多个同步节点）。InMemoryStore 没加锁——它依赖 Python 的 GIL + 这种"快照拷贝"来避免这类错误。对于真正高并发场景，应该用 PostgresStore（有数据库级事务隔离）。
另一个实跑出来的细节：_data 是 defaultdict，get 一个不存在的 namespace、或 delete 掉某个 namespace 下的最后一个 key，都会在 _data 里留下一个空字典——之后 list_namespaces() 仍会列出这些已经没有任何 Item 的 namespace。所以在 InMemoryStore 上，list_namespaces 的结果只能当"曾经被访问过的 namespace"看，不能当"有数据的 namespace"。
2、max_depth 的"截断 + set 去重"。如果你有 100 个命名空间 ("users", "alice"), ("users", "bob"), ..., ("users", "zyy")——全部以 ("users",) 开头。查 max_depth=1 会把所有 ("users", X) 截成 ("users",)——set 自然去重——最终只返回 [("users",)] 一条。这对"只看顶层结构"的场景（比如管理后台列所有用户分类）很好用——不用把全部 namespace 拉回调用方就能看到"有哪些一级分类"（InMemoryStore 内部仍会遍历全部 key）。
两个阶段的组合拳让 list_namespaces 既是一个"穷举 namespace"的 API，也是一个"做 namespace 结构分析"的 API——看你传不传 max_depth。这种用一个 API 的不同参数表达不同语义的设计，若用 SQL 类比，相当于对"截断到 max_depth 的前缀"做 SELECT DISTINCT，还是对完整 namespace 做 SELECT DISTINCT——LangGraph 用 max_depth 一个字段就把两种语义合并了。
15.8.5 _apply_operator 的真实实现：6 个算子、float 强转、JSONB 语义对齐
前面 §15.2.5 和 §15.8.2 列了 6 种 filter 算子（$eq/$ne/$gt/$gte/$lt/$lte）。真实实现（libs/checkpoint/langgraph/store/memory/init.py:577）是个朴素的 if-elif：
def _apply_operator(value: Any, operator: str, op_value: Any) -> bool:
"""Apply a comparison operator, matching PostgreSQL's JSONB behavior."""
if operator == "$eq":
return value == op_value
elif operator == "$gt":
return float(value) > float(op_value)
elif operator == "$gte":
return float(value) >= float(op_value)
elif operator == "$lt":
return float(value) < float(op_value)
elif operator == "$lte":
return float(value) <= float(op_value)
elif operator == "$ne":
return value != op_value
else:
raise ValueError(f"Unsupported operator: {operator}")三条容易被忽略的性质：
1、大小比较强制转 float。$gt/$gte/$lt/$lte 四个比较操作先把两边都 float() 再比——意味着只能比数值（int/float）。如果你对日期字符串写 {"created_at": {"$gt": "2024-01-01"}}，会抛 ValueError: could not convert string to float: '2024-02-01'（实跑，报出的是 Item 里的值）。更容易踩的是缺字段：只要前缀下有任何一个 Item 没有这个字段，item.value.get(key) 取到 None，float(None) 就抛 TypeError: float() argument must be a string or a real number, not 'NoneType'，整个 search 失败（实跑），而不是跳过那条 Item。这是 InMemoryStore 的限制——PostgresStore 的 JSONB 操作符支持字符串排序，跨后端这个差异要注意。
生产代码里的应对：日期比较要么 put 时存成 Unix timestamp、要么只用 PostgresStore。docstring 第 578 行写了 "matching PostgreSQL's JSONB behavior"——但这句话对 InMemory 来说只对了一半（只对了数值比较这部分）。
2、$eq 和 $ne 用 Python 原生 ==/!=。对嵌套 dict、list 也能比较——但这依赖 Python 的 __eq__ 语义。比如 {"tags": ["a", "b"]} 和 {"tags": ["b", "a"]} 是不相等的（list 顺序敏感）；但 {"a": 1, "b": 2} 和 {"b": 2, "a": 1} 是相等的（dict 不关心 key 顺序）。这和 JSON 序列化后的等价性一致但不等于"业务等价"——如果你的业务里 ["a", "b"] 和 ["b", "a"] 应该视为相等，filter 的 $eq 帮不上忙（也没有 $in 之类的集合算子，写了会抛 ValueError: Unsupported operator: $in），只能在业务层写入前预先排序。
3、未知算子直接 raise，不是 warn。拼错 $gT（大写 T）会立刻 ValueError: Unsupported operator: $gT（实跑）——这比静默忽略好。
_compare_values（libs/checkpoint/langgraph/store/memory/init.py:551）还有一层更精彩的设计：它检查 filter_value 是不是 "算子字典"（任一 key 以 $ 开头）——
def _compare_values(item_value: Any, filter_value: Any) -> bool:
if isinstance(filter_value, dict):
if any(k.startswith("$") for k in filter_value):
return all(
_apply_operator(item_value, op_key, op_value)
for op_key, op_value in filter_value.items()
)
if not isinstance(item_value, dict):
return False
return all(
_compare_values(item_value.get(k), v) for k, v in filter_value.items()
)
... # 省略：list/tuple 逐元素比较分支，其余情况 item_value == filter_value如果 filter_value 的任一 key 以 $ 开头，整个字典当算子字典处理（所有算子必须都满足）；否则当作嵌套字段匹配。这允许你写：
filter={"score": {"$gt": 5, "$lte": 10}} # 5 < score <= 105 < score <= 10 的复合约束在单字段上被 AND 了——靠的是 all(_apply_operator(...) for ... in filter_value.items())。这是一种非常简洁的 DSL 设计——一个 dict 同时承担"值匹配/算子组"两种语义，通过 key 前缀区分。这种写法借自 MongoDB 的 filter DSL，但只实现了 6 个比较算子，远不是 MongoDB 的全集。
15.8.6 _check_numpy：懒加载 + 一次性警告 + lru_cache
为什么要专门提这个小函数（libs/checkpoint/langgraph/store/memory/init.py:479）？
@functools.lru_cache(maxsize=1)
def _check_numpy() -> bool:
if bool(util.find_spec("numpy")):
return True
logger.warning(
"NumPy not found in the current Python environment. "
"The InMemoryStore will use a pure Python implementation for vector operations, "
"which may significantly impact performance, especially for large datasets or frequent searches. "
"For optimal speed and efficiency, consider installing NumPy: "
"pip install numpy"
)
return False一个极短的函数同时示范了 Python 库开发的三条常见实践：
1、@functools.lru_cache(maxsize=1) 防止重复警告。_check_numpy 在每次向量相似度计算时都被调用——没装 numpy 的环境每次都走 else 分支发 warning，用户 log 会被刷爆。lru_cache 把函数的返回值缓存——第一次调用时 log 一次、后续所有调用直接拿缓存返回值不重复执行函数体。这是比 "module-level boolean flag" 更干净的 memoization。
2、util.find_spec("numpy") 做存在性检测不 import。不是 try: import numpy except ImportError。find_spec 在不执行 numpy init 代码的情况下就能告诉你 numpy 是否可用——快（省掉了 import numpy 本身不小的开销）且不污染 sys.modules（真正需要 numpy 时才 import）。生产 Python 库做可选依赖检测应该优先用这个模式。
3、warning 消息给出了行动建议。不只是 "NumPy not found"——还给了 "pip install numpy"。§15.9.7 的 _check_loop 也是同一种风格——错误/警告里直接给出"下一步怎么做"，而不只是陈述事实。
这个小函数和后面 §15.9.7 的 AsyncBatchedBaseStore._run 里 weakref.ref + if s := store() 的 GC 感知模式一起，构成了 LangGraph Store 模块里一系列"看起来无关紧要但让库在长寿命进程里保持干净"的小设计。这些设计不会出现在 README、用户也永远不会直接接触——但它们是一个库从"能用"到"可信赖地长期用"的分水岭。
15.8.7 tokenize_path 与 get_text_at_path：index.fields 背后的 mini-DSL
Store 的 index.fields 允许用户写路径表达式指定哪些字段要被向量索引：
store = InMemoryStore(index={
"dims": 1536,
"embed": my_embed_fn,
"fields": [
"text", # 简单字段
"metadata.title", # 嵌套字段
"tags[*]", # 数组全部元素
"authors[0]", # 指定索引
"authors[-1]", # 倒数第一
"{title,description}", # 多字段一起索引
]
})这套表达式的解析器 tokenize_path（libs/checkpoint/langgraph/store/base/embed.py:333）是个手写的字符扫描器——while i < len(path) 循环、分支处理 [、{、.、普通字符。六十多行代码实现了一个最小 JSONPath 子集：
field1.field2→ tokenize 成["field1", "field2"]tags[*]→["tags", "[*]"]（[*]作为一个 token 传下去）{title,description}→["{title,description}"]（花括号内不拆）
取值函数 get_text_at_path（libs/checkpoint/langgraph/store/base/embed.py:233）拿这些 tokens 按顺序下潜——遇到 [*] 展开 list 所有元素、遇到 {a,b} 取多个字段都返回。最终产出 list[str]，每个字符串作为一条单独的待 embed 文本。
这层抽象解决了"多字段 + 数组 + 嵌套"混合场景——("docs",) 下一条 Item 可能是：
{
"title": "Python for LLM",
"sections": [
{"heading": "Intro", "body": "..."},
{"heading": "Chat Models", "body": "..."},
],
"authors": ["Alice", "Bob"]
}配置 fields=["title", "sections[*].body"] 就能同时索引 title 和每个 section 的 body——一个 Item 产出 3 条 embedding（1 个 title + 2 个 section body，后两条的路径名是 sections[*].body.0、sections[*].body.1）。_batch_search 里的 max pooling 会在搜索时自动把这 3 条折回单 Item。
InMemoryStore 这里有个实跑踩到的坑：_extract_texts 以文本内容为字典键汇总待嵌入文本，嵌入时只送去重后的文本，写回时却按全部路径计数。所以同一批 put 里只要有两个路径取到了完全相同的字符串（比如 fields=["a", "b"]、值是 {"a": "x", "b": "x"}），就会抛 ValueError: Number of embeddings (1) does not match number of indices (2)。配多个 fields 时，要留意字段内容可能重复。
这套小 DSL 的价值：用户不用把文档拆成多个 Item 就能分字段索引。PostgresStore 和 SqliteStore 也用同样的 tokenize_path——DSL 在 libs/checkpoint/langgraph/store/base/embed.py 一处实现，三种后端都能共用。这是 LangGraph 的典型风格：核心工具函数放 base 子包、后端只专注 I/O 差异。
15.9 设计决策
15.9.1 为什么 Store 接口基于 batch？
BaseStore 的核心是 batch 方法，而非独立的 get/put 方法。这个设计的动机是网络效率：在生产环境中，Store 通常是远程服务（如 PostgresStore），每次 RPC 调用都有网络延迟。batch 操作允许将多个读写请求合并为一次网络往返：
# 效率更高的批量操作
results = store.batch([
GetOp(("users", "alice"), "prefs"),
GetOp(("users", "alice"), "profile"),
PutOp(("logs",), "entry_1", {"action": "login"}),
])便捷方法（get、put 等）内部将单个操作包装为长度为 1 的 batch 调用，保持 API 简洁的同时不损失性能优化的可能。
15.9.2 为什么 Item.value 是 dict 而非 Any？
Item.value 固定为 dict[str, Any] 类型，不像 Send.arg 那样接受任意类型。原因有三：
- 可过滤性：
SearchOp.filter需要按 key 做精确匹配，dict 结构天然支持 - 可索引性：向量搜索需要按字段路径提取文本，dict 结构便于字段定位
- 序列化安全：dict 可以直接 JSON 序列化，而任意 Python 对象不行
15.9.3 InMemoryStore 的定位
InMemoryStore 是开发和测试的首选，但不适合生产环境——数据在进程退出时丢失。生产环境应使用 PostgresStore 或 SqliteStore：
# 开发
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
# 生产（需另装 langgraph-checkpoint-postgres）
from langgraph.store.postgres import PostgresStore
with PostgresStore.from_conn_string("postgresql://...") as store:
store.setup() # 首次使用时建表、跑迁移
graph = builder.compile(store=store)PostgresStore 的构造函数接收的是一个现成的连接（conn），并没有 connection_string 参数；从连接串创建要用上下文管理器 from_conn_string（libs/checkpoint-postgres/langgraph/store/postgres/base.py:756）。
所有实现都遵循 BaseStore 接口，节点代码无需修改即可切换后端。
15.9.4 PostgresStore 的 schema：主表 + 向量表 + ON DELETE CASCADE
为了让"内存 vs 持久化后端"的差别不止停留在口头上，打开 PostgresStore 源码 libs/checkpoint-postgres/langgraph/store/postgres/base.py:62（与本专栏 pin 同一个 tag 1.1.6，包版本 langgraph-checkpoint-postgres 3.0.5）——MIGRATIONS 里能看到它真实的表结构：
-- MIGRATIONS[0]：主表
CREATE TABLE IF NOT EXISTS store (
prefix text NOT NULL, -- namespace 拼接
key text NOT NULL, -- Item key
value jsonb NOT NULL, -- value 存成 jsonb
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (prefix, key)
);
-- MIGRATIONS[1]：前缀索引（text_pattern_ops 让 LIKE 'prefix%' 能走 index）
CREATE INDEX CONCURRENTLY IF NOT EXISTS store_prefix_idx
ON store USING btree (prefix text_pattern_ops);
-- MIGRATIONS[2]：TTL 支持（后加的）
ALTER TABLE store
ADD COLUMN IF NOT EXISTS expires_at TIMESTAMP WITH TIME ZONE,
ADD COLUMN IF NOT EXISTS ttl_minutes INT;
-- MIGRATIONS[3]：只对 expires_at IS NOT NULL 建 partial index
CREATE INDEX IF NOT EXISTS idx_store_expires_at ON store (expires_at)
WHERE expires_at IS NOT NULL;
-- VECTOR_MIGRATIONS[0]：启用 pgvector 扩展（真实写法是幂等的 DO 块）
DO $$
BEGIN
IF NOT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'vector') THEN
CREATE EXTENSION vector;
END IF;
END $$;
-- VECTOR_MIGRATIONS[1]：向量子表（每个字段一行）
CREATE TABLE IF NOT EXISTS store_vectors (
prefix text NOT NULL,
key text NOT NULL,
field_name text NOT NULL,
embedding %(vector_type)s(%(dims)s), -- 类型/维度在建表时由 index_config 填入
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (prefix, key, field_name),
FOREIGN KEY (prefix, key) REFERENCES store(prefix, key) ON DELETE CASCADE
);五条设计值得拆开细看：
1、namespace tuple 被拼成 prefix 一列。内存版用的 dict[tuple, ...] 在 Postgres 里要扁平化成一个字符串——tuple ("users", "alice") 变成 "users.alice"（这也就是 §15.3.1 禁止 label 里含 . 的根本原因）。PRIMARY KEY (prefix, key) 保证 namespace+key 的唯一性，和内存版的 _data[namespace][key] 语义等价。
2、prefix text_pattern_ops 索引。Postgres 默认 btree 索引用 locale-aware 比较——对 LIKE 'users.alice%' 这种前缀匹配不能走索引。text_pattern_ops 操作符类告诉 Postgres "这列逐字符比较、用于 LIKE 前缀匹配时能走 index"。§15.8.1 的 MatchCondition("prefix", ("users",)) 在 Postgres 后端就对应一条 WHERE prefix LIKE 'users%'（search 的 namespace 前缀也一样）——没有这个操作符类就得全表扫描。注意拼出来的模式末尾没有 .，所以前缀 ("users",) 在 Postgres 里也会匹配到 ("users2", ...)，而 InMemoryStore 按 tuple 逐段比较、不会；两个后端在这里语义不同。
3、TTL 是后加特性（MIGRATIONS[2]/[3]）。原始 schema 没有 expires_at，后来才 ALTER 加列——这与 §15.2.6 supports_ttl 默认 False 的取向一致：TTL 是 opt-in 特性、后引入、不是所有实现都支持。同一个 Store 可以在升级到新 LangGraph 时自动跑迁移开启 TTL（靠 MIGRATIONS 自增）。
4、partial index 只索引"有 TTL"的行。CREATE INDEX ... WHERE expires_at IS NOT NULL 是 Postgres partial index——不索引 NULL 行。大部分 Store 条目没 TTL（永不过期）——索引它们纯粹是浪费空间和维护成本。只给需要被后台扫描器 sweep 的那些建索引——这类 partial index 是 Postgres 优化器的一条典型实践。
5、ON DELETE CASCADE 让删除时的向量清理由数据库保证。Postgres schema 的 store_vectors 表引用 store 表的 PK——删除主表一行时，向量子表对应行会自动级联删。这对应 InMemoryStore 在 _apply_put_ops 里手动 self._vectors[namespace].pop(key, None) 的那一步，但由数据库内核保证一致性，不会因为中间出错留下半清半不清的状态。不过它只管删除：§15.4.4 说的"更新后旧字段向量残留"，因为更新走的是 ON CONFLICT DO UPDATE 而不是删行，CASCADE 管不到，Postgres 后端同样存在。
这些 schema 细节解释了"为什么 PostgresStore 在生产更稳"的每一条底层理由。如果未来你要自己写一个 Store 后端（比如 RedisStore），这份 MIGRATIONS 是最好的起点清单——MIGRATIONS 4 条 + VECTOR_MIGRATIONS 3 条，一共 7 条 SQL 换来一个生产级 Store（VECTOR_MIGRATIONS 的第 3 条是 ANN 索引，index_type 为 flat 时按 condition 跳过）。
15.9.5 TTL 与数据清理
Store 支持可选的 TTL（Time-To-Live）机制。GetOp 和 SearchOp 的 refresh_ttl 参数控制读取操作是否刷新 Item 的存活时间。这对于缓存场景特别有用——频繁访问的数据自动延长存活，冷数据自然过期。
stateDiagram-v2
[*] --> Active: store.put(..., ttl=60)（单位：分钟）
Active --> Accessed: store.get(refresh_ttl=True)
Accessed --> Active: TTL 重置
Active --> Expired: TTL 到期
Expired --> [*]: sweeper 定期删除
15.9.6 TTLConfig：3 个字段各控一条语义
TTL 的真实配置结构 TTLConfig（libs/checkpoint/langgraph/store/base/init.py:545）是个 TypedDict，只有三个字段：
class TTLConfig(TypedDict, total=False):
refresh_on_read: bool
"""若 True，读操作（get/search）默认也刷新 TTL。
可在单次操作上用 refresh_ttl 覆盖。默认 True。"""
default_ttl: float | None
"""单位是分钟。新 Item 若没显式指定 ttl，用这个默认。
None 表示不过期。"""
sweep_interval_minutes: int | None
"""后台扫描清理过期条目的间隔（分钟）。
默认 None（不扫描）。"""（上面的 docstring 是中文意译，原文见源码。）
三个字段的设计表达了三条独立但相关的语义：
1、refresh_on_read 控制读是否 sliding TTL。如果一个缓存类场景希望"热数据自动延长 TTL、冷数据自然过期"——就用默认的 refresh_on_read=True。如果是审计日志类场景"创建时间固定、到期必须消失"——设 refresh_on_read=False，读操作不重置计时器。这是 LRU 和 TTL 两种缓存语义的融合点。
2、default_ttl 是"没指定 ttl 的新 Item 用什么默认"（_ensure_ttl 在调用方没传 ttl 时取它，libs/checkpoint/langgraph/store/base/__init__.py:1288）——防止忘记指定 TTL 导致数据堆积。生产环境可以考虑设一个兜底值，比如 default_ttl=60*24*7（7 天）。注意计时从"最后一次访问"算起：在 refresh_on_read=True 下，一条每隔几天就被读一次的 Item 会一直续命，7 天兜底的是"连续 7 天没人碰"的数据，而不是"写入满 7 天"的数据。
3、sweep_interval_minutes 控制"多久扫一次、物理删除已过期的行"。要注意两点（以 pin tag 下的 PostgresStore/SqliteStore 实现为准）：一是 sweeper 不会自动启动，必须显式调用 start_ttl_sweeper()，此时这个字段为 None 则按 5 分钟（libs/checkpoint-postgres/langgraph/store/postgres/base.py:823-848）；二是 get/search 的 SQL 并不按 expires_at 过滤，已过期但还没被扫掉的行照样会被读出来——TTL 是 docstring 所说的 "best-effort" 删除，而不是读时即刻失效。InMemoryStore 则完全不支持 TTL，supports_ttl=False（§15.2.6），传 ttl 直接报错。
三个字段 + supports_ttl capability flag 合起来构成了一套极简的 TTL 系统——相比 Redis 丰富的过期相关命令，LangGraph Store 选了最小可行的特性集：没有 conditional expire，也没有区分"写时过期"和"读时过期"的双时钟。这种克制对一个面向 LLM 长期记忆场景的系统是合理的——memory 不该有毫秒级过期需求，让配置表面简洁能降低用户写错概率。
这条"最小可行"和 §15.2.2 Store 只暴露 4 种操作原语是同一套克制思路：4 种原语 + 一套可组合的 filter DSL，而不是一大堆专用 API。读到这一层，你就能理解整个 Store 模块的取向——简单 > 强大，因为强大的东西更难用对。
15.9.7 AsyncBatchedBaseStore：同一 tick 内自动聚合的后台批处理
前面讲 Store 的 batch 是"显式批量"——用户写 store.batch([op1, op2, op3]) 才会聚合。LangGraph 还有一个隐式批量类型 AsyncBatchedBaseStore（libs/checkpoint/langgraph/store/base/batch.py:58），用于把异步环境下同一个事件循环 tick 内发起的独立 op 自动聚合为一次 abatch。实现值得读：
class AsyncBatchedBaseStore(BaseStore):
__slots__ = ("_loop", "_aqueue", "_task")
def __init__(self) -> None:
super().__init__()
self._loop = asyncio.get_running_loop()
self._aqueue: asyncio.Queue[tuple[asyncio.Future, Op]] = asyncio.Queue()
self._task: asyncio.Task | None = None
self._ensure_task()
def _ensure_task(self) -> None:
if self._task is None or self._task.done():
self._task = self._loop.create_task(_run(self._aqueue, weakref.ref(self)))每个 aget/aput/asearch 都把 (future, op) 塞进 _aqueue，后台协程 _run（libs/checkpoint/langgraph/store/base/batch.py:326）负责消费（简化：省略了 abatch 抛异常时给所有 future set_exception 的分支，以及 finally: del s 释放强引用）：
async def _run(aqueue, store):
while item := await aqueue.get():
if item[0].done(): continue
if s := store(): # ← 弱引用取回 store
items = [item]
try:
while item := aqueue.get_nowait(): # ← 抽空本 tick 所有 op
if item[0].done(): continue
items.append(item)
except asyncio.QueueEmpty:
pass
futs = [item[0] for item in items]
values = [item[1] for item in items]
listen, dedupped = _dedupe_ops(values) # ← 跨 future 去重
results = await s.abatch(dedupped)
if listen is not None:
results = [results[ix] for ix in listen]
for fut, result in zip(futs, results, strict=False):
if not fut.done():
fut.set_result(result)
else:
break # ← store 被 GC 了，后台 task 自行退出四条值得看懂的设计：
1、while await aqueue.get(): ... try: while aqueue.get_nowait() 两级抓取。第一次 await get() 阻塞等第一条 op 到达；拿到后立刻用 get_nowait() 在不 await 的情况下把队列里所有剩余的 op 也抓过来。这种 "等 1 个 + 搜刮剩下所有" 是 asyncio 社区里最经典的微批（micro-batching）套路——既不因为太激进批量化而增加延迟，也不因为太保守每次只处理 1 条而失去批量优势。
2、weakref.ref(self) + _ensure_task 的生命周期策略。后台 _run 对 store 用弱引用。如果用户的 store 对象被 GC（比如测试代码局部变量超出作用域），后台协程会 if s := store(): ... else: break 自行退出——不会因为 task 持有强引用导致 store 永远不被释放。__del__ 里额外 self._task.cancel() 兜底；_run 每批处理完还会在 finally 里 del s，不让自己长期握着 store 的强引用。这套组合让用户不用手动停止这个后台 task——store 被回收，task 就跟着结束（数据库连接的关闭仍要靠 from_conn_string 的上下文管理器等方式自己管）。
3、_dedupe_ops 跨 future 去重。同一 tick 内如果有两个业务代码调用 aget(("users","alice"), "prefs")——两条 GetOp 参数完全一样。_dedupe_ops（libs/checkpoint/langgraph/store/base/batch.py:283）的做法：
if isinstance(op, (GetOp, SearchOp, ListNamespacesOp)):
try:
listen.append(dedupped.index(op)) # ← 已有相同 op，复用
except ValueError:
listen.append(len(dedupped))
dedupped.append(op)
elif isinstance(op, PutOp):
putkey = (op.namespace, op.key)
if putkey in puts:
# Overwrite previous put
ix = puts[putkey]
dedupped[ix] = op # ← 同 key 多次 put，取最后一个
listen.append(ix)读操作（Get/Search/List）看到重复直接复用已有 op 的结果；写操作（Put）看到同 (ns, key) 的两次 Put 把后者覆盖前者。listen 数组维护"每个 future 要读 results 的哪个索引"——查询完再按 listen 把结果映射回各自的 future。
这层去重对 agent 场景极其有用——同一个事件循环 tick 内落到队列里的 N 次「读同一个用户偏好」只会产生 1 次后端调用，但每个调用方的 future 都拿到正确的结果。
4、_check_loop 装饰器防死锁。同步接口（如 store.get()）如果在自己的事件循环里被调用会阻塞——_check_loop（libs/checkpoint/langgraph/store/base/batch.py:33）检测到 asyncio.get_running_loop() is store._loop 就立刻 raise InvalidStateError，错误消息直接告诉用户改成 await store.aget(...)——这种"错误消息内置迁移建议"的做法和第 16 章 ToolNode 的 TOOL_CALL_ERROR_TEMPLATE（libs/prebuilt/langgraph/prebuilt/tool_node.py:109）是一个思路：错误消息不是日志、是下一步指导。
这套 AsyncBatchedBaseStore 是异步 Store 后端的基类（AsyncPostgresStore、AsyncSqliteStore 都继承它）——用户代码写 await runtime.store.aget(...)、底层自动变成跨并发请求的批量。不改一行业务代码就把"同一 tick 内的 N 次独立请求"压成一次 abatch——高并发下省掉的往返次数随聚合度增加。
15.9.8 IndexConfig.embed：4 种类型、一套 ensure_embeddings 归一
Store 的 embed 字段定义（libs/checkpoint/langgraph/store/base/init.py:590）接受 4 种完全不同形态的输入：
embed: Embeddings | EmbeddingsFunc | AEmbeddingsFunc | str- LangChain 的
Embeddings类实例（init_embeddings("openai:text-embedding-3-small")返回的） - 同步函数
(list[str]) -> list[list[float]]（直接用 OpenAI SDK 自己写） - 异步函数
(list[str]) -> Awaitable[list[list[float]]]（用 AsyncOpenAI 写） - 字符串 provider 标识符（
"openai:text-embedding-3-small"——LangGraph 自己帮你调init_embeddings）
这四种怎么归一的？ensure_embeddings（embed.py:34）统一把它们包装成 Embeddings 接口：
- 已经是 Embeddings 实例——原样返回
- 是同步/异步函数——包成
EmbeddingsLambda（embed.py:109） - 是字符串——调用
langchain.embeddings.init_embeddings("...")——但这会可选依赖 langchain（_get_init_embeddings里 try-import，embed.py:420）
第 4 种 "provider 字符串" 的好处和陷阱：
好处：用户代码里写 index={"embed": "openai:text-embedding-3-small"}——比自己实例化 embedder 简洁得多。不用 import OpenAI、不用写 API key 管理代码。对"快速原型"场景非常友好。
陷阱：langgraph 自己既不依赖 langchain，也不依赖 langchain-openai（libs/langgraph/pyproject.toml 与 libs/checkpoint/pyproject.toml 的 dependencies 里都没有）。只装 pip install langgraph 再用 embed="openai:..."，在构造 store 时 ensure_embeddings 就会抛 ValueError: Could not load embeddings from string 'openai:...'. langchain is not installed; ...，提示安装 langchain>=0.3.9 和对应的 provider 包；装了 langchain 但缺 provider 包，则由 init_embeddings 自己报错。不预装 provider 客户端让核心包保持轻量，代价就是这类错误要到运行时才暴露。生产代码推荐用 1 或 2 的显式实例化，一是避开这个陷阱，二是让依赖关系在 pip install 阶段暴露而不是运行时。
同步 vs 异步函数的选择也有策略：InMemoryStore.batch() 走同步路径——只能用同步 embed；InMemoryStore.abatch() 优先用异步 embed、如果传的是同步函数会用 run_in_executor 包装。如果你的 LLM 调用全在 async loop 里，请传 async 函数——否则 embed 调用会占线程池 slot、拖慢其他异步任务。反过来，只传 async 函数时不能走同步的 put/search：EmbeddingsLambda.embed_documents 会抛 ValueError，提示改用异步接口。
这层"4 种形态统一到一个接口"的设计是 LangGraph 的常见作风——API 表面给用户最大自由，内部实现统一归一。读 LangGraph 源码时，看到一个接口接受多种输入类型，就去找"ensure_XXX"之类的归一函数，往往能很快看懂。
15.9.9 本章在整个专栏体系中的坐标
回到本专栏整体脉络——Store 是 LangGraph 四大系统组件（StateGraph、Pregel、Checkpoint、Store）里位置最边缘、但与"应用质量"关系最直接的一个。几个承接关系值得明示：
- 承接第 7 章（Pregel 调度）与第 14 章（Runtime）：Store 通过
Runtime.store和ToolRuntime.store两条通道注入到节点/工具里。注入时机是任务准备阶段——Runtime随CONFIG_KEY_RUNTIME和其他 config 键一起写进每个任务的 config（第 7 章 §7.2.3、第 14 章 §14.7.5）。理解了这一点，就能理解"为什么节点看到的 store 是同一个实例"——它不是每次新建的，是 compile 时传入的同一个引用传到每个任务。 - 承接第 8 章（Checkpoint）：本章 §15.5 画清了两者分工。Store 的数据生命周期不受 checkpoint 版本影响——你 rollback 到 checkpoint_id=X 不会把 Store 里写过的东西"撤销"。这个设计是刻意的——长期记忆不应该被会话级的时间旅行抹掉。
- 支撑第 16 章（预构建 agent）：
InjectedStore注解和ToolRuntime.store字段（第 16.6.2/16.6.3 节）都依赖本章 BaseStore 的 API 契约。_inject_tool_args里对 store 参数的"剥离 + 补回"安全层（第 16.3.11 节）防的就是"LLM 伪造 store"——这条防线的存在预设了 store 是框架受控资源、不是业务数据。 - 铺垫第 17 章（multi-agent）：多 agent 架构里最常见的"跨 agent 共享知识"就是 Store 的用武之地。主 agent 把 research 结果
store.put(("knowledge", topic), ...)，子 agentstore.search(("knowledge",), query=...)拿结果——Store 是 agent 之间的"共享白板"。这个用法的实现细节在第 17 章 §17.8.3 展开，本章打好的 BaseStore 四操作原语（Get/Put/Search/ListNamespaces）就是白板上的"写/读/搜/列"。
这层承接关系不是后话、而是 LangGraph 的核心设计意图。Store、Checkpoint、Pregel 三者在 compile() 时同时进入 Pregel 实例——它们天生就是配套使用的。本章把 Store 单独拎出来讲，是方便知识分层；实际写应用时你几乎总会同时用到三者。
从更宏观的软件架构视角看，Store 是 LangGraph 整个技术栈里层次最高、对业务代码最近的一层——它直接被节点/工具代码调用。Checkpoint 和 Pregel 更多是框架内部机制，用户感知较弱；Store 则是你的业务代码每天都要写的东西。理解本章揭示的 Store 内部这些工程决策——_validate_namespace 的四条铁律、AsyncBatchedBaseStore 的跨 future 去重、_cosine_similarity 的 numpy 可选路径、PostgresStore 的 ON DELETE CASCADE——能让你在写应用时做出更匹配 Store 本意的 API 调用选择。这种"框架意图驱动应用选型"的能力，是从"会用 LangGraph"到"精通 LangGraph"的关键一跃。
15.10 小结
本章详细分析了 LangGraph 的 Store 长期记忆系统。BaseStore 通过四种操作原语（Get、Put、Search、ListNamespaces）和层级化的命名空间，为 LLM 应用提供了灵活的跨线程持久化能力。InMemoryStore 以简洁的字典实现覆盖了开发测试需求，同时支持可选的向量语义搜索。
Store 与 Checkpoint 的关系是互补的：Checkpoint 管理线程内的状态历史（短期记忆），Store 管理跨线程的持久化数据（长期记忆）。通过 runtime.store 的统一访问接口，节点和工具可以无缝地读写长期记忆，而底层存储后端（内存、SQLite、PostgreSQL）可以透明切换。
batch 接口的设计体现了对生产环境网络效率的重视；dict[str, Any] 的值类型约束确保了可过滤性、可索引性和序列化安全；层级命名空间为多租户和多项目场景提供了自然的组织结构。
下一章我们将进入 LangGraph 的预构建组件层——create_react_agent 工厂函数、ToolNode 实现和 tools_condition 路由——了解框架如何将底层的图构建原语封装为开箱即用的 Agent 架构。
本章源码定位：主要引用来自 libs/checkpoint/langgraph/store/ 目录的四个 Python 文件——libs/checkpoint/langgraph/store/base/__init__.py（1314 行，定义 BaseStore/Item/Op/TTLConfig/IndexConfig 和校验函数）、libs/checkpoint/langgraph/store/base/batch.py（371 行，AsyncBatchedBaseStore 和 _dedupe_ops）、libs/checkpoint/langgraph/store/base/embed.py（433 行，ensure_embeddings/tokenize_path/get_text_at_path）、libs/checkpoint/langgraph/store/memory/__init__.py（592 行，InMemoryStore 的全部实现）。Postgres 后端细节来自同一 tag 下的 libs/checkpoint-postgres/langgraph/store/postgres/base.py（langgraph-checkpoint-postgres 3.0.5；MIGRATIONS、VECTOR_MIGRATIONS、filter 与 TTL 实现），SQLite 对照来自 libs/checkpoint-sqlite/langgraph/store/sqlite/base.py（langgraph-checkpoint-sqlite 3.0.3）。所有引用都给了准确的行号锚点——存疑可直接对照核验。
本章和第 8 章（Checkpoint）在系统组件图里是"双持久化"：短期在 Checkpoint、长期在 Store；本章和第 16 章（预构建 agent）、第 17 章（multi-agent）是"对接用户代码"的前置知识——理解了 Store 的四原语 + TTLConfig/IndexConfig 两配置 + AsyncBatchedBaseStore 自动聚合，后面读 create_react_agent 里 store= 参数传进去做什么才不会停留在"被神秘注入"的模糊状态。生产环境向量搜索通常要关注：pgvector 扩展启用、HNSW/IVF 参数调优、dims 匹配 embedding 模型输出、fields 路径覆盖实际索引的字段——这几个旋钮正是本章反复出现的配置项。