b5cfa41d36
- 添加了ShortTermMemoryManager来管理短期记忆，包括提取、决策和记忆操作。 - 集成大型语言模型（LLM），用于结构化记忆提取和决策过程。 - 基于重要性阈值，实现了从短期到长期的内存转移逻辑。 - 创建了UnifiedMemoryManager，通过统一接口整合感知记忆、短期记忆和长期记忆的管理。 - 通过法官模型评估来增强记忆提取过程的充分性。 - 增加了自动和手动内存传输功能。 - 包含内存管理操作和决策的全面日志记录。
293 lines
8.2 KiB
Python
293 lines
8.2 KiB
Python
"""
|
||
三层记忆系统测试脚本
|
||
用于验证系统各组件是否正常工作
|
||
"""
|
||
|
||
import asyncio
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
# 添加项目根目录到路径
|
||
project_root = Path(__file__).parent.parent
|
||
sys.path.insert(0, str(project_root))
|
||
|
||
|
||
async def test_perceptual_memory():
|
||
"""测试感知记忆层"""
|
||
print("\n" + "=" * 60)
|
||
print("测试1: 感知记忆层")
|
||
print("=" * 60)
|
||
|
||
from src.memory_graph.three_tier.perceptual_manager import get_perceptual_manager
|
||
|
||
manager = get_perceptual_manager()
|
||
await manager.initialize()
|
||
|
||
# 添加测试消息
|
||
test_messages = [
|
||
("user1", "今天天气真好", 1700000000.0),
|
||
("user2", "是啊，适合出去玩", 1700000001.0),
|
||
("user1", "我们去公园吧", 1700000002.0),
|
||
("user2", "好主意！", 1700000003.0),
|
||
("user1", "带上野餐垫", 1700000004.0),
|
||
]
|
||
|
||
for sender, content, timestamp in test_messages:
|
||
message = {
|
||
"message_id": f"msg_{timestamp}",
|
||
"sender": sender,
|
||
"content": content,
|
||
"timestamp": timestamp,
|
||
"platform": "test",
|
||
"stream_id": "test_stream",
|
||
}
|
||
await manager.add_message(message)
|
||
|
||
print(f"✅ 成功添加 {len(test_messages)} 条消息")
|
||
|
||
# 测试TopK召回
|
||
results = await manager.recall_blocks("公园野餐", top_k=2)
|
||
print(f"✅ TopK召回返回 {len(results)} 个块")
|
||
|
||
if results:
|
||
print(f" 第一个块包含 {len(results[0].messages)} 条消息")
|
||
|
||
# 获取统计信息
|
||
stats = manager.get_statistics() # 不是async方法
|
||
print(f"✅ 统计信息: {stats}")
|
||
|
||
return True
|
||
|
||
|
||
async def test_short_term_memory():
|
||
"""测试短期记忆层"""
|
||
print("\n" + "=" * 60)
|
||
print("测试2: 短期记忆层")
|
||
print("=" * 60)
|
||
|
||
from src.memory_graph.three_tier.models import MemoryBlock
|
||
from src.memory_graph.three_tier.short_term_manager import get_short_term_manager
|
||
|
||
manager = get_short_term_manager()
|
||
await manager.initialize()
|
||
|
||
# 创建测试块
|
||
test_block = MemoryBlock(
|
||
id="test_block_1",
|
||
messages=[
|
||
{
|
||
"message_id": "msg1",
|
||
"sender": "user1",
|
||
"content": "我明天要参加一个重要的面试",
|
||
"timestamp": 1700000000.0,
|
||
"platform": "test",
|
||
}
|
||
],
|
||
combined_text="我明天要参加一个重要的面试",
|
||
recall_count=3,
|
||
)
|
||
|
||
# 从感知块转换为短期记忆
|
||
try:
|
||
await manager.add_from_block(test_block)
|
||
print("✅ 成功将感知块转换为短期记忆")
|
||
except Exception as e:
|
||
print(f"⚠️ 转换失败（可能需要LLM）: {e}")
|
||
return False
|
||
|
||
# 测试搜索
|
||
results = await manager.search_memories("面试", top_k=3)
|
||
print(f"✅ 搜索返回 {len(results)} 条记忆")
|
||
|
||
# 获取统计
|
||
stats = manager.get_statistics()
|
||
print(f"✅ 统计信息: {stats}")
|
||
|
||
return True
|
||
|
||
|
||
async def test_long_term_memory():
|
||
"""测试长期记忆层"""
|
||
print("\n" + "=" * 60)
|
||
print("测试3: 长期记忆层")
|
||
print("=" * 60)
|
||
|
||
from src.memory_graph.three_tier.long_term_manager import get_long_term_manager
|
||
|
||
manager = get_long_term_manager()
|
||
await manager.initialize()
|
||
|
||
print("✅ 长期记忆管理器初始化成功")
|
||
print(" （需要现有记忆图系统支持）")
|
||
|
||
# 获取统计
|
||
stats = manager.get_statistics()
|
||
print(f"✅ 统计信息: {stats}")
|
||
|
||
return True
|
||
|
||
|
||
async def test_unified_manager():
|
||
"""测试统一管理器"""
|
||
print("\n" + "=" * 60)
|
||
print("测试4: 统一管理器")
|
||
print("=" * 60)
|
||
|
||
from src.memory_graph.three_tier.unified_manager import UnifiedMemoryManager
|
||
|
||
manager = UnifiedMemoryManager()
|
||
await manager.initialize()
|
||
|
||
# 添加测试消息
|
||
message = {
|
||
"message_id": "unified_test_1",
|
||
"sender": "user1",
|
||
"content": "这是一条测试消息",
|
||
"timestamp": 1700000000.0,
|
||
"platform": "test",
|
||
"stream_id": "test_stream",
|
||
}
|
||
await manager.add_message(message)
|
||
|
||
print("✅ 通过统一接口添加消息成功")
|
||
|
||
# 测试搜索
|
||
results = await manager.search_memories("测试")
|
||
print(f"✅ 统一搜索返回结果:")
|
||
print(f" 感知块: {len(results.get('perceptual_blocks', []))}")
|
||
print(f" 短期记忆: {len(results.get('short_term_memories', []))}")
|
||
print(f" 长期记忆: {len(results.get('long_term_memories', []))}")
|
||
|
||
# 获取统计
|
||
stats = manager.get_statistics() # 不是async方法
|
||
print(f"✅ 综合统计:")
|
||
print(f" 感知层: {stats.get('perceptual', {})}")
|
||
print(f" 短期层: {stats.get('short_term', {})}")
|
||
print(f" 长期层: {stats.get('long_term', {})}")
|
||
|
||
return True
|
||
|
||
|
||
async def test_configuration():
|
||
"""测试配置加载"""
|
||
print("\n" + "=" * 60)
|
||
print("测试5: 配置系统")
|
||
print("=" * 60)
|
||
|
||
from src.config.config import global_config
|
||
|
||
if not hasattr(global_config, "three_tier_memory"):
|
||
print("❌ 配置类中未找到 three_tier_memory 字段")
|
||
return False
|
||
|
||
config = global_config.three_tier_memory
|
||
|
||
if config is None:
|
||
print("⚠️ 三层记忆配置为 None（可能未在 bot_config.toml 中配置）")
|
||
print(" 请在 bot_config.toml 中添加 [three_tier_memory] 配置")
|
||
return False
|
||
|
||
print(f"✅ 配置加载成功")
|
||
print(f" 启用状态: {config.enable}")
|
||
print(f" 数据目录: {config.data_dir}")
|
||
print(f" 感知层最大块数: {config.perceptual_max_blocks}")
|
||
print(f" 短期层最大记忆数: {config.short_term_max_memories}")
|
||
print(f" 激活阈值: {config.activation_threshold}")
|
||
|
||
return True
|
||
|
||
|
||
async def test_integration():
|
||
"""测试系统集成"""
|
||
print("\n" + "=" * 60)
|
||
print("测试6: 系统集成")
|
||
print("=" * 60)
|
||
|
||
# 首先需要确保配置启用
|
||
from src.config.config import global_config
|
||
|
||
if not global_config.three_tier_memory or not global_config.three_tier_memory.enable:
|
||
print("⚠️ 配置未启用，跳过集成测试")
|
||
return False
|
||
|
||
# 测试单例模式
|
||
from src.memory_graph.three_tier.manager_singleton import (
|
||
get_unified_memory_manager,
|
||
initialize_unified_memory_manager,
|
||
)
|
||
|
||
# 初始化
|
||
await initialize_unified_memory_manager()
|
||
manager = get_unified_memory_manager()
|
||
|
||
if manager is None:
|
||
print("❌ 统一管理器初始化失败")
|
||
return False
|
||
|
||
print("✅ 单例模式正常工作")
|
||
|
||
# 测试多次获取
|
||
manager2 = get_unified_memory_manager()
|
||
if manager is not manager2:
|
||
print("❌ 单例模式失败（返回不同实例）")
|
||
return False
|
||
|
||
print("✅ 单例一致性验证通过")
|
||
|
||
return True
|
||
|
||
|
||
async def run_all_tests():
|
||
"""运行所有测试"""
|
||
print("\n" + "🔬" * 30)
|
||
print("三层记忆系统集成测试")
|
||
print("🔬" * 30)
|
||
|
||
tests = [
|
||
("配置系统", test_configuration),
|
||
("感知记忆层", test_perceptual_memory),
|
||
("短期记忆层", test_short_term_memory),
|
||
("长期记忆层", test_long_term_memory),
|
||
("统一管理器", test_unified_manager),
|
||
("系统集成", test_integration),
|
||
]
|
||
|
||
results = []
|
||
|
||
for name, test_func in tests:
|
||
try:
|
||
result = await test_func()
|
||
results.append((name, result))
|
||
except Exception as e:
|
||
print(f"\n❌ 测试 {name} 失败: {e}")
|
||
import traceback
|
||
|
||
traceback.print_exc()
|
||
results.append((name, False))
|
||
|
||
# 打印测试总结
|
||
print("\n" + "=" * 60)
|
||
print("测试总结")
|
||
print("=" * 60)
|
||
|
||
passed = sum(1 for _, result in results if result)
|
||
total = len(results)
|
||
|
||
for name, result in results:
|
||
status = "✅ 通过" if result else "❌ 失败"
|
||
print(f"{status} - {name}")
|
||
|
||
print(f"\n总计: {passed}/{total} 测试通过")
|
||
|
||
if passed == total:
|
||
print("\n🎉 所有测试通过！三层记忆系统工作正常。")
|
||
else:
|
||
print("\n⚠️ 部分测试失败，请查看上方详细信息。")
|
||
|
||
return passed == total
|
||
|
||
|
||
if __name__ == "__main__":
|
||
success = asyncio.run(run_all_tests())
|
||
sys.exit(0 if success else 1)
|