"""Live v2 后端:WorldEngine(事件驱动运行时)+ HTTP 实时推流,接口兼容 mystery.html。 运行:GENESIS_WORLD=ravenisle ~/miniconda3/envs/genesis-world/bin/python -m genesis.server.live2 浏览器:http://127.0.0.1:8000/src/genesis/viz/web/mystery.html 当前为规则桩模式(零 LLM 成本):验证独立作息/逐跳移动/事件中断/聚集链路。 """ from __future__ import annotations import asyncio import json import os import threading from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, urlparse from genesis.manifest.builder import WorldBuilder from genesis.obs.logging_setup import get_logger from genesis.runtime.engine import WorldEngine logger = get_logger("server.live2") ROOT = Path(__file__).resolve().parents[3] WORLD_NAME = os.environ.get("GENESIS_WORLD", "ravenisle") WORLD_DIR = ROOT / "worlds" / WORLD_NAME STATE: dict = {"stage": None, "frame": None, "running": True, "step": 0} LOCK = threading.Lock() _engine: WorldEngine | None = None _llm = None # MeteredLLM | None,成本仪表全局引用(/api/cost、/api/state 读) def _stage(builder: WorldBuilder, viz: dict | None) -> dict: characters = [{ "name": cm.name, "age": cm.age, "mbti": cm.mbti, "occupation": cm.occupation, "backstory": getattr(cm, "backstory", ""), "sprite_dir": cm.art.sprite_path if cm.art else None, "life_goal": "", "sub_goals": [], } for cm in builder.characters.values()] places = [{"id": pid, "name": pm.name, "geography": pm.geography, "objects": [], "adjacent": [builder.places[a].name for a in pm.adjacent if a in builder.places]} for pid, pm in builder.places.items()] killer = next((cm.name for cm in builder.characters.values() if "凶手" in (getattr(cm, "role", "") or "")), None) return {"places": places, "characters": characters, "killer": killer, "viz": viz} def _engine_thread() -> None: async def run(): global _engine while True: _engine.start() frame_task = asyncio.create_task(_frames()) await _engine.sched.run(_engine.on_wake) frame_task.cancel() if not STATE.pop("_restart", False): break _engine = _build_engine(load="") # 重开一局:全新世界 with LOCK: STATE["step"] = 0 STATE["frame"] = _engine.frame(0) STATE["running"] = False # 重开也先冻结,等用户重新登岛才推进 _engine.clock.pause() logger.info("🔄 新一局开始(已冻结,等登岛)") async def _frames(): last_autosave = 0.0 import time as _time while True: await asyncio.sleep(1.0) if _time.monotonic() - last_autosave > 300: # 每 5 分钟自动存档(灭世保险) last_autosave = _time.monotonic() try: from genesis.runtime.persistence import save_to_file save_to_file(_engine, str(WORLD_DIR / "saves" / "autosave.json")) except Exception: logger.exception("自动存档失败") if not STATE["running"]: if not _engine.clock._paused: _engine.clock.pause() continue if _engine.clock._paused and _engine.ending is None: _engine.clock.resume() # 终局冻结不被帧循环误恢复 _engine.maybe_fast_forward() nar = _engine.narrator if nar is not None and nar.due(_engine.clock.sim_seconds): nar.busy = True now_s, label = _engine.clock.sim_seconds, _engine.clock.label() pano = _engine.panorama() async def _write(): try: await asyncio.to_thread(nar.write, now_s, label, pano) finally: nar.busy = False asyncio.create_task(_write()) with LOCK: STATE["step"] += 1 STATE["frame"] = _engine.frame(STATE["step"]) asyncio.run(run()) class Handler(SimpleHTTPRequestHandler): def __init__(self, *a, **k): super().__init__(*a, directory=str(ROOT), **k) def log_message(self, *a): pass def _json(self, obj, code=200): body = json.dumps(obj, ensure_ascii=False).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_GET(self): path = urlparse(self.path).path if path == "/favicon.ico": # 浏览器自动请求,我们没图标 → 静默 204(免 404 噪音) self.send_response(204) self.end_headers() return if path == "/api/state": cost = _llm.stats() if _llm is not None else None # MeteredLLM 自带锁,LOCK 外读安全 with LOCK: self._json({"stage": STATE["stage"], "frame": STATE["frame"], "running": STATE["running"], "step": STATE["step"], "cost": cost}) return if path == "/api/cost": self._json(_llm.stats() if _llm is not None else {}) return if path == "/api/latency": # LLM 单次调用耗时统计(定位思索长尾) from genesis.runtime.cognition import latency_stats self._json(latency_stats()) return if path == "/api/agent": q = parse_qs(urlparse(self.path).query) self._json(_engine.agent_detail(q.get("name", [""])[0])) return super().do_GET() def do_POST(self): p = urlparse(self.path) q = parse_qs(p.query) if p.path == "/api/pause": STATE["running"] = False; self._json({"running": False}) elif p.path == "/api/resume": STATE["running"] = True; self._json({"running": True}) elif p.path == "/api/rate": r = float(q.get("x", ["30"])[0]); _engine.set_base_rate(r); self._json({"rate": r}) elif p.path == "/api/save": from genesis.runtime.persistence import save_to_file name = q.get("name", ["manual"])[0] path = save_to_file(_engine, str(WORLD_DIR / "saves" / f"{name}.json")) self._json({"ok": True, "path": str(path)}) elif p.path == "/api/restart": STATE["_restart"] = True _engine.sched.stop() self._json({"ok": True}) elif p.path == "/api/event": n = _engine.inject_event(q.get("desc", [""])[0], q.get("place", [""])[0]) self._json({"ok": True, "reached": n}) elif p.path == "/api/weather": _engine.set_weather(q.get("w", ["暴雨"])[0]); self._json({"ok": True}) elif p.path == "/api/goal": _engine.set_world_goal(q.get("text", [""])[0]); self._json({"ok": True}) elif p.path == "/api/whisper": ok = _engine.whisper(q.get("name", [""])[0], q.get("text", [""])[0]) self._json({"ok": ok}) else: self.send_error(404) def _build_engine(load: str) -> WorldEngine: """构建一局世界(启动与 /api/restart 共用);load 为存档路径或空=全新。""" brain = os.environ.get("GENESIS_BRAIN", "llm") # llm | stub engine = WorldEngine(WORLD_DIR, rate=float(os.environ.get("GENESIS_RATE", "40")), # 七天版:40×平衡(夜间自动快进保证可看;过高会让决策频率翻倍、思索观感变重);可用 GENESIS_RATE 调 brain=brain, llm=_llm, prologue=not load) # 新开局带登岛序幕 if _llm is not None: from genesis.runtime.narrator import Narrator nlore = getattr(engine, "_lore", {}) or {} # 叙事层解耦:说书人世界设定来自 world.json 的 lore engine.narrator = Narrator( _llm, roster=list(engine.minds.keys()), novel_title=nlore.get("novel_title", "《雾鸦》"), ghost=nlore.get("ghost", "阮青"), opening=nlore.get("narrator_opening")) engine.bus.taps.append(engine.narrator.tap) from genesis.runtime.longterm import LongTerm engine.longterm = LongTerm(engine.game_id) # mem0 长期记忆(G6,反思节拍专用) if load: from genesis.runtime.persistence import load_from_file path = WORLD_DIR / "saves" / "autosave.json" if load == "auto" else Path(load) if path.exists(): load_from_file(engine, str(path)) print(f"⏪ 已从存档恢复:{path}({engine.clock.label()})") else: print(f"⚠️ 存档不存在,新开世界:{path}") return engine def main() -> None: global _engine, _llm if os.environ.get("GENESIS_BRAIN", "llm") != "stub": # llm / langgraph 都需 LLM 客户端(后者亦用其做兜底/反思) from genesis.llm.client import LiteLLMClient from genesis.runtime.metering import MeteredLLM _llm = MeteredLLM(LiteLLMClient()) _engine = _build_engine(os.environ.get("GENESIS_LOAD", "")) viz = json.loads((WORLD_DIR / "viz.json").read_text(encoding="utf-8")) builder = WorldBuilder.from_directory(str(WORLD_DIR)) STATE["stage"] = _stage(builder, viz) STATE["frame"] = _engine.frame(0) STATE["running"] = False # 开局先冻结:用户看开场白/登岛之前,sim 不许推进(否则放映室动画会插进来) _engine.clock.pause() # 立即冻结时钟,杜绝线程起步那一瞬的抢跑(否则前1秒已推进数十 sim 秒) threading.Thread(target=_engine_thread, daemon=True).start() addr = ("127.0.0.1", 8000) print(f"Live v2 世界[{WORLD_NAME}]已启动(brain={os.environ.get('GENESIS_BRAIN', 'llm')})→ http://{addr[0]}:{addr[1]}/src/genesis/viz/web/mystery.html") ThreadingHTTPServer(addr, Handler).serve_forever() if __name__ == "__main__": main()