"""WerewolfMind:LLM 角色脑,实现引擎那套 DecisionProvider 接口(无缝替下 RandomProvider)。 设计延续平台理念:人设 + 如实状态(对局板)+ 软引导,少规则、让博弈智能涌现; 但比命案局更吃逻辑与欺骗。每个决策点 = 一次结构化输出调用;失败/越界 → 随机兜底,世界不停摆。 模型工厂 make_model 注入(真实=lc_adapter.build_chat_model;测试=FakeModel),便于零真实调用单测。 """ from __future__ import annotations import json import random import re import threading import time from pathlib import Path from typing import Callable, Optional from langchain_core.messages import HumanMessage, SystemMessage from genesis.obs.logging_setup import get_logger from genesis.werewolf import perception from genesis.werewolf import schema as SC from genesis.werewolf.boardmemory import BoardMemory from genesis.werewolf.engine import RandomProvider from genesis.werewolf.recall import lookup_transcript from genesis.werewolf.roles import Role from genesis.werewolf.state import GameState logger = get_logger("werewolf.brain") WIN_RULE = ("9 人预女猎·屠边:狼方杀光所有神 或 所有民 即胜;好人放逐所有狼即胜。") # 全局限并发 LLM 调用:并行投票/复盘会同时打 9 路,撞 API 并发限流→单次飙到 60-90s。 # 限 4 路后单次回落到正常区间;真人等 UI 不走这里,故不占名额。 _LLM_SEM = threading.Semaphore(4) # 发言/决策时回看的公开事件条数:须能容下"满人一整轮发言(最多 9 段)+ 本轮法官播报 + 一点上一轮", # 否则后发言者看不到先发者(尤其首日上警竞选 + 白天发言两段叠加),跨人推理断档 → 显得"没逻辑/不接茬"。 TRANSCRIPT_TAIL = 30 def load_personas(world_dir: str | Path) -> dict[int, str]: """从世界目录读 personas.json → {seat: 人设文本};无文件或解析失败则返回空(退回无人设)。""" p = Path(world_dir) / "personas.json" if not p.exists(): return {} try: raw = json.loads(p.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError) as e: logger.warning("读取座位人设失败:%s", str(e)[:120]) return {} out: dict[int, str] = {} for k, v in raw.items(): if str(k).isdigit() and isinstance(v, str): # 跳过 _comment 等非座位键 out[int(k)] = v return out class WerewolfMind: """某座位的 AI 大脑。维护自己的主观对局板;每次决策喂入【人设 + 如实对局板 + 本次任务】。""" def __init__(self, seat: int, role: Role, roles: dict[str, Role], make_model: Callable, *, make_think_model: Optional[Callable] = None, persona: str = "", rng: Optional[random.Random] = None, retries: int = 2) -> None: self.seat = seat self.role = role self.roles = roles self.make_model = make_model self.make_think_model = make_think_model # 思考模式模型(发言/复盘用,质量优先;不给则与 make_model 同) self.persona = persona self.retries = retries # 结构化输出失败时的重试次数(治"LLM偶发失败→傻瓜兜底发言") self.board = BoardMemory(seat) self.fallback = RandomProvider(rng or random.Random(seat)) self.stream_sink = None # 设了则白天发言走流式:每来一段 token 调 sink(seat, 累计文本)→ 网页实时显示 def _system(self, structured: bool = True) -> str: team = "狼人阵营" if self.role.team == "wolf" else "好人阵营" # 身份强反提示:实测 agent 会把自己说成要查/投/怀疑的对象(2号想听2号发言、9号说查杀9号)→ 反复钉死"你是N号" parts = [f"你是 {self.seat} 号玩家,你的身份是【{self.role.name}】({team})。" f"全程牢记你就是 {self.seat} 号本人——绝不会查验/投票/怀疑/带走你自己;提到别人时一律用对方的座位号,别把自己的号说成别人。"] if self.persona: parts.append("你的性子(用它决定说话风格):" + self.persona) if self.role.persona_hint: parts.append(self.role.persona_hint) parts.append("你在玩 9 人狼人杀。" + WIN_RULE) if structured: parts.append("凭你的身份、性子与立场,基于下面如实的对局信息做决策。只输出结构化结果。") else: # 软引导(不硬编码行为):口语化+有思考+有套路+不耿直+简洁;狼队配合 parts.append( "发言时像真人一样自然口语:可以有停顿、迟疑、『嗯…』『我想想』、改口、带口头禅和情绪,别像念稿。" "高手不耿直——会铺垫、试探、藏信息、找时机、配合队友,而不是一上来就摊牌或说『我不跳预言家』这种没信息量的废话。" "该悍跳、对跳、隐藏实力、结盟、带节奏、撒谎就来(尤其你若是狼:和狼队口径一致、互相抬轿、别公开拆队友的台)。" "每次只说有信息量或有目的的一段话,只说这段话本身,不要前缀/解释。") # 反捏造:可以伪装/撒谎,但不能断言自己根本无从得知的客观事实(最常见翻车=首夜凭空说"X是预言家") parts.append("注意:你只能基于『你确实掌握到的信息』推理。可以伪装身份、可以撒谎带节奏," "但别把『纯靠猜的』说成『我知道的事实』——尤其首夜你对别人的身份一无所知,不要凭空断言谁是预言家/女巫/神。") return "\n".join(parts) def _ask(self, schema, task: str, state: GameState, extra: str = "", think: bool = False): """渲染对局板 + 本次任务 → 结构化输出;失败重试 self.retries 次,仍失败才返回 None(随机兜底)。 think=True 用思考模式模型(发言/复盘,质量优先)。重试是稳定性关键:治"LLM偶发失败→3号发言这种傻瓜兜底"。""" user = (self.board.render(perception.view(state, self.seat, self.roles), transcript_tail=TRANSCRIPT_TAIL) + "\n\n" + task + extra) msgs = [SystemMessage(content=self._system()), HumanMessage(content=user)] make = self.make_think_model if (think and self.make_think_model) else self.make_model name = getattr(schema, "__name__", "?") for attempt in range(self.retries + 1): try: # method="function_calling":MiniMax 经 ChatLiteLLM 的默认结构化会吐 YAML 式文本导致解析失败,原生函数调用才稳。 with _LLM_SEM: # 限并发,避免并行投票/复盘撞 API 限流 out = make().with_structured_output(schema, method="function_calling").invoke(msgs) if out is not None: return out logger.warning("[%d号 %s] 第%d次返回空", self.seat, name, attempt + 1) except Exception as e: logger.warning("[%d号 %s] 第%d次失败:%s", self.seat, name, attempt + 1, str(e)[:120]) if attempt < self.retries: time.sleep(0.8 * (attempt + 1)) # 退避后重试(瞬时网络/限流抖动多能自愈) logger.warning("[%d号 %s] 重试 %d 次仍失败,随机兜底", self.seat, name, self.retries + 1) return None # ───────── DecisionProvider 接口(与引擎契约一致)───────── def wolf_discuss(self, state, seat, candidates): firstnight = ("今天是首夜,你们对所有人的身份一无所知——按直觉/座位/谁威胁大盲刀即可," "别在理由里编造『他是预言家』这种你根本没法知道的话。" if state.day <= 1 else "") out = self._ask(SC.WolfDiscussOut, f"天黑,你在狼队私聊频道里和队友商量今晚刀谁(你说的只有狼队友看得到;前面队友的提议见上方)。" f"{firstnight}可刀:{candidates}。给出你提议的 target 和一句给队友的 message。", state) if out is None or out.target not in candidates: return self.fallback.wolf_discuss(state, seat, candidates) return {"target": out.target, "message": out.message} def decide_kill(self, state, seat, candidates): out = self._ask(SC.NightKillOut, f"天黑了,和狼队商定今晚刀谁。可刀对象:{candidates}。", state) return out.target if (out and out.target in candidates) else self.fallback.decide_kill(state, seat, candidates) def decide_check(self, state, seat, candidates): out = self._ask(SC.NightCheckOut, f"你要查验一人验明好坏。可查:{candidates}。", state) return out.target if (out and out.target in candidates) else self.fallback.decide_check(state, seat, candidates) def decide_witch(self, state, seat, knife, can_save, can_poison, poison_candidates): kn = f"今夜 {knife} 号倒在刀下。" if knife is not None else "今夜你未获知刀型。" out = self._ask(SC.WitchOut, f"{kn}解药可用={can_save},毒药可用={can_poison}。决定救不救、毒不毒(可毒:{poison_candidates})。", state) if out is None: return self.fallback.decide_witch(state, seat, knife, can_save, can_poison, poison_candidates) poison = out.poison_target if out.poison_target in poison_candidates else None return {"save": bool(out.save), "poison": poison} def decide_run_sheriff(self, state, seat): out = self._ask(SC.RunSheriffOut, "警长竞选:是否上警竞选警徽(1.5 票、可定发言顺序)?", state) return bool(out.run) if out else self.fallback.decide_run_sheriff(state, seat) def decide_speech_dir(self, state, seat): out = self._ask(SC.SpeechDirOut, "你当选警长,握有发言定序权:决定本轮从你左手边(cw,座位顺位)还是右手边(ccw,逆位)起依次发言," "你自己将压轴最后发言。把你最想压制/想后置观察的可疑位置安排到对你最有利的发言序。", state) d = out.direction if (out and out.direction in ("cw", "ccw")) else "cw" return {"direction": d} def decide_withdraw(self, state, seat): out = self._ask(SC.WithdrawOut, "竞选发言结束,进入退水环节:综合场上局势与你的目的,是否退水(退出警长竞选)?" "(若你是悍跳/警上无优势/想把警徽让给队友等,可考虑退;否则不退)", state) return bool(out.withdraw) if out else False def decide_badge_pass(self, state, seat, candidates): out = self._ask(SC.BadgePassOut, f"你(警长)即将出局,临终可把警徽移交给一名存活玩家或撕毁。可移交:{candidates}。" "移交给你最信任、对好人/你阵营最有利的人(撕毁填 null)。", state) if out and out.target in candidates: return out.target return None def decide_explode(self, state, seat): out = self._ask(SC.ExplodeOut, "轮到你发言。你是狼,可选择自爆(掀桌亮身份,中止今天发言投票直接入夜)。" "自爆是极强但极少用的手段——只在能打断好人关键归票/查杀传递、或保住更重要队友时才用;" "通常不自爆,正常发言带节奏即可。现在要自爆吗?", state) return bool(out.explode) if out else False def speak(self, state, seat): v = perception.view(state, self.seat, self.roles) extra = "" if v.day >= 2 and len(v.transcript) > 12: # 历史够多才值得调阅原文(省调用):先问要不要查 rec = self._ask(SC.RecallOut, "发言前,你想调阅哪段原始记录来佐证判断?给一个 query(某座位号 / 'day:N' / 关键词);不需要就留空。", state) if rec and (rec.query or "").strip(): hit = lookup_transcript(v, rec.query) if hit: extra = "\n\n【你刚调阅的原始记录】\n" + hit # 发言走纯文本生成(非结构化):MiniMax 的 function_calling 对 SpeechOut 这种多字段 schema 会偶发吐空 {} # → 校验失败 → 兜底"N号发言"且拖慢;纯文本 invoke 无函数调用脆弱性,稳且快。重试 self.retries 次。 for attempt in range(self.retries + 1): text = self._speak_text(state, extra) if text: return text if attempt < self.retries: time.sleep(0.5 * (attempt + 1)) logger.warning("[%d号] 发言重试 %d 次仍空,随机兜底", self.seat, self.retries + 1) return self.fallback.speak(state, seat) def _speak_text(self, state, extra: str): """纯文本生成一段发言(非结构化,鲁棒);失败/空返回 None。""" # 越往后越要简洁(优化4:第二轮起发言太长);首日信息少可略展开 brevity = ("现在是首日,信息少,可以略作铺垫。" if state.day <= 1 else "已是第二天之后,务必言简意赅、抓重点,一两句到一小段就好,别长篇复述别人说过的话。") user = (self.board.render(perception.view(state, self.seat, self.roles), transcript_tail=TRANSCRIPT_TAIL) + f"\n\n轮到你({self.seat}号)发言。以你的身份和立场,像真人一样自然口语地说" "(可带停顿/迟疑/口头禅/情绪,别念稿;有套路、有目的,别耿直摊牌或说废话)。" + brevity + "过身份/报查验/站边/归票/试探/隐藏都行。只输出这段话本身,不要任何前缀、格式或解释。" + extra) try: with _LLM_SEM: out = self.make_model().invoke([SystemMessage(content=self._system(structured=False)), HumanMessage(content=user)]) return (getattr(out, "content", "") or "").strip() or None except Exception as e: logger.warning("[%d号] 发言生成失败:%s", self.seat, str(e)[:120]) return None def digest_day(self, state, seat): """每天收尾的记忆压缩(每个存活 agent 各跑一次;引擎 _digest_day 钩子触发)。 多回合、信息量大 → 双层压缩防遗忘+控长: ① key_facts = 铁的客观事实清单(死亡+死法/公开身份声明/对跳/查验/警长/票型异常),累积保留别丢旧的; ② notes = 滚动主观分析+计划,在旧 notes 上重写。 触发规则:每天收尾压一次(此处);常驻事实/票型/查验等硬信息另由 PlayerView 每回合如实下发,二者互补。 ★实现:走纯文本+宽松解析(非结构化)——实测 MiniMax 对 DayDigestOut 这种多字段 function_calling 首日约 40% 吐空/回显字段名,导致 agent 次日记忆全空;纯文本 invoke 无函数调用脆弱性,稳。失败重试。""" for attempt in range(self.retries + 1): d = self._digest_text(state) if d: self.board.apply_digest(d) return if attempt < self.retries: time.sleep(0.5 * (attempt + 1)) logger.warning("[%d号] 记忆压缩重试 %d 次仍失败,保留旧局记", self.seat, self.retries + 1) def _digest_text(self, state) -> Optional[dict]: """纯文本生成记忆压缩 → 宽松解析成 {key_facts,notes,prime_suspect,most_trusted};失败/空返回 None。""" old = "" if self.board.key_facts or self.board.notes: old = f"\n\n你已有的【事实】:{self.board.key_facts}\n你已有的【局记】:{self.board.notes}" user = (self.board.render(perception.view(state, self.seat, self.roles), transcript_tail=TRANSCRIPT_TAIL) + old + "\n\n今天结束,做记忆压缩。严格按下面 4 行纯文本输出(每行一项、冒号后写内容、没有就留空,不要任何别的话):\n" "事实:<铁的客观事实清单,累积旧的+今天新增——谁死了+死法(刀/毒/枪/放逐)、谁公开跳了什么身份/谁对跳、" "公开查验结果、警长是谁、票型异常;只记客观事实不下判断>\n" "分析:<基于事实,你这个身份/立场的好狼判断与下一步计划,在旧局记上重写,几句话>\n" "最疑:<你最怀疑是狼的座位号,或留空>\n" "最信:<你最信任的座位号,或留空>") try: with _LLM_SEM: out = self.make_model().invoke([SystemMessage(content=self._system(structured=False)), HumanMessage(content=user)]) text = (getattr(out, "content", "") or "").strip() except Exception as e: logger.warning("[%d号] 记忆压缩生成失败:%s", self.seat, str(e)[:120]) return None return self._parse_digest(text) @staticmethod def _parse_digest(text: str) -> Optional[dict]: """把『事实/分析/最疑/最信』四行纯文本宽松解析成 digest dict;至少要有事实或分析才算有效。""" labels = {"事实": "key_facts", "分析": "notes", "最疑": "prime_suspect", "最信": "most_trusted"} got: dict = {} for line in text.splitlines(): s = line.strip().lstrip("-*【 ").replace("】", "") for label, key in labels.items(): if s.startswith(label + ":") or s.startswith(label + "："): got[key] = s[len(label) + 1:].strip() break if not got.get("key_facts") and not got.get("notes"): return None for k in ("prime_suspect", "most_trusted"): m = re.search(r"\d+", got.get(k) or "") got[k] = int(m.group()) if m else None return got def decide_vote(self, state, seat, candidates): out = self._ask(SC.VoteOut, f"放逐投票:投谁出局?可投:{candidates}(弃票填 null)。", state) if out is None: return self.fallback.decide_vote(state, seat, candidates) return out.target if out.target in candidates else None def decide_hunter_shoot(self, state, seat, candidates): out = self._ask(SC.HunterShootOut, f"你(猎人)出局,可开枪带走一人。可带:{candidates}(放弃填 null)。", state) if out is None: return self.fallback.decide_hunter_shoot(state, seat, candidates) return out.target if out.target in candidates else None class MindProvider: """把每座位的 WerewolfMind 组装成引擎用的单一 DecisionProvider(按 seat 路由)。""" def __init__(self, minds: dict[int, WerewolfMind]) -> None: self.minds = minds def set_stream_sink(self, sink) -> None: """给所有 AI 脑装上流式发言回调(网页 live 用);不设则发言走非流式结构化。""" for m in self.minds.values(): m.stream_sink = sink def wolf_discuss(self, s, seat, c): return self.minds[seat].wolf_discuss(s, seat, c) def decide_kill(self, s, seat, c): return self.minds[seat].decide_kill(s, seat, c) def decide_check(self, s, seat, c): return self.minds[seat].decide_check(s, seat, c) def decide_witch(self, s, seat, k, cs, cp, pc): return self.minds[seat].decide_witch(s, seat, k, cs, cp, pc) def decide_run_sheriff(self, s, seat): return self.minds[seat].decide_run_sheriff(s, seat) def decide_speech_dir(self, s, seat): return self.minds[seat].decide_speech_dir(s, seat) def decide_withdraw(self, s, seat): return self.minds[seat].decide_withdraw(s, seat) def decide_badge_pass(self, s, seat, c): return self.minds[seat].decide_badge_pass(s, seat, c) def decide_explode(self, s, seat): return self.minds[seat].decide_explode(s, seat) def speak(self, s, seat): return self.minds[seat].speak(s, seat) def decide_vote(self, s, seat, c): return self.minds[seat].decide_vote(s, seat, c) def decide_hunter_shoot(self, s, seat, c): return self.minds[seat].decide_hunter_shoot(s, seat, c) def digest_day(self, s, seat): return self.minds[seat].digest_day(s, seat) @classmethod def for_game(cls, engine, make_model: Callable, personas: Optional[dict] = None, make_think_model: Optional[Callable] = None) -> "MindProvider": """按已发牌的引擎给每个座位建一个对应身份的脑。make_think_model:发言/复盘用的思考模式模型(可选)。""" personas = personas or {} minds = { seat: WerewolfMind(seat, engine.roles[engine.state.role_of(seat)], engine.roles, make_model, make_think_model=make_think_model, persona=personas.get(seat, ""), rng=random.Random(seat)) for seat in engine.state.roles } return cls(minds)