跳到主要内容

Memory 记忆

Memory（记忆）系统让 LangChain 应用能够记住对话历史和上下文。本章将详细介绍各种记忆组件及其使用方法。

什么是 Memory？​

Memory 是 LangChain 中用于存储和检索对话历史的组件。它让 AI 应用能够记住之前的交互，提供更连贯的对话体验。

Memory 的作用​

  • 短期记忆：记住当前对话的上下文
  • 长期记忆：跨会话存储用户偏好和历史
  • 上下文管理：为 LLM 提供完整的对话背景

Memory 工作原理​

Memory 类型​

1. ConversationBufferMemory​

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

from langchain.memory import ConversationBufferMemory
from langchain.memory.chat_message_histories.in_memory import ChatMessageHistory

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

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

# 查看历史
history = memory.load_memory_variables({"chat_history": []})
print(history)

2. ConversationTokenBufferMemory​

基于 token 数量的记忆，限制内存占用：

from langchain.memory import ConversationTokenBufferMemory

memory = ConversationTokenBufferMemory(
    llm=model,              # 用于计算 token 数
    max_token_limit=2000,   # 最大 token 数
    return_messages=True
)

适用场景：处理长对话时控制 token 消耗

3. ConversationSummaryMemory​

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

from langchain.memory import ConversationSummaryMemory

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

# 添加长对话
memory.chat_memory.add_user_message("...")
memory.chat_memory.add_ai_message("...")

# 加载时返回摘要
summary = memory.load_memory_variables({})

4. BufferWindowMemory​

只保留最近 k 轮对话：

from langchain.memory import BufferWindowMemory

memory = BufferWindowMemory(
    k=3,                   # 保留最近3轮
    return_messages=True
)

在 Chain 中使用 Memory​

基本用法​

from langchain.chat_models import init_chat_model
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain.memory import ConversationBufferMemory

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

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

# 创建 prompt
prompt = PromptTemplate.from_template("""
你是一个对话助手。

对话历史：
{history}

用户：{input}
助手：
""")

# 创建 Chain（带记忆）
chain = LLMChain(
    llm=model,
    prompt=prompt,
    memory=memory
)

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

response2 = chain.invoke({"input": "你还记得我叫什么吗?"})
print(response2["text"])

使用 LCEL​

from langchain.schema import MessagesPlaceholder
from langchain.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的助手"),
    MessagesPlaceholder(variable_name="history", optional=True),
    ("user", "{input}")
])

chain = (
    {"input": lambda x: x["input"], "history": memory.load_memory_variables}
    | prompt
    | model
)

在 Agent 中使用 Memory​

创建带记忆的 Agent​

from langchain.agents import create_agent
from langchain.memory import ConversationBufferMemory
from langchain.tools import tool

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

@tool
def get_weather(city: str) -> str:
    """获取城市天气"""
    return f"{city}天气晴朗"

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

# Agent 的 prompt 需要包含历史
system_prompt = """你是一个有帮助的助手。

历史对话：
{chat_history}

可用工具：
- get_weather: 获取天气"""

# 创建 Agent
agent = create_agent(
    model=model,
    tools=[get_weather],
    prompt={
        "system_prompt": system_prompt,
        "input": "{input}",
        "chat_history": "{chat_history}"
    }
)

# 对话
result1 = agent.invoke({"messages": [{"role": "user", "content": "北京天气怎么样?"}]})
result2 = agent.invoke({"messages": [{"role": "user", "content": "上海呢?"}]})

消息类型​

消息对象​

from langchain.schema import HumanMessage, AIMessage, SystemMessage

# 人类消息
msg = HumanMessage(content="你好")
print(msg.type)        # "human"
print(msg.content)    # "你好"

# AI 消息
msg = AIMessage(content="你好，有什么可以帮您?")
print(msg.type)        # "ai"

# 系统消息
msg = SystemMessage(content="你是一个有帮助的助手")
print(msg.type)        # "system"

消息历史存储​

from langchain.memory.chat_message_histories.in_memory 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}")

持久化记忆​

文件存储​

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
)

Redis 存储​

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

# 创建 Redis 消息历史
history = RedisChatMessageHistory(
    session_id="user_123",
    url="redis://localhost:6379"
)

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

自定义存储​

from langchain.memory import ChatMessageHistory
from langchain.schema import BaseChatMessageHistory

class MyCustomHistory(BaseChatMessageHistory):
    def __init__(self):
        self.messages = []
    
    def add_user_message(self, message):
        self.messages.append(HumanMessage(content=message))
    
    def add_ai_message(self, message):
        self.messages.append(AIMessage(content=message))
    
    def clear(self):
        self.messages = []

# 使用
history = MyCustomHistory()
memory = ConversationBufferMemory(
    chat_memory=history,
    memory_key="history"
)

多用户记忆管理​

会话隔离​

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

def get_memory_for_user(user_id: str):
    """为每个用户创建独立的记忆"""
    history = RedisChatMessageHistory(
        session_id=f"user_{user_id}",
        url="redis://localhost:6379"
    )
    
    return ConversationBufferMemory(
        chat_memory=history,
        memory_key="history"
    )

# 为不同用户创建独立记忆
memory_user1 = get_memory_for_user("user_001")
memory_user2 = get_memory_for_user("user_002")

SQLite 存储​

from langchain.memory import ConversationBufferMemory
from langchain.memory.chat_message_histories import SQLiteChatMessageHistory

# 使用 SQLite 存储
history = SQLiteChatMessageHistory("conversations.db")

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

高级用法​

摘要式记忆​

from langchain.memory import ConversationSummaryMemory
from langchain.chat_models import init_chat_model

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

memory = ConversationSummaryMemory(
    llm=model,
    memory_key="summary"
)

# 添加很多对话后，memory 会自动生成摘要
for i in range(10):
    memory.chat_memory.add_user_message(f"问题{i}")
    memory.chat_memory.add_ai_message(f"回答{i}")

# 查看摘要
summary = memory.load_memory_variables({})
print(summary["summary"])

组合多种记忆​

from langchain.memory import CombinedMemory

# 短期记忆（最近几轮）
short_memory = ConversationBufferMemory(
    memory_key="recent",
    k=3
)

# 长期记忆（用户偏好）
long_memory = ConversationBufferMemory(
    memory_key="preferences"
)

# 组合
combined = CombinedMemory(memories=[short_memory, long_memory])

实体记忆​

from langchain.memory import EntityMemory

memory = EntityMemory(
    llm=model,
    memory_key="entities"
)

# 添加对话
memory.save_context(
    {"input": "我叫张三，我住在北京"},
    {"output": "好的，我记住你叫张三，住在北京"}
)

# 查看实体
entities = memory.load_memory_variables({})
print(entities["entities"])
# {'张三': '住在这里...', '北京': '居住地...'}

实际应用示例​

1. 客服系统​

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

# 为每个用户创建独立记忆
def get_customer_memory(customer_id: str):
    history = FileChatMessageHistory(f"customer_{customer_id}.json")
    return ConversationBufferMemory(
        chat_memory=history,
        memory_key="history"
    )

# 创建对话链
memory = get_customer_memory("customer_001")
chain = LLMChain(llm=model, prompt=prompt, memory=memory)

# 处理对话
chain.invoke({"input": "我想查询订单"})
chain.invoke({"input": "订单号是12345"})

2. 多轮问答系统​

from langchain.memory import BufferWindowMemory

memory = BufferWindowMemory(k=5)  # 保留最近5轮

# 问答链
qa_prompt = PromptTemplate.from_template("""
基于以下对话历史，回答用户问题。如果历史信息不相关，只根据常识回答。

历史：
{history}

问题：{question}
回答：
""")

chain = LLMChain(llm=model, prompt=qa_prompt, memory=memory)

# 多轮问答
questions = [
    "Python 是什么?",
    "它主要用来做什么?",
    "学Python需要什么基础?",
    "学多久能入门?",
    "找不到工作怎么办?"
]

for q in questions:
    result = chain.invoke({"question": q})
    print(f"Q: {q}")
    print(f"A: {result['text']}\n")

3. 个性化推荐​

from langchain.memory import ConversationBufferMemory
from langchain.tools import tool

# 保存用户偏好
preference_memory = ConversationBufferMemory(
    memory_key="preferences"
)

@tool
def recommend_movies(genre: str) -> list:
    """推荐电影"""
    return ["电影A", "电影B", "电影C"]

# 收集偏好
preference_memory.chat_memory.add_user_message("我喜欢科幻电影")
preference_memory.chat_memory.add_ai_message("好的，我记住你喜欢科幻类型")

# 使用偏好推荐
def get_recommendations():
    prefs = preference_memory.load_memory_variables({})
    # 从偏好中提取信息并推荐
    return recommend_movies.invoke("科幻")

最佳实践​

1. 合理设置记忆长度​

# 短对话
memory = ConversationBufferMemory()

# 长对话，控制 token
memory = ConversationTokenBufferMemory(
    llm=model,
    max_token_limit=2000
)

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

2. 定期清理记忆​

# 定期清理旧记忆
def cleanup_old_sessions(days=30):
    import time
    cutoff = time.time() - days * 86400
    # 清理逻辑...

3. 敏感信息处理​

# 不存储敏感信息
class SensitiveFilterMemory(ConversationBufferMemory):
    def save_context(self, inputs, outputs):
        # 过滤敏感信息
        filtered = {k: v for k, v in inputs.items() if not is_sensitive(k, v)}
        super().save_context(filtered, outputs)

下一步​

现在你已经掌握了 Memory 的使用方法，接下来学习：