feat:实现短期内存管理器和统一内存管理器
- 添加了ShortTermMemoryManager来管理短期记忆，包括提取、决策和记忆操作。 - 集成大型语言模型（LLM），用于结构化记忆提取和决策过程。 - 基于重要性阈值，实现了从短期到长期的内存转移逻辑。 - 创建了UnifiedMemoryManager，通过统一接口整合感知记忆、短期记忆和长期记忆的管理。 - 通过法官模型评估来增强记忆提取过程的充分性。 - 增加了自动和手动内存传输功能。 - 包含内存管理操作和决策的全面日志记录。
This commit is contained in:
Windpicker-owo
292
scripts/test_three_tier_memory.py
Normal file
292
scripts/test_three_tier_memory.py
Normal file
@@ -0,0 +1,292 @@
|
||||
"""
|
||||
三层记忆系统测试脚本
|
||||
用于验证系统各组件是否正常工作
|
||||
"""
|
||||
|
||||
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)
|
||||
Reference in New Issue
Block a user