【LangGraph】langgraph.store.memory.InMemoryStore 类：基于内存的键值存储

作者：彬彬侠
langgraph.store.memory.InMemoryStore 详解

langgraph.store.memory.InMemoryStore 是 LangGraph 框架中的一个核心类，继承自 langgraph.store.base.BaseStore，提供了一个基于内存的键值存储，用于管理代理的状态和长时记忆。它适合开发和测试场景，因数据不持久化，生产环境建议使用其他存储（如 PostgresStore）。本文详细介绍其功能、参数、方法、使用方法、实际应用及注意事项。
1. 概述

1.1 什么是 InMemoryStore？

InMemoryStore 是 langgraph.store.memory 模块中的类，基于 Python 字典实现了一个高效的内存键值存储。它支持层次命名空间、批量操作和可选的语义搜索，适用于快速迭代和临时数据存储。由于数据存储在内存中，程序重启后数据会丢失，因此主要用于开发、测试或短期记忆场景。
该类在 langchain-core 0.2.14 版本中引入，是 LangGraph 存储系统的一部分，提供了轻量级的存储解决方案。相关信息可参考 LangGraph 存储文档 和 InMemoryStore API 参考。
1.2 核心功能

内存存储：基于 Python 字典，读写操作高效，适合快速访问。层次命名空间：支持结构化数据组织，如 ("users", "123", "messages")，确保数据隔离。语义搜索：通过配置嵌入模型支持语义查询（默认禁用，需显式启用）。批量操作：提供 batch 和 abatch 方法，支持高效的批量读写和删除。异步操作：支持异步方法（如 aget、aput），适合非阻塞场景。分页与过滤：支持查询结果的分页（limit 和 offset）和条件过滤（filter）。版本要求：需 langchain-core>=0.2.14，推荐使用最新版本。

2. 定义与结构

2.1 类定义

InMemoryStore 继承自 BaseStore，定义如下（简化表示）：

1. from langgraph.store.base import BaseStore
2. from typing import Dict, Any, Optional, List, Tuple, Union, Literal
3. from langchain_core.embeddings import Embeddings
5. classInMemoryStore(BaseStore):def__init__(self, index: Optional[Dict[str, Any]]=None):
6.         self.store: Dict[Tuple[Tuple[str,...],str], Any]={}
7.         self.index_config = index
8.         self.supports_ttl =False
9.         self.supports_index =bool(index)

复制代码

2.2 初始化参数

参数	类型	默认值	描述
index	Optional[Dict[str, Any]]	None	可选，嵌入索引配置，包含 dims（嵌入维度）和 embed（嵌入函数）。若提供，则启用语义搜索。

2.3 关键属性

属性	类型	默认值	描述
supports_ttl	bool	False	是否支持 TTL（时间到活），InMemoryStore 默认禁用。
supports_index	bool	False	是否支持语义搜索索引，取决于是否提供 index 参数。
store	Dict	{}	内部字典，存储键值对，键为 (namespace, key)，值为 Item。
index_config	Dict	None	嵌入索引配置，若提供则用于语义搜索。

2.4 核心方法

InMemoryStore 实现 BaseStore 的所有方法，以下是主要方法：

方法	描述	参数	返回值	异步
__init__(index=None)	初始化内存存储，可选配置嵌入索引。	index: Optional[Dict[str, Any]]	None	No
put(namespace, key, value, index=None, ttl=None)	存储键值对，可选设置索引（用于语义搜索）。	namespace: Tuple[str, ...], key: str, value: Dict[str, Any], index: Optional[Union[Literal[False], List[str]]], ttl: Optional[float]	None	No
get(namespace, key, refresh_ttl=None)	获取指定键值。	namespace: Tuple[str, ...], key: str, refresh_ttl: Optional[bool]	Optional[Item]	No
delete(namespace, key)	删除指定键值。	namespace: Tuple[str, ...], key: str	None	No
search(namespace_prefix, query=None, filter=None, limit=10, offset=0, refresh_ttl=None)	搜索匹配项，支持语义查询、过滤和分页。	namespace_prefix: Tuple[str, ...], query: Optional[str], filter: Optional[Dict[str, Any]], limit: int, offset: int, refresh_ttl: Optional[bool]	List[SearchItem]	No
list_namespaces(prefix=None, suffix=None, max_depth=None, limit=100, offset=0)	列出匹配前缀、后缀和深度的命名空间，分页控制。	prefix: Optional[Tuple[str, ...]], suffix: Optional[Tuple[str, ...]], max_depth: Optional[int], limit: int, offset: int	List[Tuple[str, ...]]	No
batch(ops)	批量执行存储操作（获取、存储、删除）。	ops: Iterable[Op] (操作列表)	List[Result]	No
abatch(ops)	异步批量执行存储操作。	ops: Iterable[Op] (操作列表)	List[Result]	Yes
aget(namespace, key)	异步获取值。	namespace: Tuple[str, ...], key: str	Optional[Item]	Yes
aput(namespace, key, value, index=None, ttl=None)	异步存储键值对。	namespace: Tuple[str, ...], key: str, value: Dict[str, Any], index: Optional[Union[Literal[False], List[str]]], ttl: Optional[float]	None	Yes
adelete(namespace, key)	异步删除值。	namespace: Tuple[str, ...], key: str	None	Yes
asearch(...)	异步搜索匹配项，参数与 search 一致。	同 search	List[SearchItem]	Yes
alist_namespaces(...)	异步列出命名空间，参数与 list_namespaces 一致。	同 list_namespaces	List[Tuple[str, ...]]	Yes

默认值：

search 和 asearch 的 limit 默认 10，offset 默认 0。list_namespaces 和 alist_namespaces 的 limit 默认 100，offset 默认 0。TTL 默认禁用（supports_ttl=False）。

3. 使用方法

以下是 InMemoryStore 的主要使用方式，结合前文内容展示具体示例：
3.1 基本用法

初始化和基本操作：

1. from langgraph.store.memory import InMemoryStore
3. # 初始化存储
4. store = InMemoryStore()# 存储键值对
5. namespace =("users","123","profile")
6. key ="info"
7. value ={"name":"Alice","age":30}
8. store.put(namespace, key, value)# 获取值
9. item = store.get(namespace, key)print(item)# 输出: {"name": "Alice", "age": 30}# 删除值
10. store.delete(namespace, key)

复制代码

3.2 启用语义搜索

配置嵌入模型以支持语义搜索：

1. from langchain_openai import OpenAIEmbeddings
2. from langgraph.store.memory import InMemoryStore
4. # 初始化嵌入模型
5. embeddings = OpenAIEmbeddings(model="text-embedding-3-small")# 初始化存储，启用语义搜索
6. store = InMemoryStore(
7.     index={"dims":1536,# 嵌入维度"embed": embeddings
8.     })# 存储带索引的数据
9. store.put(
10.     namespace=("docs",),
11.     key="doc1",
12.     value={"title":"LangChain 简介","content":"LangChain 是一个 LLM 框架"},
13.     index=["title","content"]# 索引字段)# 语义搜索
14. results = store.search(
15.     namespace_prefix=("docs",),
16.     query="LangChain 框架",
17.     limit=5)for result in results:print(result.value)# 输出: {"title": "LangChain 简介", "content": "LangChain 是一个 LLM 框架"}

复制代码

3.3 批量操作

执行批量存储和删除：

1. from langgraph.store.base import Op
3. ops =[
4.     Op(type="put", namespace=("users","123","messages"), key="msg1", value={"content":"Hello"}),
5.     Op(type="put", namespace=("users","123","messages"), key="msg2", value={"content":"World"}),
6.     Op(type="get", namespace=("users","123","messages"), key="msg1")]
7. results = store.batch(ops)print(results)# 输出: [None, None, {"content": "Hello"}]

复制代码

3.4 结合代理系统

在 LangGraph 代理中存储会话状态：

1. from langgraph.graph import StateGraph
2. from langgraph.store.memory import InMemoryStore
4. store = InMemoryStore()# 定义状态classAgentState:
5.     messages: List[Dict[str, Any]]# 创建图
6. graph = StateGraph(AgentState)# 定义节点，存储消息defstore_message(state: AgentState, store: InMemoryStore):for msg in state.messages:
7.         store.put(("users","123","messages"), msg["id"], msg)return state
9. # 添加节点
10. graph.add_node("store",lambda state: store_message(state, store))# ... 配置其他节点和边 ...# 运行图
11. result = graph.invoke({"messages":[{"id":"msg1","content":"Hello"}]}, store=store)
12. item = store.get(("users","123","messages"),"msg1")print(item)# 输出: {"id": "msg1", "content": "Hello"}

复制代码

3.5 结合检索器工具

结合前文讨论的 create_retriever_tool，存储检索索引：

1. from langchain_core.vectorstores import VectorStoreRetriever
2. from langchain_core.tools.retriever import create_retriever_tool
4. retriever = VectorStoreRetriever(...)# 假设的检索器
5. tool = create_retriever_tool(retriever,"search_docs","搜索相关文档")
6. store.put(("tools","search"),"config",{"tool_name":"search_docs"})# 获取工具配置
7. config = store.get(("tools","search"),"config")print(config)# 输出: {"tool_name": "search_docs"}

复制代码

4. 实际应用

InMemoryStore 在以下场景中广泛应用：

开发与测试：快速迭代测试代理逻辑或工作流，无需配置数据库。临时记忆存储：存储会话状态或短期上下文数据，如对话历史。语义搜索测试：在开发环境中验证嵌入模型和语义搜索效果。原型开发：构建 LangGraph 应用的原型，验证功能逻辑。轻量级应用：适合小规模、非持久化的状态管理场景。

5. 最佳实践

开发环境优先：在开发和测试阶段使用 InMemoryStore，快速验证逻辑。层次命名空间：设计结构化命名空间（如 ("users", "id", "data")），确保数据隔离。语义搜索配置：若需语义搜索，设置 index 参数，选择高效的嵌入模型（如 OpenAIEmbeddings）。批量操作优化：使用 batch 或 abatch 执行批量操作，减少性能开销。版本检查：通过 pip install -qU langgraph 确保安装最新版本（依赖 langchain-core>=0.2.14）。调试工具：使用 LangSmith 跟踪存储操作，调试复杂工作流或搜索结果。

6. 注意事项与限制

数据不持久化：程序重启后数据丢失，生产环境建议使用 PostgresStore 或其他持久化存储。内存消耗：存储大量数据可能导致高内存使用，需监控资源。语义搜索限制：默认禁用，需显式配置 index，确保嵌入模型可用。TTL 不支持：supports_ttl=False，无法设置数据过期。模块路径：确保正确导入 langgraph.store.memory.InMemoryStore。版本依赖：功能可能受版本限制，需确保兼容最新版本。

7. 结论

langgraph.store.memory.InMemoryStore 是 LangGraph 框架中提供内存键值存储的类，继承自 BaseStore，支持层次命名空间、批量操作和可选的语义搜索。其高效的内存访问和简单配置使其成为开发和测试的理想选择，适合快速迭代、临时存储和语义搜索验证。由于其非持久化特性，生产环境需选择持久化存储（如 PostgresStore）。结合 BaseStore 接口、嵌入模型和 LangGraph 生态系统，开发者可以构建高效的代理记忆和状态管理方案。
8. 参考资料

LangGraph 存储文档InMemoryStore API 参考语义搜索与 LangGraph 记忆如何在 LangGraph 中持久化 InMemoryStore 数据构建具有长期记忆的 LangGraph 代理LangGraph GitHub 仓库

原文地址：https://blog.csdn.net/u013172930/article/details/148034730