start_dashboard.py•4.96 kB
#!/usr/bin/env python3
"""
MemOS Dashboard 启动脚本
同时启动API服务器和Dashboard Web服务器
"""
import os
import sys
import time
import threading
import webbrowser
from pathlib import Path
from http.server import HTTPServer, SimpleHTTPRequestHandler
import subprocess
# 添加项目根目录到Python路径
sys.path.insert(0, str(Path(__file__).parent))
class DashboardHandler(SimpleHTTPRequestHandler):
"""自定义HTTP处理器，用于服务Dashboard文件"""
def __init__(self, *args, **kwargs):
# 设置Dashboard目录为服务根目录
super().__init__(*args, directory=str(Path(__file__).parent / "dashboard"), **kwargs)
def end_headers(self):
# 添加CORS头，允许跨域请求
self.send_header('Access-Control-Allow-Origin', '*')
self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS')
self.send_header('Access-Control-Allow-Headers', 'Content-Type')
super().end_headers()
def start_api_server():
"""启动API服务器"""
print("🚀 启动MemOS API服务器...")
try:
# 导入并启动API服务器
from official_api_server import create_official_api_app, setup_exception_handlers
import uvicorn
app, manager = create_official_api_app()
setup_exception_handlers(app)
print("✅ API服务器初始化完成")
print("📖 API文档: http://localhost:8000/docs")
# 启动API服务器
uvicorn.run(
app,
host="0.0.0.0",
port=8000,
log_level="info",
access_log=False # 减少日志输出
)
except Exception as e:
print(f"❌ API服务器启动失败: {e}")
sys.exit(1)
def start_dashboard_server():
"""启动Dashboard Web服务器"""
print("🌐 启动Dashboard Web服务器...")
try:
# 等待API服务器启动
time.sleep(3)
# 检查dashboard目录是否存在
dashboard_dir = Path(__file__).parent / "dashboard"
if not dashboard_dir.exists():
print(f"❌ Dashboard目录不存在: {dashboard_dir}")
return
# 检查必要文件是否存在
index_file = dashboard_dir / "index.html"
js_file = dashboard_dir / "dashboard.js"
if not index_file.exists():
print(f"❌ Dashboard HTML文件不存在: {index_file}")
return
if not js_file.exists():
print(f"❌ Dashboard JS文件不存在: {js_file}")
return
# 启动HTTP服务器
server = HTTPServer(('localhost', 3000), DashboardHandler)
print("✅ Dashboard服务器启动成功")
print("🎯 Dashboard地址: http://localhost:3000")
# 自动打开浏览器
try:
webbrowser.open('http://localhost:3000')
print("🌐 已自动打开浏览器")
except Exception as e:
print(f"⚠️ 无法自动打开浏览器: {e}")
print("请手动访问: http://localhost:3000")
# 启动服务器
server.serve_forever()
except KeyboardInterrupt:
print("\n🛑 Dashboard服务器已停止")
except Exception as e:
print(f"❌ Dashboard服务器启动失败: {e}")
def check_dependencies():
"""检查依赖"""
print("🔍 检查依赖...")
required_modules = ['fastapi', 'uvicorn', 'pydantic']
missing_modules = []
for module in required_modules:
try:
__import__(module)
except ImportError:
missing_modules.append(module)
if missing_modules:
print(f"❌ 缺少依赖模块: {', '.join(missing_modules)}")
print("请运行: pip install fastapi uvicorn pydantic")
return False
print("✅ 依赖检查通过")
return True
def main():
"""主函数"""
print("🧠 MemOS Dashboard 启动器")
print("=" * 50)
# 检查依赖
if not check_dependencies():
sys.exit(1)
# 检查虚拟环境
venv_path = Path(__file__).parent / "venv"
if venv_path.exists():
print("✅ 检测到虚拟环境")
else:
print("⚠️ 未检测到虚拟环境，建议使用虚拟环境")
print("\n🚀 启动服务...")
try:
# 在单独线程中启动Dashboard服务器
dashboard_thread = threading.Thread(target=start_dashboard_server, daemon=True)
dashboard_thread.start()
# 在主线程中启动API服务器
start_api_server()
except KeyboardInterrupt:
print("\n🛑 服务已停止")
print("👋 感谢使用MemOS Dashboard!")
except Exception as e:
print(f"❌ 启动失败: {e}")
sys.exit(1)
if __name__ == "__main__":
main()