diagnose_enhanced_status.py•6.8 kB
#!/usr/bin/env python3
"""
MemOS增强版状态诊断脚本
用于重现和分析MCP服务器状态显示不一致问题
"""
import os
import sys
import time
import traceback
from pathlib import Path
# 添加项目路径
sys.path.insert(0, str(Path(__file__).parent))
def test_direct_mvp_manager():
"""直接测试MVP管理器状态"""
print("=" * 60)
print("🔍 直接测试MVP管理器状态")
print("=" * 60)
try:
from mvp_memory import MVPMemoryManager
# 使用与MCP相同的数据目录
data_dir = "./mcp_memos_data"
print(f"📁 数据目录: {data_dir}")
# 创建MVP管理器实例
print("🚀 创建MVP管理器实例...")
mvp_manager = MVPMemoryManager(data_dir=data_dir, use_enhanced=True)
# 检查状态
print(f"✅ use_enhanced: {mvp_manager.use_enhanced}")
print(f"✅ memory类型: {type(mvp_manager.memory).__name__}")
# 测试功能
print("\n🧪 测试记忆功能...")
test_content = f"诊断测试记忆 - {time.time()}"
success = mvp_manager.remember(test_content, tags=["诊断", "测试"])
print(f"✅ 记忆添加: {'成功' if success else '失败'}")
# 测试检索
print("\n🔍 测试检索功能...")
results = mvp_manager.recall("诊断", top_k=3)
print(f"✅ 检索结果: {len(results)}条")
return mvp_manager
except Exception as e:
print(f"❌ 直接测试失败: {e}")
traceback.print_exc()
return None
def test_mcp_server_initialization():
"""测试MCP服务器初始化过程"""
print("\n" + "=" * 60)
print("🔍 测试MCP服务器初始化过程")
print("=" * 60)
try:
# 模拟MCP服务器的初始化过程
from memos_mcp_server import MemOSMCPServer
print("🚀 创建MCP服务器实例...")
server = MemOSMCPServer()
# 检查MVP管理器状态
print(f"✅ MCP中use_enhanced: {server.mvp_manager.use_enhanced}")
print(f"✅ MCP中memory类型: {type(server.mvp_manager.memory).__name__}")
# 模拟MCP工具调用
print("\n🧪 模拟MCP工具调用...")
# 模拟添加记忆的MCP调用
test_request = {
"jsonrpc": "2.0",
"id": "test-1",
"method": "tools/call",
"params": {
"name": "add_memos_memory",
"arguments": {
"content": f"MCP诊断测试记忆 - {time.time()}",
"tags": ["MCP", "诊断"]
}
}
}
# 直接调用处理方法
import asyncio
response = asyncio.run(server.add_memos_memory(test_request["params"]["arguments"]))
print(f"✅ MCP响应: {response}")
return server
except Exception as e:
print(f"❌ MCP测试失败: {e}")
traceback.print_exc()
return None
def analyze_initialization_timing():
"""分析初始化时序问题"""
print("\n" + "=" * 60)
print("🔍 分析初始化时序问题")
print("=" * 60)
try:
# 检查Qdrant服务状态
import requests
print("🔍 检查Qdrant服务状态...")
try:
response = requests.get("http://localhost:6335/health", timeout=5)
print(f"✅ Qdrant服务状态: {response.status_code}")
except Exception as e:
print(f"❌ Qdrant服务不可用: {e}")
return False
# 测试多次初始化
print("\n🔄 测试多次初始化...")
for i in range(3):
print(f"\n--- 第{i+1}次初始化 ---")
try:
from mvp_memory import MVPMemoryManager
# 每次都创建新实例
mvp = MVPMemoryManager(data_dir="./test_timing_data", use_enhanced=True)
print(f"✅ 第{i+1}次: use_enhanced={mvp.use_enhanced}")
# 清理
del mvp
except Exception as e:
print(f"❌ 第{i+1}次初始化失败: {e}")
return True
except Exception as e:
print(f"❌ 时序分析失败: {e}")
traceback.print_exc()
return False
def check_configuration_consistency():
"""检查配置一致性"""
print("\n" + "=" * 60)
print("🔍 检查配置一致性")
print("=" * 60)
# 检查环境变量
print("📋 环境变量:")
env_vars = ["MEMOS_CONFIG", "SILICONFLOW_API_KEY", "SILICONFLOW_BASE_URL"]
for var in env_vars:
value = os.environ.get(var, "未设置")
print(f" {var}: {value}")
# 检查配置文件
config_paths = [
"./mcp_memos_data/concurrent_config.json",
"./memos_data/concurrent_config.json",
"./concurrent_config.json"
]
print("\n📋 配置文件:")
for path in config_paths:
if os.path.exists(path):
print(f" ✅ {path}: 存在")
try:
import json
with open(path, 'r') as f:
config = json.load(f)
print(f" 内容: {config}")
except Exception as e:
print(f" ❌ 读取失败: {e}")
else:
print(f" ❌ {path}: 不存在")
def main():
"""主诊断流程"""
print("🩺 MemOS增强版状态诊断开始")
print("=" * 80)
# 1. 检查配置一致性
check_configuration_consistency()
# 2. 直接测试MVP管理器
mvp_direct = test_direct_mvp_manager()
# 3. 测试MCP服务器初始化
mcp_server = test_mcp_server_initialization()
# 4. 分析初始化时序
timing_ok = analyze_initialization_timing()
# 5. 对比结果
print("\n" + "=" * 60)
print("📊 诊断结果对比")
print("=" * 60)
if mvp_direct and mcp_server:
print(f"直接测试 use_enhanced: {mvp_direct.use_enhanced}")
print(f"MCP测试 use_enhanced: {mcp_server.mvp_manager.use_enhanced}")
if mvp_direct.use_enhanced != mcp_server.mvp_manager.use_enhanced:
print("❌ 发现状态不一致！")
print("🔍 可能原因:")
print(" 1. MCP初始化时Qdrant连接超时")
print(" 2. 不同的配置文件路径")
print(" 3. 环境变量差异")
print(" 4. 初始化时序问题")
else:
print("✅ 状态一致")
print("\n🎯 诊断完成")
if __name__ == "__main__":
main()