第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 在定位、生命周期和使用场景上的本质��别。

本章要点

1. BaseStore 接口的操作原语——Get、Put、Search、ListNamespaces 四种操作
2. Item 数据模型——key/value/namespace/created_at/updated_at
3. InMemoryStore 的实现——字典存储 + 可选向量搜索
4. Store 与 Checkpoint 的区别——跨线程 vs 线程内、键值 vs 快照
5. runtime.store ��访问模式——从节点和工具中使用 Store

15.2 BaseStore 接口

15.2.1 接口定义

BaseStore 定义在 langgraph/store/base/__init__.py 中，是一个抽象基类，定义了 Store 的全部操作原语：

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：