#!/usr/bin/env python3 """ EzVibe 项目初始化脚本 运行: python setup_project.py 功能: 自动创建项目目录树 + 各模块基础骨架（类定义 + 函数签名） """ import os import sys from pathlib import Path # ============================================================ # 项目根目录（相对于本脚本所在位置） # ============================================================ ROOT = Path(__file__).parent.resolve() MODULES = { "ui": ["pet_window"], "agent": ["brain", "emotion", "memory", "scheduler"], "api": ["server"], "perception": [], "data": [], } # ============================================================ # 每个模块的骨架模板 # key = 文件名（不含 .py） # value = (class_names: list[str], function_sigs: list[str], docstring: str) # ============================================================ SCAFFOLDS = { "pet_window": { "classes": ["PetWindow"], "funcs": [ "def init_ui(self) -> None: ...", "def update_emotion(self, emotion: str) -> None: ...", "def play_animation(self, anim_name: str) -> None: ...", "def show_reminder(self, message: str, priority: int = 0) -> None: ...", "def close(self) -> None: ...", ], "doc": "Qt 桌宠渲染与用户交互面板。", }, "brain": { "classes": ["AgentBrain"], "funcs": [ "def __init__(self, config: dict | None = None) -> None: ...", "async def think(self, user_input: str, context: dict | None = None) -> dict: ...", "async def generate_response(self, prompt: str, system_prompt: str = '') -> str: ...", "def decide_action(self, emotion: str, user_context: dict) -> dict: ...", ], "doc": "LLM 推理引擎 + 行为决策中心。", }, "emotion": { "classes": ["EmotionEngine"], "funcs": [ "def __init__(self, transition_matrix: list[list[float]] | None = None) -> None: ...", "def get_state(self) -> str: ...", "def update(self, event: str, context: dict | None = None) -> str: ...", "def _sample_next_state(self, current: str, event_boosts: dict | None = None) -> str: ...", "def get_emotion_display_name(self, state: str) -> str: ...", ], "doc": "情绪状态机：5 状态 {idle,happy,focused,annoyed,sleepy} + 概率转移矩阵。", }, "memory": { "classes": ["VectorMemory"], "funcs": [ "def __init__(self, storage_path: str = 'data/MEMORY.db', embedder: str = 'ollama') -> None: ...", "async def add(self, text: str, tags: list[str] | None = None, metadata: dict | None = None) -> str: ...", "async def search(self, query: str, top_k: int = 5) -> list[dict]: ...", "async def retrieve_context(self, query: str, top_k: int = 3) -> str: ...", "def cosine_similarity(self, a: list[float], b: list[float]) -> float: ...", "async def get_user_profile(self) -> dict: ...", ], "doc": "向量记忆系统：SQLite 存储 + RAG 语义检索。", }, "scheduler": { "classes": ["BehaviorScheduler"], "funcs": [ "def __init__(self, emotion_engine, brain) -> None: ...", "def register_behavior(self, name: str, action_fn: callable, priority: int = 1, cooldown: float = 60.0) -> None: ...", "def check_and_trigger(self, user_activity_level: float) -> list[dict]: ...", "def get_active_reminders(self) -> list[dict]: ...", "def clear_reminder(self, reminder_id: str) -> None: ...", "def _should_interrupt(self) -> bool: ...", ], "doc": "调度器：优先级控制 + 冷却时间 + 活跃度判断。", }, "server": { "classes": ["APIServer"], "funcs": [ "def __init__(self, host: str = '127.0.0.1', port: int = 8765) -> None: ...", "async def start(self) -> None: ...", "async def stop(self) -> None: ...", "async def ws_handler(self, websocket) -> None: ...", "async def broadcast(self, event: dict) -> None: ...", ], "doc": "FastAPI/WebSocket 服务：Agent Core ↔ Qt 前端 异步通信。", }, } def _cls_template(cls_name: str, funcs: list[str], doc: str) -> str: """生成单个类的骨架代码。""" func_bodies = "\n ".join(f"{f}" for f in funcs) return f'''class {cls_name}: """ {doc} """ def __init__(self) -> None: raise NotImplementedError("请在子类中实现 __init__") {func_bodies} ''' def _module_scaffold(module: str) -> str: """生成单个 Python 模块（.py）的骨架代码。""" if module not in SCAFFOLDS: return "" info = SCAFFOLDS[module] cls_blocks = "\n\n".join( _cls_template(cls, info["funcs"], info["doc"]) for cls in info["classes"] ) return cls_blocks def _init_py(package: str) -> str: """生成 __init__.py，暴露主类。""" if package not in SCAFFOLDS or not SCAFFOLDS[package]["classes"]: return "" main_cls = SCAFFOLDS[package]["classes"][0] return ( f'"""EzVibe {package} 模块。"""\n' f"from .{package} import {main_cls}\n" f"__all__ = ['{main_cls}']\n" ) def _write(path: Path, content: str) -> None: path.write_text(content, encoding="utf-8") print(f" [CREATE] {path}") def run() -> None: print(f"\n{'='*50}") print(f"EzVibe 项目初始化") print(f"根目录: {ROOT}") print(f"{'='*50}\n") # 1. 创建目录树 print("[1/3] 创建目录结构 ...") for pkg, files in MODULES.items(): pkg_dir = ROOT / pkg pkg_dir.mkdir(parents=True, exist_ok=True) for f in files: (pkg_dir / f).with_suffix(".py").touch() # 2. 写入骨架代码 for pkg, files in MODULES.items(): pkg_dir = ROOT / pkg # __init__.py if pkg in SCAFFOLDS and SCAFFOLDS[pkg].get("classes"): init_content = _init_py(pkg) _write(pkg_dir / "__init__.py", init_content) else: _write(pkg_dir / "__init__.py", '"""EzVibe package."""\n') # 模块文件 for fname in files: content = _module_scaffold(fname) _write(pkg_dir / f"{fname}.py", content) # 3. main.py print("\n[3/3] 生成 main.py ...") main_content = '''#!/usr/bin/env python3 """ EzVibe — AI 桌宠系统主入口 用法: python main.py """ import sys import asyncio from pathlib import Path # 确保项目根目录在 sys.path _ROOT = Path(__file__).parent.resolve() sys.path.insert(0, str(_ROOT)) from ui.pet_window import PetWindow from agent.brain import AgentBrain from agent.emotion import EmotionEngine from agent.memory import VectorMemory from agent.scheduler import BehaviorScheduler from api.server import APIServer class EzVibeApp: """ 应用生命周期管理。 负责初始化各层组件并启动 Qt 事件循环 + asyncio 事件循环。 """ def __init__(self) -> None: self.emotion_engine = EmotionEngine() self.memory = VectorMemory() self.brain = AgentBrain() self.scheduler = BehaviorScheduler( emotion_engine=self.emotion_engine, brain=self.brain, ) self.api_server = APIServer() # Qt 窗口在 run() 中延迟创建（必须在 QApplication 主线程） self.window: PetWindow | None = None async def async_init(self) -> None: """异步初始化：连接组件，启动 API 服务。""" await self.memory # 确保记忆库就绪 await self.api_server.start() print("[EzVibe] 系统就绪。") def run(self) -> None: """阻塞入口：启动 Qt 主循环。""" from PySide6.QtWidgets import QApplication app = QApplication(sys.argv) app.setApplicationName("EzVibe") self.window = PetWindow( emotion_engine=self.emotion_engine, brain=self.brain, scheduler=self.scheduler, ) self.window.show() # 用 QTimer 将 asyncio 事件循环嵌入 Qt timer = app.property("_aiotimer") if timer is None: from PySide6.QtCore import QTimer loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) t = QTimer(app) t.timeout.connect(lambda: loop.run_until_complete(asyncio.sleep(0))) t.setInterval(20) t.start() app.setProperty("_aiotimer", t) app.setProperty("_aioloop", loop) sys.exit(app.exec()) if __name__ == "__main__": print("EzVibe 启动中 ...") app = EzVibeApp() app.run() ''' _write(ROOT / "main.py", main_content) print(f"\n[OK] 项目初始化完成！") print(f"请运行: cd {ROOT} && pip install -r requirements.txt") print(f"启动: python main.py\n") if __name__ == "__main__": run()