MemOS-MCP

#!/usr/bin/env python3 """ 创建专家咨询代码包 打包MemOS增强版激活问题的所有相关文件 """ import os import shutil import zipfile from pathlib import Path from datetime import datetime def create_consultation_package(): """创建专家咨询代码包""" # 创建临时目录 package_name = f"MemOS增强版问题咨询包_{datetime.now().strftime('%Y%m%d_%H%M%S')}" temp_dir = Path(f"/tmp/{package_name}") temp_dir.mkdir(exist_ok=True) print(f"📦 创建咨询包: {package_name}") # 核心代码文件 core_files = [ "memos_mcp_server.py", "mvp_memory.py", "enhanced_simple_memos.py", "test_mvp_enhanced.py", "qwen_embedding_config.py", "usage_examples.py" ] # 配置文件 config_files = [ "mcp_memos_data/concurrent_config.json", "claude_desktop_config.json", "memos_mcp_config_complete.json", ".env" ] # 文档文件 doc_files = [ "专家咨询-MemOS增强版激活问题分析.md", "README.md", "QWEN_MODEL_CONFIGURATION.md" ] # 关键源码目录 src_dirs = [ "src/memos/configs", "src/memos/embedders", "src/memos/rerankers" ] # 复制核心文件 print("📄 复制核心代码文件...") for file_path in core_files: src = Path(file_path) if src.exists(): dst = temp_dir / src.name shutil.copy2(src, dst) print(f" ✅ {file_path}") else: print(f" ❌ {file_path} (不存在)") # 复制配置文件 print("⚙️ 复制配置文件...") config_dir = temp_dir / "configs" config_dir.mkdir(exist_ok=True) for file_path in config_files: src = Path(file_path) if src.exists(): dst = config_dir / src.name shutil.copy2(src, dst) print(f" ✅ {file_path}") else: print(f" ❌ {file_path} (不存在)") # 复制文档文件 print("📚 复制文档文件...") docs_dir = temp_dir / "docs" docs_dir.mkdir(exist_ok=True) for file_path in doc_files: src = Path(file_path) if src.exists(): dst = docs_dir / src.name shutil.copy2(src, dst) print(f" ✅ {file_path}") else: print(f" ❌ {file_path} (不存在)") # 复制源码目录 print("🔧 复制关键源码目录...") src_base_dir = temp_dir / "src" src_base_dir.mkdir(exist_ok=True) for dir_path in src_dirs: src = Path(dir_path) if src.exists(): dst = src_base_dir / src.relative_to("src") dst.parent.mkdir(parents=True, exist_ok=True) shutil.copytree(src, dst, dirs_exist_ok=True) print(f" ✅ {dir_path}") else: print(f" ❌ {dir_path} (不存在)") # 创建问题重现脚本 print("🧪 创建问题重现脚本...") reproduce_script = temp_dir / "reproduce_issue.py" with open(reproduce_script, 'w', encoding='utf-8') as f: f.write('''#!/usr/bin/env python3 """ 问题重现脚本 用于重现MemOS增强版激活状态显示不一致的问题 """ import sys import os from pathlib import Path # 添加当前目录到Python路径 sys.path.insert(0, str(Path(__file__).parent)) def test_mvp_manager_status(): """测试MVP管理器状态""" print("🧪 测试MVP管理器状态...") try: from mvp_memory import MVPMemoryManager # 创建MVP管理器实例 mvp = MVPMemoryManager("./test_data", use_enhanced=True) print(f"📊 MVP管理器状态:") print(f" - use_enhanced: {mvp.use_enhanced}") print(f" - memory类型: {type(mvp.memory).__name__}") return mvp.use_enhanced except Exception as e: print(f"❌ 测试失败: {e}") return False def simulate_mcp_server(): """模拟MCP服务器行为""" print("\\n🖥️ 模拟MCP服务器...") try: from memos_mcp_server import MemOSMCPServer # 创建MCP服务器实例 server = MemOSMCPServer() print(f"📊 MCP服务器状态:") print(f" - mvp_manager.use_enhanced: {server.mvp_manager.use_enhanced}") print(f" - memory类型: {type(server.mvp_manager.memory).__name__}") # 模拟状态检查逻辑 model_info = "Qwen3-Embedding-0.6B" if server.mvp_manager.use_enhanced else "哈希向量" print(f" - 显示状态: {model_info}") return server.mvp_manager.use_enhanced except Exception as e: print(f"❌ MCP服务器测试失败: {e}") return False if __name__ == "__main__": print("🚀 开始问题重现测试...") print("=" * 60) # 测试1: 直接测试MVP管理器 mvp_status = test_mvp_manager_status() # 测试2: 模拟MCP服务器 mcp_status = simulate_mcp_server() print("\\n" + "=" * 60) print("📋 测试结果总结:") print(f" - 直接MVP管理器: {'✅ 增强版' if mvp_status else '❌ 基础版'}") print(f" - MCP服务器: {'✅ 增强版' if mcp_status else '❌ 基础版'}") if mvp_status != mcp_status: print("\\n🔥 问题确认: 状态不一致!") print(" 直接测试和MCP服务器显示不同的状态") else: print("\\n✅ 状态一致") ''') # 创建README print("📖 创建README...") readme = temp_dir / "README.md" with open(readme, 'w', encoding='utf-8') as f: f.write(f'''# MemOS增强版激活问题咨询包 ## 📋 包内容 ### 核心代码文件 - `memos_mcp_server.py` - MCP服务器主文件 - `mvp_memory.py` - MVP记忆管理器 - `enhanced_simple_memos.py` - 增强版MemOS实现 - `test_mvp_enhanced.py` - 原始测试脚本 - `reproduce_issue.py` - 问题重现脚本 ### 配置文件 (configs/) - `concurrent_config.json` - 并发配置 - `claude_desktop_config.json` - MCP客户端配置 - `.env` - 环境变量 ### 文档 (docs/) - `专家咨询-MemOS增强版激活问题分析.md` - 详细问题分析 - `QWEN_MODEL_CONFIGURATION.md` - Qwen模型配置说明 ### 源码 (src/) - `src/memos/configs/` - 配置模块 - `src/memos/embedders/` - 嵌入器模块 - `src/memos/rerankers/` - 重排器模块 ## 🚀 快速重现问题 ```bash # 1. 安装依赖 pip install openai qdrant-client # 2. 运行问题重现脚本 python reproduce_issue.py # 3. 查看详细分析 cat docs/专家咨询-MemOS增强版激活问题分析.md ``` ## 🎯 核心问题 MVP管理器在直接测试中显示use_enhanced=True，但在MCP服务器上下文中可能返回False，导致状态显示不一致。 ## 📞 联系信息 - 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - 问题类型: 状态显示不一致 - 优先级: 高 ''') # 创建压缩包 print("🗜️ 创建压缩包...") zip_path = Path(f"/home/qqinshu/视频/专家咨询代码包/{package_name}.zip") zip_path.parent.mkdir(exist_ok=True) with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf: for root, dirs, files in os.walk(temp_dir): for file in files: file_path = Path(root) / file arc_path = file_path.relative_to(temp_dir) zipf.write(file_path, arc_path) # 清理临时目录 shutil.rmtree(temp_dir) print(f"✅ 咨询包创建完成: {zip_path}") print(f"📊 文件大小: {zip_path.stat().st_size / 1024:.1f} KB") return zip_path if __name__ == "__main__": create_consultation_package()