第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：

15.2.3 GetOp

class GetOp(NamedTuple):
    namespace: tuple[str, ...]
    key: str
    refresh_ttl: bool = True

GetOp 通过 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: list[str] | Literal[False] | None = None

PutOp 用于创建、更新或删除 Item。当 value 为 None 时表示删除。index 字段控制哪些值路径需要被向量索引。

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 = True

SearchOp 是最灵活的查询操作。它支持三种过滤方式：

1. 命名空间前缀：只搜索指定前缀下的 Item
2. 结构化过滤：通过 filter 字典做精确匹配或比较操��
3. 语义搜索：通过 query 字符串做向量相似度搜索

过滤操作支持多种比较运算符：

# 精确匹配
{"status": "active"}

# 比较运算符
{"score": {"$gt": 4.99}}
{"score": {"$gte": 3.0}}
{"priority": {"$lt": 5}}
{"count": {"$lte": 100}}
{"status": {"$ne": "deleted"}}

15.3 Item 数据模型

15.3.1 结构定义

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.2 命名空间的层级设计