memos_daemon.py•3.82 kB
#!/usr/bin/env python3
"""
MemOS守护进程
专为systemd服务设计的长期运行进程
"""
import os
import sys
import time
import signal
import logging
from pathlib import Path
# 添加当前目录到路径
sys.path.insert(0, str(Path(__file__).parent))
from mvp_memory import create_mvp_memory_manager
from process_manager import MemOSProcessManager
class MemOSDaemon:
"""MemOS守护进程"""
def __init__(self):
self.running = True
self.mvp_manager = None
self.process_manager = None
# 设置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
self.logger = logging.getLogger('MemOSDaemon')
# 注册信号处理
signal.signal(signal.SIGTERM, self._signal_handler)
signal.signal(signal.SIGINT, self._signal_handler)
def _signal_handler(self, signum, frame):
"""信号处理器"""
self.logger.info(f"收到信号 {signum}，准备关闭...")
self.running = False
def initialize(self):
"""初始化守护进程"""
try:
self.logger.info("初始化MemOS守护进程...")
# 初始化进程管理器
self.process_manager = MemOSProcessManager()
# 获取进程锁
if not self.process_manager.acquire_lock():
self.logger.error("无法获取进程锁，可能有其他MemOS实例正在运行")
return False
# 写入PID文件
self.process_manager.write_pid_file()
# 初始化MVP管理器
self.mvp_manager = create_mvp_memory_manager()
# 测试连接
if not self.mvp_manager.test_connection():
self.logger.error("MVP管理器连接测试失败")
return False
self.logger.info("MemOS守护进程初始化成功")
return True
except Exception as e:
self.logger.error(f"初始化失败: {e}")
return False
def run(self):
"""运行守护进程"""
if not self.initialize():
sys.exit(1)
self.logger.info("MemOS守护进程开始运行...")
try:
# 主循环
while self.running:
# 执行健康检查
self._health_check()
# 等待一段时间
time.sleep(30) # 每30秒检查一次
except KeyboardInterrupt:
self.logger.info("收到键盘中断")
except Exception as e:
self.logger.error(f"运行时错误: {e}")
finally:
self.cleanup()
def _health_check(self):
"""健康检查"""
try:
# 检查MVP管理器状态
if self.mvp_manager:
# 简单的连接测试
result = self.mvp_manager.test_connection()
self.logger.debug(f"健康检查通过，连接状态: {result}")
except Exception as e:
self.logger.warning(f"健康检查失败: {e}")
def cleanup(self):
"""清理资源"""
self.logger.info("清理资源...")
try:
# 释放进程锁
if self.process_manager:
self.process_manager.release_lock()
self.logger.info("资源清理完成")
except Exception as e:
self.logger.error(f"清理资源时出错: {e}")
def main():
"""主函数"""
daemon = MemOSDaemon()
daemon.run()
if __name__ == "__main__":
main()