From 81a5178e3c79e6f333041be127f4ff7ffefea153 Mon Sep 17 00:00:00 2001 From: better629 Date: Wed, 24 Jan 2024 20:16:04 +0800 Subject: [PATCH] update werewolf_env --- .../mincraft_env/mincraft_ext_env.py | 4 +- .../environment/werewolf_env/werewolf_env.py | 22 +- .../werewolf_env/werewolf_ext_env.py | 198 ++++++++++++++++-- 3 files changed, 206 insertions(+), 18 deletions(-) diff --git a/metagpt/environment/mincraft_env/mincraft_ext_env.py b/metagpt/environment/mincraft_env/mincraft_ext_env.py index d4fec8efa..eb7225acf 100644 --- a/metagpt/environment/mincraft_env/mincraft_ext_env.py +++ b/metagpt/environment/mincraft_env/mincraft_ext_env.py @@ -5,7 +5,7 @@ from typing import Optional import requests -from pydantic import Field, model_validator +from pydantic import Field, model_validator, ConfigDict from metagpt.const import ( MC_CKPT_DIR, @@ -20,6 +20,8 @@ from metagpt.logs import logger class MincraftExtEnv(ExtEnv): + model_config = ConfigDict(arbitrary_types_allowed=True) + mc_port: Optional[int] = Field(default=None) server_host: str = Field(default="http://127.0.0.1") server_port: str = Field(default=3000) diff --git a/metagpt/environment/werewolf_env/werewolf_env.py b/metagpt/environment/werewolf_env/werewolf_env.py index 831f8e020..502bda90a 100644 --- a/metagpt/environment/werewolf_env/werewolf_env.py +++ b/metagpt/environment/werewolf_env/werewolf_env.py @@ -2,8 +2,28 @@ # -*- coding: utf-8 -*- # @Desc : MG Werewolf Env +from pydantic import Field + from metagpt.environment.werewolf_env.werewolf_ext_env import WerewolfExtEnv +from metagpt.schema import Message class WerewolfEnv(WerewolfExtEnv): - pass + timestamp: int = Field(default=0) + + 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}" + + async def run(self, k=1): + """Process all Role runs by order""" + for _ in range(k): + for role in self.roles.values(): + await role.run() + self.timestamp += 1 diff --git a/metagpt/environment/werewolf_env/werewolf_ext_env.py b/metagpt/environment/werewolf_env/werewolf_ext_env.py index 014417009..fddfceb0d 100644 --- a/metagpt/environment/werewolf_env/werewolf_ext_env.py +++ b/metagpt/environment/werewolf_env/werewolf_ext_env.py @@ -2,37 +2,203 @@ # -*- coding: utf-8 -*- # @Desc : The werewolf game external environment to integrate with +import random +import re from enum import Enum +from typing import Optional -from pydantic import Field +from pydantic import ConfigDict, Field from metagpt.environment.base_env import ExtEnv, mark_as_readable, mark_as_writeable class RoleState(Enum): - ALIVE = "alive" - KILLED = "killed" - POISONED = "poisoned" - SAVED = "saved" + 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": "", + }, +} class WerewolfExtEnv(ExtEnv): - roles_state: dict[str, RoleState] = Field(default=dict(), description="the role's current state") + model_config = ConfigDict(arbitrary_types_allowed=True) + + roles_state: dict[str, RoleState] = Field(default=dict(), description="the role's current state by role_name") + + step_idx: int = Field(default=0) # the current step of current round + eval_step_idx: int = Field(default=0) + per_round_steps: int = Field(default=len(STEP_INSTRUCTIONS)) + + # game global states + game_setup: str = Field(default="", description="game setup including role and its num") + living_players: list[str] = Field(default=[]) + werewolf_players: list[str] = Field(default=[]) + villager_players: list[str] = Field(default=[]) + 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) + + # game current round states, a round is from closing your eyes to the next time you close your eyes + player_hunted: Optional[str] = Field(default=None) + player_protected: Optional[str] = Field(default=None) + is_hunted_player_saved: bool = Field(default=False) + player_poisoned: Optional[str] = Field(default=None) + player_current_dead: list[str] = Field(default=[]) + + 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 + ] + + # init role state + self.roles_state = {player_name: RoleState.ALIVE for player_name in self.living_players} @mark_as_readable - def get_roles_status(self): - pass + def init_game_setup( + self, + role_uniq_objs: list[object], + num_villager: int = 2, + num_werewolf: int = 2, + shuffle=True, + add_human=False, + use_reflection=True, + use_experience=False, + use_memory_selection=False, + new_experience_version="", + ) -> tuple[str, list]: + role_objs = [] + for role_obj in role_uniq_objs: + if str(role_obj) == "Villager": + role_objs.extend([role_obj] * num_villager) + elif str(role_obj) == "Werewolf": + role_objs.extend([role_obj] * num_werewolf) + else: + role_objs.append(role_obj) + if shuffle: + random.shuffle(len(role_objs)) + if add_human: + assigned_role_idx = random.randint(0, len(role_objs) - 1) + assigned_role = role_objs[assigned_role_idx] + role_objs[assigned_role_idx] = prepare_human_player(assigned_role) # TODO + + players = [ + role( + name=f"Player{i + 1}", + use_reflection=use_reflection, + use_experience=use_experience, + use_memory_selection=use_memory_selection, + new_experience_version=new_experience_version, + ) + for i, role in enumerate(role_objs) + ] + + if add_human: + logger.info(f"You are assigned {players[assigned_role_idx].name}({players[assigned_role_idx].profile})") + + game_setup = ["Game setup:"] + [f"{player.name}: {player.profile}," for player in players] + game_setup = "\n".join(game_setup) + + return game_setup, players + + @mark_as_readable + def curr_step_instruction(self) -> dict: + step_idx = self.step_idx % len(STEP_INSTRUCTIONS) + instruction = STEP_INSTRUCTIONS[step_idx] + self.step_idx += 1 + return instruction @mark_as_writeable - def wolf_kill_someone(self, role_name: str): - pass + def update_players_state(self, player_names: list[str], state: RoleState = RoleState.KILLED): + for player_name in player_names: + if player_name in self.roles_state: + self.roles_state[player_name] = state + + @mark_as_readable + def get_players_status(self, player_names: list[str]) -> dict[str, RoleState]: + roles_state = { + player_name: self.roles_state[player_name] + for player_name in player_names + if player_name in self.roles_state + } + return roles_state @mark_as_writeable - def witch_poison_someone(self, role_name: str = None): - if not role_name: + def wolf_kill_someone(self, player_name: str): + self.update_players_state([player_name], RoleState.KILLED) + + @mark_as_writeable + def witch_poison_someone(self, player_name: str = None): + self.update_players_state([player_name], RoleState.POISONED) + + @mark_as_writeable + def witch_save_someone(self, player_name: str = None): + self.update_players_state([player_name], RoleState.SAVED) + + @mark_as_writeable + def update_game_states(self, memories: list): + 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 + else: + self.eval_step_idx.append(self.step_idx) # record evaluation, avoid repetitive evaluation at the same step - @mark_as_writeable - def witch_save_someone(self, role_name: str = None): - if not role_name: - return + if step_idx == 15: # step no + # 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: # step no + # 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() # TODO