From b568b8f4a0a89ecbd2db29f191e160020934aecd Mon Sep 17 00:00:00 2001 From: better629 Date: Wed, 10 Apr 2024 14:04:37 +0800 Subject: [PATCH] update werewolf to meet WerewolfEnv --- examples/werewolf_game/start_game.py | 2 +- metagpt/environment/werewolf/const.py | 106 ++++++++ metagpt/environment/werewolf/env_space.py | 49 ++++ metagpt/environment/werewolf/werewolf_env.py | 5 +- .../environment/werewolf/werewolf_ext_env.py | 246 ++++++++---------- .../ext/werewolf/actions/moderator_actions.py | 7 +- metagpt/ext/werewolf/actions/witch_actions.py | 3 +- metagpt/ext/werewolf/roles/base_player.py | 89 ++++--- metagpt/ext/werewolf/roles/guard.py | 5 +- metagpt/ext/werewolf/roles/human_player.py | 10 +- metagpt/ext/werewolf/roles/moderator.py | 221 ++++++---------- metagpt/ext/werewolf/roles/seer.py | 5 +- metagpt/ext/werewolf/roles/villager.py | 5 +- metagpt/ext/werewolf/roles/werewolf.py | 5 +- metagpt/ext/werewolf/roles/witch.py | 9 +- metagpt/ext/werewolf/schema.py | 9 +- metagpt/ext/werewolf/werewolf_game.py | 4 +- .../werewolf_env/test_werewolf_ext_env.py | 19 +- 18 files changed, 451 insertions(+), 348 deletions(-) create mode 100644 metagpt/environment/werewolf/const.py create mode 100644 metagpt/environment/werewolf/env_space.py diff --git a/examples/werewolf_game/start_game.py b/examples/werewolf_game/start_game.py index 72023eed7..db20032e9 100644 --- a/examples/werewolf_game/start_game.py +++ b/examples/werewolf_game/start_game.py @@ -29,7 +29,7 @@ async def start_game( use_memory_selection=use_memory_selection, new_experience_version=new_experience_version, ) - logger.info(f"game_setup\n{game_setup}") + logger.info(f"{game_setup}") players = [Moderator()] + players game.hire(players) diff --git a/metagpt/environment/werewolf/const.py b/metagpt/environment/werewolf/const.py new file mode 100644 index 000000000..7fcf2c4cf --- /dev/null +++ b/metagpt/environment/werewolf/const.py @@ -0,0 +1,106 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# @Desc : + +from enum import Enum +from metagpt.const import MESSAGE_ROUTE_TO_ALL + + +class RoleType(Enum): + VILLAGER = "Villager" + WEREWOLF = "Werewolf" + GUARD = "Guard" + SEER = "Seer" + WITCH = "Witch" + MODERATOR = "Moderator" + + +class RoleState(Enum): + ALIVE = "alive" # the role is alive + DEAD = "dead" # killed or poisoned + KILLED = "killed" # killed by werewolf or voting + POISONED = "poisoned" # killed by poison + SAVED = "saved" # saved by antidote + PROTECTED = "projected" # projected by guard + + +class RoleActionRes(Enum): + SAVE = "save" + PASS = "pass" # ignore current action output + + +# the ordered rules by the moderator to announce to everyone each step +STEP_INSTRUCTIONS = { + 0: { + "content": "It’s dark, everyone close your eyes. I will talk with you/your team secretly at night.", + "send_to": {"Moderator"}, # for moderator to continue speaking + "restricted_to": {}, + }, + 1: { + "content": "Guard, please open your eyes!", + "send_to": {"Moderator"}, # for moderator to continue speaking + "restricted_to": {}, + }, + 2: { + "content": """Guard, now tell me who you protect tonight? + You only choose one from the following living options please: {living_players}. + Or you can pass. For example: Protect ...""", + "send_to": {"Guard"}, + "restricted_to": {"Moderator", "Guard"}, + }, + 3: {"content": "Guard, close your eyes", "send_to": {"Moderator"}, "restricted_to": {}}, + 4: {"content": "Werewolves, please open your eyes!", "send_to": {"Moderator"}, "restricted_to": {}}, + 5: { + "content": """Werewolves, I secretly tell you that {werewolf_players} are + all of the 2 werewolves! Keep in mind you are teammates. The rest players are not werewolves. + choose one from the following living options please: + {living_players}. For example: Kill ...""", + "send_to": {"Werewolf"}, + "restricted_to": {"Moderator", "Werewolf"}, + }, + 6: {"content": "Werewolves, close your eyes", "send_to": {"Moderator"}, "restricted_to": {}}, + 7: {"content": "Witch, please open your eyes!", "send_to": {"Moderator"}, "restricted_to": {}}, + 8: { + "content": """Witch, tonight {player_hunted} has been killed by the werewolves. + You have a bottle of antidote, would you like to save him/her? If so, say "Save", else, say "Pass".""", + "send_to": {"Witch"}, + "restricted_to": {"Moderator", "Witch"}, + }, # 要先判断女巫是否有解药，再去询问女巫是否使用解药救人 + 9: { + "content": """Witch, you also have a bottle of poison, would you like to use it to kill one of the living players? + Choose one from the following living options: {living_players}. + If so, say ONLY "Poison PlayerX", replace PlayerX with the actual player name, else, say "Pass".""", + "send_to": {"Witch"}, + "restricted_to": {"Moderator", "Witch"}, + }, # + 10: {"content": "Witch, close your eyes", "send_to": {"Moderator"}, "restricted_to": {}}, + 11: {"content": "Seer, please open your eyes!", "send_to": {"Moderator"}, "restricted_to": {}}, + 12: { + "content": """Seer, you can check one player's identity. Who are you going to verify its identity tonight? + Choose only one from the following living options:{living_players}.""", + "send_to": {"Seer"}, + "restricted_to": {"Moderator", "Seer"}, + }, + 13: {"content": "Seer, close your eyes", "send_to": {"Moderator"}, "restricted_to": {}}, + # The 1-st daytime + 14: { + "content": """It's daytime. Everyone woke up except those who had been killed.""", + "send_to": {"Moderator"}, + "restricted_to": {}, + }, + 15: {"content": "{player_current_dead} was killed last night!", "send_to": {"Moderator"}, "restricted_to": {}}, + 16: { + "content": """Living players: {living_players}, now freely talk about the current situation based on your observation and + reflection with a few sentences. Decide whether to reveal your identity based on your reflection.""", + "send_to": {MESSAGE_ROUTE_TO_ALL}, # send to all to speak in daytime + "restricted_to": {}, + }, + 17: { + "content": """Now vote and tell me who you think is the werewolf. Don’t mention your role. + You only choose one from the following living options please: + {living_players}. Say ONLY: I vote to eliminate ...""", + "send_to": {MESSAGE_ROUTE_TO_ALL}, + "restricted_to": {}, + }, + 18: {"content": """{player_current_dead} was eliminated.""", "send_to": {"Moderator"}, "restricted_to": {}}, +} diff --git a/metagpt/environment/werewolf/env_space.py b/metagpt/environment/werewolf/env_space.py new file mode 100644 index 000000000..f8cedcd55 --- /dev/null +++ b/metagpt/environment/werewolf/env_space.py @@ -0,0 +1,49 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# @Desc : werewolf observation/action space and its action definition + +from pydantic import ConfigDict, Field +from gymnasium import spaces + +from metagpt.environment.base_env_space import BaseEnvAction, BaseEnvActionType +from metagpt.environment.werewolf.const import STEP_INSTRUCTIONS, RoleState + + +class EnvActionType(BaseEnvActionType): + NONE = 0 # no action to run, just get observation + WOLF_KILL = 1 # wolf kill someone + VOTE_KILL = 2 # vote kill someone + WITCH_POISON = 3 # witch poison someone + WITCH_SAVE = 4 # witch save someone + GUARD_PROTECT = 5 # guard protect someone + PROGRESS_STEP = 6 # step increment + + +class EnvAction(BaseEnvAction): + model_config = ConfigDict(arbitrary_types_allowed=True) + + action_type: int = Field(default=EnvActionType.NONE, description="action type") + player_name: str = Field(default="", description="the name of the player to do the action") + target_player_name: str = Field(default="", description="the name of the player who take the action") + + +def get_observation_space(player_num: int) -> spaces.Dict: + space = spaces.Dict({ + "step_idx": spaces.Discrete(len(STEP_INSTRUCTIONS)), + "player_states": spaces.MultiDiscrete([len(RoleState)] * player_num), + "vote_counts": spaces.MultiDiscrete([player_num - 1] * player_num), + + "player_current_dead": None, # TODO + "winner": spaces.Text(16), + "win_reason": spaces.Text(64) + }) + return space + + +def get_action_space() -> spaces.Dict: + space = spaces.Dict({ + "action_type": spaces.Discrete(len(EnvActionType)), + "player_name": spaces.Text(16), # the player to do the action + "target_player_name": spaces.Text(16) # the target player who take the action + }) + return space diff --git a/metagpt/environment/werewolf/werewolf_env.py b/metagpt/environment/werewolf/werewolf_env.py index e22d9c46f..728f1309c 100644 --- a/metagpt/environment/werewolf/werewolf_env.py +++ b/metagpt/environment/werewolf/werewolf_env.py @@ -6,7 +6,6 @@ from pydantic import Field from metagpt.environment.base_env import Environment from metagpt.environment.werewolf.werewolf_ext_env import WerewolfExtEnv -from metagpt.logs import logger from metagpt.schema import Message @@ -15,13 +14,11 @@ class WerewolfEnv(Environment, WerewolfExtEnv): def publish_message(self, message: Message, add_timestamp: bool = True): """Post information to the current environment""" - logger.debug(f"publish_message: {message.dump()}") if add_timestamp: # Because the content of the message may be repeated, for example, killing the same person in two nights # Therefore, a unique timestamp prefix needs to be added so that the same message will not be automatically deduplicated when added to the memory. message.content = f"{self.timestamp} | " + message.content - self.memory.add(message) - self.history += f"\n{message}" + super().publish_message(message) async def run(self, k=1): """Process all Role runs by order""" diff --git a/metagpt/environment/werewolf/werewolf_ext_env.py b/metagpt/environment/werewolf/werewolf_ext_env.py index 3f2508b06..ac92a5431 100644 --- a/metagpt/environment/werewolf/werewolf_ext_env.py +++ b/metagpt/environment/werewolf/werewolf_ext_env.py @@ -4,98 +4,15 @@ import random from collections import Counter -from enum import Enum from typing import Any, Callable, Optional from pydantic import ConfigDict, Field from metagpt.environment.base_env import ExtEnv, mark_as_readable, mark_as_writeable -from metagpt.environment.base_env_space import BaseEnvAction, BaseEnvObsParams +from metagpt.environment.base_env_space import BaseEnvObsParams +from metagpt.environment.werewolf.env_space import EnvAction, EnvActionType from metagpt.logs import logger - - -class RoleState(Enum): - ALIVE = "alive" # the role is alive - KILLED = "killed" # the role is killed by werewolf or voting - POISONED = "poisoned" # the role is killed by posion - SAVED = "saved" # the role is saved by antidote - - -# the ordered rules by the moderator to announce to everyone each step -STEP_INSTRUCTIONS = { - 0: { - "content": "It’s dark, everyone close your eyes. I will talk with you/your team secretly at night.", - "send_to": "Moderator", # for moderator to continuen speaking - "restricted_to": "", - }, - 1: { - "content": "Guard, please open your eyes!", - "send_to": "Moderator", # for moderator to continuen speaking - "restricted_to": "", - }, - 2: { - "content": """Guard, now tell me who you protect tonight? - You only choose one from the following living options please: {living_players}. - Or you can pass. For example: Protect ...""", - "send_to": "Guard", - "restricted_to": "Moderator,Guard", - }, - 3: {"content": "Guard, close your eyes", "send_to": "Moderator", "restricted_to": ""}, - 4: {"content": "Werewolves, please open your eyes!", "send_to": "Moderator", "restricted_to": ""}, - 5: { - "content": """Werewolves, I secretly tell you that {werewolf_players} are - all of the 2 werewolves! Keep in mind you are teammates. The rest players are not werewolves. - choose one from the following living options please: - {living_players}. For example: Kill ...""", - "send_to": "Werewolf", - "restricted_to": "Moderator,Werewolf", - }, - 6: {"content": "Werewolves, close your eyes", "send_to": "Moderator", "restricted_to": ""}, - 7: {"content": "Witch, please open your eyes!", "send_to": "Moderator", "restricted_to": ""}, - 8: { - "content": """Witch, tonight {player_hunted} has been killed by the werewolves. - You have a bottle of antidote, would you like to save him/her? If so, say "Save", else, say "Pass".""", - "send_to": "Witch", - "restricted_to": "Moderator,Witch", - }, # 要先判断女巫是否有解药，再去询问女巫是否使用解药救人 - 9: { - "content": """Witch, you also have a bottle of poison, would you like to use it to kill one of the living players? - Choose one from the following living options: {living_players}. - If so, say ONLY "Poison PlayerX", replace PlayerX with the actual player name, else, say "Pass".""", - "send_to": "Witch", - "restricted_to": "Moderator,Witch", - }, # - 10: {"content": "Witch, close your eyes", "send_to": "Moderator", "restricted_to": ""}, - 11: {"content": "Seer, please open your eyes!", "send_to": "Moderator", "restricted_to": ""}, - 12: { - "content": """Seer, you can check one player's identity. Who are you going to verify its identity tonight? - Choose only one from the following living options:{living_players}.""", - "send_to": "Seer", - "restricted_to": "Moderator,Seer", - }, - 13: {"content": "Seer, close your eyes", "send_to": "Moderator", "restricted_to": ""}, - # The 1-st daytime - 14: { - "content": """It's daytime. Everyone woke up except those who had been killed.""", - "send_to": "Moderator", - "restricted_to": "", - }, - 15: {"content": "{player_current_dead} was killed last night!", "send_to": "Moderator", "restricted_to": ""}, - 16: { - "content": """Living players: {living_players}, now freely talk about the current situation based on your observation and - reflection with a few sentences. Decide whether to reveal your identity based on your reflection.""", - "send_to": "", # send to all to speak in daytime - "restricted_to": "", - }, - 17: { - "content": """Now vote and tell me who you think is the werewolf. Don’t mention your role. - You only choose one from the following living options please: - {living_players}. Say ONLY: I vote to eliminate ...""", - "send_to": "", - "restricted_to": "", - }, - 18: {"content": """{player_current_dead} was eliminated.""", "send_to": "Moderator", "restricted_to": ""}, -} +from metagpt.environment.werewolf.const import STEP_INSTRUCTIONS, RoleState, RoleType class WerewolfExtEnv(ExtEnv): @@ -115,13 +32,13 @@ class WerewolfExtEnv(ExtEnv): special_role_players: list[str] = Field(default=[]) winner: Optional[str] = Field(default=None) win_reason: Optional[str] = Field(default=None) - witch_poison_left: int = Field(default=1) - witch_antidote_left: int = Field(default=1) + witch_poison_left: int = Field(default=1, description="should be 1 or 0") + witch_antidote_left: int = Field(default=1, description="should be 1 or 0") # game current round states, a round is from closing your eyes to the next time you close your eyes round_hunts: dict[str, str] = Field(default=dict(), description="nighttime wolf hunt result") round_votes: dict[str, str] = Field( - default=dict(), description="daytime all players vote result, key=voteer, value=voted one" + default=dict(), description="daytime all players vote result, key=voter, value=voted one" ) player_hunted: Optional[str] = Field(default=None) player_protected: Optional[str] = Field(default=None) @@ -140,8 +57,63 @@ class WerewolfExtEnv(ExtEnv): def observe(self, obs_params: Optional[BaseEnvObsParams] = None) -> Any: pass - def step(self, action: BaseEnvAction) -> tuple[dict[str, Any], float, bool, bool, dict[str, Any]]: - pass + def _get_obs(self): + return { + "game_setup": self.game_setup, + "step_idx": self.step_idx, + + "living_players": self.living_players, + "werewolf_players": self.werewolf_players, + "player_hunted": self.player_hunted, + "player_current_dead": self.player_current_dead, + "witch_poison_left": self.witch_poison_left, + "witch_antidote_left": self.witch_antidote_left, + "winner": self.winner, + "win_reason": self.win_reason + } + + def step(self, action: EnvAction) -> tuple[dict[str, Any], float, bool, bool, dict[str, Any]]: + action_type = action.action_type + player_name = action.player_name + target_player_name = action.target_player_name + if action_type == EnvActionType.WOLF_KILL: + self.wolf_kill_someone(wolf_name=player_name, player_name=target_player_name) + elif action_type == EnvActionType.VOTE_KILL: + self.vote_kill_someone(voter_name=player_name, player_name=target_player_name) + elif action_type == EnvActionType.WITCH_POISON: + self.witch_poison_someone(witch_name=player_name, player_name=target_player_name) + elif action_type == EnvActionType.WITCH_SAVE: + self.witch_save_someone(witch_name=player_name, player_name=target_player_name) + elif action_type == EnvActionType.GUARD_PROTECT: + self.guard_protect_someone(guard_name=player_name, player_name=target_player_name) + elif action_type == EnvActionType.PROGRESS_STEP: + self.progress_step() + elif action_type == EnvActionType.NONE: + pass + else: + raise ValueError(f"not supported action_type: {action_type}") + + self.update_game_states() + terminated = self._check_game_finish() + obs = self._get_obs() + return obs, 1.0, terminated, False, {} + + def _check_game_finish(self) -> bool: + """return True if game finished else False""" + # game's termination condition + terminated = False + living_werewolf = [p for p in self.werewolf_players if p in self.living_players] + living_villagers = [p for p in self.villager_players if p in self.living_players] + living_special_roles = [p for p in self.special_role_players if p in self.living_players] + if not living_werewolf: + self.winner = "good guys" + self.win_reason = "werewolves all dead" + terminated = True + elif not living_villagers or not living_special_roles: + self.winner = "werewolf" + self.win_reason = "villagers all dead" if not living_villagers else "special roles all dead" + terminated = True + return terminated @property def living_players(self) -> list[str]: @@ -161,12 +133,12 @@ class WerewolfExtEnv(ExtEnv): @property def werewolf_players(self) -> list[str]: - player_names = self._role_type_players(role_type="Werewolf") + player_names = self._role_type_players(role_type=RoleType.WEREWOLF.value) return player_names @property def villager_players(self) -> list[str]: - player_names = self._role_type_players(role_type="Villager") + player_names = self._role_type_players(role_type=RoleType.VILLAGER.value) return player_names def _init_players_state(self, players: list["Role"]): @@ -193,14 +165,14 @@ class WerewolfExtEnv(ExtEnv): """init players using different roles' num""" role_objs = [] for role_obj in role_uniq_objs: - if str(role_obj) == "Villager": + if "Villager" in str(role_obj): role_objs.extend([role_obj] * num_villager) - elif str(role_obj) == "Werewolf": + elif "Werewolf" in str(role_obj): role_objs.extend([role_obj] * num_werewolf) else: role_objs.append(role_obj) if shuffle: - random.shuffle(len(role_objs)) + random.shuffle(role_objs) if add_human: assigned_role_idx = random.randint(0, len(role_objs) - 1) assigned_role = role_objs[assigned_role_idx] @@ -233,10 +205,12 @@ class WerewolfExtEnv(ExtEnv): roletype_state = self.players_state[player_name] self.players_state[player_name] = (roletype_state[0], state) - def _check_valid_role(self, player: "Role", role_type: str) -> bool: - return True if role_type in str(player) else False + def _check_valid_role(self, player_name: str, role_type: str) -> bool: + roletype_state = self.players_state.get(player_name) + return True if roletype_state and role_type in roletype_state[0] else False def _check_player_continue(self, player_name: str, particular_step: int = -1) -> bool: + """to check if can do the operation to the player""" step_idx = self.step_idx % self.per_round_steps if particular_step > 0 and step_idx != particular_step: # step no # particular_step = 18, not daytime vote time, ignore @@ -253,6 +227,10 @@ class WerewolfExtEnv(ExtEnv): self.step_idx += 1 return instruction + @mark_as_writeable + def progress_step(self): + self.step_idx += 1 + @mark_as_readable def get_players_state(self, player_names: list[str]) -> dict[str, RoleState]: players_state = { @@ -263,14 +241,14 @@ class WerewolfExtEnv(ExtEnv): return players_state @mark_as_writeable - def vote_kill_someone(self, voteer: "Role", player_name: str = None): + def vote_kill_someone(self, voter_name: str, player_name: str = None): """player vote result at daytime player_name: if it's None, regard as abstaining from voting """ - if not self._check_player_continue(voteer.name, particular_step=18): # 18=step no + if not self._check_player_continue(voter_name, particular_step=18): # 18=step no return - self.round_votes[voteer.name] = player_name + self.round_votes[voter_name] = player_name # check if all living players finish voting, then get the dead one if list(self.round_votes.keys()) == self.living_players: voted_all = list(self.round_votes.values()) # TODO in case of tie vote, check who was voted first @@ -279,41 +257,55 @@ class WerewolfExtEnv(ExtEnv): self._update_players_state([self.player_current_dead]) @mark_as_writeable - def wolf_kill_someone(self, wolf: "Role", player_name: str): - if not self._check_valid_role(wolf, "Werewolf"): + def wolf_kill_someone(self, wolf_name: str, player_name: str): + if not self._check_valid_role(wolf_name, RoleType.WEREWOLF.value): return - if not self._check_player_continue(wolf.name, particular_step=5): # 5=step no + if not self._check_player_continue(wolf_name, particular_step=5): # 5=step no return - self.round_hunts[wolf.name] = player_name - living_werewolf = [p for p in self.werewolf_players if p in self.living_players] + self.round_hunts[wolf_name] = player_name + # living_werewolf = [p for p in self.werewolf_players if p in self.living_players] # check if all living wolfs finish hunting, then get the hunted one - if list(self.round_hunts.keys()) == living_werewolf: - hunted_all = list(self.round_hunts.values()) - self.player_hunted = Counter(hunted_all).most_common()[0][0] + # if list(self.round_hunts.keys()) == living_werewolf: + # hunted_all = list(self.round_hunts.values()) + # self.player_hunted = Counter(hunted_all).most_common()[0][0] + self.player_hunted = player_name - @mark_as_writeable - def witch_poison_someone(self, witch: "Role", player_name: str = None): - if not self._check_valid_role(witch, "Witch"): + def _witch_poison_or_save_someone(self, witch_name: str, player_name: str = None, + state: RoleState = RoleState.POISONED): + if not self._check_valid_role(witch_name, RoleType.WITCH.value): return if not self._check_player_continue(player_name): return - self._update_players_state([player_name], RoleState.POISONED) - self.player_poisoned = player_name + assert state in [RoleState.POISONED, RoleState.SAVED] + self._update_players_state([player_name], state) + if state == RoleState.POISONED: + self.player_poisoned = player_name + self.witch_poison_left -= 1 + else: + # self.player_protected = player_name + self.is_hunted_player_saved = True + self.witch_antidote_left -= 1 @mark_as_writeable - def witch_save_someone(self, witch: "Role", player_name: str = None): - if not self._check_valid_role(witch, "Witch"): + def witch_poison_someone(self, witch_name: str, player_name: str = None): + self._witch_poison_or_save_someone(witch_name, player_name, RoleState.POISONED) + + @mark_as_writeable + def witch_save_someone(self, witch_name: str, player_name: str = None): + self._witch_poison_or_save_someone(witch_name, player_name, RoleState.SAVED) + + @mark_as_writeable + def guard_protect_someone(self, guard_name: str, player_name: str = None): + if not self._check_valid_role(guard_name, RoleType.GUARD.value): return if not self._check_player_continue(player_name): return - - self._update_players_state([player_name], RoleState.SAVED) self.player_protected = player_name @mark_as_writeable - def update_game_states(self, memories: list): + def update_game_states(self): step_idx = self.step_idx % self.per_round_steps if step_idx not in [15, 18] or self.step_idx in self.eval_step_idx: return @@ -335,16 +327,6 @@ class WerewolfExtEnv(ExtEnv): self.player_protected = None self.is_hunted_player_saved = False self.player_poisoned = None - - # game's termination condition - living_werewolf = [p for p in self.werewolf_players if p in self.living_players] - living_villagers = [p for p in self.villager_players if p in self.living_players] - living_special_roles = [p for p in self.special_role_players if p in self.living_players] - if not living_werewolf: - self.winner = "good guys" - self.win_reason = "werewolves all dead" - elif not living_villagers or not living_special_roles: - self.winner = "werewolf" - self.win_reason = "villagers all dead" if not living_villagers else "special roles all dead" - if self.winner is not None: - self._record_all_experiences() # TODO + elif step_idx == 18: + # updated use vote_kill_someone + pass diff --git a/metagpt/ext/werewolf/actions/moderator_actions.py b/metagpt/ext/werewolf/actions/moderator_actions.py index 42bd0fc4e..938337819 100644 --- a/metagpt/ext/werewolf/actions/moderator_actions.py +++ b/metagpt/ext/werewolf/actions/moderator_actions.py @@ -1,12 +1,13 @@ from metagpt.actions import Action -from metagpt.environment.werewolf.werewolf_ext_env import STEP_INSTRUCTIONS +from metagpt.environment.werewolf.const import STEP_INSTRUCTIONS class InstructSpeak(Action): name: str = "InstructSpeak" async def run(self, step_idx, living_players, werewolf_players, player_hunted, player_current_dead): - instruction_info = STEP_INSTRUCTIONS.get(step_idx, {"content": "Unknown instruction.", "send_to": {}}) + instruction_info = STEP_INSTRUCTIONS.get(step_idx, {"content": "Unknown instruction.", "send_to": {}, + "restricted_to": {}}) content = instruction_info["content"] if "{living_players}" in content and "{werewolf_players}" in content: content = content.format(living_players=living_players, werewolf_players=werewolf_players) @@ -20,7 +21,7 @@ class InstructSpeak(Action): player_current_dead = "No one" if not player_current_dead else player_current_dead content = content.format(player_current_dead=player_current_dead) - return content, instruction_info["send_to"] + return content, instruction_info["send_to"], instruction_info["restricted_to"] class ParseSpeak(Action): diff --git a/metagpt/ext/werewolf/actions/witch_actions.py b/metagpt/ext/werewolf/actions/witch_actions.py index 4ea3b27f7..20f6df591 100644 --- a/metagpt/ext/werewolf/actions/witch_actions.py +++ b/metagpt/ext/werewolf/actions/witch_actions.py @@ -1,4 +1,5 @@ from metagpt.ext.werewolf.actions.common_actions import NighttimeWhispers +from metagpt.environment.werewolf.const import RoleActionRes class Save(NighttimeWhispers): @@ -41,6 +42,6 @@ class Poison(NighttimeWhispers): async def run(self, *args, **kwargs): rsp = await super().run(*args, **kwargs) - if "pass" in rsp.lower(): + if RoleActionRes.PASS.value in rsp.lower(): action_name, rsp = rsp.split() # 带PASS，只需回复PASS，不需要带上action名，否则是Poison PlayerX，无需改动 return rsp diff --git a/metagpt/ext/werewolf/roles/base_player.py b/metagpt/ext/werewolf/roles/base_player.py index 6f459526e..770ff54bf 100644 --- a/metagpt/ext/werewolf/roles/base_player.py +++ b/metagpt/ext/werewolf/roles/base_player.py @@ -1,5 +1,7 @@ import re +from pydantic import SerializeAsAny, Field + from metagpt.ext.werewolf.actions import ( ACTIONS, AddNewExperiences, @@ -9,65 +11,72 @@ from metagpt.ext.werewolf.actions import ( RetrieveExperiences, Speak, ) -from metagpt.ext.werewolf.schema import RoleExperience +from metagpt.ext.werewolf.schema import RoleExperience, WwMessage +from metagpt.environment.werewolf.const import RoleType, RoleState from metagpt.logs import logger from metagpt.roles import Role -from metagpt.schema import Message +from metagpt.actions.action import Action class BasePlayer(Role): - def __init__( - self, - name: str = "PlayerXYZ", - profile: str = "BasePlayer", - special_action_names: list[str] = [], - use_reflection: bool = True, - use_experience: bool = False, - use_memory_selection: bool = False, - new_experience_version: str = "", - **kwargs, - ): - super().__init__(name, profile, **kwargs) - # 通过 set_status() 更新状态。 - self.status = 0 # 0代表活着，1代表死亡 + name: str = "PlayerXYZ" + profile: str = "BasePlayer" + special_action_names: list[str] = [] + use_reflection: bool = True + use_experience: bool = False + use_memory_selection: bool = False + new_experience_version: str = "" + status: RoleState = RoleState.ALIVE + special_actions: list[SerializeAsAny[Action]] = Field(default=[], validate_default=True) + experiences: list[RoleExperience] = [] + + def __init__(self, **kwargs): + super().__init__(**kwargs) # 技能和监听配置 self._watch([InstructSpeak]) # 监听Moderator的指令以做行动 - special_actions = [ACTIONS[action_name] for action_name in special_action_names] + special_actions = [ACTIONS[action_name] for action_name in self.special_action_names] capable_actions = [Speak] + special_actions self.set_actions(capable_actions) # 给角色赋予行动技能 self.special_actions = special_actions - self.use_reflection = use_reflection - if not self.use_reflection and use_experience: + if not self.use_reflection and self.use_experience: logger.warning("You must enable use_reflection before using experience") self.use_experience = False - else: - self.use_experience = use_experience - self.new_experience_version = new_experience_version - self.use_memory_selection = use_memory_selection - self.experiences = [] - - async def _observe(self) -> int: - if self.status == 1: + async def _observe(self, ignore_memory=False) -> int: + if self.status != RoleState.ALIVE: # 死者不再参与游戏 return 0 + news = [] + if not news: + news = self.rc.msg_buffer.pop_all() + old_messages = [] if ignore_memory else self.rc.memory.get() + for m in news: + if len(m.restricted_to) and self.profile not in m.restricted_to and self.name not in m.restricted_to: + # if the msg is not send to the whole audience ("") nor this role (self.profile or self.name), + # then this role should not be able to receive it and record it into its memory + continue + self.rc.memory.add(m) + self.rc.news = [ + n for n in news if (n.cause_by in self.rc.watch or self.profile in n.send_to) and n not in old_messages + ] - await super()._observe() - # 只有发给全体的（""）或发给自己的（self.profile）消息需要走下面的_react流程， - # 其他的收听到即可，不用做动作 - self.rc.news = [msg for msg in self.rc.news if msg.send_to in ["", self.profile]] + # TODO to delete + # await super()._observe() + # # 只有发给全体的（""）或发给自己的（self.profile）消息需要走下面的_react流程， + # # 其他的收听到即可，不用做动作 + # self.rc.news = [msg for msg in self.rc.news if msg.send_to in ["", self.profile]] return len(self.rc.news) async def _think(self): news = self.rc.news[0] assert news.cause_by == InstructSpeak # 消息为来自Moderator的指令时，才去做动作 - if not news.send_to: + if not news.restricted_to: # 消息接收范围为全体角色的，做公开发言（发表投票观点也算发言） self.rc.todo = Speak() - elif self.profile in news.send_to: - # FIXME: hard code to split, restricted为"Moderator"或"Moderator,角色profile" + elif self.profile in news.restricted_to: + # FIXME: hard code to split, restricted为"Moderator"或"Moderator, 角色profile" # Moderator加密发给自己的，意味着要执行角色的特殊动作 self.rc.todo = self.special_actions[0]() @@ -79,7 +88,6 @@ class BasePlayer(Role): # 可以用这个函数获取该角色的全部记忆和最新的instruction memories = self.get_all_memories() latest_instruction = self.get_latest_instruction() - # print("*" * 10, f"{self._setting}'s current memories: {memories}", "*" * 10) reflection = ( await Reflect().run( @@ -107,20 +115,21 @@ class BasePlayer(Role): reflection=reflection, experiences=experiences, ) - send_to = "" + restricted_to = {} elif isinstance(todo, NighttimeWhispers): rsp = await todo.run( profile=self.profile, name=self.name, context=memories, reflection=reflection, experiences=experiences ) - send_to = {"Moderator", self.profile} # 给Moderator发送使用特殊技能的加密消息 + restricted_to = {RoleType.MODERATOR.value, self.profile} # 给Moderator发送使用特殊技能的加密消息 - msg = Message( + msg = WwMessage( content=rsp, role=self.profile, sent_from=self.name, cause_by=type(todo), - send_to=send_to, + send_to={}, + restricted_to=restricted_to ) self.experiences.append( @@ -149,7 +158,7 @@ class BasePlayer(Role): def get_latest_instruction(self) -> str: return self.rc.important_memory[-1].content # 角色监听着Moderator的InstructSpeak，是其重要记忆，直接获取即可 - def set_status(self, new_status): + def set_status(self, new_status: RoleState): self.status = new_status def record_experiences(self, round_id: str, outcome: str, game_setup: str): diff --git a/metagpt/ext/werewolf/roles/guard.py b/metagpt/ext/werewolf/roles/guard.py index 1a4471fbd..bd0a1fa30 100644 --- a/metagpt/ext/werewolf/roles/guard.py +++ b/metagpt/ext/werewolf/roles/guard.py @@ -1,7 +1,8 @@ from metagpt.ext.werewolf.roles.base_player import BasePlayer +from metagpt.environment.werewolf.const import RoleType class Guard(BasePlayer): - name: str = "Guard" - profile: str = "Guard" + name: str = RoleType.GUARD.value + profile: str = RoleType.GUARD.value special_action_names: list[str] = ["Protect"] diff --git a/metagpt/ext/werewolf/roles/human_player.py b/metagpt/ext/werewolf/roles/human_player.py index 830bf06cf..3f294b0fb 100644 --- a/metagpt/ext/werewolf/roles/human_player.py +++ b/metagpt/ext/werewolf/roles/human_player.py @@ -1,7 +1,8 @@ from metagpt.ext.werewolf.actions import Speak from metagpt.ext.werewolf.roles import BasePlayer from metagpt.logs import logger -from metagpt.schema import Message +from metagpt.ext.werewolf.schema import WwMessage +from metagpt.environment.werewolf.const import RoleType async def _act(self): @@ -22,14 +23,15 @@ async def _act(self): rsp = input(input_instruction) # wait for human input msg_cause_by = type(todo) - msg_restricted_to = "" if isinstance(todo, Speak) else f"Moderator,{self.profile}" + msg_restricted_to = {} if isinstance(todo, Speak) else {RoleType.MODERATOR.value, self.profile} - msg = Message( + msg = WwMessage( content=rsp, role=self.profile, sent_from=self.name, cause_by=msg_cause_by, - send_to=msg_restricted_to, # 给Moderator及自身阵营发送加密消息 + send_to={}, + restricted_to=msg_restricted_to, # 给Moderator及自身阵营发送加密消息 ) logger.info(f"{self._setting}: {rsp}") diff --git a/metagpt/ext/werewolf/roles/moderator.py b/metagpt/ext/werewolf/roles/moderator.py index f0de03959..09ce0f320 100644 --- a/metagpt/ext/werewolf/roles/moderator.py +++ b/metagpt/ext/werewolf/roles/moderator.py @@ -1,10 +1,11 @@ import re -from collections import Counter +from typing import Union from datetime import datetime from metagpt.actions.add_requirement import UserRequirement from metagpt.const import DEFAULT_WORKSPACE_ROOT -from metagpt.environment.werewolf.werewolf_ext_env import STEP_INSTRUCTIONS +from metagpt.environment.werewolf.const import STEP_INSTRUCTIONS, RoleType +from metagpt.environment.werewolf.env_space import EnvAction, EnvActionType from metagpt.ext.werewolf.actions import Hunt, Poison, Protect, Save, Verify from metagpt.ext.werewolf.actions.moderator_actions import ( AnnounceGameResult, @@ -12,55 +13,29 @@ from metagpt.ext.werewolf.actions.moderator_actions import ( ParseSpeak, ) from metagpt.logs import logger -from metagpt.roles import Role -from metagpt.schema import Message +from metagpt.ext.werewolf.roles.base_player import BasePlayer +from metagpt.ext.werewolf.schema import WwMessage +from metagpt.environment.werewolf.const import RoleState, RoleActionRes -class Moderator(Role): - name: str = "Moderator" - profile: str = "Moderator" +class Moderator(BasePlayer): + name: str = RoleType.MODERATOR.value + profile: str = RoleType.MODERATOR.value - def __init__( - self, - name: str = "Moderator", - profile: str = "Moderator", - **kwargs, - ): - super().__init__(name, profile, **kwargs) + def __init__(self, **kwargs): + super().__init__(**kwargs) self._watch([UserRequirement, InstructSpeak, ParseSpeak]) self.set_actions([InstructSpeak, ParseSpeak, AnnounceGameResult]) - self.step_idx = 0 - self.eval_step_idx = [] # game states + self.step_idx = 0 self.game_setup = "" - self.living_players = [] self.werewolf_players = [] - self.villager_players = [] - self.special_role_players = [] self.winner = None self.win_reason = None self.witch_poison_left = 1 self.witch_antidote_left = 1 - # player states of current night - self.player_hunted = None - self.player_protected = None - self.is_hunted_player_saved = False - self.player_poisoned = None - self.player_current_dead = [] - - def _parse_game_setup(self, game_setup: str): - self.game_setup = game_setup - self.living_players = re.findall(r"Player[0-9]+", game_setup) - self.werewolf_players = re.findall(r"Player[0-9]+: Werewolf", game_setup) - self.werewolf_players = [p.replace(": Werewolf", "") for p in self.werewolf_players] - self.villager_players = re.findall(r"Player[0-9]+: Villager", game_setup) - self.villager_players = [p.replace(": Villager", "") for p in self.villager_players] - self.special_role_players = [ - p for p in self.living_players if p not in self.werewolf_players + self.villager_players - ] - def update_player_status(self, player_names: list[str]): if not player_names: return @@ -68,33 +43,23 @@ class Moderator(Role): for role_setting, role in roles_in_env.items(): for player_name in player_names: if player_name in role_setting: - role.set_status(new_status=1) # 更新为死亡 + role.set_status(new_status=RoleState.DEAD) # 更新为死亡 def _record_all_experiences(self): + logger.info(f"The winner of the game: {self.winner}, start to record roles' experiences") roles_in_env = self.rc.env.get_roles() timestamp = datetime.now().strftime("%Y_%m_%d_%H_%M_%S") for _, role in roles_in_env.items(): if role == self: continue if self.winner == "werewolf": - outcome = "won" if role.name in self.werewolf_players else "lost" + outcome = "won" if role.profile in RoleType.WEREWOLF.value else "lost" else: - outcome = "won" if role.name not in self.werewolf_players else "lost" + outcome = "won" if role.profile not in RoleType.WEREWOLF.value else "lost" role.record_experiences(round_id=timestamp, outcome=outcome, game_setup=self.game_setup) - async def _instruct_speak(self): - step_idx = self.step_idx % len(STEP_INSTRUCTIONS) - self.step_idx += 1 - return await InstructSpeak().run( - step_idx, - living_players=self.living_players, - werewolf_players=self.werewolf_players, - player_hunted=self.player_hunted, - player_current_dead=self.player_current_dead, - ) - async def _parse_speak(self, memories): - logger.info(self.step_idx) + logger.info(f"step_idx: {self.step_idx}") latest_msg = memories[-1] latest_msg_content = latest_msg.content @@ -104,157 +69,135 @@ class Moderator(Role): # default return msg_content = "Understood" - send_to = set() + restricted_to = set() msg_cause_by = latest_msg.cause_by if msg_cause_by == Hunt: - self.player_hunted = target + self.rc.env.step(EnvAction(action_type=EnvActionType.WOLF_KILL, player_name=latest_msg.send_from, + target_player_name=target)) elif msg_cause_by == Protect: - self.player_protected = target + self.rc.env.step(EnvAction(action_type=EnvActionType.GUARD_PROTECT, player_name=latest_msg.send_from, + target_player_name=target)) elif msg_cause_by == Verify: if target in self.werewolf_players: msg_content = f"{target} is a werewolf" else: msg_content = f"{target} is a good guy" - send_to = {"Moderator", "Seer"} + restricted_to = {RoleType.MODERATOR.value, RoleType.SEER.value} elif msg_cause_by == Save: - if "pass" in latest_msg_content.lower(): + if RoleActionRes.PASS.value in latest_msg_content.lower(): + # the role ignore to response, answer `pass` pass elif not self.witch_antidote_left: msg_content = "You have no antidote left and thus can not save the player" - send_to = {"Moderator", "Witch"} + restricted_to = {RoleType.MODERATOR.value, RoleType.WITCH.value} else: - self.witch_antidote_left -= 1 - self.is_hunted_player_saved = True + self.rc.env.step(EnvAction(action_type=EnvActionType.WITCH_SAVE, player_name=latest_msg.send_from, + target_player_name=target)) elif msg_cause_by == Poison: - if "pass" in latest_msg_content.lower(): + if RoleActionRes.PASS.value in latest_msg_content.lower(): pass elif not self.witch_poison_left: msg_content = "You have no poison left and thus can not poison the player" - send_to = {"Moderator", "Witch"} + restricted_to = {RoleType.MODERATOR.value, RoleType.WITCH.value} else: - self.witch_poison_left -= 1 - self.player_poisoned = target # "" if not poisoned and "PlayerX" if poisoned + self.rc.env.step(EnvAction(action_type=EnvActionType.WITCH_POISON, player_name=latest_msg.send_from, + target_player_name=target)) - return msg_content, send_to + return msg_content, restricted_to - def _update_game_states(self, memories): - step_idx = self.step_idx % len(STEP_INSTRUCTIONS) - if step_idx not in [15, 18] or self.step_idx in self.eval_step_idx: # FIXME: hard code - return - else: - self.eval_step_idx.append(self.step_idx) # record evaluation, avoid repetitive evaluation at the same step + def _update_player_status(self, step_idx: int, player_current_dead: list[str]): + """update dead player's status""" + if step_idx in [15, 18]: + self.update_player_status(player_current_dead) - if step_idx == 15: # FIXME: hard code - # night ends: after all special roles acted, process the whole night - self.player_current_dead = [] # reset - - if self.player_hunted != self.player_protected and not self.is_hunted_player_saved: - self.player_current_dead.append(self.player_hunted) - if self.player_poisoned: - self.player_current_dead.append(self.player_poisoned) - - self.living_players = [p for p in self.living_players if p not in self.player_current_dead] - self.update_player_status(self.player_current_dead) - # reset - self.player_hunted = None - self.player_protected = None - self.is_hunted_player_saved = False - self.player_poisoned = None - - elif step_idx == 18: # FIXME: hard code - # day ends: after all roles voted, process all votings - voting_msgs = memories[-len(self.living_players) :] - voted_all = [] - for msg in voting_msgs: - voted = re.search(r"Player[0-9]+", msg.content[-10:]) - if not voted: - continue - voted_all.append(voted.group(0)) - self.player_current_dead = [Counter(voted_all).most_common()[0][0]] # 平票时，杀最先被投的 - # print("*" * 10, "dead", self.player_current_dead) - self.living_players = [p for p in self.living_players if p not in self.player_current_dead] - self.update_player_status(self.player_current_dead) - - # game's termination condition - living_werewolf = [p for p in self.werewolf_players if p in self.living_players] - living_villagers = [p for p in self.villager_players if p in self.living_players] - living_special_roles = [p for p in self.special_role_players if p in self.living_players] - if not living_werewolf: - self.winner = "good guys" - self.win_reason = "werewolves all dead" - elif not living_villagers or not living_special_roles: - self.winner = "werewolf" - self.win_reason = "villagers all dead" if not living_villagers else "special roles all dead" - if self.winner is not None: - self._record_all_experiences() - - def _record_game_history(self): - if self.step_idx % len(STEP_INSTRUCTIONS) == 0 or self.winner is not None: + def _record_game_history(self, step_idx: int): + if step_idx % len(STEP_INSTRUCTIONS) == 0 or self.winner: logger.info("a night and day cycle completed, examine all history") - print(self.get_all_memories()) + logger.debug(f"all_memories: {self.get_all_memories()}") with open(DEFAULT_WORKSPACE_ROOT / "werewolf_transcript.txt", "w") as f: f.write(self.get_all_memories()) async def _think(self): - if self.winner is not None: + if self.winner: self.rc.todo = AnnounceGameResult() return latest_msg = self.rc.memory.get()[-1] - if latest_msg.role in ["User"]: - # 上一轮消息是用户指令，解析用户指令，开始游戏 - game_setup = latest_msg.content - self._parse_game_setup(game_setup) + if latest_msg.role in ["User", "Human", self.profile]: + # 1. 上一轮消息是用户指令，解析用户指令，开始游戏 + # 2.1. 上一轮消息是Moderator自己的指令，继续发出指令，一个事情可以分几条消息来说 + # 2.2. 上一轮消息是Moderator自己的解析消息，一个阶段结束，发出新一个阶段的指令 self.rc.todo = InstructSpeak() - - elif latest_msg.role in [self.profile]: - # 1. 上一轮消息是Moderator自己的指令，继续发出指令，一个事情可以分几条消息来说 - # 2. 上一轮消息是Moderator自己的解析消息，一个阶段结束，发出新一个阶段的指令 - self.rc.todo = InstructSpeak() - else: # 上一轮消息是游戏角色的发言，解析角色的发言 self.rc.todo = ParseSpeak() + def _init_fields_from_obj(self, obs: dict[str, Union[int, str, list[str]]]): + self.game_setup = obs.get("game_setup", "") + self.winner = obs.get("winner") + self.win_reason = obs.get("win_reason") + self.werewolf_players = obs.get("werewolf_players", []) + self.witch_poison_left = obs.get("witch_poison_left", 0) + self.witch_antidote_left = obs.get("witch_antidote_left", 0) + async def _act(self): todo = self.rc.todo logger.info(f"{self._setting} ready to {todo}") memories = self.get_all_memories(mode="msg") - # 若进行完一夜一日的循环，打印和记录一次完整发言历史 - self._record_game_history() + obs, _, _, _, _ = self.rc.env.step(action=EnvAction(action_type=EnvActionType.NONE)) + step_idx = obs["step_idx"] + living_players = obs["living_players"] + werewolf_players = obs["werewolf_players"] + player_hunted = obs["player_hunted"] + player_current_dead = obs["player_current_dead"] + self._init_fields_from_obj(obs) - # 若一晚或一日周期结束，对当晚或当日的死者进行总结，并更新游戏状态 - self._update_game_states(memories) + # 若进行完一夜一日的循环，打印和记录一次完整发言历史 + self._record_game_history(step_idx) + + # 若一晚或一日周期结束，对当晚或当日的死者进行总结，并更新玩家状态 + self._update_player_status(step_idx, player_current_dead) + if self.winner: + self._record_all_experiences() # 根据_think的结果，执行InstructSpeak还是ParseSpeak, 并将结果返回 if isinstance(todo, InstructSpeak): - msg_content, msg_to_send_to = await self._instruct_speak() + msg_content, msg_to_send_to, msg_restricted_to = await InstructSpeak().run( + step_idx, + living_players=living_players, + werewolf_players=werewolf_players, + player_hunted=player_hunted, + player_current_dead=player_current_dead, + ) # msg_content = f"Step {self.step_idx}: {msg_content}" # HACK: 加一个unique的step_idx避免记忆的自动去重 - msg = Message( + msg = WwMessage( content=msg_content, role=self.profile, sent_from=self.name, cause_by=InstructSpeak, send_to=msg_to_send_to, + restricted_to=msg_restricted_to ) + self.rc.env.step(EnvAction(action_type=EnvActionType.PROGRESS_STEP)) # to update step_idx elif isinstance(todo, ParseSpeak): - msg_content, msg_to_send_to = await self._parse_speak(memories) + msg_content, msg_restricted_to = await self._parse_speak(memories) # msg_content = f"Step {self.step_idx}: {msg_content}" # HACK: 加一个unique的step_idx避免记忆的自动去重 - msg = Message( + msg = WwMessage( content=msg_content, role=self.profile, sent_from=self.name, cause_by=ParseSpeak, - send_to=msg_to_send_to, + send_to={}, + restricted_to=msg_restricted_to ) elif isinstance(todo, AnnounceGameResult): msg_content = await AnnounceGameResult().run(winner=self.winner, win_reason=self.win_reason) - msg = Message(content=msg_content, role=self.profile, sent_from=self.name, cause_by=AnnounceGameResult) + msg = WwMessage(content=msg_content, role=self.profile, sent_from=self.name, cause_by=AnnounceGameResult) logger.info(f"{self._setting}: {msg_content}") diff --git a/metagpt/ext/werewolf/roles/seer.py b/metagpt/ext/werewolf/roles/seer.py index a7c0517fe..9a1600fe6 100644 --- a/metagpt/ext/werewolf/roles/seer.py +++ b/metagpt/ext/werewolf/roles/seer.py @@ -1,7 +1,8 @@ from metagpt.ext.werewolf.roles.base_player import BasePlayer +from metagpt.environment.werewolf.const import RoleType class Seer(BasePlayer): - name: str = "Seer" - profile: str = "Seer" + name: str = RoleType.SEER.value + profile: str = RoleType.SEER.value special_action_names: list[str] = ["Verify"] diff --git a/metagpt/ext/werewolf/roles/villager.py b/metagpt/ext/werewolf/roles/villager.py index 769782dc2..327c2376d 100644 --- a/metagpt/ext/werewolf/roles/villager.py +++ b/metagpt/ext/werewolf/roles/villager.py @@ -1,7 +1,8 @@ from metagpt.ext.werewolf.roles.base_player import BasePlayer +from metagpt.environment.werewolf.const import RoleType class Villager(BasePlayer): - name: str = "Villager" - profile: str = "Villager" + name: str = RoleType.VILLAGER.value + profile: str = RoleType.VILLAGER.value special_action_names: list[str] = [] diff --git a/metagpt/ext/werewolf/roles/werewolf.py b/metagpt/ext/werewolf/roles/werewolf.py index 4d98eb8fc..7a3237ded 100644 --- a/metagpt/ext/werewolf/roles/werewolf.py +++ b/metagpt/ext/werewolf/roles/werewolf.py @@ -1,10 +1,11 @@ from metagpt.ext.werewolf.actions import Impersonate, Speak from metagpt.ext.werewolf.roles.base_player import BasePlayer +from metagpt.environment.werewolf.const import RoleType class Werewolf(BasePlayer): - name: str = "Werewolf" - profile: str = "Werewolf" + name: str = RoleType.WEREWOLF.value + profile: str = RoleType.WEREWOLF.value special_action_names: list[str] = ["Hunt"] async def _think(self): diff --git a/metagpt/ext/werewolf/roles/witch.py b/metagpt/ext/werewolf/roles/witch.py index 207669d64..d2341c2d6 100644 --- a/metagpt/ext/werewolf/roles/witch.py +++ b/metagpt/ext/werewolf/roles/witch.py @@ -1,20 +1,21 @@ from metagpt.ext.werewolf.actions import InstructSpeak, Poison, Save, Speak from metagpt.ext.werewolf.roles.base_player import BasePlayer +from metagpt.environment.werewolf.const import RoleType class Witch(BasePlayer): - name: str = "Witch" - profile: str = "Witch" + name: str = RoleType.WITCH.value + profile: str = RoleType.WITCH.value special_action_names: list[str] = ["Save", "Poison"] async def _think(self): """女巫涉及两个特殊技能，因此在此需要改写_think进行路由""" news = self.rc.news[0] assert news.cause_by == InstructSpeak # 消息为来自Moderator的指令时，才去做动作 - if not news.send_to: + if not news.restricted_to: # 消息接收范围为全体角色的，做公开发言（发表投票观点也算发言） self.rc.todo = Speak() - elif self.profile in news.send_to: + elif self.profile in news.restricted_to: # FIXME: hard code to split, restricted为"Moderator"或"Moderator,角色profile" # Moderator加密发给自己的，意味着要执行角色的特殊动作 # 这里用关键词进行动作的选择，需要Moderator侧的指令进行配合 diff --git a/metagpt/ext/werewolf/schema.py b/metagpt/ext/werewolf/schema.py index 5d1cccac8..09e6c8184 100644 --- a/metagpt/ext/werewolf/schema.py +++ b/metagpt/ext/werewolf/schema.py @@ -1,4 +1,6 @@ -from pydantic import BaseModel +from pydantic import BaseModel, Field + +from metagpt.schema import Message class RoleExperience(BaseModel): @@ -12,3 +14,8 @@ class RoleExperience(BaseModel): round_id: str = "" game_setup: str = "" version: str = "" + + +class WwMessage(Message): + # Werewolf Message + restricted_to: set[str] = Field(default={}, validate_default=True) diff --git a/metagpt/ext/werewolf/werewolf_game.py b/metagpt/ext/werewolf/werewolf_game.py index 091429e84..b8c4c1fa5 100644 --- a/metagpt/ext/werewolf/werewolf_game.py +++ b/metagpt/ext/werewolf/werewolf_game.py @@ -3,7 +3,7 @@ from typing import Any, Optional from metagpt.actions.add_requirement import UserRequirement from metagpt.context import Context from metagpt.environment.werewolf.werewolf_env import WerewolfEnv -from metagpt.schema import Message +from metagpt.ext.werewolf.schema import WwMessage from metagpt.team import Team @@ -23,4 +23,4 @@ class WerewolfGame(Team): def run_project(self, idea): """Run a project from user instruction.""" self.idea = idea - self.env.publish_message(Message(role="User", content=idea, cause_by=UserRequirement, send_to={"Moderator"})) + self.env.publish_message(WwMessage(role="User", content=idea, cause_by=UserRequirement, restricted_to={"Moderator"})) diff --git a/tests/metagpt/environment/werewolf_env/test_werewolf_ext_env.py b/tests/metagpt/environment/werewolf_env/test_werewolf_ext_env.py index 433f59f2c..c0ba8a4c4 100644 --- a/tests/metagpt/environment/werewolf_env/test_werewolf_ext_env.py +++ b/tests/metagpt/environment/werewolf_env/test_werewolf_ext_env.py @@ -2,7 +2,8 @@ # -*- coding: utf-8 -*- # @Desc : the unittest of WerewolfExtEnv -from metagpt.environment.werewolf.werewolf_ext_env import RoleState, WerewolfExtEnv +from metagpt.environment.werewolf.werewolf_ext_env import WerewolfExtEnv +from metagpt.environment.werewolf.const import RoleState from metagpt.roles.role import Role @@ -41,9 +42,9 @@ def test_werewolf_ext_env(): assert "Werewolves, please open your eyes" in curr_instr["content"] # current step_idx = 5 - ext_env.wolf_kill_someone(wolf=Role(name="Player10"), player_name="Player4") - ext_env.wolf_kill_someone(wolf=Werewolf(name="Player0"), player_name="Player4") - ext_env.wolf_kill_someone(wolf=Werewolf(name="Player1"), player_name="Player4") + ext_env.wolf_kill_someone(wolf_name="Player10", player_name="Player4") + ext_env.wolf_kill_someone(wolf_name="Player0", player_name="Player4") + ext_env.wolf_kill_someone(wolf_name="Player1", player_name="Player4") assert ext_env.player_hunted == "Player4" assert len(ext_env.living_players) == 5 # hunted but can be saved by witch @@ -52,11 +53,11 @@ def test_werewolf_ext_env(): # current step_idx = 18 assert ext_env.step_idx == 18 - ext_env.vote_kill_someone(voteer=Werewolf(name="Player0"), player_name="Player2") - ext_env.vote_kill_someone(voteer=Werewolf(name="Player1"), player_name="Player3") - ext_env.vote_kill_someone(voteer=Villager(name="Player2"), player_name="Player3") - ext_env.vote_kill_someone(voteer=Witch(name="Player3"), player_name="Player4") - ext_env.vote_kill_someone(voteer=Guard(name="Player4"), player_name="Player2") + ext_env.vote_kill_someone(voter_name="Player0", player_name="Player2") + ext_env.vote_kill_someone(voter_name="Player1", player_name="Player3") + ext_env.vote_kill_someone(voter_name="Player2", player_name="Player3") + ext_env.vote_kill_someone(voter_name="Player3", player_name="Player4") + ext_env.vote_kill_someone(voter_name="Player4", player_name="Player2") assert ext_env.player_current_dead == "Player2" assert len(ext_env.living_players) == 4