test_time_aware_ranking.py•6.78 kB
#!/usr/bin/env python3
"""
时间感知检索排序功能测试脚本
测试时间衰减函数的实现和效果
"""
import os
import sys
import time
from datetime import datetime, timedelta
from pathlib import Path
# 添加项目根目录到Python路径
sys.path.insert(0, str(Path(__file__).parent))
def test_time_aware_ranking():
"""测试时间感知检索排序功能"""
print("🧪 时间感知检索排序功能测试")
print("=" * 50)
try:
from mvp_memory import MVPMemoryManager
# 初始化测试数据目录
test_data_dir = "./test_time_aware_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"],
"timestamp": datetime.now().isoformat() # 今天
},
{
"content": "昨天完成了机器学习项目的数据预处理",
"tags": ["项目", "机器学习"],
"timestamp": (datetime.now() - timedelta(days=1)).isoformat() # 1天前
},
{
"content": "一周前开始学习深度学习理论",
"tags": ["学习", "深度学习"],
"timestamp": (datetime.now() - timedelta(days=7)).isoformat() # 7天前
},
{
"content": "一个月前参加了AI技术会议",
"tags": ["会议", "AI"],
"timestamp": (datetime.now() - timedelta(days=30)).isoformat() # 30天前
},
{
"content": "三个月前开始关注人工智能领域",
"tags": ["AI", "兴趣"],
"timestamp": (datetime.now() - timedelta(days=90)).isoformat() # 90天前
}
]
# 添加记忆
for memory in test_memories:
success = mvp_manager.remember(
memory["content"],
tags=memory["tags"],
metadata={"timestamp": memory["timestamp"]}
)
if success:
print(f"✅ 已添加: {memory['content'][:30]}...")
else:
print(f"❌ 添加失败: {memory['content'][:30]}...")
print(f"\n📊 总共添加了 {len(test_memories)} 条测试记忆")
# 等待一下确保记忆被索引
time.sleep(2)
print("\n🔍 测试时间感知检索排序...")
print("-" * 40)
# 测试查询
test_query = "学习"
print(f"查询: '{test_query}'")
print("\n1️⃣ 启用时间感知排序 (τ=30天):")
results_with_time = mvp_manager.recall(
test_query,
top_k=5,
use_time_decay=True,
time_decay_tau=30.0
)
for i, result in enumerate(results_with_time, 1):
content = result.get('content', '')[:50]
score = result.get('score', 0)
time_decay_factor = result.get('time_decay_factor', 1.0)
days_ago = result.get('days_ago', 0)
print(f" {i}. {content}...")
print(f" 分数: {score:.4f}, 时间权重: {time_decay_factor:.3f}, {days_ago:.1f}天前")
print("\n2️⃣ 禁用时间感知排序:")
results_without_time = mvp_manager.recall(
test_query,
top_k=5,
use_time_decay=False
)
for i, result in enumerate(results_without_time, 1):
content = result.get('content', '')[:50]
score = result.get('score', 0)
print(f" {i}. {content}...")
print(f" 分数: {score:.4f}")
print("\n3️⃣ 测试不同τ值的影响:")
for tau in [7.0, 30.0, 90.0]:
print(f"\nτ = {tau}天:")
results = mvp_manager.recall(
test_query,
top_k=3,
use_time_decay=True,
time_decay_tau=tau
)
for i, result in enumerate(results, 1):
content = result.get('content', '')[:30]
score = result.get('score', 0)
time_decay_factor = result.get('time_decay_factor', 1.0)
days_ago = result.get('days_ago', 0)
print(f" {i}. {content}... (分数: {score:.4f}, 权重: {time_decay_factor:.3f})")
print("\n✅ 时间感知检索排序功能测试完成!")
# 验证时间感知效果
print("\n📈 验证时间感知效果:")
if results_with_time and results_without_time:
# 检查最新记忆是否排在前面
with_time_first = results_with_time[0]
without_time_first = results_without_time[0]
with_time_days = with_time_first.get('days_ago', float('inf'))
print(f"启用时间感知时,第一条记忆是 {with_time_days:.1f} 天前的")
if with_time_days < 2: # 最新的记忆应该排在前面
print("✅ 时间感知排序正常工作 - 最新记忆排在前面")
else:
print("⚠️ 时间感知排序可能需要调整")
return True
except Exception as e:
print(f"❌ 测试失败: {e}")
import traceback
traceback.print_exc()
return False
def test_time_decay_formula():
"""测试时间衰减公式的数学正确性"""
print("\n🧮 测试时间衰减公式")
print("-" * 30)
import math
tau = 30.0 # 30天
test_days = [0, 1, 7, 15, 30, 60, 90]
print(f"时间衰减公式: exp(-Δt/τ), τ = {tau}天")
print("天数\t衰减因子")
print("-" * 20)
for days in test_days:
decay_factor = math.exp(-days / tau)
print(f"{days}\t{decay_factor:.4f}")
print("\n✅ 公式验证完成")
if __name__ == "__main__":
print("🚀 开始时间感知检索排序测试")
# 测试时间衰减公式
test_time_decay_formula()
# 测试完整功能
success = test_time_aware_ranking()
if success:
print("\n🎉 所有测试通过!")
sys.exit(0)
else:
print("\n💥 测试失败!")
sys.exit(1)