记忆持久化的核心挑战
Agent的会话记忆在对话结束后就消失了。要让Agent"记住"用户偏好、历史交互和学到的知识，需要将记忆持久化到外部存储。如何存储、如何检索、如何遗忘——这是记忆持久化的三大问题。
存储方案对比
方案一：向量数据库
最主流的记忆存储方案：
class VectorMemory:
def __init__(self, vector_db):
self.db = vector_db
async def store(self, content, metadata=None):
embedding = await self.embed(content)
self.db.insert({
"content": content,
"embedding": embedding,
"metadata": metadata,
"timestamp": datetime.now()
})
async def retrieve(self, query, top_k=5):
query_vec = await self.embed(query)
results = self.db.search(query_vec, top_k=top_k)
return results
优点： 语义检索、灵活查询 缺点： 缺乏结构化关系、时间感知弱
方案二：关系数据库+摘要
class RelationalMemory:
def __init__(self, db):
self.db = db
async def store_interaction(self, user_id, session_data):
# 存储完整交互
self.db.insert("interactions", {
"user_id": user_id,
"session_id": session_data.id,
"summary": await self.summarize(session_data),
"key_points": json.dumps(session_data.key_points),
"timestamp": session_data.start_time
})
async def get_user_context(self, user_id):
"""获取用户上下文"""
interactions = self.db.query(
"SELECT * FROM interactions WHERE user_id = ? ORDER BY timestamp DESC LIMIT 10",
[user_id]
)
return self.build_context(interactions)
优点： 结构化、可精确查询 缺点： 无法语义检索
方案三：知识图谱
class GraphMemory:
def __init__(self, graph_db):
self.db = graph_db
async def store_fact(self, subject, predicate, obj):
"""存储三元组"""
self.db.run(
"MERGE (s:Entity {name: $subject}) "
"MERGE (o:Entity {name: $object}) "
"MERGE (s)-[r:RELATION {type: $predicate}]->(o)",
subject=subject, predicate=predicate, object=obj
)
async def query_relations(self, entity):
"""查询实体关系"""
return self.db.run(
"MATCH (e:Entity {name: $entity})-[r]->(related) "
"RETURN e, r, related",
entity=entity
)
优点： 关系推理能力强、结构化 缺点： 构建成本高、灵活性低
方案四：混合方案（推荐）
class HybridMemory:
def __init__(self):
self.vector_db = VectorDB() # 语义检索
self.relational_db = PostgresDB() # 结构化查询
self.graph_db = Neo4jDB() # 关系推理
self.cache = Redis() # 热数据缓存
async def store(self, interaction):
# 1. 完整交互存入关系数据库
await self.relational_db.store(interaction)
# 2. 摘要存入向量数据库
summary = await self.summarize(interaction)
await self.vector_db.store(summary, metadata={
"user_id": interaction.user_id,
"timestamp": interaction.timestamp
})
# 3. 提取的知识存入图谱
facts = await self.extract_facts(interaction)
for fact in facts:
await self.graph_db.store_fact(*fact)
记忆类型与存储策略
用户偏好记忆
存储: "用户喜欢简洁的回答" → 存入用户配置
检索: 每次生成前读取用户偏好
更新: 显式更新（用户说"我想要更详细"）
持久性: 永久
交互历史记忆
存储: 每次对话的摘要
检索: 基于当前话题语义检索相关历史
更新: 追加
持久性: 30天完整，之后仅保留摘要
知识记忆
存储: 从交互中提取的事实和规则
检索: 基于实体和关系查询
更新: 增量更新，可修正
持久性: 永久（除非被更正）
技能记忆
存储: 学会的工具调用模式
检索: 遇到类似任务时调用
更新: 新的成功案例
持久性: 永久
遗忘机制
基于时间的遗忘
def time_based_decay(memory_item):
age_days = (datetime.now() - memory_item.timestamp).days
decay_factor = math.exp(-age_days / 30) # 30天半衰期
memory_item.weight *= decay_factor
if memory_item.weight < 0.01:
delete(memory_item) # 权重过低则删除
基于使用的遗忘
def usage_based_cleanup(memories):
for m in memories:
if m.last_accessed < datetime.now() - timedelta(days=90):
if m.access_count < 3: # 很少被检索
archive_or_delete(m)
基于质量的遗忘
def quality_based_filtering(memories):
for m in memories:
if m.feedback_score < 0: # 被标记为错误
delete(m)
elif m.feedback_score < 0.3: # 低质量
degrade(m) # 降级为摘要
合并压缩
async def merge_similar_memories(self):
"""合并相似记忆条目"""
all_memories = await self.get_all()
# 聚类
clusters = cluster(all_memories, threshold=0.85)
for cluster in clusters:
if len(cluster) > 1:
# 生成合并摘要
merged = await self.llm.summarize(cluster)
# 删除原条目，存入合并后的摘要
for m in cluster:
await self.delete(m)
await self.store(merged)
检索策略
多路召回
async def retrieve_memories(self, query, user_id):
# 1. 向量检索（语义相关）
semantic_results = await self.vector_db.search(
embed(query),
filter={"user_id": user_id},
top_k=10
)
# 2. 图谱检索（实体相关）
entities = extract_entities(query)
graph_results = await self.graph_db.query_relations(entities)
# 3. 时间检索（最近交互）
recent_results = await self.relational_db.query_recent(
user_id, limit=5
)
# 4. 融合排序
all_results = semantic_results + graph_results + recent_results
return self.rerank(all_results, query)
上下文构建
async def build_context(self, query, user_id):
memories = await self.retrieve_memories(query, user_id)
context = "## 相关记忆\n"
# 用户偏好
context += f"用户偏好: {memories.preferences}\n"
# 相关历史
for m in memories.history[:3]:
context += f"之前讨论过: {m.summary}\n"
# 相关知识
for fact in memories.facts[:5]:
context += f"已知: {fact}\n"
return context
隐私与安全
数据最小化
只存储必要的信息：
- 不存储完整的原始对话
- 只存摘要和关键信息
- 敏感信息脱敏后存储
访问控制
class MemoryAccessControl:
def check_access(self, user_id, memory):
# 只能访问自己的记忆
if memory.user_id != user_id:
return False
# 敏感记忆需额外验证
if memory.sensitivity == "high":
return self.verify_identity(user_id)
return True
记忆删除权
用户有权要求删除所有记忆：
async def delete_all_user_data(self, user_id):
await self.vector_db.delete_by_user(user_id)
await self.relational_db.delete_by_user(user_id)
await self.graph_db.delete_by_user(user_id)
await self.cache.delete_by_user(user_id)
# 符合GDPR"被遗忘权"
总结
Agent记忆持久化是"记住过去以改善未来"的技术。混合存储方案——向量数据库做语义检索、关系数据库做结构化存储、知识图谱做关系推理——是当前最全面的方案。但记忆管理的核心不是"存什么"，而是"忘什么"——好的遗忘机制让记忆保持高效和有用。记忆系统是Agent"人格连续性"的基础，没有持久化记忆的Agent，每次对话都是从零开始。