e2hang
2a844e83a8
- EmotionEngine: 5状态马尔可夫情绪机 + 蒙特卡洛转移 - VectorMemory: TF-IDF向量记忆 + SQLite持久化 + RAG检索 - AgentBrain: Ollama/OpenAI/Dummy三后端LLM - BehaviorScheduler: 优先级/冷却/活跃度调度 - FastAPI服务器 + WebSocket实时推送 - perception: 键鼠监控 + 屏幕截图 - ui/pet_window: PySide6桌宠窗口 + 像素动画 - assets/pet: 5情绪各2帧像素艺术资源
269 lines
8.9 KiB
Python
269 lines
8.9 KiB
Python
#!/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()
|