存储允许代理在线程之间持久化信息，包括用户偏好、积累的知识以及应该超越单个对话留存的事实。与 检查点不同，检查点保存的是限定在单个线程范围内的完整图状态，而存储则保存可从任何线程访问的任意键值数据。
基本用法
以下代码片段单独展示 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：字符串元组，此记忆类型的命名空间
- *
created_at：此记忆创建时的时间戳 - *
updated_at：此记忆更新时的时间戳
列出命名空间中的项目
调用 store.search （或异步 store.asearch），不指定 query 且不指定 filter 将返回存储在 namespace_prefix, up to limit下的项目。当您不需要语义排序时，使用此方法枚举命名空间中的所有内容。
请牢记三个行为：
- - **
namespace_prefix按前缀匹配，而非精确匹配。**("alice",)也会返回("alice", "memories"),("alice", "preferences")下的项目，依此类推。要限制为单个级别，请传递完整的命名空间或在客户端按item.namespace. - - **超出
limit的结果将被静默截断。** 没有溢出信号——将limit设置为预期的上限，或使用offset. - 默认排序取决于存储后端。
PostgresStore和AsyncPostgresStore按updated_at降序返回结果（最近更新的排在最前面）。InMemoryStore按插入顺序返回结果（最近插入的排在最后）。请勿依赖不同实现之间的特定顺序；如需排序，请在客户端按item.updated_at进行排序（如果顺序重要）。
要分页浏览大型命名空间：
要发现存在的命名空间（例如，在列出用户记忆前迭代每个用户），请使用 store.list_namespaces or store.alist_namespaces:
语义搜索
除了简单的检索，存储还支持语义搜索，允许您根据含义而非精确匹配来查找记忆。要启用此功能，请使用嵌入模型配置存储：
from langchain.embeddings import init_embeddings
store = InMemoryStore(
index={
"embed": init_embeddings("openai:text-embedding-3-small"), # Embedding provider
"dims": 1536, # Embedding dimensions
"fields": ["food_preference", "$"] # Fields to embed
}
)现在搜索时，您可以使用自然语言查询来查找相关记忆：
# Find memories about food preferences
# (This can be done after putting memories into the store)
memories = store.search(
namespace_for_memory,
query="What does the user like to eat?",
limit=3 # Return top 3 matches
)您可以通过配置 fields 参数或通过在存储记忆时指定 index 参数来控制记忆的哪些部分被嵌入：
# Store with specific fields to embed
store.put(
namespace_for_memory,
str(uuid.uuid4()),
{
"food_preference": "I love Italian cuisine",
"context": "Discussing dinner plans"
},
index=["food_preference"] # Only embed "food_preferences" field
)
# Store without embedding (still retrievable, but not searchable)
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
# We need this because we want to enable threads (conversations)
checkpointer = InMemorySaver()
# ... Define the graph ...
# Compile the graph with the checkpointer and store
builder = StateGraph(MessagesState, context_schema=Context)
# ... add nodes and edges ...
graph = builder.compile(checkpointer=checkpointer, store=store)然后使用 thread_id，与之前相同，并且也带有 user_id，它作为该特定用户的记忆命名空间，与之前相同。
# Invoke the graph
config = {"configurable": {"thread_id": "1"}}
# First let's just say hi to the AI
for update in graph.stream(
{"messages": [{"role": "user", "content": "hi"}]},
config,
stream_mode="updates",
context=Context(user_id="1"),
):
print(update)您可以从任何节点访问 store 和 user_id 从 _任何节点_ 通过使用 Runtime 对象。 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]):
# Get the user id from the runtime context
user_id = runtime.context.user_id
# Namespace the memory
namespace = (user_id, "memories")
# ... Analyze conversation and create a new memory
# Create a new memory ID
memory_id = str(uuid.uuid4())
# We create a new memory
await runtime.store.aput(namespace, memory_id, {"memory": memory})您还可以从任何节点访问 store，并使用 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]):
# Get the user id from the runtime context
user_id = runtime.context.user_id
# Namespace the memory
namespace = (user_id, "memories")
# Search based on the most recent message
memories = await runtime.store.asearch(
namespace,
query=state["messages"][-1].content,
limit=3
)
info = "\n".join([d.value["memory"] for d in memories])
# ... Use memories in the model call如果您创建新线程，只要 user_id 相同，您仍然可以访问相同的记忆。
# Invoke the graph on a new thread
config = {"configurable": {"thread_id": "2"}}
# Let's say hi again
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) or 托管环境中），基础 store 可默认使用，在编译图时无需指定。但是，要启用语义搜索，您 **do** 需要在您的 langgraph.json 文件中配置索引设置。例如：
{
...
"store": {
"index": {
"embed": "openai:text-embeddings-3-small",
"dims": 1536,
"fields": ["$"]
}
}
}请参阅 部署指南 了解更多详情和配置选项。
构建自定义 store
要使用内置实现以外的存储后端，请对 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) | List namespaces matching a prefix/suffix pattern |
在实现之前请查看确切的签名：
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) 或接近 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 对象。
语义搜索支持
如果你的后端支持向量搜索，请实现 query 参数于 asearch:
- - 接收
query: str | Noneargument. - - 当
query不是None时，嵌入它并按余弦相似度对结果进行排序。 - - 结果应包含每个
score上的Item字段，当提供query时。
如果你的后端不支持向量搜索，请在传入 NotImplementedError 时抛出 query 。
测试
目前没有针对自定义存储的合规性套件。请以 InMemoryStore 作为参考进行测试：
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 — 部署你的实现
- - 检查点器 — 线程作用域的状态持久化