记忆系统
记忆系统是 AI Agent 的重要组成部分，它让 Agent 能够记住对话历史、用户偏好和任务状态，从而提供连贯、个性化的服务。本章将深入介绍记忆系统的设计原理和实现方式。
为什么需要记忆系统
没有记忆系统的 Agent 就像一个"健忘"的助手，每次对话都从零开始：
用户：我叫张三
Agent：你好，张三！
用户：我叫什么名字？
Agent：抱歉，我不知道您的名字。
有了记忆系统，Agent 能够记住之前的对话内容：
用户：我叫张三
Agent：你好，张三！
用户：我叫什么名字？
Agent：您叫张三。
记忆系统解决了以下问题：
- 上下文连贯性：让对话保持连贯，避免重复询问
- 用户个性化：记住用户偏好，提供个性化服务
- 任务连续性：在多轮对话中持续推进任务
- 知识积累：存储和检索知识信息
记忆类型
短期记忆
短期记忆存储当前对话的上下文信息，通常通过在提示词中包含对话历史来实现。
特点：
- 容量有限，受 LLM 上下文窗口限制
- 对话结束后通常清空
- 实现简单，直接拼接历史消息
长期记忆
长期记忆持久化存储重要信息，支持跨会话检索。
特点：
- 容量可扩展
- 支持语义检索
- 需要额外的存储系统
LangChain 记忆组件
LangChain 提供了多种记忆组件。
ConversationBufferMemory
最简单的记忆形式，保存所有对话历史：
from langchain.memory import ConversationBufferMemory
from langchain_openai import ChatOpenAI
from langchain.chains import ConversationChain
memory = ConversationBufferMemory()
llm = ChatOpenAI(model="gpt-4")
conversation = ConversationChain(
llm=llm,
memory=memory,
verbose=True
)
conversation.predict(input="你好，我叫张三")
conversation.predict(input="我喜欢打篮球")
conversation.predict(input="我叫什么名字？我喜欢什么？")
print(memory.load_memory_variables({}))
输出：
{'history': 'Human: 你好，我叫张三\nAI: 你好，张三！很高兴认识你。\nHuman: 我喜欢打篮球\nAI: 篮球是一项很棒的运动！\nHuman: 我叫什么名字？我喜欢什么？\nAI: 你叫张三，你喜欢打篮球。'}
ConversationBufferWindowMemory
只保留最近 N 轮对话，避免上下文过长：
from langchain.memory import ConversationBufferWindowMemory
memory = ConversationBufferWindowMemory(k=2)
conversation = ConversationChain(
llm=llm,
memory=memory,
verbose=True
)
conversation.predict(input="第一条消息")
conversation.predict(input="第二条消息")
conversation.predict(input="第三条消息")
conversation.predict(input="第四条消息")
print(memory.load_memory_variables({}))
只保留最近 2 轮对话（4 条消息）。
ConversationTokenBufferMemory
根据 token 数量限制记忆大小：
from langchain.memory import ConversationTokenBufferMemory
memory = ConversationTokenBufferMemory(
llm=llm,
max_token_limit=100
)
conversation = ConversationChain(
llm=llm,
memory=memory
)
ConversationSummaryMemory
使用 LLM 自动总结对话历史：
from langchain.memory import ConversationSummaryMemory
memory = ConversationSummaryMemory(llm=llm)
conversation = ConversationChain(
llm=llm,
memory=memory,
verbose=True
)
conversation.predict(input="你好，我叫张三，我是一名软件工程师")
conversation.predict(input="我擅长 Python 和 JavaScript")
conversation.predict(input="我最近在学习人工智能")
print(memory.load_memory_variables({}))
输出会是对话的总结，而不是原始对话。
VectorStoreRetrieverMemory
使用向量数据库存储和检索记忆：
from langchain.memory import VectorStoreRetrieverMemory
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_texts([""], embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
memory = VectorStoreRetrieverMemory(retriever=retriever)
memory.save_context(
{"input": "我叫张三"},
{"output": "你好，张三！"}
)
memory.save_context(
{"input": "我喜欢打篮球"},
{"output": "篮球是一项很棒的运动！"}
)
memory.save_context(
{"input": "我的生日是3月15日"},
{"output": "我会记住的，你的生日是3月15日。"}
)
result = memory.load_memory_variables({"prompt": "关于张三的信息"})
print(result)
自定义记忆系统
实现一个完整的记忆管理器
from typing import Dict, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
import json
@dataclass
class Message:
"""消息记录"""
role: str
content: str
timestamp: datetime = field(default_factory=datetime.now)
metadata: Dict = field(default_factory=dict)
@dataclass
class Session:
"""会话记录"""
session_id: str
messages: List[Message] = field(default_factory=list)
user_info: Dict = field(default_factory=dict)
created_at: datetime = field(default_factory=datetime.now)
updated_at: datetime = field(default_factory=datetime.now)
class MemoryManager:
"""记忆管理器"""
def __init__(self, max_messages: int = 100):
self.max_messages = max_messages
self.sessions: Dict[str, Session] = {}
self.user_memories: Dict[str, Dict] = {}
def get_or_create_session(self, session_id: str, user_id: str = None) -> Session:
"""获取或创建会话"""
if session_id not in self.sessions:
self.sessions[session_id] = Session(
session_id=session_id,
user_info={"user_id": user_id} if user_id else {}
)
return self.sessions[session_id]
def add_message(
self,
session_id: str,
role: str,
content: str,
metadata: Dict = None
):
"""添加消息到会话"""
session = self.get_or_create_session(session_id)
message = Message(
role=role,
content=content,
metadata=metadata or {}
)
session.messages.append(message)
session.updated_at = datetime.now()
if len(session.messages) > self.max_messages:
session.messages = session.messages[-self.max_messages:]
def get_messages(
self,
session_id: str,
limit: int = None
) -> List[Dict]:
"""获取会话消息"""
session = self.sessions.get(session_id)
if not session:
return []
messages = session.messages
if limit:
messages = messages[-limit:]
return [
{"role": m.role, "content": m.content}
for m in messages
]
def save_user_memory(self, user_id: str, key: str, value: any):
"""保存用户长期记忆"""
if user_id not in self.user_memories:
self.user_memories[user_id] = {}
self.user_memories[user_id][key] = {
"value": value,
"timestamp": datetime.now().isoformat()
}
def get_user_memory(self, user_id: str, key: str = None) -> any:
"""获取用户长期记忆"""
if user_id not in self.user_memories:
return None
if key:
return self.user_memories[user_id].get(key)
return self.user_memories[user_id]
def extract_important_info(self, session_id: str) -> List[Dict]:
"""从对话中提取重要信息"""
messages = self.get_messages(session_id)
important_patterns = [
("名字", r"我叫(.+?)。"),
("职业", r"我是一名(.+?)。"),
("爱好", r"我喜欢(.+?)。"),
("生日", r"我的生日是(.+?)。"),
]
import re
extracted = []
for msg in messages:
content = msg.get("content", "")
for label, pattern in important_patterns:
match = re.search(pattern, content)
if match:
extracted.append({
"type": label,
"value": match.group(1).strip(),
"source": content[:50]
})
return extracted
def clear_session(self, session_id: str):
"""清除会话记忆"""
if session_id in self.sessions:
del self.sessions[session_id]
def export_session(self, session_id: str) -> str:
"""导出会话记录"""
session = self.sessions.get(session_id)
if not session:
return "{}"
return json.dumps({
"session_id": session.session_id,
"messages": [
{
"role": m.role,
"content": m.content,
"timestamp": m.timestamp.isoformat()
}
for m in session.messages
],
"user_info": session.user_info,
"created_at": session.created_at.isoformat(),
"updated_at": session.updated_at.isoformat()
}, ensure_ascii=False, indent=2)
memory_manager = MemoryManager()
memory_manager.add_message("session_001", "user", "你好，我叫张三")
memory_manager.add_message("session_001", "assistant", "你好，张三！很高兴认识你。")
memory_manager.add_message("session_001", "user", "我是一名软件工程师")
memory_manager.add_message("session_001", "assistant", "软件工程师是很棒的职业！")
print(memory_manager.get_messages("session_001"))
print(memory_manager.extract_important_info("session_001"))
memory_manager.save_user_memory("user_001", "name", "张三")
memory_manager.save_user_memory("user_001", "job", "软件工程师")
print(memory_manager.get_user_memory("user_001"))
结合向量数据库的长期记忆
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.schema import Document
from typing import List, Dict
import uuid
class LongTermMemory:
"""基于向量数据库的长期记忆"""
def __init__(self, persist_directory: str = "./memory_db"):
self.embeddings = OpenAIEmbeddings()
self.vectorstore = Chroma(
embedding_function=self.embeddings,
persist_directory=persist_directory
)
def save_memory(
self,
content: str,
metadata: Dict = None,
user_id: str = None
) -> str:
"""保存记忆"""
memory_id = str(uuid.uuid4())
doc = Document(
page_content=content,
metadata={
"id": memory_id,
"user_id": user_id,
**(metadata or {})
}
)
self.vectorstore.add_documents([doc])
return memory_id
def search_memories(
self,
query: str,
user_id: str = None,
k: int = 5
) -> List[Dict]:
"""搜索相关记忆"""
filter_dict = {}
if user_id:
filter_dict["user_id"] = user_id
results = self.vectorstore.similarity_search(
query,
k=k,
filter=filter_dict if filter_dict else None
)
return [
{
"content": doc.page_content,
"metadata": doc.metadata
}
for doc in results
]
def save_conversation_memory(
self,
user_input: str,
assistant_response: str,
user_id: str,
important: bool = False
):
"""保存对话记忆"""
content = f"用户：{user_input}\n助手：{assistant_response}"
self.save_memory(
content=content,
metadata={
"type": "conversation",
"important": important
},
user_id=user_id
)
def save_fact(self, fact: str, user_id: str, category: str = "general"):
"""保存事实性记忆"""
self.save_memory(
content=fact,
metadata={
"type": "fact",
"category": category
},
user_id=user_id
)
def get_user_facts(self, user_id: str) -> List[str]:
"""获取用户的所有事实性记忆"""
results = self.vectorstore.similarity_search(
"",
k=100,
filter={"user_id": user_id, "type": "fact"}
)
return [doc.page_content for doc in results]
long_term_memory = LongTermMemory()
long_term_memory.save_fact("用户的名字是张三", "user_001", "personal")
long_term_memory.save_fact("用户是一名软件工程师", "user_001", "professional")
long_term_memory.save_fact("用户喜欢打篮球", "user_001", "hobby")
results = long_term_memory.search_memories("用户喜欢什么", "user_001")
for r in results:
print(r["content"])
记忆压缩与总结
当对话历史过长时，需要对记忆进行压缩。
滑动窗口 + 总结
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
class CompressedMemory:
"""压缩记忆管理器"""
def __init__(self, window_size: int = 5, llm=None):
self.window_size = window_size
self.llm = llm or ChatOpenAI(model="gpt-4")
self.recent_messages: List[Dict] = []
self.summary: str = ""
def add_message(self, role: str, content: str):
"""添加消息"""
self.recent_messages.append({"role": role, "content": content})
if len(self.recent_messages) > self.window_size * 2:
self._compress()
def _compress(self):
"""压缩旧消息"""
messages_to_compress = self.recent_messages[:-self.window_size]
self.recent_messages = self.recent_messages[-self.window_size:]
new_summary = self._summarize(messages_to_compress)
if self.summary:
self.summary = f"{self.summary}\n{new_summary}"
else:
self.summary = new_summary
def _summarize(self, messages: List[Dict]) -> str:
"""总结消息"""
prompt = PromptTemplate(
template="""
请总结以下对话的关键信息，保留重要的事实和上下文：
对话内容：
{messages}
总结（用简洁的语言，保留关键信息）：
""",
input_variables=["messages"]
)
messages_text = "\n".join([
f"{m['role']}: {m['content']}"
for m in messages
])
return self.llm.invoke(prompt.format(messages=messages_text)).content
def get_context(self) -> str:
"""获取完整上下文"""
context_parts = []
if self.summary:
context_parts.append(f"[历史总结]\n{self.summary}")
if self.recent_messages:
recent_text = "\n".join([
f"{m['role']}: {m['content']}"
for m in self.recent_messages
])
context_parts.append(f"[最近对话]\n{recent_text}")
return "\n\n".join(context_parts)
memory = CompressedMemory(window_size=3)
for i in range(10):
memory.add_message("user", f"这是第 {i+1} 条用户消息")
memory.add_message("assistant", f"这是第 {i+1} 条助手回复")
print(memory.get_context())
记忆检索策略
基于相关性的检索
def retrieve_relevant_memories(
query: str,
vectorstore,
k: int = 5,
recency_weight: float = 0.3
) -> List[Document]:
"""
结合相关性和时效性的记忆检索
参数：
- query: 查询文本
- vectorstore: 向量存储
- k: 返回数量
- recency_weight: 时效性权重（0-1）
"""
from datetime import datetime
results = vectorstore.similarity_search_with_score(query, k=k*2)
scored_results = []
for doc, similarity_score in results:
timestamp = doc.metadata.get("timestamp")
if timestamp:
age_days = (datetime.now() - datetime.fromisoformat(timestamp)).days
recency_score = 1 / (1 + age_days * 0.1)
else:
recency_score = 0.5
final_score = (1 - recency_weight) * (1 - similarity_score) + recency_weight * recency_score
scored_results.append((doc, final_score))
scored_results.sort(key=lambda x: x[1], reverse=True)
return [doc for doc, _ in scored_results[:k]]
基于实体关联的检索
import re
from collections import defaultdict
class EntityMemory:
"""基于实体的记忆管理"""
def __init__(self):
self.memories: List[Dict] = []
self.entity_index: Dict[str, List[int]] = defaultdict(list)
def extract_entities(self, text: str) -> List[str]:
"""提取文本中的实体（简化版）"""
patterns = [
r'[\u4e00-\u9fa5]{2,4}',
]
entities = []
for pattern in patterns:
entities.extend(re.findall(pattern, text))
return list(set(entities))
def add_memory(self, content: str, metadata: Dict = None):
"""添加记忆"""
memory_id = len(self.memories)
entities = self.extract_entities(content)
memory = {
"id": memory_id,
"content": content,
"metadata": metadata or {},
"entities": entities
}
self.memories.append(memory)
for entity in entities:
self.entity_index[entity].append(memory_id)
def search_by_entity(self, entity: str) -> List[Dict]:
"""根据实体搜索记忆"""
memory_ids = self.entity_index.get(entity, [])
return [self.memories[mid] for mid in memory_ids]
def search_by_entities(self, entities: List[str]) -> List[Dict]:
"""根据多个实体搜索记忆"""
memory_id_sets = [
set(self.entity_index.get(entity, []))
for entity in entities
]
if not memory_id_sets:
return []
common_ids = set.intersection(*memory_id_sets)
return [self.memories[mid] for mid in common_ids]
entity_memory = EntityMemory()
entity_memory.add_memory("张三是一名软件工程师，他在北京工作")
entity_memory.add_memory("张三喜欢打篮球，每周都会去打球")
entity_memory.add_memory("李四是一名设计师，她和张三是同事")
print("关于张三的记忆：")
for m in entity_memory.search_by_entity("张三"):
print(f" - {m['content']}")
print("\n关于张三和篮球的记忆：")
for m in entity_memory.search_by_entities(["张三", "篮球"]):
print(f" - {m['content']}")
最佳实践
记忆分层
将记忆分为不同层次，根据重要性选择存储方式：
class LayeredMemory:
"""分层记忆系统"""
def __init__(self):
self.working_memory = []
self.session_memory = {}
self.long_term_memory = None
def add_to_working(self, message: Dict):
"""添加到工作记忆（当前对话）"""
self.working_memory.append(message)
def save_to_session(self, session_id: str):
"""保存到会话记忆"""
self.session_memory[session_id] = self.working_memory.copy()
def archive_to_long_term(self, session_id: str, important_facts: List[str]):
"""归档到长期记忆"""
for fact in important_facts:
self.long_term_memory.save_memory(fact, {"session_id": session_id})
隐私保护
在存储记忆前进行脱敏处理：
import re
def sanitize_for_memory(text: 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:
text = re.sub(pattern, replacement, text)
return text
小结
记忆系统是 AI Agent 保持上下文连贯性的关键：
- 短期记忆用于当前对话，长期记忆用于跨会话持久化
- LangChain 提供了多种开箱即用的记忆组件
- 自定义记忆系统可以实现更灵活的功能
- 记忆压缩和总结可以处理长对话
- 注意隐私保护，对敏感信息进行脱敏
下一章我们将学习如何使用 LangChain 构建 Agent。