update_to_server_mode.py•7.03 kB
#!/usr/bin/env python3
"""
更新MemOS到Qdrant服务器模式
简化版本，适合个人使用
"""
import os
import re
import shutil
from pathlib import Path
def backup_files():
"""备份原始文件"""
print("📦 备份原始文件...")
files_to_backup = [
"advanced_examples.py",
"memos_cli.py",
"memos_mcp_server.py",
"concurrent_manager.py"
]
backup_dir = Path("./backup_embedded_mode")
backup_dir.mkdir(exist_ok=True)
for file in files_to_backup:
if Path(file).exists():
shutil.copy2(file, backup_dir / file)
print(f" ✅ 备份: {file}")
print(f"📁 备份目录: {backup_dir.absolute()}")
def update_advanced_examples():
"""更新advanced_examples.py使用服务器模式"""
file_path = Path("advanced_examples.py")
if not file_path.exists():
print(f"❌ 文件不存在: {file_path}")
return False
print("🔧 更新advanced_examples.py...")
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
# 替换QdrantClient初始化
old_pattern = r'self\.vector_db = QdrantClient\(path=str\(self\.data_dir / "vectors"\)\)'
new_code = '''# 服务器模式配置
qdrant_url = os.environ.get('QDRANT_URL', 'http://localhost:6333')
self.vector_db = QdrantClient(url=qdrant_url)'''
if old_pattern in content:
content = re.sub(old_pattern, new_code, content)
print(" ✅ 更新QdrantClient初始化")
else:
print(" ⚠️ 未找到QdrantClient初始化模式")
# 添加命名空间支持
namespace_code = '''
def _get_collection_name(self, namespace="default"):
"""获取带命名空间的集合名"""
return f"memos_{namespace}"
def _add_namespace_to_payload(self, payload, namespace="default"):
"""添加命名空间到payload"""
if payload is None:
payload = {}
payload["namespace"] = namespace
payload["source"] = namespace
return payload
'''
# 在类定义后添加命名空间方法
class_pattern = r'(class AdvancedMemOS:.*?\n def __init__.*?\n)'
if re.search(class_pattern, content, re.DOTALL):
content = re.sub(
r'(class AdvancedMemOS:.*?\n)',
r'\1' + namespace_code,
content,
flags=re.DOTALL
)
print(" ✅ 添加命名空间支持")
# 保存更新后的文件
with open(file_path, 'w', encoding='utf-8') as f:
f.write(content)
print(" ✅ advanced_examples.py更新完成")
return True
def update_cli_and_mcp():
"""更新CLI和MCP使用统一数据目录"""
print("🔧 更新CLI和MCP配置...")
# 更新memos_cli.py
cli_file = Path("memos_cli.py")
if cli_file.exists():
with open(cli_file, 'r', encoding='utf-8') as f:
content = f.read()
# 统一数据目录
content = content.replace(
'concurrent_manager = MemOSConcurrentManager("./memos_data")',
'concurrent_manager = MemOSConcurrentManager("./memos_data")'
)
with open(cli_file, 'w', encoding='utf-8') as f:
f.write(content)
print(" ✅ 更新memos_cli.py")
# 更新memos_mcp_server.py
mcp_file = Path("memos_mcp_server.py")
if mcp_file.exists():
with open(mcp_file, 'r', encoding='utf-8') as f:
content = f.read()
# 统一数据目录 - 从mcp_memos_data改为memos_data
content = content.replace(
'concurrent_manager = MemOSConcurrentManager("./mcp_memos_data")',
'concurrent_manager = MemOSConcurrentManager("./memos_data")'
)
content = content.replace(
'self.memos = AdvancedMemOS("./mcp_memos_data")',
'self.memos = AdvancedMemOS("./memos_data")'
)
with open(mcp_file, 'w', encoding='utf-8') as f:
f.write(content)
print(" ✅ 更新memos_mcp_server.py")
def create_migration_script():
"""创建数据迁移脚本"""
migration_script = '''#!/usr/bin/env python3
"""
数据迁移脚本：从embedded模式迁移到服务器模式
"""
import sys
import json
from pathlib import Path
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
def migrate_data():
"""迁移数据"""
print("🔄 开始数据迁移...")
# 连接新的Qdrant服务器
client = QdrantClient(url="http://localhost:6333")
# 创建统一集合
collection_name = "memos_unified"
try:
# 检查集合是否存在
collections = client.get_collections()
collection_exists = any(c.name == collection_name for c in collections.collections)
if not collection_exists:
print(f"📝 创建集合: {collection_name}")
client.create_collection(
collection_name=collection_name,
vectors_config=VectorParams(size=768, distance=Distance.COSINE)
)
else:
print(f"✅ 集合已存在: {collection_name}")
print("🎉 数据迁移准备完成！")
print("📋 下一步:")
print("1. 启动新的MemOS CLI")
print("2. 重新添加重要记忆")
print("3. 验证CLI和MCP可以同时运行")
except Exception as e:
print(f"❌ 迁移失败: {e}")
return False
return True
if __name__ == "__main__":
migrate_data()
'''
script_path = Path("migrate_to_server.py")
with open(script_path, 'w', encoding='utf-8') as f:
f.write(migration_script)
# 设置执行权限
os.chmod(script_path, 0o755)
print(f"✅ 创建迁移脚本: {script_path}")
def main():
"""主函数"""
print("🔧 更新MemOS到Qdrant服务器模式")
print("=" * 40)
try:
# 1. 备份文件
backup_files()
# 2. 更新代码
update_advanced_examples()
update_cli_and_mcp()
# 3. 创建迁移脚本
create_migration_script()
print("\n🎉 代码更新完成！")
print("\n📋 下一步操作:")
print("1. 运行: python3 lightweight_qdrant_setup.py")
print("2. 运行: python3 migrate_to_server.py")
print("3. 测试: python3 memos_cli.py")
print("4. 测试: python3 memos_mcp_server.py")
print("\n⚠️ 重要提醒:")
print("- 原始文件已备份到 backup_embedded_mode/")
print("- 如有问题可以从备份恢复")
print("- 建议先在测试环境验证")
return True
except Exception as e:
print(f"❌ 更新失败: {e}")
return False
if __name__ == "__main__":
success = main()
if not success:
sys.exit(1)