Memory Store：跨线程长期记忆系统

1 记忆类型回顾：短期 vs. 长期记忆

认知科学中，记忆分为多种类型。在 AI 应用中，最重要的区分是短期记忆（within-thread）和长期记忆（cross-thread）。

2 LangGraph Store 核心 API：put / search / get

LangGraph Store 是一个开源的持久化键值存储基类（BaseStore），提供三个核心方法来管理跨线程记忆。

初始化 InMemoryStore

import uuid
from langgraph.store.memory import InMemoryStore

# 创建内存中的 Store 实例（生产环境可替换为数据库后端）
in_memory_store = InMemoryStore()

put()：写入记忆

# 定义命名空间：(user_id, 分类名)
user_id = "1"
namespace_for_memory = (user_id, "memories")

# 生成唯一 key（每条记忆一个 UUID）
key = str(uuid.uuid4())

# value 必须是字典
value = {"food_preference": "I like pizza"}

# 写入 Store
in_memory_store.put(namespace_for_memory, key, value)

search()：按命名空间批量检索

# 检索命名空间下的所有记忆
memories = in_memory_store.search(namespace_for_memory)

# 每条记忆的完整结构示例：
# {
#   'value': {'food_preference': 'I like pizza'},
#   'key': 'a754b8c5-...',
#   'namespace': ['1', 'memories'],
#   'created_at': '2024-11-04T22:48:16.727572+00:00',
#   'updated_at': '2024-11-04T22:48:16.727574+00:00'
# }

get()：按 namespace + key 精确检索

# 精确获取单条记忆
memory = in_memory_store.get(namespace_for_memory, key)
memory.dict()  # 返回与 search()[0] 相同的结构

3 命名空间设计：像文件系统一样组织记忆

Store 的 namespace 是一个元组（tuple），工作方式类似文件系统的目录路径，允许你按层级隔离不同用户、不同类型的记忆。

# 命名空间：按 user_id 隔离，"memory" 是分类前缀
namespace = ("memory", user_id)

# 固定 key：每个用户只保存一条"汇总画像"记忆
key = "user_memory"

# 读取：先检查记忆是否存在
existing_memory = store.get(namespace, key)

if existing_memory:
    content = existing_memory.value.get('memory')
else:
    content = "No existing memory found."

4 聊天机器人架构：两个记忆节点的职责

整个图非常简洁：START → call_model → write_memory → END。每次用户发送消息，都会经历"响应"和"记忆更新"两个阶段。

5 call_model 节点：从 Store 读取记忆并个性化响应

MODEL_SYSTEM_MESSAGE = """You are a helpful assistant with memory that provides information about the user.
If you have memory for this user, use it to personalize your responses.
Here is the memory (it may be empty): {memory}"""

from langgraph.store.base import BaseStore
from langchain_core.runnables.config import RunnableConfig

def call_model(state: MessagesState, config: RunnableConfig, store: BaseStore):
    # Step 1：从 config 中取出 user_id
    user_id = config["configurable"]["user_id"]

    # Step 2：构造命名空间，从 Store 读取记忆
    namespace = ("memory", user_id)
    existing_memory = store.get(namespace, "user_memory")

    # Step 3：提取记忆内容（首次使用时为 None）
    if existing_memory:
        existing_memory_content = existing_memory.value.get('memory')
    else:
        existing_memory_content = "No existing memory found."

    # Step 4：将记忆注入系统提示词，调用 LLM
    system_msg = MODEL_SYSTEM_MESSAGE.format(memory=existing_memory_content)
    response = model.invoke(
        [SystemMessage(content=system_msg)] + state["messages"]
    )
    return {"messages": response}

6 write_memory 节点：反思对话并更新长期记忆

CREATE_MEMORY_INSTRUCTION = """You are collecting information about the user to personalize your responses.

CURRENT USER INFORMATION:
{memory}

INSTRUCTIONS:
1. Review the chat history below carefully
2. Identify new information about the user, such as:
   - Personal details (name, location)
   - Preferences (likes, dislikes)
   - Interests and hobbies
3. Merge any new information with existing memory
4. Format the memory as a clear, bulleted list
5. If new information conflicts with existing memory, keep the most recent version

Based on the chat history below, please update the user information:"""

def write_memory(state: MessagesState, config: RunnableConfig, store: BaseStore):
    user_id = config["configurable"]["user_id"]
    namespace = ("memory", user_id)
    existing_memory = store.get(namespace, "user_memory")

    if existing_memory:
        existing_memory_content = existing_memory.value.get('memory')
    else:
        existing_memory_content = "No existing memory found."

    # LLM 反思对话，提取并合并用户信息
    system_msg = CREATE_MEMORY_INSTRUCTION.format(memory=existing_memory_content)
    new_memory = model.invoke(
        [SystemMessage(content=system_msg)] + state['messages']
    )

    # 用相同 key 覆盖写入（合并更新）
    store.put(namespace, "user_memory", {"memory": new_memory.content})

7 双存储编译模式与配置注入

图需要同时接受 checkpointer（短期记忆）和 store（长期记忆）两个参数进行编译。

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.store.memory import InMemoryStore

builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_node("write_memory", write_memory)
builder.add_edge(START, "call_model")
builder.add_edge("call_model", "write_memory")
builder.add_edge("write_memory", END)

# 长期记忆：跨线程 Store
across_thread_memory = InMemoryStore()
# 短期记忆：线程内 Checkpointer
within_thread_memory = MemorySaver()

# 关键：同时传入 checkpointer（短期）和 store（长期）
graph = builder.compile(
    checkpointer=within_thread_memory,
    store=across_thread_memory
)

# thread_id → MemorySaver；user_id → Store
config = {
    "configurable": {
        "thread_id": "1",
        "user_id": "1"
    }
}

for chunk in graph.stream(
    {"messages": [HumanMessage(content="Hi, my name is Lance")]},
    config,
    stream_mode="values"
):
    chunk["messages"][-1].pretty_print()

8 完整运行演示：跨线程记忆持久化

第一次对话（Thread 1）：建立用户画像

第二次对话（Thread 2）：跨线程记忆生效

开启全新的 thread_id，但使用相同的 user_id——机器人应该能记住 Lance 的信息：

设计总结