test_default_qwen.py•6.19 kB
#!/usr/bin/env python3
"""
测试默认Qwen模型配置
验证系统默认使用Qwen3-Embedding-0.6B和Qwen3-Reranker-0.6B
"""
import os
import sys
from pathlib import Path
# 添加当前目录到路径
sys.path.insert(0, str(Path(__file__).parent))
from usage_examples import load_env_file
def test_default_mvp_manager():
"""测试默认MVP记忆管理器配置"""
print("🧪 测试默认MVP记忆管理器配置")
print("=" * 60)
# 加载环境变量
load_env_file()
try:
from mvp_memory import MVPMemoryManager
# 使用默认配置创建MVP管理器
print("🔄 创建默认MVP记忆管理器...")
mvp_manager = MVPMemoryManager("./test_default_data")
# 检查是否使用增强版
if mvp_manager.use_enhanced:
print("✅ 默认使用增强版MemOS (Qwen3-Embedding-0.6B + Qwen3-Reranker-0.6B)")
else:
print("⚠️ 降级到基础版MemOS (哈希向量)")
# 测试连接
if mvp_manager.test_connection():
print("✅ 系统连接测试通过")
else:
print("❌ 系统连接测试失败")
return False
# 测试添加记忆
print("\n📝 测试添加记忆...")
test_content = "这是一个测试记忆，用于验证Qwen模型的默认配置"
test_tags = ["测试", "Qwen", "默认配置"]
success = mvp_manager.remember(test_content, tags=test_tags)
if success:
print("✅ 记忆添加成功")
else:
print("❌ 记忆添加失败")
return False
# 测试智能检索
print("\n🔍 测试智能检索...")
query = "Qwen模型测试"
results = mvp_manager.recall(query, top_k=3)
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]}...")
else:
print("⚠️ 未找到相关记忆")
return True
except Exception as e:
print(f"❌ 测试失败: {e}")
import traceback
traceback.print_exc()
return False
def test_default_mcp_server():
"""测试默认MCP服务器配置"""
print("\n🧪 测试默认MCP服务器配置")
print("=" * 60)
try:
from memos_mcp_server import MemOSMCPServer
# 创建MCP服务器实例
print("🔄 创建MCP服务器实例...")
server = MemOSMCPServer()
# 检查MVP管理器配置
if hasattr(server, 'mvp_manager'):
if server.mvp_manager.use_enhanced:
print("✅ MCP服务器默认使用增强版MemOS (Qwen模型)")
else:
print("⚠️ MCP服务器降级到基础版MemOS")
else:
print("❌ MCP服务器未使用MVP管理器")
return False
# 模拟查询测试
print("\n🔍 模拟MCP查询测试...")
test_args = {
"query": "MemOS智能记忆管理",
"max_memories": 3
}
# 这里只是验证方法存在，不实际执行异步调用
if hasattr(server, 'query_memos_context'):
print("✅ MCP查询方法可用")
else:
print("❌ MCP查询方法不可用")
return False
return True
except Exception as e:
print(f"❌ MCP服务器测试失败: {e}")
return False
def test_cli_integration():
"""测试CLI集成"""
print("\n🧪 测试CLI集成")
print("=" * 60)
try:
# 测试CLI中的记忆功能
print("🔄 测试CLI记忆功能集成...")
# 模拟CLI中的记忆管理器创建
from mvp_memory import MVPMemoryManager
cli_manager = MVPMemoryManager()
if cli_manager.use_enhanced:
print("✅ CLI默认使用增强版MemOS")
else:
print("⚠️ CLI降级到基础版MemOS")
return True
except Exception as e:
print(f"❌ CLI集成测试失败: {e}")
return False
def main():
"""主测试函数"""
print("🚀 默认Qwen模型配置测试")
print("=" * 80)
# 测试计数
total_tests = 3
passed_tests = 0
# 1. MVP管理器默认配置测试
if test_default_mvp_manager():
passed_tests += 1
# 2. MCP服务器默认配置测试
if test_default_mcp_server():
passed_tests += 1
# 3. CLI集成测试
if test_cli_integration():
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("\n📋 当前默认配置:")
print("- 嵌入模型: Qwen/Qwen3-Embedding-0.6B (1024维)")
print("- 重排模型: Qwen/Qwen3-Reranker-0.6B")
print("- 向量数据库: Qdrant服务器模式 (端口6335)")
print("- API服务: SiliconFlow")
print("- 自动降级: 支持降级到基础版哈希向量")
print("\n🔧 使用方式:")
print("- CLI: python memos_cli.py (自动使用Qwen模型)")
print("- MCP: python memos_mcp_server.py (自动使用Qwen模型)")
print("- API: MVPMemoryManager() (默认use_enhanced=True)")
return True
else:
print(f"\n❌ 部分测试失败，请检查Qwen模型配置")
return False
if __name__ == "__main__":
success = main()
sys.exit(0 if success else 1)