有条理的详细介绍 langgraph.store.base 模块

langgraph.store.base 模块是 LangGraph 框架中用于定义持久化键值存储的核心模块，提供了标准化的接口和数据结构，以支持状态管理和长时记忆存储。它是 LangGraph 的重要组成部分，特别适合构建复杂、状态化的多代理应用。本文将从背景、功能、主要组件、使用方法、实际应用及注意事项等方面，详细介绍该模块，帮助开发者理解其设计和应用场景。

---

背景与定义

LangGraph 是 LangChain 生态系统中的一个框架，专注于构建可靠、状态化的 AI 系统，支持代理编排、长时记忆、循环工作流和人类交互。作为低层次的控制工具，它为开发者提供了高度的扩展性，特别适用于多代理系统和复杂任务。在这些应用中，状态和记忆管理是关键需求，确保代理能够在不同会话和线程中保持上下文。

langgraph.store.base 模块是 LangGraph 中负责定义存储接口的核心模块，提供了 BaseStore 抽象基类以及一系列辅助类和类型。它旨在为开发者提供创建自定义存储的接口，支持数据的持久化存储和高效管理。该模块在 langchain-core 0.2.14 版本中引入，相关信息可参考官方文档：LangGraph 存储文档。

---

模块功能

langgraph.store.base 模块的主要功能是为 LangGraph 提供持久化的键值存储，支持跨会话和线程的数据管理。其核心特性包括：

• 持久化存储：确保数据在会话间持久存在，适合长时记忆存储。
• 层次命名空间：通过元组（如 ("users", "123", "messages")）组织数据，支持结构化隔离。
• 批量操作：支持高效的批量读写和删除操作。
• 可选语义搜索：提供基于查询和过滤的语义搜索功能（默认禁用）。
• TTL 支持：支持数据自动过期（默认禁用）。
• 分页与过滤：支持查询结果的分页和条件过滤，优化性能。

这些功能使其成为管理代理状态、会话历史或中间结果的理想选择。

---

主要组件

以下是 langgraph.store.base 模块的核心组件及其功能：

1. BaseStore 抽象基类

• 定义：BaseStore 是存储的核心接口，继承自 Python 的 ABC（抽象基类），开发者需要通过继承它来实现具体存储。
• 属性：
  • supports_ttl: bool = False：指示是否支持 TTL（时间到期）。
  • supports_index: bool = False：指示是否支持语义搜索索引。
• 主要方法：
  • batch(ops: Iterable[Op]) -> List[Result]：批量执行存储操作。
  • get(namespace: Tuple[str, ...], key: str) -> Optional[Item]：获取指定键值。
  • put(namespace: Tuple[str, ...], key: str, value: Dict[str, Any]) -> None：存储键值对。
  • delete(namespace: Tuple[str, ...], key: str) -> None：删除指定键值。
  • search(namespace_prefix: Tuple[str, ...], query: Optional[str], filter: Optional[Dict[str, Any]]) -> List[SearchItem]：搜索键值对。
  • list_namespaces(prefix: Optional[NamespacePath]) -> List[Tuple[str, ...]]：列出命名空间。
  • 异步版本：如 abatch、aget、aput 等，提供异步支持。
• 作用：定义存储的基本行为，需子类实现具体逻辑。

2. Op 类

• 定义：表示存储操作的数据类。
• 属性：
  • type: Literal["get", "put", "delete"]：操作类型。
  • namespace: Tuple[str, ...]：命名空间路径。
  • key: str：键。
  • value: Optional[Dict[str, Any]]：值（仅用于 put）。
• 作用：封装操作细节，用于批量操作。

3. Item 类型

• 定义：类型别名，通常为 Dict[str, Any]，表示存储的键值对。
• 作用：作为存储和获取操作的基本数据单位。

4. SearchItem 类

• 定义：表示搜索结果的数据类。
• 属性：
  • namespace: Tuple[str, ...]：命名空间。
  • key: str：键。
  • value: Item：值。
• 作用：封装搜索返回的结果。

5. NamespacePath 类型

• 定义：类型别名，为 Tuple[str, ...]，表示层次命名空间路径。
• 示例：("users", "123", "documents")。
• 作用：支持数据的结构化组织。

6. NOT_PROVIDED 常量

• 定义：特殊标记，用于表示未提供 TTL。
• 作用：在 put 方法中区分是否设置过期时间。

---

使用方法

开发者可以通过继承 BaseStore 创建自定义存储，或使用 LangGraph 提供的现有实现（如 PostgresStore）。以下是两种使用方式的示例：

自定义存储实现

创建一个简单的内存存储：

from langgraph.store.base import BaseStore, Op, Result
from typing import Iterable, List, Optional, Tuple, Dict, Any

class MyInMemoryStore(BaseStore):
    def __init__(self):
        self.store = {}  # 使用字典存储数据
        self.supports_ttl = False
        self.supports_index = False

    def batch(self, ops: Iterable[Op]) -> List[Result]:
        results = []
        for op in ops:
            if op.type == "get":
                results.append(self.store.get((op.namespace, op.key)))
            elif op.type == "put":
                self.store[(op.namespace, op.key)] = op.value
                results.append(None)
            elif op.type == "delete":
                if (op.namespace, op.key) in self.store:
                    del self.store[(op.namespace, op.key)]
                results.append(None)
        return results

    async def abatch(self, ops: Iterable[Op]) -> List[Result]:
        return self.batch(ops)  # 简单异步实现

# 使用示例
store = MyInMemoryStore()
store.put(("users", "123", "messages"), "msg1", {"content": "Hello"})
item = store.get(("users", "123", "messages"), "msg1")
print(item)  # 输出: {"content": "Hello"}

使用现有实现

使用 PostgresStore 连接 PostgreSQL 数据库：

from langgraph.store.postgres import PostgresStore

store = PostgresStore(connection_string="postgresql://user:password@localhost/db")
store.put(("users", "123", "messages"), "msg1", {"content": "Hello"})
item = store.get(("users", "123", "messages"), "msg1")
print(item)  # 输出: {"content": "Hello"}

安装依赖：

pip install -qU langgraph

---

实际应用

langgraph.store.base 模块在以下场景中具有广泛应用：

• 代理记忆管理：存储代理的会话历史、用户偏好或上下文数据，确保跨会话一致性。
• 检索增强生成（RAG）应用：存储检索文档的索引或元数据，支持语义搜索。
• 状态化工作流：在 LangGraph 图中存储中间状态，支持循环或多步任务。
• 分布式系统：通过数据库后端（如 PostgresStore）实现跨节点的记忆共享。

---

注意事项与最佳实践

注意事项

• 抽象方法实现：自定义存储必须实现 batch 和 abatch 方法。
• 语义搜索和 TTL：默认禁用，需在子类中明确启用（设置 supports_index 或 supports_ttl 为 True）。
• 版本依赖：确保使用 langchain-core>=0.2.14，通过 pip install -qU langgraph 更新。
• 性能考虑：高并发场景下需优化批量操作和异步实现。

最佳实践

• 命名空间设计：使用层次命名空间（如 ("users", "id", "data")）确保数据隔离。
• 启用高级功能：根据需求启用语义搜索或 TTL。
• 调试工具：结合 LangSmith 跟踪存储操作，便于调试。
• 测试实现：验证自定义存储在目标场景下的性能和安全性。

---

结论

langgraph.store.base 模块是 LangGraph 框架中用于定义持久化键值存储的核心组件，通过 BaseStore 抽象基类及辅助类（如 Op、Item 等），提供了灵活的存储接口。它支持层次命名空间、批量操作和可选的语义搜索，适用于代理记忆管理、RAG 应用和状态化工作流等场景。开发者可以通过继承 BaseStore 或使用现有实现（如 PostgresStore），构建高效、状态化的 AI 系统。

如需进一步探索，可参考：

• LangGraph 存储文档
• LangGraph GitHub 仓库