存储 - Docs by LangChain中文

存储允许智能体跨线程持久化信息，包括用户偏好、积累的知识以及应超越单次对话而存活的事实。与检查点保存器将完整图状态保存到单个线程不同，存储持有可从任何线程访问的任意键值数据。共享状态模型

Agent Server 自动处理存储 使用 Agent Server 时，您无需手动实现或配置存储。API 会在后台为您处理所有存储基础设施。

InMemoryStore 适用于开发和测试。对于生产环境，请使用 PostgresStore、MongoDBStore 或 RedisStore 等持久化存储。所有实现都扩展了 BaseStore，这是节点函数签名中应使用的类型注解。

基本用法

以下代码片段展示了独立使用 InMemoryStore，而不使用 LangGraph：

from langgraph.store.memory import InMemoryStore
store = InMemoryStore()

记忆通过一个 tuple 来划分命名空间，在以下示例中为 (<user_id>, "memories")。命名空间可以是任意长度并表示任何内容，不必特定于用户。

user_id = "1"
namespace_for_memory = (user_id, "memories")

使用 store.put 方法将记忆保存到存储中的命名空间。指定如上定义的命名空间，以及一个记忆的键值对：键只是记忆的唯一标识符 (memory_id)，值（一个字典）就是记忆本身。

memory_id = str(uuid.uuid4())
memory = {"food_preference" : "I like pizza"}
store.put(namespace_for_memory, memory_id, memory)

使用 store.search 方法从您的命名空间中读取记忆，该方法返回给定用户的记忆列表，最多到 limit 参数（默认 10）。使用 InMemoryStore 时，项目按插入顺序返回，因此最新的记忆在列表末尾；其他后端可能以不同方式排序记忆（参见列出命名空间中的项目）。

memories = store.search(namespace_for_memory)
memories[-1].dict()
{'value': {'food_preference': 'I like pizza'},
 'key': '07e0caf4-1631-47b7-b15f-65515d4c1843',
 'namespace': ['1', 'memories'],
 'created_at': '2024-10-02T17:22:31.590602+00:00',
 'updated_at': '2024-10-02T17:22:31.590605+00:00'}

每种记忆类型都是一个 Python 类 (Item)，具有某些属性。我们可以通过使用 .dict 转换来将其作为字典访问。它具有的属性是：

value: 此记忆的值（本身是一个字典）
key: 此记忆在此命名空间中的唯一键
namespace: 字符串的元组，此记忆类型的命名空间
虽然类型是 tuple[str, ...]，但在转换为 JSON 时可能会序列化为列表（例如 ['1', 'memories']）。
created_at: 此记忆创建的时间戳
updated_at: 此记忆更新的时间戳

列出命名空间中的项目

调用 store.search（或异步的 store.asearch）时，不带 query 和 filter 参数，将返回存储在 namespace_prefix 下的项目，最多返回 limit 个。当您不需要语义排序时，使用此方法来枚举命名空间中的所有内容。

# Return up to 100 items stored under ("alice", "memories").
items = store.search(("alice", "memories"), limit=100)

需要注意的三种行为：

namespace_prefix 通过前缀匹配，不是精确匹配。 ("alice",) 也会返回 ("alice", "memories")、("alice", "preferences") 等下的项目。要限制到单个级别，请传递完整的命名空间或在客户端根据 item.namespace 过滤返回的项目。
超过 limit 的结果将被静默截断。 没有溢出信号——将 limit 设置为高于您预期的最大值，或使用 offset 进行分页。
默认排序取决于存储后端。 PostgresStore 和 AsyncPostgresStore 返回按 updated_at 降序排列的结果（最近更新的在前）。InMemoryStore 返回按插入顺序排列的结果（最近插入的在最后）。不要依赖特定实现间的顺序；如果顺序重要，请在客户端根据 item.updated_at 进行排序。

要对大型命名空间进行分页：

page_size = 50
offset = 0
while True:
    page = store.search(("alice", "memories"), limit=page_size, offset=offset)
    if not page:
        break
    for item in page:
        pass
    offset += page_size

要发现存在哪些命名空间（例如，在列出每个用户的记忆之前迭代所有用户），请使用 store.list_namespaces 或 store.alist_namespaces：

# All namespaces that start with ("alice",), truncated to two levels deep.
namespaces = store.list_namespaces(prefix=("alice",), max_depth=2)

语义搜索

除了简单的检索，存储还支持语义搜索，允许您根据含义而不是精确匹配来查找记忆。要启用此功能，请使用嵌入模型配置存储：

from langchain.embeddings import init_embeddings

store = InMemoryStore(
    index={
        "embed": init_embeddings("openai:text-embedding-3-small"),  # 嵌入提供者
        "dims": 1536,                              # 嵌入维度
        "fields": ["food_preference", "$"]              # 要嵌入的字段
    }
)

现在搜索时，您可以使用自然语言查询来查找相关的记忆：

# 查找关于食物偏好的记忆
# （这可以在将记忆放入存储后完成）
memories = store.search(
    namespace_for_memory,
    query="What does the user like to eat?",
    limit=3  # 返回前 3 个匹配项
)

您可以通过配置 fields 参数或在存储记忆时指定 index 参数来控制记忆的哪些部分被嵌入：

# 存储时指定要嵌入的特定字段
store.put(
    namespace_for_memory,
    str(uuid.uuid4()),
    {
        "food_preference": "I love Italian cuisine",
        "context": "Discussing dinner plans"
    },
    index=["food_preference"]  # 仅嵌入 "food_preferences" 字段
)

# 存储时不进行嵌入（仍可检索，但不可搜索）
store.put(
    namespace_for_memory,
    str(uuid.uuid4()),
    {"system_info": "Last updated: 2024-01-01"},
    index=False
)

在 LangGraph 中使用

存储与检查点保存器协同工作：检查点保存器将状态保存到线程（如上所述），而存储允许您存储任意信息以便跨线程访问。如下所示，同时使用检查点保存器和存储来编译图。

from dataclasses import dataclass
from langgraph.checkpoint.memory import InMemorySaver

@dataclass
class Context:
    user_id: str

# 我们需要这个是因为我们想要启用线程（对话）
checkpointer = InMemorySaver()

# ... 定义图 ...

# 使用检查点保存器和存储编译图
builder = StateGraph(MessagesState, context_schema=Context)
# ... 添加节点和边 ...
graph = builder.compile(checkpointer=checkpointer, store=store)

然后像之前一样使用 thread_id 调用图，同时也使用 user_id，如前所述，它充当此特定用户的记忆的命名空间。

# 调用图
config = {"configurable": {"thread_id": "1"}}

# 首先，让我们向 AI 打个招呼
for update in graph.stream(
    {"messages": [{"role": "user", "content": "hi"}]},
    config,
    stream_mode="updates",
    context=Context(user_id="1"),
):
    print(update)

您可以使用 Runtime 对象从任何节点访问存储和 user_id。当您将 Runtime 作为参数添加到节点函数时，LangGraph 会自动注入它。您可以使用它来保存记忆：

from langgraph.runtime import Runtime
from dataclasses import dataclass

@dataclass
class Context:
    user_id: str

async def update_memory(state: MessagesState, runtime: Runtime[Context]):

    # 从运行时上下文中获取用户 ID
    user_id = runtime.context.user_id

    # 为记忆划分命名空间
    namespace = (user_id, "memories")

    # ... 分析对话并创建新记忆

    # 创建新的记忆 ID
    memory_id = str(uuid.uuid4())

    # 我们创建一个新记忆
    await runtime.store.aput(namespace, memory_id, {"memory": memory})

您还可以从任何节点访问存储，并使用 store.search 方法获取记忆。记忆以对象列表形式返回，可以转换为字典。

memories[-1].dict()
{'value': {'food_preference': 'I like pizza'},
 'key': '07e0caf4-1631-47b7-b15f-65515d4c1843',
 'namespace': ['1', 'memories'],
 'created_at': '2024-10-02T17:22:31.590602+00:00',
 'updated_at': '2024-10-02T17:22:31.590605+00:00'}

您访问记忆并在模型调用中使用它们。

from dataclasses import dataclass
from langgraph.runtime import Runtime

@dataclass
class Context:
    user_id: str

async def call_model(state: MessagesState, runtime: Runtime[Context]):
    # 从运行时上下文中获取用户 ID
    user_id = runtime.context.user_id

    # 为记忆划分命名空间
    namespace = (user_id, "memories")

    # 基于最新消息进行搜索
    memories = await runtime.store.asearch(
        namespace,
        query=state["messages"][-1].content,
        limit=3
    )
    info = "\n".join([d.value["memory"] for d in memories])

    # ... 在模型调用中使用记忆

如果您创建一个新线程，只要 user_id 相同，您仍然可以访问相同的记忆。

# 在新线程上调用图
config = {"configurable": {"thread_id": "2"}}

# 让我们再次打个招呼
for update in graph.stream(
    {"messages": [{"role": "user", "content": "hi, tell me about my memories"}]},
    config,
    stream_mode="updates",
    context=Context(user_id="1"),
):
    print(update)

当您在本地使用 LangSmith（例如在 Studio 中）或托管时，默认情况下可以使用基础存储，并且您无需在图编译期间指定它。但是，要启用语义搜索，您确实需要在 langgraph.json 文件中配置索引设置。例如：

{
    ...
    "store": {
        "index": {
            "embed": "openai:text-embeddings-3-small",
            "dims": 1536,
            "fields": ["$"]
        }
    }
}

有关更多详细信息和配置选项，请参阅部署指南。

构建自定义存储

要使用内置实现以外的存储后端，请子类化 BaseStore 并实现其必需的方法。内置的 InMemoryStore 是最简单的参考实现。

基础契约

所有五个异步方法都是必需的。同步对应方法（put、get、delete、search、list_namespaces）是可选的，但建议实现以兼容同步图执行。

方法	描述
`aput(namespace, key, value, index=None)`	存储或覆盖单个项目
`aget(namespace, key)`	通过键检索单个项目；如果缺失则返回 `None`
`adelete(namespace, key)`	删除单个项目
`asearch(namespace_prefix, *, query=None, filter=None, limit=10, offset=0)`	在命名空间前缀下搜索项目；可选通过语义查询
`alist_namespaces(*, prefix=None, suffix=None, max_depth=None, limit=100, offset=0)`	列出匹配前缀/后缀模式的命名空间

在实现前查找确切的签名：

import inspect
from langgraph.store.base import BaseStore
print(inspect.getsource(BaseStore))

命名空间设计

命名空间是字符串的元组，例如 ("user_id", "memories")。存储实现必须支持：

前缀匹配：asearch(("alice",)) 返回 ("alice",)、("alice", "memories") 以及任何其他子命名空间下的项目。
精确键查找：aget(("alice", "memories"), "some-key") 必须是 O(1) 或接近。

对于 SQL 后端，一个常见的模式：

CREATE TABLE store_items (
    namespace   TEXT[] NOT NULL,
    key         TEXT NOT NULL,
    value       JSONB NOT NULL,
    created_at  TIMESTAMPTZ DEFAULT now(),
    updated_at  TIMESTAMPTZ DEFAULT now(),
    PRIMARY KEY (namespace, key)
);

CREATE INDEX ON store_items USING gin(namespace);

序列化

存储值是普通的 Python 字典——不需要特殊的序列化器。直接使用 json.dumps / json.loads 或 JSONB 列进行序列化。不要存储不可 JSON 序列化的原始 Python 对象。

语义搜索支持

如果您的后端支持向量搜索，请在 asearch 上实现 query 参数：

接受 query: str | None 参数。
当 query 不为 None 时，嵌入它并按余弦相似度对结果排序。
当提供 query 时，结果中的每个 Item 应该包含一个 score 字段。

如果您的后端不支持向量搜索，当传递 query 时抛出 NotImplementedError。

测试

目前没有针对自定义存储的合规性套件。以 InMemoryStore 作为参考进行测试：

import pytest
from langgraph.store.memory import InMemoryStore
from your_module import YourStore

@pytest.fixture
async def store():
    async with YourStore.create() as s:
        yield s

@pytest.fixture
def reference():
    return InMemoryStore()

async def test_put_and_get(store, reference):
    ns = ("test", "ns")
    for s in [store, reference]:
        await s.aput(ns, "k1", {"val": 1})
        item = await s.aget(ns, "k1")
        assert item is not None
        assert item.value == {"val": 1}

async def test_delete(store, reference):
    ns = ("test", "ns")
    for s in [store, reference]:
        await s.aput(ns, "k1", {"val": 1})
        await s.adelete(ns, "k1")
        assert await s.aget(ns, "k1") is None

async def test_search_prefix(store, reference):
    for s in [store, reference]:
        await s.aput(("user", "memories"), "m1", {"text": "likes pizza"})
        results = await s.asearch(("user",))
        assert any(r.key == "m1" for r in results)

后续步骤

向 Agent Server 添加自定义存储 — 部署您的实现
检查点保存器 — 线程范围的状态持久化

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.

Get started

Capabilities

Production

Frontend

LangGraph APIs

存储

基本用法

列出命名空间中的项目

语义搜索

在 LangGraph 中使用

构建自定义存储

基础契约

命名空间设计

序列化

语义搜索支持

测试

后续步骤

​基本用法

​列出命名空间中的项目

​语义搜索

​在 LangGraph 中使用

​构建自定义存储

​基础契约

​命名空间设计

​序列化

​语义搜索支持

​测试

​后续步骤

基本用法

列出命名空间中的项目

语义搜索

在 LangGraph 中使用

构建自定义存储

基础契约

命名空间设计

序列化

语义搜索支持

测试

后续步骤