test_qwen_integration.py•7.18 kB
#!/usr/bin/env python3
"""
测试Qwen模型集成
验证嵌入模型和重排模型的完整功能
"""
import os
import sys
from pathlib import Path
# 添加当前目录到路径
sys.path.insert(0, str(Path(__file__).parent))
from usage_examples import load_env_file
from mvp_memory import MVPMemoryManager
def test_qwen_models_basic():
"""测试Qwen模型基础功能"""
print("🧪 测试Qwen模型基础功能")
print("=" * 60)
# 加载环境变量
load_env_file()
try:
# 测试Qwen配置
from qwen_embedding_config import test_qwen_models
success = test_qwen_models()
if success:
print("✅ Qwen模型基础测试通过")
return True
else:
print("❌ Qwen模型基础测试失败")
return False
except Exception as e:
print(f"❌ Qwen模型测试异常: {e}")
return False
def test_mvp_with_qwen():
"""测试MVP记忆管理器与Qwen模型集成"""
print("\n🧪 测试MVP记忆管理器与Qwen模型集成")
print("=" * 60)
try:
# 创建MVP记忆管理器，启用增强版
mvp_manager = MVPMemoryManager("./test_qwen_mvp_data", use_enhanced=True)
# 测试连接
if not mvp_manager.test_connection():
print("❌ MVP管理器连接测试失败")
return False
# 测试添加记忆
print("\n📝 测试添加记忆...")
test_memories = [
("Qwen3-Embedding-0.6B是阿里巴巴开发的高质量嵌入模型", ["Qwen", "嵌入模型", "阿里巴巴"]),
("Qwen3-Reranker-0.6B提供精准的文档重排功能", ["Qwen", "重排模型", "文档检索"]),
("MemOS集成Qwen模型后，搜索精度显著提升", ["MemOS", "Qwen", "搜索优化"]),
("向量数据库使用1024维嵌入向量存储语义信息", ["向量数据库", "嵌入向量", "语义搜索"]),
("SiliconFlow API提供稳定的模型推理服务", ["SiliconFlow", "API", "模型服务"])
]
for content, tags in test_memories:
success = mvp_manager.remember(content, tags=tags)
if success:
print(f"✅ 记忆添加成功: {content[:30]}...")
else:
print(f"❌ 记忆添加失败: {content[:30]}...")
# 测试智能检索
print("\n🔍 测试智能检索...")
test_queries = [
"Qwen模型的功能特点",
"向量搜索和语义检索",
"API服务和模型推理",
"文档重排和搜索优化"
]
for query in test_queries:
print(f"\n查询: {query}")
results = mvp_manager.recall(query, top_k=3, use_reranker=True)
if results:
print(f"找到 {len(results)} 条相关记忆:")
for i, result in enumerate(results, 1):
print(f" {i}. 向量分数: {result.get('score', 0):.4f}")
if 'rerank_score' in result:
print(f" 重排分数: {result['rerank_score']:.4f}")
print(f" 内容: {result['content'][:50]}...")
print(f" 标签: {result.get('tags', [])}")
else:
print(" 未找到相关记忆")
print("\n✅ MVP与Qwen模型集成测试完成")
return True
except Exception as e:
print(f"❌ MVP与Qwen集成测试失败: {e}")
import traceback
traceback.print_exc()
return False
def test_performance_comparison():
"""测试性能对比：基础版 vs 增强版"""
print("\n🧪 测试性能对比：基础版 vs 增强版")
print("=" * 60)
try:
# 测试基础版
print("🔄 测试基础版MemOS...")
basic_manager = MVPMemoryManager("./test_basic_data", use_enhanced=False)
# 测试增强版
print("🔄 测试增强版MemOS...")
enhanced_manager = MVPMemoryManager("./test_enhanced_data", use_enhanced=True)
# 添加相同的测试数据
test_content = "这是一个测试记忆，用于比较基础版和增强版的性能差异"
test_tags = ["测试", "性能对比"]
# 基础版添加记忆
basic_success = basic_manager.remember(test_content, tags=test_tags)
print(f"基础版添加记忆: {'成功' if basic_success else '失败'}")
# 增强版添加记忆
enhanced_success = enhanced_manager.remember(test_content, tags=test_tags)
print(f"增强版添加记忆: {'成功' if enhanced_success else '失败'}")
# 搜索测试
query = "性能测试记忆"
print(f"\n🔍 搜索查询: {query}")
# 基础版搜索
basic_results = basic_manager.recall(query, top_k=3)
print(f"基础版搜索结果: {len(basic_results)} 条")
# 增强版搜索
enhanced_results = enhanced_manager.recall(query, top_k=3, use_reranker=True)
print(f"增强版搜索结果: {len(enhanced_results)} 条")
# 显示结果对比
if basic_results:
print(f"基础版最佳匹配分数: {basic_results[0].get('score', 0):.4f}")
if enhanced_results:
print(f"增强版向量分数: {enhanced_results[0].get('score', 0):.4f}")
if 'rerank_score' in enhanced_results[0]:
print(f"增强版重排分数: {enhanced_results[0]['rerank_score']:.4f}")
print("\n✅ 性能对比测试完成")
return True
except Exception as e:
print(f"❌ 性能对比测试失败: {e}")
return False
def main():
"""主测试函数"""
print("🚀 Qwen模型集成完整测试")
print("=" * 80)
# 测试计数
total_tests = 3
passed_tests = 0
# 1. 基础功能测试
if test_qwen_models_basic():
passed_tests += 1
# 2. MVP集成测试
if test_mvp_with_qwen():
passed_tests += 1
# 3. 性能对比测试
if test_performance_comparison():
passed_tests += 1
# 总结
print(f"\n📊 测试总结")
print("=" * 80)
print(f"总测试数: {total_tests}")
print(f"通过测试: {passed_tests}")
print(f"失败测试: {total_tests - passed_tests}")
print(f"成功率: {passed_tests/total_tests*100:.1f}%")
if passed_tests == total_tests:
print("\n🎉 所有测试通过！Qwen模型集成成功！")
print("现在可以使用以下功能:")
print("- Qwen3-Embedding-0.6B 高质量嵌入向量")
print("- Qwen3-Reranker-0.6B 精准文档重排")
print("- 增强版MVP记忆管理器")
print("- 智能检索和语义搜索")
return True
else:
print(f"\n❌ 部分测试失败，请检查配置和环境")
return False
if __name__ == "__main__":
success = main()
sys.exit(0 if success else 1)