记忆（Memory） #
记忆系统让 AI 应用能够记住之前的对话内容，实现连贯的多轮对话。LangChain 提供了多种记忆类型，满足不同场景的需求。
记忆系统概览 #
text
┌─────────────────────────────────────────────────────────────┐
│ LangChain 记忆系统 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 为什么需要记忆？ │
│ │
│ 无记忆： │
│ 用户: 我叫张三 │
│ AI: 你好，张三！ │
│ 用户: 我叫什么？ │
│ AI: 抱歉，我不知道你的名字... │
│ │
│ 有记忆： │
│ 用户: 我叫张三 │
│ AI: 你好，张三！ │
│ 用户: 我叫什么？ │
│ AI: 你叫张三。 │
│ │
└─────────────────────────────────────────────────────────────┘
记忆类型 #
1. ConversationBufferMemory #
保存完整的对话历史：
python
from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory()
# 添加对话
memory.save_context(
{"input": "你好，我叫张三"},
{"output": "你好，张三！很高兴认识你"}
)
memory.save_context(
{"input": "我喜欢编程"},
{"output": "编程是一项很有趣的技能！"}
)
# 获取历史
print(memory.load_memory_variables({}))
# {'history': 'Human: 你好，我叫张三\nAI: 你好，张三！很高兴认识你\nHuman: 我喜欢编程\nAI: 编程是一项很有趣的技能！'}
2. ConversationBufferWindowMemory #
只保留最近 N 轮对话：
python
from langchain.memory import ConversationBufferWindowMemory
# 只保留最近 2 轮对话
memory = ConversationBufferWindowMemory(k=2)
memory.save_context({"input": "消息1"}, {"output": "回复1"})
memory.save_context({"input": "消息2"}, {"output": "回复2"})
memory.save_context({"input": "消息3"}, {"output": "回复3"})
print(memory.load_memory_variables({}))
# 只显示最近 2 轮：消息2 和 消息3
text
┌─────────────────────────────────────────────────────────────┐
│ 滑动窗口记忆 │
├─────────────────────────────────────────────────────────────┤
│ │
│ k=2 时的记忆窗口： │
│ │
│ 时间线： │
│ ─────────────────────────────────────────────> │
│ │
│ [消息1] ─> [消息2] ─> [消息3] ─> [消息4] │
│ ❌ ✅ ✅ ✅ │
│ 丢弃 保留 保留 新增 │
│ │
│ 窗口始终只保留最近的 k 轮对话 │
│ │
└─────────────────────────────────────────────────────────────┘
3. ConversationSummaryMemory #
使用 LLM 自动总结历史对话：
python
from langchain.memory import ConversationSummaryMemory
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o-mini")
memory = ConversationSummaryMemory(llm=model)
memory.save_context(
{"input": "你好，我是张三，一名 Python 开发者"},
{"output": "你好张三！很高兴认识一位 Python 开发者"}
)
memory.save_context(
{"input": "我最近在学习机器学习"},
{"output": "机器学习是个很好的方向！"}
)
print(memory.load_memory_variables({}))
# {'history': '用户张三是一名 Python 开发者，最近在学习机器学习。'}
4. ConversationSummaryBufferMemory #
结合总结和缓冲区，在 token 限制内保留最多信息：
python
from langchain.memory import ConversationSummaryBufferMemory
memory = ConversationSummaryBufferMemory(
llm=model,
max_token_limit=100 # 最大 token 数
)
# 当对话超过 token 限制时，旧对话会被总结
memory.save_context({"input": "长对话1..."}, {"output": "长回复1..."})
memory.save_context({"input": "长对话2..."}, {"output": "长回复2..."})
5. ConversationTokenBufferMemory #
基于 token 数量限制记忆：
python
from langchain.memory import ConversationTokenBufferMemory
memory = ConversationTokenBufferMemory(
llm=model,
max_token_limit=1000
)
6. VectorStoreRetrieverMemory #
将记忆存储在向量数据库中，支持语义检索：
python
from langchain.memory import VectorStoreRetrieverMemory
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
# 创建向量存储
vectorstore = Chroma(
embedding_function=OpenAIEmbeddings(),
persist_directory="./chroma_memory"
)
# 创建检索器
retriever = vectorstore.as_retriever(
search_kwargs={"k": 3} # 检索最相关的 3 条记忆
)
memory = VectorStoreRetrieverMemory(retriever=retriever)
# 保存记忆
memory.save_context(
{"input": "我最喜欢的颜色是蓝色"},
{"output": "好的，我记住了你喜欢蓝色"}
)
memory.save_context(
{"input": "我喜欢吃披萨"},
{"output": "披萨确实很美味！"}
)
# 基于语义检索相关记忆
print(memory.load_memory_variables({"prompt": "颜色"}))
# 会返回关于颜色的记忆
text
┌─────────────────────────────────────────────────────────────┐
│ 记忆类型对比 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 类型 │ 特点 │ 适用场景 │
│ ───────────────────────────────────────────────────────── │
│ Buffer │ 完整保存 │ 短对话 │
│ BufferWindow │ 固定轮数 │ 有限上下文 │
│ Summary │ 自动总结 │ 长对话 │
│ SummaryBuffer │ 总结+缓冲 │ 平衡方案 │
│ TokenBuffer │ Token 限制 │ 精确控制 │
│ VectorStoreRetriever │ 语义检索 │ 大规模记忆 │
│ │
└─────────────────────────────────────────────────────────────┘
在链中使用记忆 #
基础用法 #
python
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
model = ChatOpenAI(model="gpt-4o-mini")
memory = ConversationBufferMemory(return_messages=True)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个友好的助手"),
MessagesPlaceholder(variable_name="history"),
("human", "{input}")
])
def chat(input_text: str) -> str:
# 加载历史
history = memory.load_memory_variables({})["history"]
# 调用链
chain = prompt | model | StrOutputParser()
result = chain.invoke({
"history": history,
"input": input_text
})
# 保存对话
memory.save_context(
{"input": input_text},
{"output": result}
)
return result
# 测试
print(chat("你好，我叫张三"))
print(chat("我叫什么？")) # 会记住名字
使用 RunnableWithMessageHistory #
python
from langchain_core.runnables import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
model = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个有帮助的助手"),
MessagesPlaceholder(variable_name="history"),
("human", "{input}")
])
chain = prompt | model
# 会话存储
store = {}
def get_session_history(session_id: str) -> BaseChatMessageHistory:
if session_id not in store:
store[session_id] = ChatMessageHistory()
return store[session_id]
# 创建带历史的链
chain_with_history = RunnableWithMessageHistory(
chain,
get_session_history,
input_messages_key="input",
history_messages_key="history"
)
# 使用
result = chain_with_history.invoke(
{"input": "你好，我叫张三"},
config={"configurable": {"session_id": "user-123"}}
)
result = chain_with_history.invoke(
{"input": "我叫什么？"},
config={"configurable": {"session_id": "user-123"}}
)
text
┌─────────────────────────────────────────────────────────────┐
│ 会话管理流程 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 用户请求 ───> 获取会话历史 ───> 构建提示 ───> 调用模型 │
│ │ │
│ ▼ │
│ 返回响应 <─── 保存对话历史 <─── 获取响应 │
│ │
│ session_id 用于区分不同用户/会话 │
│ │
└─────────────────────────────────────────────────────────────┘
持久化存储 #
文件存储 #
python
from langchain_community.chat_message_histories import FileChatMessageHistory
# 创建文件存储的历史记录
history = FileChatMessageHistory("chat_history.json")
# 添加消息
from langchain_core.messages import HumanMessage, AIMessage
history.add_message(HumanMessage(content="你好"))
history.add_message(AIMessage(content="你好！"))
# 读取消息
print(history.messages)
Redis 存储 #
python
from langchain_community.chat_message_histories import RedisChatMessageHistory
history = RedisChatMessageHistory(
session_id="user-123",
url="redis://localhost:6379"
)
history.add_message(HumanMessage(content="你好"))
SQLite 存储 #
python
from langchain_community.chat_message_histories import SQLChatMessageHistory
history = SQLChatMessageHistory(
session_id="user-123",
connection="sqlite:///chat_history.db"
)
MongoDB 存储 #
python
from langchain_community.chat_message_histories import MongoDBChatMessageHistory
history = MongoDBChatMessageHistory(
session_id="user-123",
connection_string="mongodb://localhost:27017",
database_name="chat_db",
collection_name="chat_history"
)
自定义记忆 #
创建自定义记忆类 #
python
from langchain.memory import BaseChatMemory
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from typing import List, Dict, Any
class CustomMemory(BaseChatMemory):
"""自定义记忆：只记住包含关键词的对话"""
def __init__(self, keywords: List[str]):
self.keywords = keywords
self.messages: List[BaseMessage] = []
@property
def memory_variables(self) -> List[str]:
return ["history"]
def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, str]:
# 只返回包含关键词的消息
relevant_messages = []
for msg in self.messages:
if any(kw in msg.content for kw in self.keywords):
relevant_messages.append(msg)
return {"history": "\n".join([m.content for m in relevant_messages])}
def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None:
input_text = inputs.get("input", "")
output_text = outputs.get("output", "")
# 只保存包含关键词的对话
if any(kw in input_text or kw in output_text for kw in self.keywords):
self.messages.append(HumanMessage(content=input_text))
self.messages.append(AIMessage(content=output_text))
def clear(self) -> None:
self.messages = []
# 使用
memory = CustomMemory(keywords=["名字", "喜欢", "重要"])
带权重的记忆 #
python
from collections import defaultdict
from datetime import datetime
class WeightedMemory:
"""带权重的记忆：重要信息权重更高"""
def __init__(self):
self.memories = []
self.importance_keywords = {
"名字": 10,
"喜欢": 5,
"讨厌": 5,
"重要": 8,
"记住": 7
}
def add_memory(self, text: str, response: str):
# 计算重要性
importance = 0
for keyword, weight in self.importance_keywords.items():
if keyword in text:
importance += weight
self.memories.append({
"text": text,
"response": response,
"importance": importance,
"timestamp": datetime.now()
})
def get_relevant_memories(self, query: str, top_k: int = 5):
# 按重要性排序
sorted_memories = sorted(
self.memories,
key=lambda x: x["importance"],
reverse=True
)
return sorted_memories[:top_k]
记忆管理策略 #
1. 清理过期记忆 #
python
from datetime import datetime, timedelta
def clean_old_memories(memories: list, days: int = 7):
"""清理超过指定天数的记忆"""
cutoff = datetime.now() - timedelta(days=days)
return [m for m in memories if m["timestamp"] > cutoff]
2. 压缩记忆 #
python
from langchain.chains import LLMChain
from langchain_core.prompts import PromptTemplate
def compress_memories(model, memories: list) -> str:
"""使用 LLM 压缩多条记忆"""
memory_text = "\n".join([
f"用户: {m['text']}\nAI: {m['response']}"
for m in memories
])
prompt = PromptTemplate.from_template(
"请将以下对话历史压缩成简短的摘要：\n\n{history}"
)
chain = prompt | model | StrOutputParser()
return chain.invoke({"history": memory_text})
3. 记忆优先级 #
python
class PriorityMemory:
"""按优先级管理记忆"""
def __init__(self, max_items: int = 100):
self.max_items = max_items
self.memories = []
def add(self, text: str, priority: int = 0):
self.memories.append({
"text": text,
"priority": priority,
"access_count": 0
})
# 超过限制时，移除优先级最低的
if len(self.memories) > self.max_items:
self.memories.sort(key=lambda x: x["priority"])
self.memories = self.memories[-self.max_items:]
def access(self, index: int):
"""访问记忆时增加访问计数"""
if 0 <= index < len(self.memories):
self.memories[index]["access_count"] += 1
# 热门记忆提升优先级
self.memories[index]["priority"] += 1
完整示例 #
智能客服系统 #
python
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationSummaryBufferMemory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableWithMessageHistory
from langchain_community.chat_message_histories import RedisChatMessageHistory
model = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_messages([
("system", """你是一个专业的客服助手。
请记住用户的重要信息，包括：
- 姓名
- 订单号
- 问题描述
- 联系方式
请用专业、友好的语气回复。"""),
MessagesPlaceholder(variable_name="history"),
("human", "{input}")
])
def get_session_history(session_id: str):
return RedisChatMessageHistory(
session_id=session_id,
url="redis://localhost:6379"
)
chain = prompt | model | StrOutputParser()
chain_with_history = RunnableWithMessageHistory(
chain,
get_session_history,
input_messages_key="input",
history_messages_key="history"
)
def chat(session_id: str, message: str) -> str:
return chain_with_history.invoke(
{"input": message},
config={"configurable": {"session_id": session_id}}
)
# 使用
session_id = "customer-001"
print(chat(session_id, "你好，我叫张三，订单号是12345"))
print(chat(session_id, "我的订单还没收到，能帮我查一下吗？"))
print(chat(session_id, "我的订单号是多少？")) # 会记住订单号
最佳实践 #
1. 选择合适的记忆类型 #
python
# 短对话 - Buffer
memory = ConversationBufferMemory()
# 长对话 - Summary
memory = ConversationSummaryMemory(llm=model)
# 有限上下文 - Window
memory = ConversationBufferWindowMemory(k=5)
# 大规模应用 - VectorStore
memory = VectorStoreRetrieverMemory(retriever=retriever)
2. 设置合理的限制 #
python
# Token 限制
memory = ConversationTokenBufferMemory(
llm=model,
max_token_limit=2000 # 根据模型上下文长度设置
)
# 轮数限制
memory = ConversationBufferWindowMemory(k=10)
3. 定期清理 #
python
# 设置过期时间
# 使用 Redis 时可以设置 TTL
history = RedisChatMessageHistory(
session_id="user-123",
url="redis://localhost:6379",
ttl=86400 # 24小时后过期
)
下一步 #
- 代理（Agents） - 让模型自主决策
- 工具（Tools） - 扩展模型能力
- RAG 应用 - 构建知识库应用
最后更新：2026-03-30