"""仿真线(E2E 整夜仿真):真 WorldEngine + FakeTime + 剧本脑,几秒钟跑完一整夜。 定位:舞台机械的整车下线检测——验证「当心智自由地做出决定时,世界忠实响应; 心智需要知道的真相,送达每一种思考模式」。不测角色像不像人(那是真 LLM 的事)。 架构: - ScriptedBrain 站在 llm.complete 的位置:**真实的 prompt 组装/JSON 解析链路全部走到**, 它收到的就是生产环境的 prompt(并全部留底供送达断言); - 决策由 policy 白盒读引擎状态产生(上帝视角),确定性强、不靠正则扒 prompt; - FakeTime 拨快时钟,8.5 模拟小时 ≈ 数秒真实时间。 """ from __future__ import annotations import json import re NIGHT_ANCHORS = { # 游走锚点:堡内(夜话15分钟能赶到门厅)且各有独处点(贴近真实局里各奔私罪任务) "顾长风": ["暗房", "书房"], # 暗房=独处点(找母带) "邵铭轩": ["书房", "大客厅"], "程亦深": ["大客厅", "餐厅"], # 黏人,基本不独处 "唐曼": ["唐曼的房间", "二层走廊"], # 房间=独处点 "白聿": ["白聿的房间", "餐厅"], # 凶手不杀他 "温言": ["酒窖", "书房"], # 酒窖=独处点 "江离": ["琴房", "道具间"], } def _decision(thought="", intent="", speech=None, action=None, minutes=6, goal_update=None, leave=None): """拼一份生产协议的决策 JSON(L2 或 L1)。""" if leave is not None: # L1 对话协议 return json.dumps({"thought": thought, "speech": speech or "嗯,先这样。", "gesture": "", "leave": leave}, ensure_ascii=False) d = {"thought": thought, "intent": intent or thought, "speech": ({"to": speech[0], "text": speech[1]} if speech else None), "gesture": "", "action": ({"kind": action[0], "arg": action[1]} if action else None), "minutes": minutes, "goal_update": goal_update} return json.dumps(d, ensure_ascii=False) HUNT_VENUES = ["储藏室", "暗房", "酒窖"] # 凶手按幕把猎物引向的僻静处(地下,移动快,僻静可独处) class ScriptedBrain: """剧本脑:实现 llm.complete 契约;按提示词类型路由到对应 policy。""" def __init__(self): self.engine = None # 引擎建好后回填 self.prompts: list[tuple] = [] # (name, kind, system, user) 全量留底 self._step: dict[str, int] = {} # name → 游走步数(确定性轮换锚点) self.hunt = None # 进行中的狩猎:(victim, venue)——凶手引诱猎物赴僻静处 # ── llm 契约 ── def complete(self, messages, **kw): sys_p, user_p = messages[0]["content"], messages[-1]["content"] m = re.match(r"你是(.+?)。", sys_p) name = m.group(1) if m else "?" if "执笔者" in sys_p: # 说书人(本仿真不接) self.prompts.append((name, "narrator", sys_p, user_p)) return '{"prose": "夜雨如注。", "summary": "略"}' if "私人目标" in user_p: # 目标自举 self.prompts.append((name, "bootstrap", sys_p, user_p)) return '{"goals": ["完成今晚必须办的事", "活到天亮"]}' if "沉淀出判断" in user_p: # 反思 self.prompts.append((name, "reflect", sys_p, user_p)) return '{"insights": [], "relations": []}' if "你正在交谈中" in sys_p: # L1 对话 self.prompts.append((name, "dialogue", sys_p, user_p)) return self._dialogue(name, user_p) self.prompts.append((name, "l2", sys_p, user_p)) # L2 决策 return self._l2(name, user_p) # ── L1:简短接话,凶手被钟点催促/聊满两轮就收场 ── def _dialogue(self, name, user_p): eng = self.engine sess = eng._session_of.get(name) rounds = sess.rounds if sess else 0 # 狩猎/赴约中的人(凶手或被引诱的猎物)绝不被闲聊缠住:一句话脱身,各赴各的局 hunting = (name == eng.killer_name and "剧本的钟点" in user_p) \ or bool(self.hunt and self.hunt[0] == name) return _decision(thought="接话", speech="嗯,先这样,回头再说。", leave=bool(hunting or rounds >= 4)) # ── L2:夜话 > 凶手狩猎 > 黏人/游走 ── def _l2(self, name, user_p): eng = self.engine pos = eng.state.positions here = pos.get(name, "门厅") a = eng.assembly # 1) 夜话/对质压倒一切 if a is not None: if a["phase"] == "summon" and here != "门厅": return _decision(thought="去门厅", action=("move", "门厅")) if a["phase"] == "vote" and name in a.get("attendees", []) and name not in a["votes"]: target = self._vote_for(name) return _decision(thought="表态", action=("accuse", target)) return _decision(thought="等对质", action=("idle", ""), minutes=4) # 1.5) 被凶手引诱的猎物:乖乖赴约去僻静处(模拟"我知道母带在X"的引诱) if self.hunt and self.hunt[0] == name and name not in eng.state.dead: venue = self.hunt[1] if here != venue: return _decision(thought="赴约", action=("move", venue)) return _decision(thought="到了", action=("idle", ""), minutes=2) if name == eng.killer_name: return self._killer(name, here, user_p) # 2) 程亦深黏人(人设):隔次搭话,留出独处窗口;凶手被拉入时靠对话压力脱身(测对话穿透) others = [x for x in eng.state.occupants(here) if x != name and x not in eng.state.dead] if name == "程亦深" and others and eng._session_of.get(name) is None: cnt = self._step["程亦深_t"] = self._step.get("程亦深_t", 0) + 1 if cnt % 2 == 0: return _decision(thought="搭话", speech=(others[0], "这鬼天气,陪我说说话。"), action=None, minutes=4) # 3) 确定性游走:锚点轮换,走一步歇一步 step = self._step[name] = self._step.get(name, 0) + 1 anchors = NIGHT_ANCHORS.get(name, ["门厅"]) if step % 2: dest = anchors[(step // 2) % len(anchors)] if dest != here: return _decision(thought="走动", action=("move", dest)) return _decision(thought="观察", action=("idle", ""), minutes=5) def _killer(self, name, here, user_p): eng = self.engine pressured = "复仇的进度" in user_p # 逐日复仇压力(七天版)穿透与否在此被验证 if not pressured: # 平时:演无害的编剧,四处采风布置 step = self._step[name] = self._step.get(name, 0) + 1 dest = NIGHT_ANCHORS[name][step % 2] return (_decision(thought="布置", action=("move", dest)) if dest != here else _decision(thought="采风", action=("idle", ""), minutes=5)) # 狩猎:锁定一个非白聿的活人,引诱去僻静处,会合即落幕(全走真实 move/kill 链路) if self.hunt is None: done = len([d for d in eng.state.dead if d not in eng.executed]) # 选好引诱的猎物:非白聿、活着、且当下不在会话里(会话中的人在 L1 模式,收不到引诱) victim = next((v for v in ("唐曼", "邵铭轩", "温言", "顾长风", "程亦深") if v != "白聿" and v not in eng.state.dead and eng._session_of.get(v) is None), None) if victim is None: return _decision(thought="无人可下手", action=("idle", ""), minutes=3) self.hunt = (victim, HUNT_VENUES[min(done, len(HUNT_VENUES) - 1)]) victim, venue = self.hunt if victim in eng.state.dead: self.hunt = None return _decision(thought="收敛", action=("idle", ""), minutes=3) if here != venue: return _decision(thought="去取景地候着", action=("move", venue)) occ = [x for x in eng.state.occupants(venue) if x not in eng.state.dead] if victim in occ and len([x for x in occ if x != name]) == 1: self.hunt = None return _decision(thought="落幕", action=("kill", victim), minutes=4) return _decision(thought="等他赴约", action=("idle", ""), minutes=2) def _vote_for(self, name): """投票策略(七天版):血案尚少时分裂以测『无果继续』;血案累积(≥2)幸存者合力指认真凶。""" eng = self.engine killer = eng.killer_name deaths = len([d for d in eng.state.dead if d not in eng.executed]) if deaths < 2: # 信息不足:分裂 → 无过半散场 if name == "顾长风": return killer # 只有他咬对了 if name == killer: return "顾长风" # 凶手反咬嫁祸 return "" # 其余弃权 return "顾长风" if name == killer else killer # 血案累累:合力指认真凶 async def run_night(until_hour: float = 29.6, max_events: int = 200000): """跑一整夜:返回 (engine, brain, milestones)。 **纯事件驱动**(不并发跑 sched.run、不依赖真实时间):时钟逐个跳到下一个调度事件, 同步抽干该时刻的到期任务。这保证 100% 确定性——同代码同输出,回归测试的地基。 """ import heapq from genesis.runtime.engine import WorldEngine from tests.test_runtime2 import FakeTime ft = FakeTime() brain = ScriptedBrain() eng = WorldEngine("worlds/ravenisle", rate=600.0, time_fn=ft, brain="llm", llm=brain) brain.engine = eng accusations = [] # 当众指认事件留底 eng.bus.taps.append(lambda e: accusations.append(e.content) if "当众指认" in e.content or "弃权" in e.content else None) eng.start() clk, heap = eng.clock, eng.sched._heap n = 0 while heap and eng.ending is None and n < max_events: next_t = heap[0][0] if next_t > until_hour * 3600.0: break ft.t = clk._anchor_real + (next_t - clk._anchor_sim) / clk.rate # 跳到事件时刻(确定) while heap and heap[0][0] <= clk.sim_seconds + 1e-6: _, _, aid, reason = heapq.heappop(heap) await eng.on_wake(aid, reason) n += 1 def _h(label): # "第1天 22:33 深夜" → 连续小时 m = re.search(r"第(\d+)天 (\d+):(\d+)", label) return (int(m.group(1)) - 1) * 24 + int(m.group(2)) + int(m.group(3)) / 60 if m else 0 kills = [(n["text"], _h(n["t"]), n.get("place", "")) for n in eng.narrative if "☠" in n["text"]] courts = [(n["text"], _h(n["t"])) for n in eng.narrative if "🕯" in n["text"]] milestones = {"kills": kills, "courts": courts, "accusations": accusations, "ending": eng.ending, "executed": list(eng.executed), "end_hour": eng.clock.sim_hours} return eng, brain, milestones