update werewolf_env

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

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)

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

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

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
 
 

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))