use WerewolfEnv

metagpt/ext/werewolf/actions/moderator_actions.py

@@ -1,90 +1,12 @@
 from metagpt.actions import Action
-
-STEP_INSTRUCTIONS = {
-    # 上帝需要介入的全部步骤和对应指令
-    # The 1-st night
-    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.werewolf_ext_env 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": "", "restricted_to": ""}
-        )
+        instruction_info = STEP_INSTRUCTIONS.get(step_idx, {"content": "Unknown instruction.", "send_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)
@@ -98,7 +20,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"], instruction_info["restricted_to"]
+        return content, instruction_info["send_to"]
 
 
 class ParseSpeak(Action):

metagpt/ext/werewolf/roles/base_player.py

@@ -57,23 +57,23 @@ class BasePlayer(Role):
         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)
+        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]
+        news = self.rc.news[0]
         assert news.cause_by == InstructSpeak  # 消息为来自Moderator的指令时，才去做动作
-        if not news.restricted_to:
+        if not news.send_to:
             # 消息接收范围为全体角色的，做公开发言（发表投票观点也算发言）
-            self._rc.todo = Speak()
-        elif self.profile in news.restricted_to.split(","):
+            self.rc.todo = Speak()
+        elif self.profile in news.send_to:
             # FIXME: hard code to split, restricted为"Moderator"或"Moderator,角色profile"
             # Moderator加密发给自己的，意味着要执行角色的特殊动作
-            self._rc.todo = self.special_actions[0]()
+            self.rc.todo = self.special_actions[0]()
 
     async def _act(self):
         # todo为_think时确定的，有两种情况，Speak或Protect
-        todo = self._rc.todo
+        todo = self.rc.todo
         logger.info(f"{self._setting}: ready to {str(todo)}")
 
         # 可以用这个函数获取该角色的全部记忆和最新的instruction
@@ -107,21 +107,20 @@ class BasePlayer(Role):
                 reflection=reflection,
                 experiences=experiences,
             )
-            restricted_to = ""
+            send_to = ""
 
         elif isinstance(todo, NighttimeWhispers):
             rsp = await todo.run(
                 profile=self.profile, name=self.name, context=memories, reflection=reflection, experiences=experiences
             )
-            restricted_to = f"Moderator,{self.profile}"  # 给Moderator发送使用特殊技能的加密消息
+            send_to = {"Moderator", self.profile}  # 给Moderator发送使用特殊技能的加密消息
 
         msg = Message(
             content=rsp,
             role=self.profile,
             sent_from=self.name,
             cause_by=type(todo),
-            send_to="",
-            restricted_to=restricted_to,
+            send_to=send_to,
         )
 
         self.experiences.append(
@@ -140,7 +139,7 @@ class BasePlayer(Role):
         return msg
 
     def get_all_memories(self) -> str:
-        memories = self._rc.memory.get()
+        memories = self.rc.memory.get()
         time_stamp_pattern = r"[0-9]+ \| "
         # NOTE: 除Moderator外，其他角色使用memory，只能用m.sent_from（玩家名）不能用m.role（玩家角色），因为他们不知道说话者的身份
         memories = [f"{m.sent_from}: {re.sub(time_stamp_pattern, '', m.content)}" for m in memories]  # regex去掉时间戳
@@ -148,7 +147,7 @@ class BasePlayer(Role):
         return memories
 
     def get_latest_instruction(self) -> str:
-        return self._rc.important_memory[-1].content  # 角色监听着Moderator的InstructSpeak，是其重要记忆，直接获取即可
+        return self.rc.important_memory[-1].content  # 角色监听着Moderator的InstructSpeak，是其重要记忆，直接获取即可
 
     def set_status(self, new_status):
         self.status = new_status

metagpt/ext/werewolf/roles/human_player.py

@@ -5,7 +5,7 @@ from metagpt.schema import Message
 
 
 async def _act(self):
-    todo = self._rc.todo
+    todo = self.rc.todo
 
     memories = self.get_all_memories()
 
@@ -29,8 +29,7 @@ async def _act(self):
         role=self.profile,
         sent_from=self.name,
         cause_by=msg_cause_by,
-        send_to="",
-        restricted_to=msg_restricted_to,  # 给Moderator及自身阵营发送加密消息
+        send_to=msg_restricted_to,  # 给Moderator及自身阵营发送加密消息
     )
 
     logger.info(f"{self._setting}: {rsp}")

metagpt/ext/werewolf/roles/moderator.py

@@ -4,9 +4,9 @@ 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.ext.werewolf.actions import Hunt, Poison, Protect, Save, Verify
 from metagpt.ext.werewolf.actions.moderator_actions import (
-    STEP_INSTRUCTIONS,
     AnnounceGameResult,
     InstructSpeak,
     ParseSpeak,
@@ -64,14 +64,14 @@ class Moderator(Role):
     def update_player_status(self, player_names: list[str]):
         if not player_names:
             return
-        roles_in_env = self._rc.env.get_roles()
+        roles_in_env = self.rc.env.get_roles()
         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)  # 更新为死亡
 
     def _record_all_experiences(self):
-        roles_in_env = self._rc.env.get_roles()
+        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:
@@ -104,7 +104,7 @@ class Moderator(Role):
 
         # default return
         msg_content = "Understood"
-        restricted_to = ""
+        send_to = set()
 
         msg_cause_by = latest_msg.cause_by
         if msg_cause_by == Hunt:
@@ -116,13 +116,13 @@ class Moderator(Role):
                 msg_content = f"{target} is a werewolf"
             else:
                 msg_content = f"{target} is a good guy"
-            restricted_to = "Moderator,Seer"
+            send_to = {"Moderator", "Seer"}
         elif msg_cause_by == Save:
             if "pass" in latest_msg_content.lower():
                 pass
             elif not self.witch_antidote_left:
                 msg_content = "You have no antidote left and thus can not save the player"
-                restricted_to = "Moderator,Witch"
+                send_to = {"Moderator", "Witch"}
             else:
                 self.witch_antidote_left -= 1
                 self.is_hunted_player_saved = True
@@ -131,12 +131,12 @@ class Moderator(Role):
                 pass
             elif not self.witch_poison_left:
                 msg_content = "You have no poison left and thus can not poison the player"
-                restricted_to = "Moderator,Witch"
+                send_to = {"Moderator", "Witch"}
             else:
                 self.witch_poison_left -= 1
                 self.player_poisoned = target  # "" if not poisoned and "PlayerX" if poisoned
 
-        return msg_content, restricted_to
+        return msg_content, send_to
 
     def _update_game_states(self, memories):
         step_idx = self.step_idx % len(STEP_INSTRUCTIONS)
@@ -198,27 +198,27 @@ class Moderator(Role):
 
     async def _think(self):
         if self.winner is not None:
-            self._rc.todo = AnnounceGameResult()
+            self.rc.todo = AnnounceGameResult()
             return
 
-        latest_msg = self._rc.memory.get()[-1]
+        latest_msg = self.rc.memory.get()[-1]
         if latest_msg.role in ["User"]:
             # 上一轮消息是用户指令，解析用户指令，开始游戏
             game_setup = latest_msg.content
             self._parse_game_setup(game_setup)
-            self._rc.todo = InstructSpeak()
+            self.rc.todo = InstructSpeak()
 
         elif latest_msg.role in [self.profile]:
             # 1. 上一轮消息是Moderator自己的指令，继续发出指令，一个事情可以分几条消息来说
             # 2. 上一轮消息是Moderator自己的解析消息，一个阶段结束，发出新一个阶段的指令
-            self._rc.todo = InstructSpeak()
+            self.rc.todo = InstructSpeak()
 
         else:
             # 上一轮消息是游戏角色的发言，解析角色的发言
-            self._rc.todo = ParseSpeak()
+            self.rc.todo = ParseSpeak()
 
     async def _act(self):
-        todo = self._rc.todo
+        todo = self.rc.todo
         logger.info(f"{self._setting} ready to {todo}")
 
         memories = self.get_all_memories(mode="msg")
@@ -231,7 +231,7 @@ class Moderator(Role):
 
         # 根据_think的结果，执行InstructSpeak还是ParseSpeak, 并将结果返回
         if isinstance(todo, InstructSpeak):
-            msg_content, msg_to_send_to, msg_restriced_to = await self._instruct_speak()
+            msg_content, msg_to_send_to = await self._instruct_speak()
             # msg_content = f"Step {self.step_idx}: {msg_content}" # HACK: 加一个unique的step_idx避免记忆的自动去重
             msg = Message(
                 content=msg_content,
@@ -239,19 +239,17 @@ class Moderator(Role):
                 sent_from=self.name,
                 cause_by=InstructSpeak,
                 send_to=msg_to_send_to,
-                restricted_to=msg_restriced_to,
             )
 
         elif isinstance(todo, ParseSpeak):
-            msg_content, msg_restriced_to = await self._parse_speak(memories)
+            msg_content, msg_to_send_to = await self._parse_speak(memories)
             # msg_content = f"Step {self.step_idx}: {msg_content}" # HACK: 加一个unique的step_idx避免记忆的自动去重
             msg = Message(
                 content=msg_content,
                 role=self.profile,
                 sent_from=self.name,
                 cause_by=ParseSpeak,
-                send_to="",
-                restricted_to=msg_restriced_to,
+                send_to=msg_to_send_to,
             )
 
         elif isinstance(todo, AnnounceGameResult):
@@ -263,7 +261,7 @@ class Moderator(Role):
         return msg
 
     def get_all_memories(self, mode="str") -> str:
-        memories = self._rc.memory.get()
+        memories = self.rc.memory.get()
         if mode == "str":
             memories = [f"{m.sent_from}({m.role}): {m.content}" for m in memories]
             memories = "\n".join(memories)

metagpt/ext/werewolf/roles/werewolf.py

@@ -10,5 +10,5 @@ class Werewolf(BasePlayer):
     async def _think(self):
         """狼人白天发言时需要伪装，与其他角色不同，因此需要重写_think"""
         await super()._think()
-        if isinstance(self._rc.todo, Speak):
-            self._rc.todo = Impersonate()
+        if isinstance(self.rc.todo, Speak):
+            self.rc.todo = Impersonate()

metagpt/ext/werewolf/roles/witch.py

@@ -9,18 +9,18 @@ class Witch(BasePlayer):
 
     async def _think(self):
         """女巫涉及两个特殊技能，因此在此需要改写_think进行路由"""
-        news = self._rc.news[0]
+        news = self.rc.news[0]
         assert news.cause_by == InstructSpeak  # 消息为来自Moderator的指令时，才去做动作
-        if not news.restricted_to:
+        if not news.send_to:
             # 消息接收范围为全体角色的，做公开发言（发表投票观点也算发言）
-            self._rc.todo = Speak()
-        elif self.profile in news.restricted_to.split(","):
+            self.rc.todo = Speak()
+        elif self.profile in news.send_to:
             # FIXME: hard code to split, restricted为"Moderator"或"Moderator,角色profile"
             # Moderator加密发给自己的，意味着要执行角色的特殊动作
             # 这里用关键词进行动作的选择，需要Moderator侧的指令进行配合
             if "save" in news.content.lower():
-                self._rc.todo = Save()
+                self.rc.todo = Save()
             elif "poison" in news.content.lower():
-                self._rc.todo = Poison()
+                self.rc.todo = Poison()
             else:
                 raise ValueError("Moderator's instructions must include save or poison keyword")

metagpt/ext/werewolf/werewolf_game.py

@@ -1,42 +1,26 @@
+from typing import Any, Optional
+
 from metagpt.actions.add_requirement import UserRequirement
-from metagpt.environment import Environment
+from metagpt.context import Context
+from metagpt.environment.werewolf.werewolf_env import WerewolfEnv
 from metagpt.schema import Message
 from metagpt.team import Team
 
 
-class WerewolfEnvironment(Environment):
-    timestamp: int = 0
-
-    def publish_message(self, message: Message, add_timestamp: bool = True):
-        """向当前环境发布信息
-        Post information to the current environment
-        """
-        # self.message_queue.put(message)
-        if add_timestamp:
-            # 因消息内容可能重复，例如，连续两晚杀同一个人，
-            # 因此需要加一个unique的time_stamp以使得相同的message在加入记忆时不被自动去重
-            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 at once
-        """
-        for _ in range(k):
-            for role in self.roles.values():
-                await role.run()
-            self.timestamp += 1
-
-
 class WerewolfGame(Team):
     """Use the "software company paradigm" to hold a werewolf game"""
 
-    environment = WerewolfEnvironment()
+    env: Optional[WerewolfEnv] = None
 
-    def start_project(self, idea):
-        """Start a project from user instruction."""
+    def __init__(self, context: Context = None, **data: Any):
+        super(Team, self).__init__(**data)
+        ctx = context or Context()
+        if not self.env:
+            self.env = WerewolfEnv(context=ctx)
+        else:
+            self.env.context = ctx  # The `env` object is allocated by deserialization
+
+    def run_project(self, idea):
+        """Run a project from user instruction."""
         self.idea = idea
-        self.environment.publish_message(
-            Message(role="User", content=idea, cause_by=UserRequirement, restricted_to="Moderator")
-        )
+        self.env.publish_message(Message(role="User", content=idea, cause_by=UserRequirement, send_to={"Moderator"}))