跳到主要内容

Memory 记忆

Memory（记忆）系统让 AI 应用能够记住对话历史和上下文。本章将详细介绍记忆的概念、类型以及如何在 LangChain 和 LangGraph 中实现记忆功能。

什么是 Memory？​

Memory 是 AI 应用中用于存储和检索对话历史的组件。它让应用能够：

  • 记住之前的对话：用户不需要重复已经说过的话
  • 保持上下文连贯：AI 能够理解指代和省略
  • 积累用户信息：记住用户的偏好、姓名等

为什么需要记忆？​

想象一下和没有记忆的人对话：

用户：我叫张三
AI：你好张三！
用户：我叫什么名字？
AI：（没有记忆）我不知道你的名字。

有了记忆：

用户：我叫张三
AI：你好张三！
用户：我叫什么名字？
AI：（有记忆）你叫张三呀！

记忆的两种类型​

类型说明示例
短期记忆当前对话的上下文对话历史、最近的消息
长期记忆跨会话存储的信息用户偏好、重要事实

LangGraph 中的记忆（推荐）​

LangGraph 提供了现代化的记忆机制，通过 Checkpoint（检查点） 实现状态持久化。

基础概念​

在 LangGraph 中，记忆是通过 checkpointer 实现的：

每次对话 → 自动保存状态（checkpoint）→ 下次对话恢复状态

使用 InMemorySaver（内存记忆）​

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langchain.chat_models import init_chat_model
from langchain.tools import tool

# 定义工具
@tool
def get_weather(city: str) -> str:
    """获取城市天气"""
    return f"{city}天气晴朗，25°C"

# 初始化模型
model = init_chat_model(model="gpt-4o-mini", model_provider="openai")

# 创建内存检查点（短期记忆）
checkpointer = InMemorySaver()

# 创建带记忆的 Agent
app = create_agent(
    model=model,
    tools=[get_weather],
    checkpointer=checkpointer
)

# 配置 thread_id 来区分不同对话
config = {"configurable": {"thread_id": "user_123"}}

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

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

关键点：

  • thread_id 是区分不同对话的标识
  • 相同的 thread_id 会共享记忆
  • 不同的 thread_id 是完全独立的对话

持久化记忆到 SQLite​

内存记忆在程序重启后会丢失，生产环境建议使用持久化存储：

from langgraph.checkpoint.sqlite import SqliteSaver
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool

model = init_chat_model(model="gpt-4o-mini", model_provider="openai")

# 定义工具
@tool
def save_note(content: str) -> str:
    """保存笔记"""
    return f"已保存：{content}"

tools = [save_note]

# 使用 SQLite 持久化
with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
    app = create_agent(
        model=model,
        tools=tools,
        checkpointer=checkpointer
    )
    
    config = {"configurable": {"thread_id": "user_123"}}
    result = app.invoke(
        {"messages": [{"role": "user", "content": "记住我喜欢Python"}]},
        config=config
    )

# 程序重启后，记忆仍然存在
with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
    app = create_agent(
        model=model,
        tools=tools,
        checkpointer=checkpointer
    )
    
    # 仍然记得之前的内容
    result = app.invoke(
        {"messages": [{"role": "user", "content": "我喜欢什么编程语言？"}]},
        config=config
    )

多用户记忆管理​

为每个用户创建独立的记忆：

from langgraph.checkpoint.sqlite import SqliteSaver
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model

model = init_chat_model(model="gpt-4o-mini", model_provider="openai")

def get_agent_for_user(user_id: str):
    """为每个用户创建独立的 Agent 和记忆"""
    checkpointer = SqliteSaver.from_conn_string(f"user_{user_id}.db")
    
    return create_agent(
        model=model,
        tools=tools,
        checkpointer=checkpointer
    ), {"configurable": {"thread_id": f"user_{user_id}"}}

# 用户 1
app1, config1 = get_agent_for_user("user_001")
result = app1.invoke(
    {"messages": [{"role": "user", "content": "我叫李四"}]},
    config=config1
)

# 用户 2
app2, config2 = get_agent_for_user("user_002")
result = app2.invoke(
    {"messages": [{"role": "user", "content": "我叫什么？"}]},
    config=config2
)  # 不会知道李四，因为是独立的记忆

传统 Memory 组件（LangChain）​

除了 LangGraph 的检查点机制，LangChain 还提供了多种 Memory 组件，适用于不同的场景。

ConversationBufferMemory​

最常用的记忆类型，保存完整的对话历史：

from langchain.memory import ConversationBufferMemory

# 创建记忆
memory = ConversationBufferMemory(
    memory_key="history",  # 在 prompt 中引用的键名
    return_messages=True   # 返回消息对象而非字符串
)

# 添加对话
memory.chat_memory.add_user_message("你好，我叫张三")
memory.chat_memory.add_ai_message("你好张三！很高兴认识你")

# 查看历史
history = memory.load_memory_variables({})
print(history["history"])
# [HumanMessage(content='你好，我叫张三'), AIMessage(content='你好张三！很高兴认识你')]

ConversationTokenBufferMemory​

基于 token 数量的记忆，限制内存占用，适合长对话：

from langchain.memory import ConversationTokenBufferMemory
from langchain_openai import ChatOpenAI

# 需要 LLM 来计算 token 数
llm = ChatOpenAI(model="gpt-4o-mini")

memory = ConversationTokenBufferMemory(
    llm=llm,              # 用于计算 token 数
    max_token_limit=500,  # 最大 token 数，超过会丢弃旧消息
    return_messages=True
)

# 添加很多对话
for i in range(20):
    memory.chat_memory.add_user_message(f"问题{i}")
    memory.chat_memory.add_ai_message(f"回答{i}")

# 只保留最近的、token 数不超过 500 的对话
history = memory.load_memory_variables({})

适用场景：

  • 长对话场景
  • 需要控制 token 消耗（节省 API 费用）
  • 上下文窗口有限的模型

ConversationSummaryMemory​

自动生成对话摘要，适合超长对话：

from langchain.memory import ConversationSummaryMemory
from langchain_openai import ChatOpenAI

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

memory = ConversationSummaryMemory(
    llm=llm,              # 用于生成摘要
    memory_key="summary",
    return_messages=True
)

# 添加长对话
memory.chat_memory.add_user_message("我叫张三，住在北京")
memory.chat_memory.add_ai_message("好的，我记住了")
memory.chat_memory.add_user_message("我喜欢吃川菜")
memory.chat_memory.add_ai_message("川菜很辣，你喜欢吃什么川菜？")

# 加载时返回摘要而非完整对话
summary = memory.load_memory_variables({})
print(summary["summary"])
# "张三住在北京，喜欢吃川菜。"

工作原理：

  • 当对话超过一定长度时，自动调用 LLM 生成摘要
  • 后续对话基于摘要进行
  • 大幅减少 token 消耗

BufferWindowMemory​

只保留最近 k 轮对话，简单高效：

from langchain.memory import ConversationBufferWindowMemory

memory = ConversationBufferWindowMemory(
    k=3,                   # 只保留最近 3 轮对话
    memory_key="history",
    return_messages=True
)

# 添加 5 轮对话
for i in range(5):
    memory.chat_memory.add_user_message(f"问题{i}")
    memory.chat_memory.add_ai_message(f"回答{i}")

# 只保留问题2、3、4和回答2、3、4
history = memory.load_memory_variables({})

在 Chain 中使用 Memory​

基础用法​

from langchain_openai import ChatOpenAI
from langchain.chains import LLMChain
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationBufferMemory

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

# 创建记忆
memory = ConversationBufferMemory(
    memory_key="history",
    return_messages=True
)

# 创建 prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好的对话助手。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

# 创建 Chain
chain = LLMChain(
    llm=model,
    prompt=prompt,
    memory=memory,
    verbose=True
)

# 对话
response1 = chain.invoke({"input": "我叫李四"})
print(response1["text"])

response2 = chain.invoke({"input": "你还记得我叫什么吗？"})
print(response2["text"])  # 会回答"李四"

使用 LCEL（推荐）​

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationBufferMemory
from langchain_core.runnables import RunnablePassthrough

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

# 创建记忆
memory = ConversationBufferMemory(
    memory_key="history",
    return_messages=True
)

# 创建 prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好的对话助手。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

# 定义加载记忆的函数
def load_memory(input_dict):
    return memory.load_memory_variables({})["history"]

# 使用 LCEL 构建 Chain
chain = (
    RunnablePassthrough.assign(history=load_memory)
    | prompt
    | model
)

# 对话并保存到记忆
from langchain.schema import HumanMessage, AIMessage

# 第一次对话
input1 = {"input": "我叫王五"}
response1 = chain.invoke(input1)
print(response1.content)

# 保存到记忆
memory.chat_memory.add_user_message(input1["input"])
memory.chat_memory.add_ai_message(response1.content)

# 第二次对话
input2 = {"input": "我叫什么名字？"}
response2 = chain.invoke(input2)
print(response2.content)

消息类型详解​

基本消息类型​

from langchain.schema import HumanMessage, AIMessage, SystemMessage, ToolMessage

# 人类消息（用户输入）
human_msg = HumanMessage(content="你好")
print(human_msg.type)      # "human"
print(human_msg.content)   # "你好"

# AI 消息（模型回复）
ai_msg = AIMessage(content="你好，有什么可以帮您？")
print(ai_msg.type)         # "ai"

# 系统消息（设定角色和行为）
system_msg = SystemMessage(content="你是一个专业的编程助手")
print(system_msg.type)     # "system"

# 工具消息（工具执行结果）
tool_msg = ToolMessage(content="25°C", tool_call_id="call_123")
print(tool_msg.type)       # "tool"

消息历史管理​

from langchain.memory.chat_message_histories import ChatMessageHistory

# 创建消息历史
history = ChatMessageHistory()

# 添加消息
history.add_user_message("第一个问题")
history.add_ai_message("第一个回答")
history.add_user_message("第二个问题")
history.add_ai_message("第二个回答")

# 查看所有消息
for msg in history.messages:
    print(f"{msg.type}: {msg.content}")

# 清空历史
history.clear()

持久化存储方案​

文件存储​

from langchain.memory import ConversationBufferMemory
from langchain.memory.chat_message_histories import FileChatMessageHistory

# 使用文件存储消息
history = FileChatMessageHistory("chat_history.json")

memory = ConversationBufferMemory(
    chat_memory=history,
    memory_key="history",
    return_messages=True
)

# 添加对话
memory.chat_memory.add_user_message("你好")
memory.chat_memory.add_ai_message("你好！")

# 自动保存到文件

Redis 存储​

from langchain.memory import ConversationBufferMemory
from langchain.memory.chat_message_histories.redis import RedisChatMessageHistory

# 创建 Redis 消息历史
history = RedisChatMessageHistory(
    session_id="user_123",           # 会话 ID
    url="redis://localhost:6379"     # Redis 连接地址
)

memory = ConversationBufferMemory(
    chat_memory=history,
    memory_key="history",
    return_messages=True
)

# 添加对话会自动同步到 Redis
memory.chat_memory.add_user_message("你好")

自定义存储​

from langchain.memory.chat_message_histories import ChatMessageHistory
from langchain.schema import HumanMessage, AIMessage

class DatabaseChatMessageHistory(ChatMessageHistory):
    """自定义数据库存储"""
    
    def __init__(self, session_id: str, db_connection):
        self.session_id = session_id
        self.db = db_connection
        self.messages = self._load_messages()
    
    def _load_messages(self):
        """从数据库加载消息"""
        # 实现数据库查询逻辑
        return []
    
    def add_user_message(self, message):
        """添加用户消息到数据库"""
        super().add_user_message(message)
        # 实现数据库插入逻辑
        self.db.execute(
            "INSERT INTO messages (session_id, role, content) VALUES (?, 'user', ?)",
            (self.session_id, message)
        )
    
    def add_ai_message(self, message):
        """添加 AI 消息到数据库"""
        super().add_ai_message(message)
        # 实现数据库插入逻辑
        self.db.execute(
            "INSERT INTO messages (session_id, role, content) VALUES (?, 'ai', ?)",
            (self.session_id, message)
        )
    
    def clear(self):
        """清空历史"""
        super().clear()
        # 实现数据库删除逻辑
        self.db.execute("DELETE FROM messages WHERE session_id = ?", (self.session_id,))

实际应用案例​

案例 1：客服系统​

from langchain.agents import create_agent
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain.chat_models import init_chat_model
from langchain.tools import tool

model = init_chat_model(model="gpt-4o-mini", model_provider="openai")

@tool
def get_order_status(order_id: str) -> str:
    """查询订单状态"""
    orders = {
        "ORD-12345": {"status": "已发货", "tracking": "SF123456"},
        "ORD-67890": {"status": "处理中", "eta": "2天后发货"}
    }
    return str(orders.get(order_id, "订单不存在"))

# 为每个客户创建独立记忆
def get_customer_service_agent(customer_id: str):
    checkpointer = SqliteSaver.from_conn_string(f"customer_{customer_id}.db")
    
    return create_agent(
        model=model,
        tools=[get_order_status],
        checkpointer=checkpointer
    ), {"configurable": {"thread_id": f"customer_{customer_id}"}}

# 客户 001 的会话
app, config = get_customer_service_agent("001")

# 多轮对话，保持上下文
result1 = app.invoke(
    {"messages": [{"role": "user", "content": "我的订单 ORD-12345 到哪了？"}]},
    config=config
)

result2 = app.invoke(
    {"messages": [{"role": "user", "content": "那大概什么时候能到？"}]},
    config=config
)  # Agent 记得是在问 ORD-12345

案例 2：个性化推荐​

from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool

model = init_chat_model(model="gpt-4o-mini", model_provider="openai")

@tool
def recommend_movies(genre: str) -> str:
    """推荐电影"""
    movies = {
        "科幻": ["星际穿越", "盗梦空间", "银翼杀手2049"],
        "喜剧": ["功夫", "夏洛特烦恼", "西虹市首富"],
        "动作": ["战狼2", "红海行动", "流浪地球"]
    }
    return str(movies.get(genre, ["暂无推荐"]))

# 创建带记忆的推荐 Agent
checkpointer = InMemorySaver()
recommendation_agent = create_agent(
    model=model,
    tools=[recommend_movies],
    checkpointer=checkpointer
)

config = {"configurable": {"thread_id": "user_pref_001"}}

# 收集用户偏好
result1 = recommendation_agent.invoke(
    {"messages": [{"role": "user", "content": "我喜欢科幻电影"}]},
    config=config
)

# 基于偏好推荐
result2 = recommendation_agent.invoke(
    {"messages": [{"role": "user", "content": "给我推荐几部电影"}]},
    config=config
)  # 会推荐科幻电影

案例 3：多轮问答系统​

from langchain.memory import BufferWindowMemory
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

# 使用窗口记忆，只保留最近5轮
memory = BufferWindowMemory(k=5, return_messages=True)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个知识问答助手。基于对话历史回答问题。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{question}")
])

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

def answer_question(question: str) -> str:
    # 加载历史
    history = memory.load_memory_variables({})["history"]
    
    # 生成回答
    messages = prompt.format_messages(history=history, question=question)
    response = model.invoke(messages)
    
    # 保存到记忆
    memory.chat_memory.add_user_message(question)
    memory.chat_memory.add_ai_message(response.content)
    
    return response.content

# 多轮问答
questions = [
    "Python 是什么？",
    "它主要用来做什么？",  # "它"指代 Python
    "学它需要什么基础？",   # "它"仍然指代 Python
    "学多久能入门？",
    "有什么学习建议？"
]

for q in questions:
    answer = answer_question(q)
    print(f"Q: {q}")
    print(f"A: {answer}\n")

最佳实践​

1. 选择合适的记忆类型​

场景推荐方案理由
短对话InMemorySaver简单、够用
长对话ConversationTokenBufferMemory控制 token 消耗
超长对话ConversationSummaryMemory自动摘要
生产环境SqliteSaver / Redis持久化

2. 合理设置记忆长度​

# 短对话（10轮以内）
memory = ConversationBufferMemory()

# 长对话，控制 token
memory = ConversationTokenBufferMemory(
    llm=model,
    max_token_limit=2000  # 根据模型上下文窗口调整
)

# 只保留最近几轮
memory = BufferWindowMemory(k=3)

3. 敏感信息处理​

import re

class SensitiveFilterMemory(ConversationBufferMemory):
    """过滤敏感信息的记忆"""
    
    SENSITIVE_PATTERNS = [
        r'\b\d{16,19}\b',  # 银行卡号
        r'\b\d{18}\b',     # 身份证号
        r'password[=:]\s*\S+',  # 密码
    ]
    
    def is_sensitive(self, text: str) -> bool:
        for pattern in self.SENSITIVE_PATTERNS:
            if re.search(pattern, text):
                return True
        return False
    
    def save_context(self, inputs, outputs):
        # 过滤输入中的敏感信息
        filtered_inputs = {
            k: "[FILTERED]" if self.is_sensitive(v) else v
            for k, v in inputs.items()
        }
        super().save_context(filtered_inputs, outputs)

4. 定期清理旧记忆​

import time

def cleanup_old_sessions(storage_path: str, days: int = 30):
    """清理超过指定天数的旧会话"""
    cutoff = time.time() - days * 86400
    
    # 遍历所有存储文件
    for filename in os.listdir(storage_path):
        filepath = os.path.join(storage_path, filename)
        if os.path.getmtime(filepath) < cutoff:
            os.remove(filepath)
            print(f"已删除旧会话: {filename}")

下一步​

参考资源​