test_topic_drift_integration.py•6.27 kB
#!/usr/bin/env python3
"""
主题漂移检测集成测试脚本
测试主题漂移检测器与MVP记忆管理器的集成效果
"""
import os
import sys
import time
from pathlib import Path
# 添加项目根目录到Python路径
sys.path.insert(0, str(Path(__file__).parent))
def test_topic_drift_integration():
"""测试主题漂移检测与MVP管理器的集成"""
print("🧪 主题漂移检测集成测试")
print("=" * 50)
try:
from mvp_memory import MVPMemoryManager
# 初始化测试数据目录
test_data_dir = "./test_topic_drift_integration_data"
if os.path.exists(test_data_dir):
import shutil
shutil.rmtree(test_data_dir)
# 创建MVP管理器
mvp_manager = MVPMemoryManager(test_data_dir, use_enhanced=True)
print("\n📝 添加测试记忆（不同主题）...")
# 添加不同主题的测试记忆
test_memories = [
{
"content": "Python是一种高级编程语言，语法简洁易学",
"tags": ["编程", "Python"]
},
{
"content": "机器学习是人工智能的一个重要分支",
"tags": ["AI", "机器学习"]
},
{
"content": "深度学习使用神经网络进行模式识别",
"tags": ["AI", "深度学习"]
},
{
"content": "今天天气很好，适合出门散步",
"tags": ["天气", "生活"]
},
{
"content": "晚餐吃了美味的意大利面",
"tags": ["美食", "生活"]
}
]
# 添加记忆
for i, memory in enumerate(test_memories, 1):
success = mvp_manager.remember(
memory["content"],
tags=memory["tags"]
)
if success:
print(f"✅ 已添加记忆 {i}: {memory['content'][:30]}...")
else:
print(f"❌ 添加失败: {memory['content'][:30]}...")
print(f"\n📊 总共添加了 {len(test_memories)} 条测试记忆")
# 等待一下确保记忆被索引
time.sleep(2)
print("\n🔍 测试主题漂移检测...")
print("-" * 40)
# 测试查询序列 - 模拟主题漂移场景
test_queries = [
"Python编程语言特点", # 编程主题
"Python函数定义方法", # 继续编程主题
"机器学习算法原理", # 主题开始漂移到AI
"深度学习神经网络", # 继续AI主题
"今天天气如何", # 主题漂移到生活
"晚餐吃什么好", # 继续生活主题
"Python web开发框架", # 主题回到编程
]
print("查询序列:")
for i, query in enumerate(test_queries, 1):
print(f" {i}. {query}")
print("\n🔄 执行查询并观察主题漂移检测:")
print("-" * 50)
for i, query in enumerate(test_queries, 1):
print(f"\n{i}. 查询: '{query}'")
# 启用主题漂移检测
results_with_drift = mvp_manager.recall(
query,
top_k=3,
use_topic_drift_detection=True
)
print(f" 检索到 {len(results_with_drift)} 条结果")
# 显示前2条结果的简要信息
for j, result in enumerate(results_with_drift[:2], 1):
content = result.get('content', '')[:40]
score = result.get('score', 0)
print(f" {j}. {content}... (分数: {score:.3f})")
print("\n📈 主题漂移检测统计:")
if mvp_manager.topic_drift_detector:
stats = mvp_manager.topic_drift_detector.get_statistics()
for key, value in stats.items():
print(f" {key}: {value}")
else:
print(" 主题漂移检测器未初始化")
print("\n✅ 主题漂移检测集成测试完成!")
return True
except Exception as e:
print(f"❌ 测试失败: {e}")
import traceback
traceback.print_exc()
return False
def test_topic_drift_detector_standalone():
"""测试独立的主题漂移检测器"""
print("\n🔧 独立主题漂移检测器测试")
print("-" * 30)
try:
from topic_drift_detector import TopicDriftDetector
detector = TopicDriftDetector(window_size=3, drift_threshold=0.5, min_similarity=0.3)
# 测试查询序列
test_queries = [
"Python编程基础",
"Python函数使用",
"机器学习算法", # 主题漂移
"深度学习网络",
"今天天气很好", # 主题漂移
"晚餐计划安排",
"Python web开发" # 主题回归
]
print("测试查询序列:")
for i, query in enumerate(test_queries, 1):
should_clear, reason = detector.should_clear_candidates(query)
status = "🚨 清空候选集" if should_clear else "✅ 保持候选集"
print(f" {i}. {query}")
print(f" {status}: {reason}")
print("\n📊 检测统计:")
stats = detector.get_statistics()
for key, value in stats.items():
print(f" {key}: {value}")
return True
except Exception as e:
print(f"❌ 独立测试失败: {e}")
return False
if __name__ == "__main__":
print("🚀 开始主题漂移检测测试")
# 测试独立检测器
standalone_success = test_topic_drift_detector_standalone()
# 测试集成功能
integration_success = test_topic_drift_integration()
if standalone_success and integration_success:
print("\n🎉 所有测试通过!")
sys.exit(0)
else:
print("\n💥 部分测试失败!")
sys.exit(1)