Memory 记忆体系

LangGraph 的 Memory（记忆）体系让 Agent 能够跨对话保持状态和知识。结合 Checkpointer 的持久化机制，Memory 分为短期工作记忆和长期跨会话记忆。

记忆的层次

类型	范围	存储	生命周期
工作记忆	单次对话	State + Checkpoint	对话期间
长期记忆	跨对话	外部存储（向量库、数据库）	持久化

工作记忆：通过 State 实现

单轮对话记忆

from langgraph.graph import MessagesState, StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

def chat_node(state: MessagesState):
    """LLM 可以访问所有历史消息"""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

builder = StateGraph(MessagesState)
builder.add_node("chat", chat_node)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)

graph = builder.compile(checkpointer=InMemorySaver())

# 第一次对话
config = {"configurable": {"thread_id": "alice"}}
graph.invoke(
    {"messages": [{"role": "user", "content": "我叫小明"}]},
    config
)

# 第二次对话 - AI 还记得你叫小明
graph.invoke(
    {"messages": [{"role": "user", "content": "我叫什么名字?"}]},
    config
)

自定义 State 记忆

from typing import TypedDict, List, Annotated
import operator

class MemoryState(TypedDict):
    messages: Annotated[List[dict], operator.add]
    user_summary: str           # 对用户的总结
    preferences: dict           # 偏好信息
    facts: Annotated[List[str], operator.add]  # 已确认的事实

def update_summary(state: MemoryState):
    """定期更新用户总结"""
    if len(state["messages"]) > 10:
        summary = summarize_conversation(state["messages"])
        return {"user_summary": summary}
    return {}

长期记忆：外部存储

方案 1：向量库记忆

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma

class VectorMemory:
    def __init__(self):
        self.vectorstore = Chroma(
            collection_name="agent_memory",
            embedding_function=OpenAIEmbeddings()
        )

    def remember(self, fact: str, metadata: dict = None):
        """存储一条记忆"""
        self.vectorstore.add_texts(
            texts=[fact],
            metadatas=[metadata or {}]
        )

    def recall(self, query: str, k: int = 5):
        """检索相关记忆"""
        return self.vectorstore.similarity_search(query, k=k)

在图中使用：

memory_store = VectorMemory()

def memory_aware_node(state: State, config: RunnableConfig):
    user_id = config["configurable"]["user_id"]

    # 检索相关记忆
    relevant_memories = memory_store.recall(state["input"])
    # 将记忆加入上下文
    memory_context = "\n".join([m.page_content for m in relevant_memories])

    prompt = f"""
    用户历史记忆：
    {memory_context}

    当前输入：
    {state['input']}
    """

    response = llm.invoke(prompt)
    return {"response": response}

方案 2：结构化数据库记忆

import sqlite3

class StructuredMemory:
    def __init__(self, db_path: str = "agent_memory.db"):
        self.conn = sqlite3.connect(db_path)
        self._init_db()

    def _init_db(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS memory (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id TEXT,
                key TEXT,
                value TEXT,
                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
            )
        """)

    def set(self, user_id: str, key: str, value: str):
        self.conn.execute(
            "INSERT INTO memory (user_id, key, value) VALUES (?, ?, ?)",
            (user_id, key, value)
        )
        self.conn.commit()

    def get(self, user_id: str, key: str):
        cursor = self.conn.execute(
            "SELECT value FROM memory WHERE user_id = ? AND key = ? ORDER BY timestamp DESC LIMIT 1",
            (user_id, key)
        )
        row = cursor.fetchone()
        return row[0] if row else None

方案 3：使用 LangGraph 官方 Store

from langgraph.store import Store

# 构建 Store
store = Store(
    index={
        "dims": 1536,       # embedding 维度
        "embed": "openai",  # embedding 模型
    },
    client=...               # 持久化后端
)

# 在节点中访问
def node_with_store(state, config, *, store: Store):
    # 写入记忆
    store.put(
        ("users", "{user_id}", "memories"),
        "key_1",
        {"text": "用户喜欢编程"}
    )

    # 检索记忆
    items = store.search(
        ("users", "{user_id}", "memories"),
        query="编程"
    )
    return {"context": items}

记忆管理策略

窗口策略

只保留最近 N 条消息：

def trim_memory(state: State) -> State:
    if len(state["messages"]) > 20:
        # 保留系统提示 + 最近 10 轮对话
        state["messages"] = (
            [state["messages"][0]] +  # 系统提示
            state["messages"][-20:]   # 最近 20 条
        )
    return state

摘要策略

def summarize_memory(state: State) -> State:
    if len(state["messages"]) > 30:
        summary_prompt = f"总结以下对话: {state['messages'][:-10]}"
        summary = llm.invoke(summary_prompt)
        return {
            "messages": [
                {"role": "system", "content": f"历史摘要: {summary}"}
            ] + state["messages"][-10:]
        }
    return {}

最佳实践

1. 工作记忆用 State + Checkpoint，简单直接
2. 长期记忆用外部存储，避免每次加载全部历史
3. 为记忆打标签（如主题、重要性），提升检索质量
4. 定期压缩记忆，避免 context window 溢出
5. 区分事实性记忆和对话风格记忆，分别处理

参考

• 官方文档：Memory
• 官方文档：Persistence