跳到主要内容

记忆系统

记忆系统是 AI Agent 的重要组成部分，它让 Agent 能够记住对话历史、用户偏好和任务状态，从而提供连贯、个性化的服务。LangGraph 提供了一套完整的记忆系统解决方案，包括短期记忆和长期记忆。

为什么需要记忆系统​

没有记忆系统的 Agent 就像一个"健忘"的助手，每次对话都从零开始：

用户：我叫张三
Agent：你好，张三！

用户：我叫什么名字？
Agent：抱歉，我不知道您的名字。

有了记忆系统，Agent 能够记住之前的对话内容：

用户：我叫张三
Agent：你好，张三！

用户：我叫什么名字？
Agent：您叫张三。

记忆系统解决了以下问题：

  • 上下文连贯性：让对话保持连贯，避免重复询问
  • 用户个性化：记住用户偏好，提供个性化服务
  • 任务连续性：在多轮对话中持续推进任务
  • 知识积累：存储和检索知识信息

LangGraph 记忆架构​

LangGraph 将记忆系统分为两个层次：

短期记忆 vs 长期记忆​

特性短期记忆长期记忆
作用域单个会话（thread）跨会话、跨用户
存储内容对话历史、临时状态用户偏好、知识事实
持久化通过 checkpointer通过 store
检索方式按 thread_id 加载语义搜索、过滤查询
生命周期会话期间永久保存

短期记忆​

短期记忆（Short-term Memory）用于跟踪当前会话中的对话历史。LangGraph 通过 checkpointer 实现短期记忆的持久化。

基本使用​

使用 MemorySaver 作为最简单的 checkpointer：

from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI

# 创建 checkpointer
checkpointer = MemorySaver()

# 创建模型
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 创建 Agent
agent = create_react_agent(
    model=llm,
    tools=[],  # 可添加工具
    prompt="你是一个友好的助手。",
    checkpointer=checkpointer
)

# 配置会话
config = {"configurable": {"thread_id": "session-001"}}

# 第一轮对话
result1 = agent.invoke(
    {"messages": [{"role": "user", "content": "你好，我叫张三"}]},
    config=config
)
print(result1["messages"][-1].content)

# 第二轮对话（Agent 会记住之前的内容）
result2 = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么名字？"}]},
    config=config
)
print(result2["messages"][-1].content)

输出：

你好张三！很高兴认识你。有什么我可以帮助你的吗？

根据我们之前的对话，你告诉我你的名字是张三。

生产级 Checkpointer​

在生产环境中，需要使用数据库支持的 checkpointer：

PostgreSQL Checkpointer：

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

# 配置数据库连接
DB_URI = "postgresql://user:password@localhost:5432/mydb"

# 创建 checkpointer
async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # 首次使用需要初始化表结构
    await checkpointer.setup()
    
    agent = create_react_agent(
        model="openai:gpt-4o-mini",
        tools=[...],
        checkpointer=checkpointer
    )
    
    result = await agent.ainvoke(
        {"messages": [...]},
        config={"configurable": {"thread_id": "session-001"}}
    )

SQLite Checkpointer：

from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
import aiosqlite

async def main():
    async with aiosqlite.connect("checkpoints.db") as conn:
        checkpointer = AsyncSqliteSaver(conn)
        await checkpointer.setup()
        
        agent = create_react_agent(
            model="openai:gpt-4o-mini",
            tools=[...],
            checkpointer=checkpointer
        )

Redis Checkpointer：

from langgraph.checkpoint.redis.aio import AsyncRedisSaver

DB_URI = "redis://localhost:6379"

async with AsyncRedisSaver.from_conn_string(DB_URI) as checkpointer:
    await checkpointer.setup()
    
    agent = create_react_agent(
        model="openai:gpt-4o-mini",
        tools=[...],
        checkpointer=checkpointer
    )

管理对话历史​

当对话历史过长时，需要主动管理以避免超出 LLM 上下文限制：

裁剪消息：

from langchain_core.messages import trim_messages
from langgraph.prebuilt import create_react_agent

# 在自定义节点中裁剪消息
def trim_messages_node(state):
    messages = state["messages"]
    
    # 保留最后 10 条消息
    trimmed = trim_messages(
        messages,
        max_tokens=4000,
        strategy="last",
        token_counter=len,  # 或使用实际的 token 计数器
        start_on="human",
        end_on=("human", "tool"),
    )
    
    return {"messages": trimmed}

总结历史：

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

async def summarize_if_needed(state):
    """如果对话过长，总结旧消息"""
    messages = state["messages"]
    
    if len(messages) > 10:
        # 提取需要总结的消息
        old_messages = messages[:-5]
        recent_messages = messages[-5:]
        
        # 生成总结
        summary_prompt = f"""请总结以下对话内容，保留关键信息：
        
        {chr(10).join([f"{m.type}: {m.content}" for m in old_messages])}
        """
        
        summary = await llm.ainvoke([{"role": "user", "content": summary_prompt}])
        
        # 用总结替换旧消息
        from langchain_core.messages import SystemMessage
        return {
            "messages": [
                SystemMessage(content=f"[历史摘要] {summary.content}"),
                *recent_messages
            ]
        }
    
    return state

长期记忆​

长期记忆（Long-term Memory）用于跨会话存储用户偏好、事实知识等信息。LangGraph 通过 Store 实现。

基本使用​

使用 InMemoryStore 作为最简单的 store：

from langgraph.prebuilt import create_react_agent
from langgraph.store.memory import InMemoryStore
from langchain_core.tools import tool
from langchain_core.runnables import RunnableConfig
from langgraph.config import get_store
import uuid

# 创建 store
store = InMemoryStore()

# 定义工具
@tool
def save_memory(content: str, config: RunnableConfig) -> str:
    """保存重要信息到长期记忆"""
    memory_store = get_store()
    user_id = config.get("configurable", {}).get("user_id", "default")
    
    memory_store.put(
        ("memories", user_id),
        str(uuid.uuid4()),
        {"content": content}
    )
    return f"已保存：{content}"

@tool
def recall_memories(config: RunnableConfig) -> str:
    """从长期记忆中检索信息"""
    memory_store = get_store()
    user_id = config.get("configurable", {}).get("user_id", "default")
    
    memories = memory_store.search(("memories", user_id))
    if not memories:
        return "暂无记忆"
    return "\n".join([f"- {m.value['content']}" for m in memories])

# 创建 Agent
agent = create_react_agent(
    model="openai:gpt-4o-mini",
    tools=[save_memory, recall_memories],
    store=store,
    prompt="你是一个智能助手，可以记住用户的信息。"
)

# 用户 A 的会话
config_a = {"configurable": {"user_id": "user-a", "thread_id": "session-1"}}
agent.invoke(
    {"messages": [{"role": "user", "content": "请记住：我喜欢喝咖啡"}]},
    config=config_a
)

# 用户 A 的新会话（仍能回忆起记忆）
config_a2 = {"configurable": {"user_id": "user-a", "thread_id": "session-2"}}
result = agent.invoke(
    {"messages": [{"role": "user", "content": "我记得什么？"}]},
    config=config_a2
)
print(result["messages"][-1].content)  # 包含"喜欢喝咖啡"

语义搜索​

启用语义搜索可以更智能地检索记忆：

from langgraph.store.memory import InMemoryStore
from langchain_openai import OpenAIEmbeddings

# 创建带语义搜索的 store
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

def embed_texts(texts: list) -> list:
    return embeddings.embed_documents(texts)

store = InMemoryStore(
    index={
        "embed": embed_texts,
        "dims": 1536,  # embedding 维度
    }
)

# 保存记忆
store.put(("user_123", "memories"), "1", {"text": "用户喜欢吃披萨"})
store.put(("user_123", "memories"), "2", {"text": "用户是一名软件工程师"})
store.put(("user_123", "memories"), "3", {"text": "用户的生日是3月15日"})

# 语义搜索
results = store.search(
    ("user_123", "memories"),
    query="关于用户工作的信息",
    limit=2
)

for item in results:
    print(f"记忆: {item.value['text']}")
    print(f"相似度: {item.score}")

生产级 Store​

PostgreSQL Store：

from langgraph.store.postgres import PostgresStore

DB_URI = "postgresql://user:password@localhost:5432/mydb"

store = PostgresStore.from_conn_string(DB_URI)
store.setup()  # 首次使用

agent = create_react_agent(
    model="openai:gpt-4o-mini",
    tools=[...],
    store=store
)

MongoDB Store：

from langgraph.store.mongodb import MongoDBStore

DB_URI = "mongodb://localhost:27017"

store = MongoDBStore.from_conn_string(DB_URI)

agent = create_react_agent(
    model="openai:gpt-4o-mini",
    tools=[...],
    store=store
)

记忆类型​

根据认知心理学，记忆可分为三种类型：

类型说明Agent 应用
语义记忆存储事实和知识用户偏好、个人信息
情景记忆存储经历和事件过去的对话、完成的任务
程序记忆存储规则和技能Agent 的行为模式、提示词

语义记忆实现​

语义记忆用于存储关于用户的事实信息：

from langgraph.prebuilt import create_react_agent
from langgraph.store.memory import InMemoryStore
from langchain_core.tools import tool
from langchain_core.runnables import RunnableConfig
from langgraph.config import get_store
from pydantic import BaseModel

store = InMemoryStore()

class UserProfile(BaseModel):
    """用户档案"""
    name: str = ""
    preferences: list = []
    facts: list = []

@tool
def update_profile(key: str, value: str, config: RunnableConfig) -> str:
    """更新用户档案"""
    memory_store = get_store()
    user_id = config.get("configurable", {}).get("user_id", "default")
    namespace = ("profiles", user_id)
    
    # 获取现有档案
    existing = memory_store.get(namespace, "profile")
    if existing:
        profile = UserProfile(**existing.value)
    else:
        profile = UserProfile()
    
    # 更新字段
    if key == "name":
        profile.name = value
    elif key == "preference":
        profile.preferences.append(value)
    elif key == "fact":
        profile.facts.append(value)
    
    # 保存更新后的档案
    memory_store.put(namespace, "profile", profile.model_dump())
    return f"已更新档案：{key} = {value}"

@tool
def get_profile(config: RunnableConfig) -> str:
    """获取用户档案"""
    memory_store = get_store()
    user_id = config.get("configurable", {}).get("user_id", "default")
    
    existing = memory_store.get(("profiles", user_id), "profile")
    if existing:
        return str(existing.value)
    return "暂无档案信息"

agent = create_react_agent(
    model="openai:gpt-4o-mini",
    tools=[update_profile, get_profile],
    store=store,
    prompt="你是一个智能助手，可以记住用户的信息。"
)

情景记忆实现​

情景记忆用于存储过去的经历，常用于 few-shot 学习：

from langgraph.store.memory import InMemoryStore
from langchain_openai import ChatOpenAI
from datetime import datetime
import uuid

store = InMemoryStore()
llm = ChatOpenAI(model="gpt-4o-mini")

async def save_episode(user_id: str, task: str, approach: str, success: bool):
    """保存情景记忆"""
    namespace = ("episodes", user_id)
    
    episode = {
        "task": task,
        "approach": approach,
        "success": success,
        "timestamp": datetime.now().isoformat()
    }
    
    store.put(namespace, str(uuid.uuid4()), episode)

async def retrieve_episodes(user_id: str, current_task: str, limit: int = 3):
    """检索相关情景记忆作为 few-shot 示例"""
    namespace = ("episodes", user_id)
    
    # 可以使用语义搜索
    episodes = store.search(namespace, query=current_task, limit=limit)
    
    return [
        {
            "task": e.value["task"],
            "approach": e.value["approach"],
            "success": e.value["success"]
        }
        for e in episodes
    ]

async def agent_with_episodes(user_input: str, user_id: str):
    """使用情景记忆的 Agent"""
    # 检索相关经历
    episodes = await retrieve_episodes(user_id, user_input)
    
    # 构建 few-shot 示例
    examples_text = ""
    if episodes:
        examples_text = "\n\n相关的成功案例：\n"
        for i, ep in enumerate(episodes, 1):
            examples_text += f"{i}. 任务：{ep['task']}\n   方法：{ep['approach']}\n"
    
    system_msg = f"""你是一个智能助手。
    {examples_text}
    
    请参考过去的成功经验来处理当前任务。"""
    
    response = await llm.ainvoke([
        {"role": "system", "content": system_msg},
        {"role": "user", "content": user_input}
    ])
    
    return response.content

程序记忆实现​

程序记忆用于存储 Agent 的行为规则，可以通过"反思"机制动态优化：

from langgraph.store.memory import InMemoryStore
from langchain_openai import ChatOpenAI

store = InMemoryStore()
llm = ChatOpenAI(model="gpt-4o-mini")

async def update_instructions(user_id: str, feedback: str) -> str:
    """根据反馈更新程序记忆（指令）"""
    namespace = ("instructions", user_id)
    
    # 获取当前指令
    existing = store.get(namespace, "system_prompt")
    current_instructions = existing.value["text"] if existing else "你是一个友好的助手。"
    
    reflection_prompt = f"""当前指令：
    {current_instructions}
    
    用户反馈：
    {feedback}
    
    请根据反馈优化指令，使 Agent 能更好地服务用户。
    返回优化后的完整指令。"""
    
    response = await llm.ainvoke([{"role": "user", "content": reflection_prompt}])
    
    # 保存更新后的指令
    store.put(namespace, "system_prompt", {"text": response.content})
    
    return response.content

async def agent_with_procedural_memory(user_input: str, user_id: str) -> str:
    """使用程序记忆的 Agent"""
    namespace = ("instructions", user_id)
    
    # 获取自定义指令
    existing = store.get(namespace, "system_prompt")
    system_prompt = existing.value["text"] if existing else "你是一个友好的助手。"
    
    response = await llm.ainvoke([
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_input}
    ])
    
    return response.content

记忆更新策略​

热路径更新（In the Hot Path）​

在对话过程中实时更新记忆：

from langgraph.prebuilt import create_react_agent
from langchain_core.tools import tool
from langchain_core.runnables import RunnableConfig
from langgraph.config import get_store
import uuid

@tool
def remember(content: str, config: RunnableConfig) -> str:
    """记住重要信息（实时保存）"""
    store = get_store()
    user_id = config.get("configurable", {}).get("user_id", "default")
    
    store.put(
        ("memories", user_id),
        str(uuid.uuid4()),
        {"content": content}
    )
    return f"我会记住：{content}"

agent = create_react_agent(
    model="openai:gpt-4o-mini",
    tools=[remember],
    prompt="如果用户告诉你重要的信息，使用 remember 工具保存。"
)

优点：

  • 实时性强，新记忆立即可用
  • 用户可以知道记忆何时被保存

缺点：

  • 增加 Agent 响应延迟
  • Agent 需要同时处理任务和记忆管理

后台更新（Background）​

在后台异步更新记忆：

import asyncio
from datetime import datetime

async def background_memory_updater(user_id: str, messages: list, store):
    """后台记忆更新任务"""
    namespace = ("memories", user_id)
    
    # 分析对话，提取记忆
    extraction_prompt = f"""从以下对话中提取重要信息：
    
    {chr(10).join([f"{m['role']}: {m['content']}" for m in messages])}
    
    返回需要记住的信息列表。"""
    
    llm = ChatOpenAI(model="gpt-4o-mini")
    response = await llm.ainvoke([{"role": "user", "content": extraction_prompt}])
    
    # 保存提取的记忆
    store.put(namespace, str(uuid.uuid4()), {
        "content": response.content,
        "timestamp": datetime.now().isoformat()
    })

优点：

  • 不影响主流程响应时间
  • 可以进行更复杂的记忆分析

缺点：

  • 新记忆不会立即可用
  • 需要额外的调度机制

完整示例：带记忆的智能助手​

from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver
from langgraph.store.memory import InMemoryStore
from langchain_core.tools import tool
from langchain_core.runnables import RunnableConfig
from langgraph.config import get_store
import uuid
from datetime import datetime

# ========== 配置 ==========

# 创建 checkpointer（短期记忆）
checkpointer = MemorySaver()

# 创建 store（长期记忆）
store = InMemoryStore()

# ========== 工具定义 ==========

@tool
def save_memory(content: str, config: RunnableConfig) -> str:
    """保存重要信息到长期记忆"""
    memory_store = get_store()
    user_id = config.get("configurable", {}).get("user_id", "default")
    
    memory_store.put(
        ("memories", user_id),
        str(uuid.uuid4()),
        {
            "content": content,
            "timestamp": datetime.now().isoformat()
        }
    )
    return f"已保存：{content}"

@tool
def recall_memories(config: RunnableConfig) -> str:
    """检索长期记忆"""
    memory_store = get_store()
    user_id = config.get("configurable", {}).get("user_id", "default")
    
    memories = memory_store.search(("memories", user_id))
    if not memories:
        return "暂无记忆"
    
    return "\n".join([
        f"- {m.value['content']} ({m.value.get('timestamp', 'unknown')})"
        for m in memories
    ])

# ========== 创建 Agent ==========

agent = create_react_agent(
    model="openai:gpt-4o-mini",
    tools=[save_memory, recall_memories],
    prompt="""你是一个智能助手，能够记住用户的信息和偏好。

如果用户告诉你一些值得记住的信息（如偏好、个人信息等），
请使用 save_memory 工具保存。

如果用户询问之前的信息，请使用 recall_memories 工具检索。""",
    checkpointer=checkpointer,
    store=store
)

# ========== 使用示例 ==========

def chat(user_input: str, user_id: str, thread_id: str) -> str:
    """与助手对话"""
    config = {
        "configurable": {
            "user_id": user_id,
            "thread_id": thread_id
        }
    }
    
    result = agent.invoke(
        {"messages": [{"role": "user", "content": user_input}]},
        config=config
    )
    
    return result["messages"][-1].content

# 会话 1：告诉 Agent 一些信息
print("=== 会话 1 ===")
print(chat("你好，我叫张三，我是一名软件工程师", "user-001", "session-001"))

# 会话 2：继续对话（短期记忆）
print("\n=== 会话 2 ===")
print(chat("我刚才说我叫什么？", "user-001", "session-001"))

# 会话 3：新会话，但长期记忆保留
print("\n=== 会话 3（新会话）===")
print(chat("你知道关于我的什么信息？", "user-001", "session-002"))

最佳实践​

1. 选择合适的存储后端​

场景推荐后端
开发/测试MemorySaver / InMemoryStore
单机生产SQLite
分布式生产PostgreSQL / Redis
高并发Redis

2. 合理设置命名空间​

# 按用户隔离
namespace = ("user", user_id, "memories")

# 按类型隔离
namespace = ("user", user_id, "preferences")  # 偏好
namespace = ("user", user_id, "facts")        # 事实
namespace = ("user", user_id, "episodes")     # 经历

3. 定期清理过期记忆​

from datetime import datetime, timedelta

async def cleanup_old_memories(store, user_id: str, max_age_days: int = 90):
    """清理过期的记忆"""
    namespace = ("memories", user_id)
    
    # 获取所有记忆
    all_memories = store.search(namespace, limit=1000)
    
    cutoff_date = datetime.now() - timedelta(days=max_age_days)
    
    for memory in all_memories:
        timestamp = memory.value.get("timestamp")
        if timestamp:
            memory_date = datetime.fromisoformat(timestamp)
            if memory_date < cutoff_date:
                # 删除过期记忆
                store.delete(namespace, memory.key)

4. 隐私保护​

import re

def sanitize_memory(content: str) -> str:
    """脱敏处理敏感信息"""
    patterns = [
        (r'\b\d{17}[\dXx]\b', '[身份证号]'),
        (r'\b1[3-9]\d{9}\b', '[手机号]'),
        (r'\b[\w\.-]+@[\w\.-]+\.\w+\b', '[邮箱]'),
        (r'\b\d{16,19}\b', '[银行卡号]'),
    ]
    
    for pattern, replacement in patterns:
        content = re.sub(pattern, replacement, content)
    
    return content

小结​

LangGraph 的记忆系统提供了完整的解决方案：

  • 短期记忆：通过 checkpointer 实现会话级状态持久化
  • 长期记忆：通过 store 实现跨会话数据存储和语义检索
  • 三种记忆类型：语义、情景、程序记忆，满足不同场景需求
  • 灵活的更新策略：热路径或后台更新
  • 生产级支持：PostgreSQL、Redis、MongoDB 等多种后端

设计良好的记忆系统是构建智能 Agent 的关键，它让 Agent 能够提供连贯、个性化的服务体验。

参考资料​