增强代理长期记忆的多重记忆系统

增强代理长期记忆的多重记忆系统

ArXiv ID: 2508.15294
作者: Gaoke Zhang, Bo Wang, Yunlong Ma, Dongming Zhao, Zifei Yu
机构: Shandong University, Alibaba Group
发布日期: 2025-08-21


摘要

由大型语言模型驱动的代理取得了令人印象深刻的成果,但有效处理交互过程中产生的大量历史数据仍然是一个挑战。现有方法如 MemoryBank 和 A-MEM 存在记忆内容质量差的问题,影响了检索性能和响应质量。本文提出 MMS(Multiple Memory Systems) 架构,灵感来自人类认知科学的多重记忆理论。系统包含三个独立但协同工作的记忆子系统:程序性记忆语义记忆情节记忆。实验表明,MMS 相比单一记忆系统在响应相关性上提升35%,事实准确性提升40%


问题背景

单一记忆系统的局限

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
当前代理记忆系统的问题:

单一记忆池架构:
┌─────────────────────────────────────────┐
│ 记忆存储 (扁平列表) │
│ ├─ "用户喜欢咖啡" │
│ ├─ "项目截止日期是周五" │
│ ├─ "如何写 Python 函数" │
│ ├─ "昨天讨论了 API 设计" │
│ └─ "2+2=4" │
│ │
│ 问题: │
│ • 不同类型信息混在一起 │
│ • 检索时难以精确定位 │
│ • 无法针对性优化 │
│ • 记忆质量参差不齐 │
└─────────────────────────────────────────┘

核心问题

  • 技能知识、事实信息、个人经历混为一谈
  • 使用相同的存储和检索策略
  • 无法模拟人类的记忆组织方式

人类记忆系统的启示

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
人类多重记忆系统:

┌─────────────────────────────────────────────────────────┐
│ Human Memory Systems │
│ │
│ ┌─────────────────┐ │
│ │ Procedural │ "如何做" (How-to) │
│ │ 程序性记忆 │ 技能、习惯、操作流程 │
│ └─────────────────┘ │
│ │
│ ┌─────────────────┐ │
│ │ Semantic │ "是什么" (What-is) │
│ │ 语义记忆 │ 事实、概念、一般知识 │
│ └─────────────────┘ │
│ │
│ ┌─────────────────┐ │
│ │ Episodic │ "何时何地" (When-Where) │
│ │ 情节记忆 │ 个人经历、具体事件 │
│ └─────────────────┘ │
│ │
│ 特点: │
│ • 分离存储,协同工作 │
│ • 针对不同类型优化 │
│ • 相互验证和增强 │
└─────────────────────────────────────────────────────────┘

MMS 架构

整体设计

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
┌─────────────────────────────────────────────────────────┐
│ Multiple Memory Systems │
│ (MMS Framework) │
│ │
│ 用户输入 │
│ │ │
│ ▼ │
│ ┌─────────────────┐ │
│ │ 记忆分类器 │ │
│ │ Memory │ │
│ │ Classifier │ │
│ └─────────────────┘ │
│ │ │
│ ├──────────────┬──────────────┐ │
│ ▼ ▼ ▼ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │程序性记忆│ │语义记忆 │ │情节记忆 │ │
│ │Procedural│ │Semantic │ │Episodic │ │
│ │ │ │ │ │ │ │
│ │ 技能库 │ │ 知识库 │ │ 经历库 │ │
│ │ 操作流程 │ │ 事实概念│ │ 时间线 │ │
│ └─────────┘ └─────────┘ └─────────┘ │
│ │ │ │ │
│ └──────────────┼──────────────┘ │
│ ▼ │
│ ┌─────────────────┐ │
│ │ 记忆融合模块 │ │
│ │ Memory Fusion │ │
│ │ & Verification │ │
│ └─────────────────┘ │
│ │ │
│ ▼ │
│ 响应生成 │
└─────────────────────────────────────────────────────────┘

组件 1:程序性记忆

存储技能和操作流程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
class ProceduralMemory:
"""程序性记忆:存储"如何做"的知识"""

def __init__(self):
self.skills = {} # 技能库
self.success_rates = {} # 成功率统计

def store_skill(self, skill_name: str, procedure: dict):
"""
存储技能

procedure 格式:
{
"steps": ["步骤 1", "步骤 2", ...],
"conditions": {"前置条件": ...},
"expected_outcome": "预期结果"
}
"""
self.skills[skill_name] = {
'procedure': procedure,
'usage_count': 0,
'success_count': 0,
'last_used': time.time()
}

def retrieve_skill(self, task_description: str) -> dict:
"""检索相关技能"""
# 基于任务描述匹配技能
matched_skills = []

for skill_name, skill_data in self.skills.items():
similarity = self._compute_similarity(
task_description, skill_name
)

if similarity > 0.7: # 阈值
# 考虑成功率
success_rate = (
skill_data['success_count'] /
max(1, skill_data['usage_count'])
)
matched_skills.append((skill_name, skill_data, similarity * success_rate))

# 按综合分数排序
matched_skills.sort(key=lambda x: x[2], reverse=True)

return matched_skills[0] if matched_skills else None

def update_success(self, skill_name: str, success: bool):
"""更新技能成功率"""
if skill_name in self.skills:
self.skills[skill_name]['usage_count'] += 1
if success:
self.skills[skill_name]['success_count'] += 1

示例技能

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 存储示例
memory.store_skill(
"写 Python 函数",
{
"steps": [
"1. 定义函数名和参数",
"2. 编写 docstring 说明功能",
"3. 实现函数体逻辑",
"4. 添加错误处理",
"5. 编写测试用例"
],
"best_practices": [
"使用有意义的变量名",
"保持函数单一职责",
"添加类型注解"
]
}
)

组件 2:语义记忆

知识图谱组织

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
import networkx as nx

class SemanticMemory:
"""语义记忆:存储事实性知识"""

def __init__(self):
self.knowledge_graph = nx.Graph()

def store_fact(self, subject: str, predicate: str, object_: str):
"""
存储事实

示例:
store_fact("Python", "是", "编程语言")
store_fact("咖啡", "含有", "咖啡因")
"""
# 添加节点
if subject not in self.knowledge_graph:
self.knowledge_graph.add_node(subject)
if object_ not in self.knowledge_graph:
self.knowledge_graph.add_node(object_)

# 添加关系边
self.knowledge_graph.add_edge(subject, object_, relation=predicate)

def query(self, subject: str = None, predicate: str = None, object_: str = None):
"""
查询知识

支持部分匹配,返回所有匹配的事实
"""
results = []

for edge in self.knowledge_graph.edges(data=True):
s, o, data = edge[0], edge[1], edge[2]

match = True
if subject and s != subject and subject not in s:
match = False
if predicate and data.get('relation') != predicate:
match = False
if object_ and o != object_ and object_ not in o:
match = False

if match:
results.append((s, data.get('relation'), o))

return results

def get_related_concepts(self, concept: str, depth=2):
"""获取相关概念(BFS 遍历)"""
visited = set()
queue = [(concept, 0)]
related = []

while queue:
current, current_depth = queue.pop(0)

if current in visited or current_depth > depth:
continue

visited.add(current)
related.append(current)

# 获取邻居节点
for neighbor in self.knowledge_graph.neighbors(current):
if neighbor not in visited:
queue.append((neighbor, current_depth + 1))

return related

组件 3:情节记忆

时序索引的经历存储

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
class EpisodicMemory:
"""情节记忆:存储具体经历和事件"""

def __init__(self):
self.episodes = [] # 情节列表
self.time_index = {} # 时间索引

def store_episode(self, content: str, timestamp: float, context: dict = None):
"""
存储情节

Args:
content: 情节内容
timestamp: 时间戳
context: 上下文信息(地点、参与者、情绪等)
"""
episode = {
'id': len(self.episodes),
'content': content,
'timestamp': timestamp,
'context': context or {},
'embedding': self._embed(content),
'access_count': 0,
'last_accessed': None
}

self.episodes.append(episode)

# 建立时间索引(按小时、天、月)
hour_key = int(timestamp // 3600)
day_key = int(timestamp // 86400)

if hour_key not in self.time_index:
self.time_index[hour_key] = []
self.time_index[hour_key].append(episode['id'])

return episode['id']

def retrieve_by_time(self, start_time: float, end_time: float) -> list:
"""按时间范围检索情节"""
results = []

for episode in self.episodes:
if start_time <= episode['timestamp'] <= end_time:
results.append(episode)

return sorted(results, key=lambda x: x['timestamp'])

def retrieve_by_similarity(self, query: str, top_k: int = 5) -> list:
"""基于语义相似度检索"""
query_embedding = self._embed(query)

scores = []
for episode in self.episodes:
similarity = cosine_similarity(query_embedding, episode['embedding'])

# 考虑近因效应(recent episodes get bonus)
recency_bonus = self._compute_recency_bonus(episode['timestamp'])
adjusted_score = similarity * (1 + recency_bonus)

scores.append((episode, adjusted_score))

scores.sort(key=lambda x: x[1], reverse=True)
return [(ep, score) for ep, score in scores[:top_k]]

def _compute_recency_bonus(self, timestamp: float) -> float:
"""计算近因奖励(越近的事件奖励越高)"""
elapsed = time.time() - timestamp
hours_elapsed = elapsed / 3600

# 指数衰减:24 小时后减半
return 0.5 ** (hours_elapsed / 24)

记忆融合与验证

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
class MemoryFusion:
"""融合和验证来自不同记忆系统的信息"""

def __init__(self, procedural, semantic, episodic):
self.procedural = procedural
self.semantic = semantic
self.episodic = episodic

def retrieve_comprehensive(self, query: str, context: dict = None) -> dict:
"""
综合检索所有记忆系统

Returns:
包含三个子系统检索结果的字典
"""
results = {}

# 1. 程序性记忆检索(技能)
skill_result = self.procedural.retrieve_skill(query)
results['procedural'] = skill_result

# 2. 语义记忆检索(事实)
# 从查询中提取实体进行查询
entities = self._extract_entities(query)
semantic_results = []
for entity in entities:
facts = self.semantic.query(subject=entity)
semantic_results.extend(facts)
results['semantic'] = semantic_results

# 3. 情节记忆检索(经历)
episode_results = self.episodic.retrieve_by_similarity(query)
results['episodic'] = episode_results

# 4. 交叉验证
verified_results = self._cross_verify(results)

return verified_results

def _cross_verify(self, results: dict) -> dict:
"""
交叉验证:不同记忆系统的信息相互验证

例如:
- 情节记忆中的技能使用记录验证程序性记忆
- 语义记忆中的事实验证情节记忆的准确性
"""
verified = {}

# 验证程序性记忆
if results['procedural']:
skill_name = results['procedural'][0]
# 检查情节记忆中是否有成功使用该技能的记录
related_episodes = self.episodic.retrieve_by_similarity(skill_name)
success_evidence = [
ep for ep, _ in related_episodes
if ep['context'].get('outcome') == 'success'
]
verified['procedural_confidence'] = (
len(success_evidence) / max(1, len(related_episodes))
)

# 验证语义记忆
# ...

return verified

实验结果

实验设置

任务场景

  • 客户服务对话(500+ 轮)
  • 项目管理协作(30 天 +)
  • 个人助理(长期交互)

对比方法

  • MemoryBank
  • A-MEM
  • 单一记忆池
  • RAG 基线

评估指标

  • 响应相关性(1-5 分)
  • 事实准确性(%)
  • 技能应用成功率(%)
  • 用户满意度(1-5 分)

主要结果

响应质量

方法 相关性 准确性 一致性
MemoryBank 3.2 68% 72%
A-MEM 3.4 71% 75%
单一记忆池 3.1 65% 70%
MMS 4.5 88% 91%

提升:相比最佳基线 +35% 相关性,+40% 准确性

技能应用成功率

任务类型 MemoryBank A-MEM MMS
代码写作 62% 65% 82%
文档撰写 58% 61% 78%
数据分析 55% 58% 75%
问题诊断 52% 55% 72%

长时记忆保持

1
2
3
4
5
6
7
8
9
10
11
12
记忆保持率 vs 时间:

保持率

│ MMS: ●─────────●

│ A-MEM: ╲
│ ●
│ MemoryBank: ╲
│ ●
└─────────────────────────
1 天 7 天 14 天 30 天

消融实验

各记忆系统贡献

配置 相关性 准确性
MMS (完整) 4.5 88%
- 程序性记忆 4.1 85%
- 语义记忆 4.0 78%
- 情节记忆 4.2 82%
单一记忆池 3.1 65%

总结

MMS 通过模拟人类多重记忆系统,实现了更高效的代理记忆管理:

核心贡献

  1. 程序性、语义、情节三系统分离设计
  2. 针对性存储和检索策略
  3. 交叉验证提高记忆质量

实际价值

  • 35% 响应相关性提升
  • 40% 事实准确性提升
  • 适用于长期交互代理

资源


评分: 4.4/5.0 ⭐⭐⭐⭐

推荐度: 推荐。长期代理记忆管理的创新架构。

© 2026 Generative AI Discovery All Rights Reserved.
Theme by hiero