From 7fe2a291f2fd8fb0b6646953863fcffa64853a30 Mon Sep 17 00:00:00 2001 From: better629 Date: Wed, 24 Jan 2024 20:16:04 +0800 Subject: [PATCH] update werewolf_env --- metagpt/const.py | 41 +++- .../mincraft_env/mincraft_ext_env.py | 4 +- .../environment/werewolf_env/werewolf_env.py | 22 +- .../werewolf_env/werewolf_ext_env.py | 198 ++++++++++++++++-- metagpt/logs.py | 5 +- .../werewolf_env/test_werewolf_ext_env.py | 29 +++ 6 files changed, 278 insertions(+), 21 deletions(-) create mode 100644 tests/metagpt/environment/werewolf_env/test_werewolf_ext_env.py diff --git a/metagpt/const.py b/metagpt/const.py index e9297aab1..b21bbc282 100644 --- a/metagpt/const.py +++ b/metagpt/const.py @@ -127,5 +127,44 @@ GENERALIZATION = "Generalize" COMPOSITION = "Composite" AGGREGATION = "Aggregate" -# Others +# For Android App Agent ADB_EXEC_FAIL = "FAILED" + +# For Mincraft Game Agent +MC_CKPT_DIR = METAGPT_ROOT / "data/mincraft/ckpt" +MC_LOG_DIR = METAGPT_ROOT / "logs" +MC_DEFAULT_WARMUP = { + "context": 15, + "biome": 10, + "time": 15, + "nearby_blocks": 0, + "other_blocks": 10, + "nearby_entities": 5, + "health": 15, + "hunger": 15, + "position": 0, + "equipment": 0, + "inventory": 0, + "optional_inventory_items": 7, + "chests": 0, + "completed_tasks": 0, + "failed_tasks": 0, +} +MC_CURRICULUM_OB = [ + "context", + "biome", + "time", + "nearby_blocks", + "other_blocks", + "nearby_entities", + "health", + "hunger", + "position", + "equipment", + "inventory", + "chests", + "completed_tasks", + "failed_tasks", +] +MC_CORE_INVENTORY_ITEMS = r".*_log|.*_planks|stick|crafting_table|furnace" +r"|cobblestone|dirt|coal|.*_pickaxe|.*_sword|.*_axe", # curriculum_agent: only show these items in inventory before optional_inventory_items reached in warm up \ No newline at end of file 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 diff --git a/metagpt/logs.py b/metagpt/logs.py index fb0fdd553..659d46bcf 100644 --- a/metagpt/logs.py +++ b/metagpt/logs.py @@ -15,14 +15,15 @@ from loguru import logger as _logger from metagpt.const import METAGPT_ROOT -def define_log_level(print_level="INFO", logfile_level="DEBUG"): +def define_log_level(print_level="INFO", logfile_level="DEBUG", name: str = None): """Adjust the log level to above level""" current_date = datetime.now() formatted_date = current_date.strftime("%Y%m%d") + log_name = f"{name}_{formatted_date}" if name else formatted_date _logger.remove() _logger.add(sys.stderr, level=print_level) - _logger.add(METAGPT_ROOT / f"logs/{formatted_date}.txt", level=logfile_level) + _logger.add(METAGPT_ROOT / f"logs/{log_name}.txt", level=logfile_level) return _logger diff --git a/tests/metagpt/environment/werewolf_env/test_werewolf_ext_env.py b/tests/metagpt/environment/werewolf_env/test_werewolf_ext_env.py new file mode 100644 index 000000000..a24328143 --- /dev/null +++ b/tests/metagpt/environment/werewolf_env/test_werewolf_ext_env.py @@ -0,0 +1,29 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# @Desc : the unittest of WerewolfExtEnv + + +from metagpt.environment.werewolf_env.werewolf_ext_env import RoleState, WerewolfExtEnv + + +def test_werewolf_ext_env(): + ext_env = WerewolfExtEnv() + + game_setup = """Game setup: + Player0: Werewolf, + Player1: Werewolf, + Player2: Villager, + Player3: Guard, + """ + ext_env.parse_game_setup(game_setup) + assert len(ext_env.living_players) == 4 + assert len(ext_env.special_role_players) == 1 + assert len(ext_env.werewolf_players) == 2 + + curr_instr = ext_env.curr_step_instruction() + assert ext_env.step_idx == 1 + assert "close your eyes" in curr_instr["content"] + + player_names = ["Player0", "Player2"] + ext_env.update_players_state(player_names, RoleState.KILLED) + assert ext_env.get_players_status(player_names) == dict(zip(player_names, [RoleState.KILLED] * 2))