test_performance_optimization.py•6.75 kB
#!/usr/bin/env python3
"""
MemOS性能优化测试脚本
测试Qdrant参数优化和缓存效果
"""
import time
import statistics
from typing import List, Dict, Any
from mvp_memory import create_mvp_memory_manager
from enhanced_simple_memos import EnhancedSimpleMemOS
def test_search_performance(mvp_manager, queries: List[str], iterations: int = 3) -> Dict[str, Any]:
"""测试搜索性能"""
print(f"🧪 测试搜索性能 ({iterations}次迭代)")
# 清空缓存确保公平测试
if hasattr(mvp_manager.memory, 'clear_caches'):
mvp_manager.memory.clear_caches()
# 第一次搜索（无缓存）
first_search_times = []
for query in queries:
start_time = time.time()
results = mvp_manager.recall(query, top_k=5)
duration = time.time() - start_time
first_search_times.append(duration)
print(f" 查询: '{query}' - {duration:.3f}s ({len(results)}条结果)")
# 重复搜索（有缓存）
print(f"\n🎯 测试缓存效果...")
cached_search_times = []
for query in queries:
start_time = time.time()
results = mvp_manager.recall(query, top_k=5)
duration = time.time() - start_time
cached_search_times.append(duration)
print(f" 查询: '{query}' - {duration:.3f}s (缓存)")
# 计算统计信息
avg_first = statistics.mean(first_search_times)
avg_cached = statistics.mean(cached_search_times)
improvement = ((avg_first - avg_cached) / avg_first * 100) if avg_first > 0 else 0
return {
"first_search_times": first_search_times,
"cached_search_times": cached_search_times,
"avg_first_search": avg_first,
"avg_cached_search": avg_cached,
"performance_improvement": improvement,
"target_met": avg_first < 1.0 # 目标：1秒以内
}
def test_different_data_sizes():
"""测试不同数据量下的性能"""
print("📊 测试不同数据量下的性能表现")
data_sizes = [100, 500, 1000, 2000]
test_queries = [
"MemOS智能记忆系统",
"Qdrant向量数据库",
"性能优化技术",
"缓存机制实现"
]
results = {}
for size in data_sizes:
print(f"\n🔢 测试数据量: {size}条记忆")
# 创建测试数据
memos = EnhancedSimpleMemOS(f"./test_data_{size}")
# 添加测试记忆
for i in range(size):
content = f"测试记忆{i}: 这是关于技术{i%10}的内容，包含性能优化和系统设计的相关信息"
tags = [f"技术{i%10}", "测试", "性能"]
memos.add_memory(content, tags=tags)
# 测试搜索性能
search_times = []
for query in test_queries:
start_time = time.time()
search_results = memos.search_memories(query, limit=5, use_reranker=True)
duration = time.time() - start_time
search_times.append(duration)
avg_time = statistics.mean(search_times)
results[size] = {
"avg_search_time": avg_time,
"search_times": search_times,
"target_met": avg_time < 1.0
}
print(f" 平均搜索时间: {avg_time:.3f}s")
print(f" 目标达成: {'✅' if avg_time < 1.0 else '❌'} (目标<1s)")
# 显示缓存统计
if hasattr(memos, 'print_performance_stats'):
memos.print_performance_stats()
return results
def test_cache_hit_rates():
"""测试缓存命中率"""
print("🎯 测试缓存命中率")
mvp = create_mvp_memory_manager()
# 准备测试查询
test_queries = [
"MemOS系统架构",
"Qdrant性能优化",
"缓存机制设计",
"向量搜索算法",
"智能记忆管理"
]
# 清空缓存
if hasattr(mvp.memory, 'clear_caches'):
mvp.memory.clear_caches()
# 第一轮搜索（建立缓存）
print("第一轮搜索（建立缓存）:")
for query in test_queries:
results = mvp.recall(query, top_k=3)
print(f" {query}: {len(results)}条结果")
# 第二轮搜索（测试缓存命中）
print("\n第二轮搜索（测试缓存命中）:")
cache_hit_count = 0
for query in test_queries:
start_time = time.time()
results = mvp.recall(query, top_k=3)
duration = time.time() - start_time
# 如果搜索时间很短，说明命中缓存
is_cache_hit = duration < 0.1
if is_cache_hit:
cache_hit_count += 1
print(f" {query}: {duration:.3f}s {'🎯' if is_cache_hit else '❌'}")
hit_rate = (cache_hit_count / len(test_queries)) * 100
print(f"\n缓存命中率: {hit_rate:.1f}% ({cache_hit_count}/{len(test_queries)})")
print(f"目标达成: {'✅' if hit_rate >= 50 else '❌'} (目标≥50%)")
return hit_rate
def main():
"""主测试函数"""
print("🚀 MemOS性能优化测试")
print("=" * 50)
# 测试1: 基础搜索性能
print("\n📋 测试1: 基础搜索性能")
mvp = create_mvp_memory_manager()
test_queries = [
"MemOS智能记忆系统",
"Qdrant向量数据库性能",
"缓存优化技术"
]
perf_results = test_search_performance(mvp, test_queries)
print(f"\n📊 性能测试结果:")
print(f" 平均首次搜索: {perf_results['avg_first_search']:.3f}s")
print(f" 平均缓存搜索: {perf_results['avg_cached_search']:.3f}s")
print(f" 性能提升: {perf_results['performance_improvement']:.1f}%")
print(f" 目标达成: {'✅' if perf_results['target_met'] else '❌'} (目标<1s)")
# 测试2: 缓存命中率
print(f"\n📋 测试2: 缓存命中率")
hit_rate = test_cache_hit_rates()
# 测试3: 不同数据量性能
print(f"\n📋 测试3: 不同数据量性能")
size_results = test_different_data_sizes()
# 总结报告
print(f"\n🎯 性能优化总结报告")
print("=" * 50)
print(f"✅ 搜索延迟优化: {perf_results['avg_first_search']:.3f}s (目标<1s)")
print(f"✅ 缓存命中率: {hit_rate:.1f}% (目标≥50%)")
print(f"✅ 性能提升: {perf_results['performance_improvement']:.1f}%")
# 检查是否达到所有目标
all_targets_met = (
perf_results['target_met'] and
hit_rate >= 50 and
perf_results['performance_improvement'] > 80
)
print(f"\n🏆 总体评估: {'✅ 所有目标达成' if all_targets_met else '⚠️ 部分目标未达成'}")
if __name__ == "__main__":
main()