# store/base/__init__.py:700
class BaseStore(ABC):
    """Abstract base class for persistent key-value stores.

    Stores enable persistence and memory that can be shared across threads,
    scoped to user IDs, assistant IDs, or other arbitrary namespaces.
    Some implementations may support semantic search capabilities through
    an optional `index` configuration.
    """
    supports_ttl: bool = False        # 是否支持"到期自动删"，默认关

Store 的每条记忆由 namespace（文件夹路径）+ key（文件名） 定位。看 PutOp 里对 namespace 的定义（store/base/__init__.py:431-475）：

# store/base/__init__.py:438
class PutOp(NamedTuple):
    namespace: tuple[str, ...]
    """Hierarchical path that identifies the location of the item.
       The namespace acts as a folder-like structure to organize items.
       Examples:
           ("documents",)                    # 根级
           ("documents", "user123")          # 用户专属
           ("cache", "embeddings", "v1")     # 多层嵌套
    """
    key: str
    """Unique identifier within the namespace."""
    value: dict[str, Any] | None
    """The data to store, or None to mark the item for deletion."""   # ← value=None 表示删除
    index: Literal[False] | list[str] | None = None
    """要不要为语义搜索建索引、索引哪些字段（L05）"""

BaseStore 只强制子类实现两个抽象方法：batch 和 abatch。其它 get/search/put 全是"包装成 op 丢给 batch"（store/base/__init__.py:724-769）：

# store/base/__init__.py:724
@abstractmethod
def batch(self, ops: Iterable[Op]) -> list[Result]:
    """Execute multiple operations synchronously in a single batch."""

@abstractmethod
async def abatch(self, ops: Iterable[Op]) -> list[Result]: ...

# store/base/__init__.py:748  —— get 只是包一个 GetOp 丢给 batch
def get(self, namespace, key, *, refresh_ttl=None) -> Item | None:
    return self.batch(
        [GetOp(namespace, str(key), _ensure_refresh(self.ttl_config, refresh_ttl))]
    )[0]                                      # 取结果列表第 0 个

看最简实现 InMemoryStore 的 batch 主干，理解 put/search 怎么落地（store/memory/__init__.py:206-219）：

# store/memory/__init__.py:206
def batch(self, ops: Iterable[Op]) -> list[Result]:
    results, put_ops, search_ops = self._prepare_ops(ops)   # ① 按类型分组 + 先处理 get
    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))  # 批量算 embedding
        self._insertinmem_store(to_embed, embeddings)                 # 存进向量表
    self._apply_put_ops(put_ops)                                   # ④ 真正写/删数据
    return results

# store/memory/__init__.py:404  —— put 的落地：value=None 就删，否则存 Item
def _apply_put_ops(self, put_ops) -> None:
    for (namespace, key), op in put_ops.items():
        if op.value is None:
            self._data[namespace].pop(key, None)     # value=None → 删除
        else:
            self._data[namespace][key] = Item(...)   # 否则存/覆盖一个 Item

Store 最有魔力的是"按意思找"。核心就是把查询和候选都变成向量，算余弦相似度（store/memory/__init__.py:493-519）：

# store/memory/__init__.py:493
def _cosine_similarity(X: list[float], Y: list[list[float]]) -> list[float]:
    """Compute cosine similarity between a vector X and a matrix Y."""
    try:
        import numpy as np
        X_arr = np.array(X); Y_arr = np.array(Y)
        X_norm = np.linalg.norm(X_arr)
        Y_norm = np.linalg.norm(Y_arr, axis=1)
        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()
    except ImportError:                                     # 没装 numpy → 纯 Python 兜底
        # norm1 = sum(a*a for a in X)**0.5 ; 逐个算 dot/(norm1*norm2)
        ...

真实搜索不止"算相似度"，还要先过滤、再排序、去重、分页。看 _batch_search 的排序段（store/memory/__init__.py:324-350）：

# store/memory/__init__.py:324
scores = _cosine_similarity(query_embedding, flat_vectors)
sorted_results = sorted(zip(scores, flat_items), key=lambda x: x[0], reverse=True)  # 按分降序
# max pooling：同一条记忆可能有多个向量（多字段/多段），只保留它最高的那次命中
seen: set[tuple[tuple[str, ...], str]] = 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))

阶段 6 八天走完，用一张表把两套持久化设施钉在一起，这是你脑中该常驻的对照：

✋ 10 分钟动手

# 1. BaseStore 抽象 + get/search/put 签名
sed -n '700,769p' libs/checkpoint/langgraph/store/base/__init__.py

# 2. InMemoryStore 的 batch 分派 + 余弦相似度
sed -n '206,219p' libs/checkpoint/langgraph/store/memory/__init__.py
sed -n '493,519p' libs/checkpoint/langgraph/store/memory/__init__.py

# 3. 亲手用 Store 存取（不带语义搜索，无需 embedding）
python - <<'PY'
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
store.put(("users","u42","prefs"), "diet", {"text":"不吃辣"})
print("精确取:", store.get(("users","u42","prefs"), "diet").value)
print("按前缀搜:", [i.key for i in store.search(("users","u42"))])
PY