本文介绍了如何在LangGraph中实现跨线程的信息持久化，特别是在多线程环境下，如何通过InMemoryStore来存储用户的特定信息，从而实现跨线程的记忆共享。文章详细阐述了使用InMemoryStore存储和检索记忆的步骤，以及如何结合checkpointer实现状态的持久化，并通过代码示例展示了如何在LangGraph中配置和使用这一功能，为构建多Agent对话系统提供了有力的支持。

💡LangGraph通过InMemoryStore实现了跨线程的记忆存储功能，允许在不同的线程中共享用户的特定信息，例如聊天机器人的对话历史和用户偏好。

🔑InMemoryStore的使用方法包括：定义命名空间、使用store.put方法存储记忆、使用store.search方法检索记忆。记忆以键值对的形式存储，并支持语义搜索，通过嵌入模型实现基于语义相似度的检索。

🛠️在LangGraph中，InMemoryStore与checkpointer协同工作，checkpointer负责状态的保存，而InMemoryStore则支持跨线程的数据存储和访问。通过graph.compile方法将二者集成。

💬代码示例展示了如何在LangGraph的节点中使用InMemoryStore，包括获取用户ID、定义命名空间、检索记忆，并在模型调用中使用这些记忆，从而实现对用户信息的个性化响应。

在之前的文章我们有讲过状态可以由 checkpointer 在图形执行的每一个步骤写入线程，从而实现状态持久性。而图形执行时这只是其中一个线程，但如果我们想跨线程保留一些信息怎么办？加入有一个聊天机器人，我们希望在机器人与该用户的所有聊天对话（例如，线程）中保留有关该用户的特定信息！这种情况下仅使用 checkpointer，我们无法跨线程共享信息。例如，我们可以定义一个 InMemoryStore 来跨线程存储有关用户的信息。像以前一样，我们只需使用 checkpointer 和新的 in_memory_store 变量编译我们的图。我们下面来看一下怎么实现。首先，让我们在不使用 LangGraph 的情况下单独展示它。

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

记忆（Memories）通过一个元组进行命名空间划分，在此特定示例中，该元组格式为（<用户ID>, "memories"）。命名空间的长度可以任意，并能表示任何含义，不必仅限于用户特定场景。

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

我们使用 store.put 方法将记忆存储到指定命名空间中。调用该方法时需传入两个参数：上文定义的命名空间（namespace），以及表示记忆内容的键值对——其中键（key）是记忆的唯一标识符（memory_id），值（value）则是记忆本体（以字典形式存储）。

import uuidmemory_id = str(uuid.uuid4())memory = {"食物推荐" : "我喜欢吃辣的食物", "天气" : "今天的天气很热", "运动" : "我喜欢打篮球"}in_memory_store.put(namespace_for_memory, memory_id, memory)我们可通过 store.search 方法读取指定命名空间内的记忆数据，该方法会以列表

形式返回对应用户的所有记忆记录，其中最新生成的记忆位于列表末尾。

memories = in_memory_store.search(namespace_for_memory)for memory in memories:    print(memory.dict())

得到下面结果

{'namespace': ['1', 'memories'], 'key': '8abdd488-918d-437c-8463-4178f3a33b27', 'value': {'食物推荐': '我喜欢吃辣的食物', '天气': '今天的天气很热', '运动': '我喜欢打篮球'}, 'created_at': '2025-06-22T03:42:08.172059+00:00', 'updated_at': '2025-06-22T03:42:08.172059+00:00', 'score': None}

可以看到输出的结构每个内存类型都是一个具有特定属性的 Python类 (Item)。我们可以像上面一样通过 .dict 进行转换，将其作为字典访问。 它每个字段的含义是：

value：此内存的值（本身是一个字典）key：此命名空间中此内存的唯一 keynamespace：字符串列表，此内存类型的命名空间created_at：创建此内存时的时间戳updated_at：更新此内存时的时间戳

一 语义检索

除了基础检索功能外，该存储系统还支持语义搜索，允许基于语义相似度（而非精确匹配）查找记忆内容。启用该功能需通过以下配置为存储实例指定嵌入模型：

from langchain.embeddings import init_embeddingsstore = InMemoryStore(    index={        "embed": init_embeddings("openai:text-embedding-3-small"),  # onsEmbedding provider        "dims": 1536,                              # Embedding dimensi        "fields": ["食物推荐", "$"]              # Fields to embed    })

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

memories = store.search(    namespace_for_memory,    query="你喜欢吃什么食物?",    limit=3  # Return top 3 matches)

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

# 对特定字段进行词嵌入store.put(    namespace_for_memory,    str(uuid.uuid4()),    {        "食物推荐": "我喜欢意大利面",        "context": "提供晚餐计划"    },    index=["food_preference"]  # Only embed "food_preferences" field)# 不进行词嵌入store.put(    namespace_for_memory,    str(uuid.uuid4()),    {"system_info": "Last updated: 2024-01-01"},    index=False)

二 在langgraph中使用

完成上面准备工作之后，我们将在LangGraph中使用in_memory_store内存存储组件。该组件与检查点处理器（checkpointer）协同工作：如之前所述，检查点处理器负责将状态保存至线程，而in_memory_store则支持跨线程存储和访问任意数据信息。最终我们按以下方式同时集成检查点处理器和内存存储组件来编译图结构：

from langgraph.checkpoint.memory import InMemorySavercheckpointer = InMemorySaver()# ... Define the graph ...graph = graph.compile(checkpointer=checkpointer, store=in_memory_store)

像之前一样，我们使用 thread_id 和 user_id 调用图形，我们将使用它来为这个特定用户命名我们的内存，如上所示。

# Invoke the graphuser_id = "1"config = {"configurable": {"thread_id": "1", "user_id": user_id}}# First let's just say hi to the AIfor update in graph.stream(    {"messages": [{"role": "user", "content": "你好"}]}, config, stream_mode="updates"):    print(update)

我们可以通过传递 store: BaseStore 和 config: RunnableConfig 作为节点参数来访问任何节点中的 in_memory_store 和 user_id。以下是我们如何在节点中使用语义搜索来查找相关记忆：

def update_memory(state: MessagesState, config: RunnableConfig, *, store: BaseStore):    user_id = config["configurable"]["user_id"]    namespace = (user_id, "memories")    memory_id = str(uuid.uuid4())    store.put(namespace, memory_id, {"memory": memory})

正如我们上面展示的，我们还可以在任何节点中访问 storestore.search 方法来获取记忆。我们可以访问这些记忆并在我们的 Model 调用中使用它们。

def call_model(state: MessagesState, config: RunnableConfig, *, store: BaseStore):    # Get the user id from the config    user_id = config["configurable"]["user_id"]    # Namespace the memory    namespace = (user_id, "memories")    # Search based on the most recent message    memories = store.search(        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 graphconfig = {"configurable": {"thread_id": "2", "user_id": "1"}}# Let's say hi againfor update in graph.stream(    {"messages": [{"role": "user", "content": "hi, tell me about my memories"}]}, config, stream_mode="updates"):    print(update)

下面是完整的代码：

import uuidimport osfrom langchain.chat_models import init_chat_modelos.environ["OPENAI_API_KEY"] = "xxxx"os.environ["OPENAI_API_BASE"] = "https://openkey.cloud/v1"from typing_extensions import Annotated, TypedDictfrom langchain_core.runnables import RunnableConfigfrom langgraph.graph import StateGraph, MessagesState, STARTfrom langgraph.checkpoint.memory import InMemorySaver# highlight-next-linefrom langgraph.store.memory import InMemoryStorefrom langgraph.store.base import BaseStoremodel = init_chat_model(model="openai:gpt-4.1-mini")def call_model(    state: MessagesState,    config: RunnableConfig,    *,    store: BaseStore,  #):    user_id = config["configurable"]["user_id"]    print("user_id: ", user_id)     namespace = ("memories", user_id)    memories = store.search(namespace, query=str(state["messages"][-1].content))          # Fix: Convert dictionary to string format    info_parts = []    for d in memories:        memory_data = d.value["data"]        if isinstance(memory_data, dict):            # Convert dict to readable string format            memory_str = ", ".join([f"{k}: {v}" for k, v in memory_data.items()])            info_parts.append(memory_str)        else:            info_parts.append(str(memory_data))        info = "\n".join(info_parts)    print("memories: ", memories)    print("formatted info: ", info)        system_msg = f"You are a helpful assistant talking to the user. User info: {info}"    last_message = state["messages"][-1]    print("last_message: ", last_message)    response = model.invoke(        [{"role": "system", "content": system_msg}] + state["messages"]    )    return {"messages": response}builder = StateGraph(MessagesState)builder.add_node(call_model)builder.add_edge(START, "call_model")checkpointer = InMemorySaver()from langchain.embeddings import init_embeddingsstore = InMemoryStore(    index={        "embed": init_embeddings("openai:text-embedding-3-small"),  # onsEmbedding provider        "dims": 1536,                              # Embedding dimensi        "fields": ["食物推荐", "$"]              # Fields to embed    })# store = InMemoryStore()user_id = "12024"memory = {"食物偏好" : "我喜欢吃麻婆豆腐", "天气偏好" : "我喜欢下雨天", "运动偏好" : "我喜欢打篮球"}namespace = ("memories", user_id)store.put(namespace, str(uuid.uuid4()), {"data": memory})graph = builder.compile(    checkpointer=checkpointer,    store=store,)config = {    "configurable": {        # highlight-next-line        "thread_id": "1",        # highlight-next-line        "user_id": user_id,    }}for chunk in graph.stream(    {"messages": [{"role": "user", "content": "我喜欢吃什么"}]},    # highlight-next-line    config,    stream_mode="values",):    chunk["messages"][-1].pretty_print()

得到如下结果通过上面例子可以看到，只要我们多个线程维护同一个用户id，就能做到跨线程读取持久化记忆，这是我们开发多agent对话时一个非常强大的机制。