# -*- coding: utf-8 -*- """ 三国狼人杀 - 基于AgentScope的中文版狼人杀游戏 融合三国演义角色和传统狼人杀玩法 """ import asyncio import os import random from typing import List, Dict, Optional from agentscope.agent import ReActAgent from agentscope.model import DashScopeChatModel from agentscope.pipeline import MsgHub, sequential_pipeline, fanout_pipeline from agentscope.formatter import DashScopeMultiAgentFormatter from prompt_cn import ChinesePrompts from game_roles import GameRoles from structured_output_cn import ( DiscussionModelCN, get_vote_model_cn, WitchActionModelCN, get_seer_model_cn, get_hunter_model_cn, WerewolfKillModelCN ) from utils_cn import ( check_winning_cn, majority_vote_cn, get_chinese_name, format_player_list, GameModerator, MAX_GAME_ROUND, MAX_DISCUSSION_ROUND, ) class ThreeKingdomsWerewolfGame: """三国狼人杀游戏主类""" def __init__(self): self.players: Dict[str, ReActAgent] = {} self.roles: Dict[str, str] = {} self.moderator = GameModerator() self.alive_players: List[ReActAgent] = [] self.werewolves: List[ReActAgent] = [] self.villagers: List[ReActAgent] = [] self.seer: List[ReActAgent] = [] self.witch: List[ReActAgent] = [] self.hunter: List[ReActAgent] = [] # 女巫道具状态 self.witch_has_antidote = True self.witch_has_poison = True async def create_player(self, role: str, character: str) -> ReActAgent: """创建具有三国背景的玩家""" name = get_chinese_name(character) self.roles[name] = role agent = ReActAgent( name=name, sys_prompt=ChinesePrompts.get_role_prompt(role, character), model=DashScopeChatModel( model_name="qwen-max", api_key=os.environ["DASHSCOPE_API_KEY"], enable_thinking=True, ), formatter=DashScopeMultiAgentFormatter(), ) # 角色身份确认 await agent.observe( await self.moderator.announce( f"【{name}】你在这场三国狼人杀中扮演{GameRoles.get_role_desc(role)}，" f"你的角色是{character}。{GameRoles.get_role_ability(role)}" ) ) self.players[name] = agent return agent async def setup_game(self, player_count: int = 6): """设置游戏""" print("🎮 开始设置三国狼人杀游戏...") # 获取角色配置 roles = GameRoles.get_standard_setup(player_count) characters = random.sample([ "刘备", "关羽", "张飞", "诸葛亮", "赵云", "曹操", "司马懿", "周瑜", "孙权" ], player_count) # 创建玩家 for i, (role, character) in enumerate(zip(roles, characters)): agent = await self.create_player(role, character) self.alive_players.append(agent) # 分配到对应阵营 if role == "狼人": self.werewolves.append(agent) elif role == "预言家": self.seer.append(agent) elif role == "女巫": self.witch.append(agent) elif role == "猎人": self.hunter.append(agent) else: self.villagers.append(agent) # 游戏开始公告 await self.moderator.announce( f"三国狼人杀游戏开始！参与者：{format_player_list(self.alive_players)}" ) print(f"✅ 游戏设置完成，共{len(self.alive_players)}名玩家") async def werewolf_phase(self, round_num: int): """狼人阶段""" if not self.werewolves: return None await self.moderator.announce(f"🐺 狼人请睁眼，选择今晚要击杀的目标...") # 狼人讨论 async with MsgHub( self.werewolves, enable_auto_broadcast=True, announcement=await self.moderator.announce( f"狼人们，请讨论今晚的击杀目标。存活玩家：{format_player_list(self.alive_players)}" ), ) as werewolves_hub: # 讨论阶段 for _ in range(MAX_DISCUSSION_ROUND): for wolf in self.werewolves: await wolf(structured_model=DiscussionModelCN) # 投票击杀 werewolves_hub.set_auto_broadcast(False) kill_votes = await fanout_pipeline( self.werewolves, msg=await self.moderator.announce("请选择击杀目标"), structured_model=WerewolfKillModelCN, enable_gather=False, ) # 统计投票 votes = {} for i, vote_msg in enumerate(kill_votes): votes[self.werewolves[i].name] = vote_msg.metadata.get("target") killed_player, _ = majority_vote_cn(votes) return killed_player async def seer_phase(self): """预言家阶段""" if not self.seer: return seer_agent = self.seer[0] await self.moderator.announce("🔮 预言家请睁眼，选择要查验的玩家...") check_result = await seer_agent( structured_model=get_seer_model_cn(self.alive_players) ) target_name = check_result.metadata.get("target") target_role = self.roles.get(target_name, "村民") # 告知预言家结果 result_msg = f"查验结果：{target_name}是{'狼人' if target_role == '狼人' else '好人'}" await seer_agent.observe(await self.moderator.announce(result_msg)) async def witch_phase(self, killed_player: str): """女巫阶段""" if not self.witch: return killed_player, None witch_agent = self.witch[0] await self.moderator.announce("🧙‍♀️ 女巫请睁眼...") # 告知女巫死亡信息 death_info = f"今晚{killed_player}被狼人击杀" if killed_player else "今晚平安无事" await witch_agent.observe(await self.moderator.announce(death_info)) # 女巫行动 witch_action = await witch_agent(structured_model=WitchActionModelCN) saved_player = None poisoned_player = None if witch_action.metadata.get("use_antidote") and self.witch_has_antidote: if killed_player: saved_player = killed_player self.witch_has_antidote = False await witch_agent.observe(await self.moderator.announce(f"你使用解药救了{killed_player}")) if witch_action.metadata.get("use_poison") and self.witch_has_poison: poisoned_player = witch_action.metadata.get("target_name") if poisoned_player: self.witch_has_poison = False await witch_agent.observe(await self.moderator.announce(f"你使用毒药毒杀了{poisoned_player}")) # 确定最终死亡玩家 final_killed = killed_player if not saved_player else None return final_killed, poisoned_player async def hunter_phase(self, shot_by_hunter: str): """猎人阶段""" if not self.hunter: return None hunter_agent = self.hunter[0] if hunter_agent.name == shot_by_hunter: await self.moderator.announce("🏹 猎人发动技能，可以带走一名玩家...") hunter_action = await hunter_agent( structured_model=get_hunter_model_cn(self.alive_players) ) if hunter_action.metadata.get("shoot"): target = hunter_action.metadata.get("target") await self.moderator.announce(f"猎人{hunter_agent.name}开枪带走了{target}") return target return None def update_alive_players(self, dead_players: List[str]): """更新存活玩家列表""" for dead_name in dead_players: if dead_name: # 从存活列表移除 self.alive_players = [p for p in self.alive_players if p.name != dead_name] # 从各阵营移除 self.werewolves = [p for p in self.werewolves if p.name != dead_name] self.villagers = [p for p in self.villagers if p.name != dead_name] self.seer = [p for p in self.seer if p.name != dead_name] self.witch = [p for p in self.witch if p.name != dead_name] self.hunter = [p for p in self.hunter if p.name != dead_name] async def day_phase(self, round_num: int): """白天阶段""" await self.moderator.day_announcement(round_num) # 讨论阶段 async with MsgHub( self.alive_players, enable_auto_broadcast=True, announcement=await self.moderator.announce( f"现在开始自由讨论。存活玩家：{format_player_list(self.alive_players)}" ), ) as all_hub: # 每人发言一轮 await sequential_pipeline(self.alive_players) # 投票阶段 all_hub.set_auto_broadcast(False) vote_msgs = await fanout_pipeline( self.alive_players, await self.moderator.announce("请投票选择要淘汰的玩家"), structured_model=get_vote_model_cn(self.alive_players), enable_gather=False, ) # 统计投票 votes = {} for i, vote_msg in enumerate(vote_msgs): votes[self.alive_players[i].name] = vote_msg.metadata.get("vote") voted_out, vote_count = majority_vote_cn(votes) await self.moderator.vote_result_announcement(voted_out, vote_count) return voted_out async def run_game(self): """运行游戏主循环""" try: await self.setup_game() for round_num in range(1, MAX_GAME_ROUND + 1): print(f"\n🌙 === 第{round_num}轮游戏开始 ===") # 夜晚阶段 await self.moderator.night_announcement(round_num) # 狼人击杀 killed_player = await self.werewolf_phase(round_num) # 预言家查验 await self.seer_phase() # 女巫行动 final_killed, poisoned_player = await self.witch_phase(killed_player) # 更新死亡玩家 night_deaths = [p for p in [final_killed, poisoned_player] if p] self.update_alive_players(night_deaths) # 死亡公告 await self.moderator.death_announcement(night_deaths) # 检查胜利条件 winner = check_winning_cn(self.alive_players, self.roles) if winner: await self.moderator.game_over_announcement(winner) return # 白天阶段 voted_out = await self.day_phase(round_num) # 猎人技能 hunter_shot = await self.hunter_phase(voted_out) # 更新死亡玩家 day_deaths = [p for p in [voted_out, hunter_shot] if p] self.update_alive_players(day_deaths) # 检查胜利条件 winner = check_winning_cn(self.alive_players, self.roles) if winner: await self.moderator.game_over_announcement(winner) return print(f"第{round_num}轮结束，存活玩家：{format_player_list(self.alive_players)}") except Exception as e: print(f"❌ 游戏运行出错：{e}") import traceback traceback.print_exc() async def main(): """主函数""" # 检查环境变量 if "DASHSCOPE_API_KEY" not in os.environ: print("❌ 请设置环境变量 DASHSCOPE_API_KEY") return print("🎮 欢迎来到三国狼人杀！") # 创建并运行游戏 game = ThreeKingdomsWerewolfGame() await game.run_game() if __name__ == "__main__": asyncio.run(main())