跳转至

voice-werewolf

第10章 · 多 Agent 协作 · 配套项目 chapter10/voice-werewolf

项目说明

实验 10-8:语音狼人杀 Agent 系统

配套《深入理解 AI Agent》第 10 章「实验 10-8:语音狼人杀 Agent 系统」。

目的

用一个多 Agent 狼人杀系统,演示多 Agent 协作里最能体现「上下文不共享」的一种 形态——信息权限控制(Information Asymmetry):不同角色天生只能看到自己该看 的信息。系统包含三部分:

  1. 多 Agent:每个玩家 = 一个独立的 LLM Agent(Chat Completions,默认 gpt-5.6-luna),各自维护严格隔离的私有上下文,只能依据自己看到的信息推理。
  2. 信息权限控制:由法官决定每条信息投递进哪些玩家 Agent 的上下文——狼人 才知道队友、预言家才知道查验结果、公开发言进所有人。每次投递都登记审计,游戏 结束后打印审计表并自动校验隔离是否正确。
  3. 法官编排:法官是代码驱动(非 LLM)的确定性编排器,编排昼夜循环并结算胜负。

语音是可选增强,不是跑通的必需。 默认「文本模式」即可完整、可复现地跑完 一局并验证信息隔离;加 --voice 才会用 OpenAI tts-1 把公开发言合成语音。

离线模式(--offline / --mock):无需任何 API Key、零成本、完全可复现。 此时每个玩家 Agent 用一套规则策略代替 LLM 做发言/投票/用技能的决策。关键在于: 离线策略同样只读自己的私有上下文memory),绝不访问其他 Agent 的信息—— 因此信息权限控制这一核心教学点在离线模式下依然成立、依然能通过审计校验。 想零成本先看清楚「法官编排 + 信息隔离」的全貌,用 python demo.py --offline 即可。

本 demo 为控成本默认采用 7 人局(全部由 AI 扮演,保证可复现可验证): 2 狼人 + 1 预言家 + 1 女巫 + 3 村民;也可用 --players / --wolves 自定义。

角色与信息权限矩阵

信息类别 谁能看到(进入谁的上下文) 说明
自己的身份 仅本人 每人只知道自己是什么角色
狼人队友身份 仅全体狼人 狼人互相知道队友,好人不知道任何人的身份
狼人夜间共识 仅全体狼人 今晚决定刀谁,只进狼人上下文
预言家查验结果 仅预言家本人 每晚查验一人的阵营,结果独享
女巫夜间信息 / 用药 仅女巫本人 今晚谁被刀、是否用解药/毒药,独享
死讯(天亮公布) 所有人 公开信息
白天发言 所有人 公开信息,进入所有玩家上下文
投票与放逐结果 所有人 公开信息,放逐时公布出局者身份

「上帝视角」的真实身份表只打印给人类观察,不进入任何 Agent 上下文

法官编排(昼夜循环)

阶段0 分配身份:私发每人身份;仅向狼人群发「队友身份」
每回合:
  夜晚:狼人共同选择击杀目标(狼人共识只进狼人上下文)
        → 预言家查验一人(结果只进预言家上下文)
        → 女巫决定是否解药救人 / 毒药毒人(只进女巫上下文)
        → 结算今晚死亡
  判定胜负;未分胜负则进入白天
  白天:公布死讯(公开)
        → 按座位顺序依次公开发言(进所有人上下文)
        → 全员投票放逐,公布出局者身份(公开)
  判定胜负
胜负规则:狼人全部出局 → 好人胜;狼人数 ≥ 好人数 → 狼人胜(屠边简化)

运行

pip install -r requirements.txt
cp env.example .env        # 填入 OPENAI_API_KEY;或直接 export OPENAI_API_KEY=sk-...

# 离线模式:无需 API Key,规则决策,零成本、可复现,最适合先跑通看全貌
python demo.py --offline

# 在线模式(LLM 决策,需 OPENAI_API_KEY)
python demo.py             # 文本模式跑完整一局(默认,推荐)
python demo.py --seed 7    # 换一局身份分布(可复现)
python demo.py --voice     # 额外用 OpenAI tts-1 把公开发言合成语音到 audio/
python demo.py --voice --play   # 合成并播放(macOS afplay)
python demo.py --model gpt-5.6-luna   # 覆盖模型

# 通用参数(离线/在线均可)
python demo.py --offline --players 9 --wolves 3   # 自定义人数与狼人数
python demo.py --offline --max-rounds 8           # 调整回合上限
python demo.py --offline --log game.log           # 把完整对局日志另存到文件

python demo.py --help      # 查看全部参数(中文说明)

运行会依次打印:每个阶段的流程(带「谁能看到什么」的信息隔离标注)→ 最终胜负 → 信息可见性审计表信息隔离自动校验(对照展示狼人/村民/预言家三方各自的 私有上下文,证明各看各的)。

文件说明

voice-werewolf/
├── demo.py              # 入口:跑完整一局 + 打印审计表 + 自动校验信息隔离
├── werewolf/
│   ├── roles.py         # 角色、阵营、各角色策略提示词
│   ├── agent.py         # PlayerAgent:每玩家一个 Agent + 私有上下文 + LLM/离线规则决策
│   ├── game.py          # Judge:法官编排 + 信息投递原语(含审计登记)
│   ├── audit.py         # 信息可见性审计日志
│   └── tts.py           # 可选语音合成(OpenAI tts-1)
├── requirements.txt
├── env.example
└── audio/               # --voice 时合成的语音文件

信息隔离的落点在 game.py 的三个投递原语:broadcast(进所有人)、 private_send(只进一人)、wolves_send(只进狼人);每个原语都同步写入对应 Agent 的 memory 并在审计里登记「进了谁的上下文」。Agent 每次思考只读自己的 memory,物理上不可能看到别人的私密信息。

真实运行样例(python demo.py,seed=42,gpt-5.6-luna)

夜晚(私密行动,均带可见范围标注):

【第 1 回合 · 夜晚】天黑请闭眼。
  [仅法官+狼人可见] 狼人 P1 提议击杀 → P3
  [仅法官+狼人可见] 狼人 P6 提议击杀 → P3
  → 狼人共识:击杀 P3(此共识只进狼人上下文)
  [仅法官+预言家 P4 可见] 预言家查验 P1 → 狼人
  [仅法官+女巫 P2 可见] 女巫得知今晚被刀者:P3
  [仅法官+女巫可见] 女巫使用解药救 P3

白天预言家 P4 直接跳出报查杀(第 1 回合白天,公开发言进所有人上下文):

  P4(发言):我是预言家,昨晚查验P1是狼人。平安夜不影响验人结果,今天请大家优先投P1;若P1悍跳,请他解释具体验人信息和逻辑。
  —— 投票阶段 ——
  → 投票结果:P1 被放逐出局,其真实身份是【狼人】。计票:P1=5票,P4=2票

第 2 回合夜晚,预言家再验出 P6 是狼人、女巫用毒药毒杀 P6,好人阵营在放逐/毒杀 两只狼人后获胜(存活:P2 女巫、P3 村民、P5 村民、P7 村民)。

信息隔离自动校验(节选):

[校验1] 『狼人队友身份』只进狼人上下文:通过 ✓
   - 存在非狼人上下文含队友身份?False(应为 False)
[校验2] 『预言家查验结果』只进预言家(P4)上下文:通过 ✓
   - 存在其他玩家上下文含查验结果?False(应为 False)
[校验3] 审计日志中狼人专属信息的可见集合 == 狼人集合 ['P1', 'P6']:通过 ✓
[校验4] 审计日志中所有『公开-*』信息可见集合 == 全体玩家:通过 ✓
信息隔离总校验:全部通过 ✓✓✓

对照可见:狼人 P1 的上下文里有「狼人阵营的玩家是:P1、P6」和「狼人决定击杀 P3」, 而村民 P3 的上下文里没有任何他人身份、预言家 P4 的上下文里独有「第 1 回合 你查验了 P1,结果为【狼人】」。这就是信息权限控制生效的直接证据。

局限

  • 在线模式 LLM 决策默认用便宜旗舰 gpt-5.6-lunaOPENAI_API_KEY)。通用回退: 若未设置 OPENAI_API_KEY 但设了 OPENROUTER_API_KEY,则自动改走 OpenRouter,并把 模型名映射到其命名空间(gpt-5.6-lunaopenai/gpt-5.6-luna);提示 gpt-5.6 系列 直连 OpenAI 需组织验证,只填 OPENROUTER_API_KEY 即可强制走 OpenRouter。 注意:可选的语音合成(--voice,OpenAI tts-1)仍只支持 OPENAI_API_KEY,OpenRouter 无 TTS 端点。 想零成本、无 Key 跑通全流程,请用 --offline(规则决策);离线策略较为简单, 发言/推理不如 LLM 生动,仅用于演示编排与信息隔离机制,不代表真实博弈水平。
  • 为可复现与控成本,本 demo 全部由 AI 扮演、默认文本模式;书中「真人语音连线」 与实时对话在此简化为:法官按座位顺序编排发言、AI 用文本推理投票,语音仅作可选 的单向 TTS 输出(--voice),未做真人 ASR 语音输入与实时打断。
  • 胜负采用「屠边简化」规则(狼人数 ≥ 好人数即狼人胜),未实现猎人、警长、 自爆等进阶机制;女巫解药默认允许自救,均可按需扩展。
  • 每局结果随机,长度取决于 AI 的推理(通常 2~4 回合分出胜负);用 --seed 复现 或切换局面。AI 的伪装/推理质量受模型能力限制。

源代码

demo.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""实验 10-8:语音狼人杀 Agent 系统 —— 一键跑完整一局。

配套《深入理解 AI Agent》第 10 章「实验 10-8:语音狼人杀 Agent 系统」。

本 demo 演示三件事(对应书中架构设计):
1. **多 Agent**:每个玩家 = 一个独立 LLM Agent(OpenAI,默认 gpt-5.6-luna)。
2. **信息权限控制**:法官按角色把信息投递进各 Agent 的私有上下文——狼人才知道
   队友、预言家才知道查验结果、公开发言进所有人。游戏后打印审计表 + 自动校验,
   客观证明信息隔离正确。
3. **法官编排**:确定性法官编排夜晚(刀/验/用药)→ 白天(死讯/发言/投票)→ 结算。

语音是**可选增强**(--voice,用 OpenAI tts-1 合成公开发言),默认文本模式即可
完整、可复现地跑完一局。

用法:
    export OPENAI_API_KEY=sk-...
    python demo.py                 # 文本模式跑完整一局(LLM 决策,默认)
    python demo.py --offline       # 离线模式:规则决策,零成本、可复现,无需 API Key
    python demo.py --seed 7        # 换一局身份分布
    python demo.py --players 9 --wolves 3   # 自定义人数与狼人数
    python demo.py --voice         # 额外把公开发言合成语音到 audio/
    python demo.py --voice --play  # 合成并播放(macOS afplay)
    python demo.py --offline --log game.log  # 把完整对局日志另存一份到文件
"""

import argparse
import os
import sys


class _Tee:
    """把写入同时分发到多个流(用于 --log:既打印到终端又落盘到文件)。"""

    def __init__(self, *streams):
        self.streams = streams

    def write(self, data):
        for s in self.streams:
            s.write(data)

    def flush(self):
        for s in self.streams:
            s.flush()

try:
    from dotenv import load_dotenv
    load_dotenv()  # 若存在 .env 则加载(可选)
except Exception:
    pass

from werewolf.game import Judge, create_players
from werewolf.roles import Role


def verify_isolation(judge: Judge):
    """自动校验信息隔离是否正确,并打印证据。返回是否全部通过。"""
    print("\n" + "=" * 78)
    print("信息隔离自动校验(证明每条敏感信息只进了它该进的上下文)")
    print("=" * 78)
    ok = True

    wolves = judge.wolves()
    wolf_names = {w.name for w in wolves}
    non_wolves = [p for p in judge.players if p.role != Role.WEREWOLF]
    seer = next((p for p in judge.players if p.role == Role.SEER), None)

    # 证据 1:狼人队友身份只在狼人上下文里
    team_line_marker = "狼人阵营的玩家是"
    wolves_have = all(any(team_line_marker in m for m in w.memory) for w in wolves)
    nonwolves_have = any(any(team_line_marker in m for m in p.memory) for p in non_wolves)
    check1 = wolves_have and not nonwolves_have
    ok &= check1
    print(f"\n[校验1] 『狼人队友身份』只进狼人上下文:{'通过 ✓' if check1 else '失败 ✗'}")
    print(f"   - 每个狼人上下文都含队友身份?{wolves_have}")
    print(f"   - 存在非狼人上下文含队友身份?{nonwolves_have}(应为 False)")

    # 证据 2:预言家查验结果只在预言家本人上下文里
    if seer:
        seer_marker = "你查验了"
        seer_has = any(seer_marker in m for m in seer.memory)
        others_have = any(any(seer_marker in m for m in p.memory)
                          for p in judge.players if p.name != seer.name)
        check2 = seer_has and not others_have
        ok &= check2
        print(f"\n[校验2] 『预言家查验结果』只进预言家({seer.name})上下文:{'通过 ✓' if check2 else '失败 ✗'}")
        print(f"   - 预言家上下文含查验结果?{seer_has}")
        print(f"   - 存在其他玩家上下文含查验结果?{others_have}(应为 False)")

    # 证据 3:审计日志里每条记录的 visible_to 与类别相符
    def cat_visible(cat):
        return [set(r.visible_to) for r in judge.audit.records if r.category == cat]
    check3 = all(v == wolf_names for v in cat_visible("狼人队友身份")) and \
             all(v == wolf_names for v in cat_visible("狼人夜间共识"))
    ok &= check3
    print(f"\n[校验3] 审计日志中狼人专属信息的可见集合 == 狼人集合 {sorted(wolf_names)}:"
          f"{'通过 ✓' if check3 else '失败 ✗'}")

    check4 = all(set(r.visible_to) == set(judge.names)
                 for r in judge.audit.records if r.category.startswith("公开"))
    ok &= check4
    print(f"[校验4] 审计日志中所有『公开-*』信息可见集合 == 全体玩家:"
          f"{'通过 ✓' if check4 else '失败 ✗'}")

    # 对照展示:一个狼人 vs 一个村民的完整私有上下文
    villager = next((p for p in judge.players if p.role == Role.VILLAGER), None)
    a_wolf = wolves[0] if wolves else None
    print("\n—— 对照:同一时刻两名玩家的私有上下文(证明各看各的)——")
    if a_wolf:
        print(f"\n【狼人 {a_wolf.name} 的私有上下文】(含队友身份、夜间共识)")
        for m in a_wolf.memory:
            print(f"   · {m}")
    if villager:
        print(f"\n【村民 {villager.name} 的私有上下文】(不含任何他人身份/查验结果)")
        for m in villager.memory:
            print(f"   · {m}")
    if seer:
        print(f"\n【预言家 {seer.name} 的私有上下文】(含独享的查验结果)")
        for m in seer.memory:
            print(f"   · {m}")

    print("\n" + "=" * 78)
    print(f"信息隔离总校验:{'全部通过 ✓✓✓' if ok else '存在失败 ✗'}")
    print("=" * 78)
    return ok


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="实验 10-8:语音狼人杀 Agent 系统 —— 法官编排 + 信息权限控制 + 多 Agent。",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=(
            "示例:\n"
            "  python demo.py                 文本模式跑完整一局(LLM 决策,需 API Key)\n"
            "  python demo.py --offline       离线规则决策,零成本、可复现,无需 API Key\n"
            "  python demo.py --players 9 --wolves 3   自定义人数与狼人数\n"
            "  python demo.py --voice --play  额外合成并播放公开发言(需 API Key)\n"))
    parser.add_argument("--offline", "--mock", dest="offline", action="store_true",
                        help="离线模式:用规则策略代替 LLM,无需 API Key,零成本且可复现")
    parser.add_argument("--seed", type=int, default=42,
                        help="随机种子(决定身份分布与离线决策,可复现,默认 42)")
    parser.add_argument("--players", type=int, default=7,
                        help="玩家总数(默认 7)")
    parser.add_argument("--wolves", type=int, default=None,
                        help="狼人数量(默认按玩家总数自动推导:7 人为 2 狼)")
    parser.add_argument("--max-rounds", type=int, default=6, dest="max_rounds",
                        help="昼夜循环的最大回合数上限(默认 6)")
    parser.add_argument("--model", type=str, default=None,
                        help="覆盖 LLM 模型(默认 gpt-5.6-luna,仅在线模式有效)")
    parser.add_argument("--voice", action="store_true",
                        help="用 OpenAI tts-1 把公开发言合成语音(需 API Key;默认关,即纯文本模式)")
    parser.add_argument("--play", action="store_true",
                        help="合成语音后立即播放(macOS afplay;需配合 --voice)")
    parser.add_argument("--log", type=str, default=None, metavar="PATH",
                        help="把完整对局日志(含审计表)另存一份到指定文件")
    return parser


def run_game(args):
    if args.model:
        os.environ["OPENAI_MODEL"] = args.model

    mode = "离线(规则决策)" if args.offline else "在线(LLM 决策)"
    roles_note = "" if args.wolves is None else f"(狼人数={args.wolves})"
    print("=" * 78)
    print("实验 10-8:语音狼人杀 Agent 系统")
    print(f"模式:{mode} | 模型:{os.environ.get('OPENAI_MODEL', 'gpt-5.6-luna') if not args.offline else '—'} | "
          f"种子:{args.seed} | 语音:{'开' if args.voice else '关(文本模式)'}")
    print(f"配置:{args.players} 人局{roles_note} | 最大回合:{args.max_rounds}")
    print("=" * 78)

    tts = None
    if args.voice:
        from werewolf.tts import TTS
        tts = TTS(os.path.join(os.path.dirname(__file__), "audio"), play=args.play)

    players = create_players(seed=args.seed, players=args.players,
                             wolves=args.wolves, offline=args.offline)
    judge = Judge(players, seed=args.seed, tts=tts, max_rounds=args.max_rounds)
    winner = judge.run()

    # 打印信息可见性审计表 + 自动校验
    judge.audit.print_table(judge.names)
    ok = verify_isolation(judge)

    print(f"\n最终结果:{winner.value} 获胜。")
    return ok


def main():
    args = build_parser().parse_args()

    # 在线模式(LLM 决策 / 语音合成)才需要 API Key;离线模式不需要。
    # LLM 决策支持 OPENAI_API_KEY 或(回退)OPENROUTER_API_KEY;语音合成(--voice,
    # OpenAI tts-1)目前只支持 OPENAI_API_KEY,OpenRouter 无 TTS 端点。
    has_llm_key = os.environ.get("OPENAI_API_KEY") or os.environ.get("OPENROUTER_API_KEY")
    if args.voice and not os.environ.get("OPENAI_API_KEY"):
        print("错误:语音合成(--voice,OpenAI tts-1)需要 OPENAI_API_KEY。"
              "请先 export OPENAI_API_KEY=sk-...(见 env.example),或去掉 --voice 跑纯文本模式。")
        sys.exit(1)
    if not args.offline and not has_llm_key:
        print("错误:LLM 决策需要 OPENAI_API_KEY 或 OPENROUTER_API_KEY。"
              "请先 export(见 env.example),或改用离线模式:python demo.py --offline")
        sys.exit(1)

    log_file = None
    orig_stdout = sys.stdout
    if args.log:
        log_file = open(args.log, "w", encoding="utf-8")
        sys.stdout = _Tee(orig_stdout, log_file)
    try:
        run_game(args)
    except ValueError as e:
        print(f"错误:{e}")
        sys.exit(2)
    finally:
        if log_file:
            sys.stdout = orig_stdout
            log_file.close()
            print(f"(完整对局日志已保存到 {args.log})")


if __name__ == "__main__":
    main()

werewolf/__init__.py

# -*- coding: utf-8 -*-
"""实验 10-8:语音狼人杀 Agent 系统(多 Agent + 信息权限控制 + 法官编排)。"""

werewolf/agent.py

# -*- coding: utf-8 -*-
"""玩家 Agent:每个玩家 = 一个独立的 LLM Agent,拥有**严格隔离的私有上下文**。

信息隔离的实现要点:
- 每个 PlayerAgent 只维护自己的 `memory`(一串它「观察到 / 被告知」的事件)。
- 法官(judge.py)决定把哪条信息推给哪个 Agent 的 memory——狼人才会收到「队友
  身份」,预言家才会收到「查验结果」,公开发言才会推给所有人。
- Agent 每次思考(发言 / 投票 / 用技能)时,只能看到自己 memory 里的内容,
  因此不可能「偷看」到本不该看到的信息。这就是信息权限控制的落点。

离线(--offline / --mock)策略:当没有 OpenAI Key、或想零成本可复现地跑完整一局时,
Agent 用一套**规则驱动**的决策代替 LLM。关键在于:离线策略同样**只读自己的 memory**
(不碰其他 Agent 的私有上下文),因此信息权限控制这一教学要点在离线模式下依然成立、
依然可被审计校验。
"""

import json
import os
import re
from typing import List, Optional

from .roles import Role, ROLE_STRATEGY, faction_of


# 全局唯一的 LLM 客户端。模型默认当前便宜旗舰 gpt-5.6-luna。
# 通用回退:优先 OPENAI_API_KEY 直连 OpenAI;没有则用 OPENROUTER_API_KEY 走 OpenRouter。
_MODEL = os.environ.get("OPENAI_MODEL", "gpt-5.6-luna")
_client = None


def _to_openrouter_model(model: str) -> str:
    """把模型名映射到 OpenRouter 命名空间(用于无 OPENAI_API_KEY 的回退路径)。"""
    if "/" in model:
        return model                      # 已是 OpenRouter 命名空间,原样使用
    if model.startswith("gpt-"):
        return "openai/" + model          # gpt-* -> openai/gpt-*
    if model.startswith("claude-"):
        return "anthropic/claude-opus-4.8"
    return "openai/gpt-5.6-luna"          # 兜底:当前便宜旗舰


def _safe_create(client, **kwargs):
    """调用 Chat Completions;对推理型模型(如 gpt-5.x)的参数限制做自动降级重试:
    - 不支持 max_tokens 时改用 max_completion_tokens;
    - 不支持非默认 temperature 时移除该参数(回退到模型默认 1)。
    这样同一份代码既能跑传统对话模型(接受 temperature=0.8),也能跑推理模型。"""
    for _ in range(3):
        try:
            return client.chat.completions.create(**kwargs)
        except Exception as e:
            msg = str(e)
            if "max_completion_tokens" in msg and "max_tokens" in kwargs:
                kwargs["max_completion_tokens"] = kwargs.pop("max_tokens")
                continue
            if "temperature" in msg and "temperature" in kwargs:
                kwargs.pop("temperature", None)
                continue
            raise
    return client.chat.completions.create(**kwargs)


def get_client():
    """返回全局共享的 LLM 客户端(懒加载,进程内单例)。

    仅在线模式(真实调用 LLM)才会用到;离线模式不导入 openai、不构造客户端。
      1) 有 OPENAI_API_KEY -> 直连 OpenAI;
      2) 否则有 OPENROUTER_API_KEY -> 走 OpenRouter,并把 _MODEL 映射到其命名空间;
      3) 都没有则报清晰错误。
    """
    global _client, _MODEL
    if _client is None:
        from openai import OpenAI  # 懒导入:离线模式无需安装 openai
        if os.environ.get("OPENAI_API_KEY"):
            _client = OpenAI()  # 自动读取环境变量 OPENAI_API_KEY
        elif os.environ.get("OPENROUTER_API_KEY"):
            _MODEL = _to_openrouter_model(_MODEL)
            _client = OpenAI(
                api_key=os.environ["OPENROUTER_API_KEY"],
                base_url="https://openrouter.ai/api/v1",
            )
        else:
            raise RuntimeError(
                "未设置 OPENAI_API_KEY 或 OPENROUTER_API_KEY,请参考 env.example 配置,"
                "或改用离线模式:python demo.py --offline"
            )
    return _client


class PlayerAgent:
    """一个玩家 Agent,封装其身份、私有上下文与决策(LLM 或离线规则)。"""

    def __init__(self, name: str, role: Role, offline: bool = False, rng=None):
        self.name = name          # 玩家名,如 "P3"
        self.role = role          # 真实身份(只有本人和法官知道)
        self.faction = faction_of(role)
        self.alive = True
        self.offline = offline    # True 时用规则策略代替 LLM(零成本、可复现)
        # 离线策略的私有随机源(按玩家名种子化,保证可复现且各玩家独立)
        import random as _random
        self._rng = rng or _random.Random(hash(name) & 0xFFFF)
        # 私有上下文:这个 Agent「看得到」的全部信息。别的 Agent 无法访问。
        self.memory: List[str] = []

    # ---- 上下文注入:只有法官会调用,用来把信息投递进这个 Agent 的私有上下文 ----
    def observe(self, event: str):
        """把一条信息写入本 Agent 的私有上下文。"""
        self.memory.append(event)

    # ---- system prompt:角色设定 + 策略。狼人的队友身份不写在这里,而是由法官
    #      在游戏开始时通过 observe() 投递,以便审计能记录「谁看到了队友身份」。 ----
    def _system_prompt(self, players: List[str]) -> str:
        return (
            f"你正在玩一局狼人杀。你是玩家 {self.name}\n"
            f"你的真实身份是【{self.role.value}】,属于【{self.faction.value}】。\n"
            f"本局玩家共 {len(players)} 人:{'、'.join(players)}\n\n"
            f"{ROLE_STRATEGY[self.role]}\n\n"
            "重要:只能依据你已知的信息推理,不要臆造你无从得知的身份。发言要像真人,"
            "简洁自然,有理有据。"
        )

    def _context_block(self) -> str:
        """把私有上下文拼成给 LLM 的一段文字。"""
        if not self.memory:
            return "(暂无信息)"
        return "\n".join(f"- {m}" for m in self.memory)

    def _chat(self, instruction: str, players: List[str], max_tokens: int,
              json_mode: bool = False) -> str:
        """组装 system + user 消息并调用 LLM;user 消息里只拼接本 Agent 自己的
        私有上下文(`_context_block`),绝不包含其他玩家的私密信息。"""
        messages = [
            {"role": "system", "content": self._system_prompt(players)},
            {"role": "user", "content":
                f"【你目前掌握的信息(仅你可见)】\n{self._context_block()}\n\n"
                f"【当前任务】\n{instruction}"},
        ]
        # 给推理型模型(如 gpt-5.6 系列)留足输出预算:其内部推理 token 也计入
        # max_tokens,预算过小会导致 content 被截断为空。设一个下限兜底。
        kwargs = dict(model=_MODEL, messages=messages, temperature=0.8,
                      max_tokens=max(max_tokens, 512))
        if json_mode:
            kwargs["response_format"] = {"type": "json_object"}
        resp = _safe_create(get_client(), **kwargs)
        # content 可能为 None(如被截断);用空串兜底,交由上层解析做降级处理。
        return (resp.choices[0].message.content or "").strip()

    # ---------- 三种对外能力:发言 / 决策(选目标)/ 投票 ----------

    def speak(self, players: List[str]) -> str:
        """白天公开发言。返回一段发言文本(公开信息)。"""
        if self.offline:
            return self._offline_speak(candidates=[p for p in players if p != self.name])
        instruction = (
            "现在轮到你在白天公开发言。请结合你掌握的信息,发表一段简短的发言"
            "(2~4 句话,60 字以内)。符合你的身份与策略。直接输出发言内容,不要加引号。"
        )
        return self._chat(instruction, players, max_tokens=180)

    def choose_target(self, prompt: str, candidates: List[str],
                      players: List[str], allow_none: bool = False) -> Optional[str]:
        """让 Agent 从候选人中选一个目标(夜间刀人 / 查验 / 用毒 / 救人判断等)。

        用 JSON 模式返回,鲁棒地解析出目标玩家名。
        """
        if self.offline:
            return self._offline_choose_target(candidates, allow_none)
        opt = ",也可以选择放弃(target 填 \"none\")" if allow_none else ""
        instruction = (
            f"{prompt}\n候选玩家:{'、'.join(candidates)}{opt}\n"
            "请只返回 JSON:{\"target\": \"玩家名或none\", \"reason\": \"一句话理由\"}"
        )
        raw = self._chat(instruction, players, max_tokens=120, json_mode=True)
        target = self._parse_target(raw, candidates, allow_none)
        return target

    def vote(self, candidates: List[str], players: List[str]) -> Optional[str]:
        """投票放逐。返回票投给谁(或弃票 none)。"""
        if self.offline:
            return self._offline_vote(candidates)
        instruction = (
            "现在是白天投票放逐环节。请根据全场发言与你的推理,投出你认为最可能是"
            "狼人的玩家。\n候选玩家:" + "、".join(candidates) + "。\n"
            "请只返回 JSON:{\"target\": \"玩家名\", \"reason\": \"一句话理由\"}"
        )
        raw = self._chat(instruction, players, max_tokens=120, json_mode=True)
        return self._parse_target(raw, candidates, allow_none=True)

    # ---------- 离线(规则)策略:只读自己的 memory,绝不访问他人上下文 ----------
    def _known_teammates(self) -> set:
        """狼人从自己的私有上下文里解析出队友名单(好人解析不到,返回空)。"""
        mates = set()
        for m in self.memory:
            hit = re.search(r"狼人阵营的玩家是:([^((]+)", m)
            if hit:
                mates |= set(re.findall(r"P\d+", hit.group(1)))
        return mates

    def _known_wolves(self) -> set:
        """从自己的私有上下文里收集『已知是狼人』的玩家:预言家的查验结果 + 狼人的队友。

        好人平民无从得知任何人身份 → 返回空集合,只能随机投票。这正是信息不对称。
        """
        known = set(self._known_teammates())
        for m in self.memory:
            hit = re.search(r"你查验了\s*(P\d+),结果为【狼人】", m)
            if hit:
                known.add(hit.group(1))
        return known

    def _offline_vote(self, candidates: List[str]) -> Optional[str]:
        """离线投票:优先投自己『确知的狼人』(预言家验人 / 狼人不投队友),否则随机。"""
        if not candidates:
            return None
        if self.role == Role.WEREWOLF:
            # 狼人:投一个非队友的好人,尽量隐藏自己
            mates = self._known_teammates()
            targets = [c for c in candidates if c not in mates] or candidates
            return self._rng.choice(targets)
        # 好人:预言家有验人结果就投确认的狼;其余平民只能随机(信息不对称的代价)
        wolves = [c for c in candidates if c in self._known_wolves()]
        if wolves:
            return self._rng.choice(wolves)
        return self._rng.choice(candidates)

    def _offline_choose_target(self, candidates: List[str],
                               allow_none: bool) -> Optional[str]:
        """离线夜间选目标:狼人/预言家等必选场景优先选『已知狼人之外』的目标;
        女巫解药/毒药等可放弃场景按概率决定。"""
        if not candidates:
            return None
        if allow_none:
            # 女巫用药:约一半概率行动(救/毒),使对局有变化又能收敛
            if self._rng.random() < 0.5:
                return None
            return self._rng.choice(candidates)
        if self.role == Role.SEER:
            # 预言家:优先查验尚未确认身份的玩家(避免重复查验已知狼人)
            unknown = [c for c in candidates if c not in self._known_wolves()]
            return self._rng.choice(unknown or candidates)
        if self.role == Role.WEREWOLF:
            mates = self._known_teammates()
            targets = [c for c in candidates if c not in mates] or candidates
            return self._rng.choice(targets)
        return self._rng.choice(candidates)

    def _offline_speak(self, candidates: List[str]) -> str:
        """离线发言:按角色生成一句符合身份、且不泄露私密信息的模板发言。"""
        wolves = [c for c in candidates if c in self._known_wolves()]
        suspect = self._rng.choice(candidates) if candidates else "大家"
        if self.role == Role.SEER and wolves:
            return f"我是预言家,昨晚查验到 {wolves[0]} 是狼人,请大家把票投给他。"
        if self.role == Role.WEREWOLF:
            return f"我是好人,从发言看 {suspect} 有点可疑,建议重点关注他。"
        if self.role == Role.WITCH:
            return f"我暂时观望,觉得 {suspect} 的发言站不住脚,先留意一下。"
        if self.role == Role.SEER:
            return "我还没有决定性的信息,先听大家发言,谨慎投票。"
        return f"我是村民,没有夜间信息,只能靠推理,感觉 {suspect} 稍微可疑。"

    # ---------- 解析工具 ----------
    @staticmethod
    def _parse_target(raw: str, candidates: List[str], allow_none: bool) -> Optional[str]:
        target = None
        try:
            data = json.loads(raw)
            target = str(data.get("target", "")).strip()
        except Exception:
            # 兜底:直接从文本里正则找候选玩家名
            target = raw
        if allow_none and target.lower() in ("none", "", "弃票", "放弃"):
            return None
        # 归一化:精确匹配优先,否则模糊匹配(找出现的候选名)
        if target in candidates:
            return target
        for c in candidates:
            if c in (target or ""):
                return c
        # 最后兜底:从原始串里搜 Pn
        m = re.search(r"P\d+", target or "")
        if m and m.group(0) in candidates:
            return m.group(0)
        # 实在解析不出:好人默认弃票,狼人/必须选的场景由调用方兜底
        return None if allow_none else (candidates[0] if candidates else None)

werewolf/audit.py

# -*- coding: utf-8 -*-
"""信息可见性审计(Information Visibility Audit)。

这是本实验「信息权限控制可验证」的核心工具:法官每向某个(或某些)玩家的
上下文投递一条信息时,都会在这里登记一条记录——这条信息属于哪个类别、内容
摘要是什么、**进入了谁的上下文**。游戏结束后打印这张审计表,即可客观证明
信息隔离是否正确(例如「狼人队友身份」只进狼人上下文、「预言家查验结果」只进
预言家本人上下文、「公开发言」进所有人上下文)。
"""

from dataclasses import dataclass, field
from typing import List


@dataclass
class AuditRecord:
    round_no: int          # 第几回合
    phase: str             # 阶段(夜晚/白天/...)
    category: str          # 信息类别(如「狼人队友身份」「预言家查验结果」「公开发言」)
    content: str           # 信息内容摘要
    visible_to: List[str]  # 该信息进入了哪些玩家的上下文(玩家名列表)


@dataclass
class AuditLog:
    records: List[AuditRecord] = field(default_factory=list)

    def add(self, round_no, phase, category, content, visible_to):
        self.records.append(AuditRecord(round_no, phase, category, content, list(visible_to)))

    def print_table(self, all_players):
        """打印完整的信息可见性审计表。"""
        print("\n" + "=" * 78)
        print("信息可见性审计表(每条信息进入了谁的上下文)")
        print("=" * 78)
        header = f"{'回合':<4}{'阶段':<6}{'类别':<14}{'可见玩家':<20}内容"
        print(header)
        print("-" * 78)
        for r in self.records:
            vis = "所有人" if set(r.visible_to) == set(all_players) else "、".join(r.visible_to)
            content = r.content if len(r.content) <= 30 else r.content[:29] + "…"
            print(f"{r.round_no:<5}{r.phase:<7}{r.category:<15}{vis:<21}{content}")
        print("=" * 78)

werewolf/game.py

# -*- coding: utf-8 -*-
"""法官(主持人)Agent:代码驱动的游戏编排与信息权限控制中枢。

法官不是 LLM——它是**确定性的编排器**,负责:
1. 维护中心化游戏状态(身份、阵营、生死、阶段、历史)。
2. **信息权限控制**:决定每条信息投递给哪些玩家 Agent 的私有上下文
   (狼人才知道队友、预言家才知道查验结果、公开发言进所有人),并登记审计。
3. 编排昼夜循环:夜晚(狼人刀人 → 预言家查验 → 女巫用药)→ 白天(公布死讯 →
   依次发言 → 投票放逐)→ 结算胜负。
"""

import random
from collections import Counter
from typing import List, Optional

from .agent import PlayerAgent
from .audit import AuditLog
from .roles import Role, Faction


def build_roles(players: int = 7, wolves: Optional[int] = None) -> List[Role]:
    """按玩家总数推导身份组成:默认 7 人 = 2 狼人 + 1 预言家 + 1 女巫 + 3 村民。

    - wolves 未指定时按 max(1, players // 3) 估算(7 人得 2 狼,与书中默认一致)。
    - 4 人及以上配 1 预言家,5 人及以上再配 1 女巫,其余全为村民。
    """
    if players < 3:
        raise ValueError("玩家总数至少为 3")
    wolves = wolves if wolves is not None else max(1, players // 3)
    seer = 1 if players >= 4 else 0
    witch = 1 if players >= 5 else 0
    villagers = players - wolves - seer - witch
    if wolves < 1 or villagers < 0:
        raise ValueError(
            f"身份组成非法:{players} 人无法容纳 {wolves} 狼 + {seer} 预言家 + "
            f"{witch} 女巫(剩余村民 {villagers})。请调小 --wolves 或调大 --players。")
    return ([Role.WEREWOLF] * wolves + [Role.SEER] * seer
            + [Role.WITCH] * witch + [Role.VILLAGER] * villagers)


def create_players(seed: int = 42, players: int = 7, wolves: Optional[int] = None,
                   offline: bool = False) -> List[PlayerAgent]:
    """创建一局游戏(默认 7 人:2 狼人 + 1 预言家 + 1 女巫 + 3 村民,控成本)。

    身份随机洗牌后分配给 P1~Pn,保证每局身份分布不同但可用 seed 复现。
    offline=True 时每个 Agent 用规则策略代替 LLM(零成本、可复现);每个 Agent
    还会拿到一个按 seed 与序号种子化的独立随机源,保证离线对局完全可复现。
    """
    rng = random.Random(seed)
    roles = build_roles(players, wolves)
    rng.shuffle(roles)
    return [PlayerAgent(f"P{i+1}", roles[i], offline=offline,
                        rng=random.Random(seed * 1000 + i))
            for i in range(len(roles))]


class Judge:
    """法官:编排 + 信息权限控制。"""

    def __init__(self, players: List[PlayerAgent], seed: int = 42,
                 tts=None, max_rounds: int = 6):
        self.players = players
        self.names = [p.name for p in players]
        self.audit = AuditLog()
        self.rng = random.Random(seed + 1)
        self.tts = tts                 # 可选的 TTS 合成器(--voice 时注入)
        self.max_rounds = max_rounds
        self.round_no = 0
        self.phase = "初始化"
        # 女巫药剂状态
        self.witch_heal_available = True
        self.witch_poison_available = True

    # ------------------------------------------------------------------
    # 信息投递原语:每个原语都同时 (a) 写入相应 Agent 的私有上下文;
    #               (b) 在审计日志里登记「这条信息进了谁的上下文」。
    # ------------------------------------------------------------------
    def _log(self, category, content, visible_to):
        self.audit.add(self.round_no, self.phase, category, content, visible_to)

    def broadcast(self, category: str, content: str):
        """公开信息:进入**所有玩家**(含已出局者)的上下文。"""
        for p in self.players:
            p.observe(content)
        self._log(category, content, self.names)

    def private_send(self, player: PlayerAgent, category: str, content: str):
        """私密信息:只进入**指定单个玩家**的上下文。"""
        player.observe(content)
        self._log(category, content, [player.name])

    def wolves_send(self, category: str, content: str):
        """狼人专属信息:只进入**所有狼人**的上下文。"""
        wolves = self.wolves()
        for w in wolves:
            w.observe(content)
        self._log(category, content, [w.name for w in wolves])

    # ------------------------------------------------------------------
    # 状态查询
    # ------------------------------------------------------------------
    def alive(self) -> List[PlayerAgent]:
        return [p for p in self.players if p.alive]

    def wolves(self, alive_only=False) -> List[PlayerAgent]:
        ws = [p for p in self.players if p.role == Role.WEREWOLF]
        return [w for w in ws if w.alive] if alive_only else ws

    def by_name(self, name: str) -> Optional[PlayerAgent]:
        for p in self.players:
            if p.name == name:
                return p
        return None

    # ------------------------------------------------------------------
    # 阶段 0:分配身份并投递「谁知道谁」的初始信息
    # ------------------------------------------------------------------
    def assign_identities(self):
        self.phase = "身份分配"
        print("\n" + "#" * 78)
        print("【阶段 0 · 身份分配】法官私下告知每人身份;狼人额外被告知队友是谁。")
        print("  信息隔离:每人只知道自己的身份;只有狼人上下文里有『队友身份』。")
        print("#" * 78)
        # 每个玩家私下知道自己的身份(只进本人上下文)
        for p in self.players:
            self.private_send(p, "身份分配", f"你的身份是:{p.role.value}")
        # 狼人互相知道队友是谁(只进狼人上下文)——这是信息不对称的关键
        wolves = self.wolves()
        team = "、".join(w.name for w in wolves)
        self.wolves_send("狼人队友身份", f"狼人阵营的玩家是:{team}(你们互为队友,夜晚共同行动)")
        # 打印真实身份表(这是「上帝视角」,仅供人类观察,不进任何 Agent 上下文)
        print("  [上帝视角/仅人类可见] 真实身份表:")
        for p in self.players:
            print(f"    {p.name}: {p.role.value}{p.faction.value})")
        print(f"  狼队友(仅狼人 {team} 的上下文里有这条信息)")

    # ------------------------------------------------------------------
    # 夜晚
    # ------------------------------------------------------------------
    def night(self) -> List[str]:
        """执行一个夜晚,返回今晚出局玩家名列表。"""
        self.phase = "夜晚"
        print("\n" + "=" * 78)
        print(f"【第 {self.round_no} 回合 · 夜晚】天黑请闭眼。")
        print("  信息隔离:以下所有行动与结果都是私密的——狼人共识只进狼人上下文、")
        print("  预言家查验结果只进预言家上下文、女巫用药只进女巫上下文。")
        print("=" * 78)

        killed = self._wolves_act()
        self._seer_act()
        poisoned, saved = self._witch_act(killed)

        # 结算今晚死亡:被刀且未被救 + 被毒
        deaths = []
        if killed and not saved:
            deaths.append(killed)
        if poisoned and poisoned not in deaths:
            deaths.append(poisoned)
        for name in deaths:
            self.by_name(name).alive = False
        return deaths

    def _wolves_act(self) -> Optional[str]:
        wolves = self.wolves(alive_only=True)
        if not wolves:
            return None
        # 候选:所有存活的非狼人(狼人不刀自己人)
        candidates = [p.name for p in self.alive() if p.role != Role.WEREWOLF]
        if not candidates:
            return None
        votes = []
        for w in wolves:
            t = w.choose_target(
                "现在是夜晚,狼人行动。请与队友一致,选择今晚要击杀的一名好人玩家。",
                candidates, self.names, allow_none=False)
            if t:
                votes.append(t)
                print(f"  [仅法官+狼人可见] 狼人 {w.name} 提议击杀 → {t}")
        if not votes:
            return None
        # 汇总:最高票;平票取第一名狼人的意见
        tally = Counter(votes)
        top = tally.most_common()
        best = [n for n, c in top if c == top[0][1]]
        killed = votes[0] if len(best) > 1 else top[0][0]
        # 把「今晚狼人共识」写进狼人共享上下文(只有狼人看得到)
        self.wolves_send("狼人夜间共识", f"第{self.round_no}回合夜晚,狼人决定击杀 {killed}")
        print(f"  → 狼人共识:击杀 {killed}(此共识只进狼人上下文)")
        return killed

    def _seer_act(self):
        seers = [p for p in self.alive() if p.role == Role.SEER]
        if not seers:
            return
        seer = seers[0]
        candidates = [p.name for p in self.alive() if p.name != seer.name]
        target = seer.choose_target(
            "现在是夜晚,预言家行动。请选择一名玩家查验其真实阵营。",
            candidates, self.names, allow_none=False)
        if not target:
            target = self.rng.choice(candidates)
        tgt = self.by_name(target)
        result = "狼人" if tgt.role == Role.WEREWOLF else "好人"
        # 查验结果只进预言家本人上下文——这是预言家独享的关键信息
        self.private_send(seer, "预言家查验结果",
                          f"第{self.round_no}回合你查验了 {target},结果为【{result}】")
        print(f"  [仅法官+预言家 {seer.name} 可见] 预言家查验 {target}{result}")

    def _witch_act(self, killed: Optional[str]):
        witches = [p for p in self.alive() if p.role == Role.WITCH]
        if not witches:
            return None, False
        witch = witches[0]
        saved = False
        poisoned = None

        # 告知女巫今晚谁被刀(只进女巫上下文)
        if killed:
            self.private_send(witch, "女巫夜间信息", f"第{self.round_no}回合,今晚被狼人袭击的是 {killed}")
            print(f"  [仅法官+女巫 {witch.name} 可见] 女巫得知今晚被刀者:{killed}")
            # 解药:是否救
            if self.witch_heal_available and killed != witch.name:
                dec = witch.choose_target(
                    f"今晚 {killed} 被狼人袭击。你是否使用【解药】救他?"
                    "(救则 target 填该玩家名,不救填 none)",
                    [killed], self.names, allow_none=True)
                if dec == killed:
                    saved = True
                    self.witch_heal_available = False
                    self.private_send(witch, "女巫用药", f"你在第{self.round_no}回合使用了解药,救活了 {killed}")
                    print(f"  [仅法官+女巫可见] 女巫使用解药救 {killed}")
        else:
            self.private_send(witch, "女巫夜间信息", f"第{self.round_no}回合是平安夜(无人被狼人击杀,或你无从得知)")

        # 毒药:是否毒一人
        if self.witch_poison_available:
            candidates = [p.name for p in self.alive() if p.name != witch.name]
            dec = witch.choose_target(
                "你是否使用【毒药】毒死一名你怀疑是狼人的玩家?(毒则填玩家名,不毒填 none)",
                candidates, self.names, allow_none=True)
            if dec and dec in candidates:
                poisoned = dec
                self.witch_poison_available = False
                self.private_send(witch, "女巫用药", f"你在第{self.round_no}回合使用了毒药,毒杀了 {poisoned}")
                print(f"  [仅法官+女巫可见] 女巫使用毒药毒 {poisoned}")
        return poisoned, saved

    # ------------------------------------------------------------------
    # 白天
    # ------------------------------------------------------------------
    def day(self, night_deaths: List[str]) -> Optional[str]:
        """白天:公布死讯 → 依次发言 → 投票放逐。返回被放逐者名(或 None)。"""
        self.phase = "白天"
        print("\n" + "=" * 78)
        print(f"【第 {self.round_no} 回合 · 白天】天亮请睁眼。")
        print("  信息隔离:死讯、发言、投票结果都是公开信息,进入所有人上下文。")
        print("=" * 78)

        # 公布死讯(公开)
        if night_deaths:
            msg = f"天亮了。昨晚出局的玩家是:{'、'.join(night_deaths)}"
        else:
            msg = "天亮了。昨晚是平安夜,无人出局"
        self.broadcast("公开-死讯", msg)
        print(f"  法官宣布:{msg}")

        if self._check_winner():
            return None

        # 依次发言(公开)
        print("\n  —— 发言阶段(按座位顺序,公开发言进入所有人上下文)——")
        for p in self.alive():
            speech = p.speak(self.names)
            line = f"{p.name}(发言):{speech}"
            self.broadcast("公开发言", f"{p.name} 说:{speech}")
            print(f"  {line}")
            if self.tts:  # 可选:把发言合成语音
                self.tts.synth(p.name, speech, self.round_no)

        # 投票放逐(公开)
        print("\n  —— 投票阶段 ——")
        exiled = self._vote()
        return exiled

    def _vote(self) -> Optional[str]:
        alive = self.alive()
        tally = Counter()
        for p in alive:
            candidates = [q.name for q in alive if q.name != p.name]
            t = p.vote(candidates, self.names)
            if t:
                tally[t] += 1
                print(f"  {p.name} 投票 → {t}")
            else:
                print(f"  {p.name} 弃票")
        if not tally:
            self.broadcast("公开-放逐", "本轮无人被放逐(全部弃票)")
            print("  本轮无人被放逐")
            return None
        top = tally.most_common()
        best = [n for n, c in top if c == top[0][1]]
        exiled = self.rng.choice(best) if len(best) > 1 else top[0][0]
        ex = self.by_name(exiled)
        ex.alive = False
        result = f"投票结果:{exiled} 被放逐出局,其真实身份是【{ex.role.value}】。计票:" + \
                 ",".join(f"{n}={c}票" for n, c in top)
        self.broadcast("公开-放逐", result)
        print(f"  → {result}")
        return exiled

    # ------------------------------------------------------------------
    # 结算
    # ------------------------------------------------------------------
    def _check_winner(self) -> Optional[Faction]:
        """判定当前是否已分出胜负:狼人全灭则好人胜;狼人数≥好人数则狼人胜;
        否则返回 None(继续游戏)。"""
        w = len(self.wolves(alive_only=True))
        g = len([p for p in self.alive() if p.role != Role.WEREWOLF])
        if w == 0:
            return Faction.GOOD
        if w >= g:  # 狼人数不少于好人数 → 狼人胜(屠边简化规则)
            return Faction.WEREWOLF
        return None

    def run(self) -> Faction:
        """跑完整一局,返回获胜阵营。"""
        self.assign_identities()
        winner = None
        while winner is None and self.round_no < self.max_rounds:
            self.round_no += 1
            deaths = self.night()
            winner = self._check_winner()
            if winner:
                break
            self.day(deaths)
            winner = self._check_winner()
        if winner is None:
            # 达到回合上限,按存活人数判定(好人多则好人赢)
            winner = self._check_winner() or Faction.GOOD
        self._announce(winner)
        return winner

    def _announce(self, winner: Faction):
        self.phase = "结算"
        print("\n" + "#" * 78)
        print("【游戏结束 · 结算】")
        alive = [f"{p.name}({p.role.value})" for p in self.alive()]
        print(f"  存活玩家:{'、'.join(alive) if alive else '无'}")
        print(f"  >>> 获胜阵营:{winner.value} <<<")
        print("#" * 78)

werewolf/roles.py

# -*- coding: utf-8 -*-
"""角色定义、阵营、以及每个角色的策略提示词。

狼人杀的核心是**信息不对称**:不同角色天生知道不同的信息,且拥有不同的
夜间行动能力。这里集中定义角色的元数据与提示词,供 agent.py 构造每个玩家
Agent 的 system prompt。
"""

from enum import Enum


class Role(str, Enum):
    """游戏中的四种角色。"""
    WEREWOLF = "狼人"
    SEER = "预言家"
    WITCH = "女巫"
    VILLAGER = "村民"


class Faction(str, Enum):
    """两大阵营。预言家、女巫、村民都属于好人阵营(好人 = 神职 + 平民)。"""
    WEREWOLF = "狼人阵营"
    GOOD = "好人阵营"


# 角色 -> 阵营
ROLE_FACTION = {
    Role.WEREWOLF: Faction.WEREWOLF,
    Role.SEER: Faction.GOOD,
    Role.WITCH: Faction.GOOD,
    Role.VILLAGER: Faction.GOOD,
}


# 各角色的策略提示词(对应书中「Agent 推理与策略」小节)。
ROLE_STRATEGY = {
    Role.WEREWOLF: (
        "你是狼人。你的目标是隐藏身份,误导好人,最终让狼人数量不少于好人。\n"
        "策略:像普通村民一样发言,可以表达对某些玩家的合理怀疑,但不要过于激进以免暴露。\n"
        "如果有预言家跳出来说验到你是狼人,你可以考虑反咬对方是悍跳的假预言家。\n"
        "投票时尽量跟大多数好人的票,避免成为异类。绝不要主动暴露自己或队友是狼人。"
    ),
    Role.SEER: (
        "你是预言家(好人阵营)。每晚你可以查验一名玩家的真实阵营(好人/狼人)。\n"
        "策略:在合适时机(通常查到狼人或局势危急时)跳出来公布身份与验人信息,带领好人。\n"
        "若有人悍跳预言家,请对比双方验人信息,指出对方逻辑中的矛盾或不合理之处。\n"
        "你的查验结果是你独有的关键信息,只有你自己知道,请善用它引导投票。"
    ),
    Role.WITCH: (
        "你是女巫(好人阵营)。你有一瓶解药和一瓶毒药,各只能用一次。\n"
        "解药可以在夜晚救活被狼人刀的玩家;毒药可以在夜晚毒死一名你怀疑的玩家。\n"
        "策略:解药通常留给关键好人(如预言家)或前期不要浪费;毒药在你较确定某人是狼时使用。\n"
        "你的用药信息只有你自己知道,白天发言时注意保护自己,不要轻易暴露女巫身份。"
    ),
    Role.VILLAGER: (
        "你是村民(好人阵营),没有夜间技能,只能靠逻辑推理找出狼人。\n"
        "策略:分析每个玩家的发言是否自洽,留意急于带节奏、模糊身份、频繁改变立场的玩家。\n"
        "关注投票行为——狼人往往集中票数投给对好人威胁最大的人。\n"
        "不要随机怀疑,每一个推理都应基于具体的发言和投票事实。"
    ),
}


def faction_of(role: Role) -> Faction:
    """返回某个角色所属的阵营。"""
    return ROLE_FACTION[role]

werewolf/tts.py

# -*- coding: utf-8 -*-
"""可选的语音合成(TTS)——把玩家公开发言合成为语音。

语音是本实验的**可选增强**,不是跑通的必需:默认文本模式即可完整跑完一局并
验证信息隔离。加 --voice 时才启用,用 OpenAI tts-1 把每条公开发言合成 mp3,
存到 audio/ 目录;在 macOS 上可用 afplay 顺带播放(--play)。
"""

import os
import subprocess

from .agent import get_client


# 给不同玩家分配不同音色,便于区分
_VOICES = ["alloy", "echo", "fable", "onyx", "nova", "shimmer", "coral"]


class TTS:
    def __init__(self, out_dir: str, play: bool = False):
        self.out_dir = out_dir
        self.play = play
        os.makedirs(out_dir, exist_ok=True)
        self._idx = 0

    def synth(self, speaker: str, text: str, round_no: int):
        voice = _VOICES[(int(speaker.lstrip("P")) - 1) % len(_VOICES)]
        path = os.path.join(self.out_dir, f"r{round_no}_{speaker}_{self._idx}.mp3")
        self._idx += 1
        try:
            resp = get_client().audio.speech.create(
                model="tts-1", voice=voice, input=text)
            resp.stream_to_file(path)
            print(f"    [TTS] {speaker} 发言已合成语音(音色 {voice})→ {path}")
            if self.play:
                # macOS 自带 afplay;其它平台请自行改播放器
                subprocess.run(["afplay", path], check=False)
        except Exception as e:
            print(f"    [TTS] 合成失败(不影响游戏进行):{e}")