Merge pull request #383 from garylin2099/werewolf_game

add human player

examples/werewolf_game/roles/human_player.py

@@ -0,0 +1,40 @@
+from examples.werewolf_game.actions import Speak
+from examples.werewolf_game.roles import BasePlayer
+from metagpt.schema import Message
+from metagpt.logs import logger
+
+async def _act(self):
+    todo = self._rc.todo
+
+    memories = self.get_all_memories()
+
+    input_instruction = f"""
+    ## As a reminder, you have access to the following game history:
+    {memories}
+    ## You are {self.name}({self.profile})
+    ## Guidance:
+    1. If you are performing a special action or exercising a vote,
+    end your response with "PlayerX" where X is the player index, e.g., "..., kill/protect/poison/.../vote Player1".
+    2. If it is a daytime free speech, you can speak in whatever format.
+    Now, please speak:
+    """
+    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 = Message(
+        content=rsp, role=self.profile, sent_from=self.name,
+        cause_by=msg_cause_by, send_to="",
+        restricted_to=msg_restricted_to, # 给Moderator及自身阵营发送加密消息
+    )
+
+    logger.info(f"{self._setting}: {rsp}")
+
+    return msg
+
+def prepare_human_player(player_class: BasePlayer):
+    # Dynamically define a human player class that inherits from a certain role class
+    HumanPlayer = type('HumanPlayer', (player_class,), {'_act': _act})
+    return HumanPlayer

examples/werewolf_game/roles/moderator.py

@@ -109,8 +109,15 @@ class Moderator(Role):
                 self.witch_poison_left -= 1
                 self.player_poisoned = target # "" if not poisoned and "PlayerX" if poisoned
 
+        return msg_content, restricted_to
+
+    def _update_game_states(self, memories):
+
         step_idx = self.step_idx % len(STEP_INSTRUCTIONS)
-        if step_idx == 13: # FIXME: hard code
+        if step_idx not in [15, 18]: # FIXME: hard code
+            return
+
+        if step_idx == 15: # FIXME: hard code
             # night ends: after all special roles acted, process the whole night
             self.player_current_dead = [] # reset
 
@@ -136,11 +143,9 @@ class Moderator(Role):
                 if not voted:
                     continue
                 voted_all.append(voted.group(0))
-            # breakpoint()
             self.player_current_dead = [Counter(voted_all).most_common()[0][0]] # 平票时，杀序号小的
             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)
-            msg_content = "Voting done"
 
         # game's termination condition
         living_werewolf = [p for p in self.werewolf_players if p in self.living_players]
@@ -150,7 +155,12 @@ class Moderator(Role):
         elif not living_good_guys:
             self.winner = "werewolf"
 
-        return msg_content, restricted_to
+    def _record_game_history(self):
+        if self.step_idx % len(STEP_INSTRUCTIONS) == 0 or self.winner is not None:
+            logger.info("a night and day cycle completed, examine all history")
+            print(self.get_all_memories())
+            with open(WORKSPACE_ROOT / 'werewolf_transcript.txt', "w") as f:
+                f.write(self.get_all_memories())
 
     async def _think(self):
 
@@ -179,13 +189,12 @@ class Moderator(Role):
         logger.info(f"{self._setting} ready to {todo}")
 
         memories = self.get_all_memories(mode="msg")
-        # print("*" * 10, f"{self._setting}'s current memories: {memories}", "*" * 10)
-        if self.step_idx % len(STEP_INSTRUCTIONS) == 0 or self.winner is not None:
-            # 进行完一夜一日的循环，打印一次完整发言历史
-            logger.info("a night and day cycle completed, examine all history")
-            print(self.get_all_memories())
-            with open(WORKSPACE_ROOT / 'werewolf_transcript.txt', "w") as f:
-                f.write(self.get_all_memories())
+
+        # 若进行完一夜一日的循环，打印和记录一次完整发言历史
+        self._record_game_history()
+
+        # 若一晚或一日周期结束，对当晚或当日的死者进行总结，并更新游戏状态
+        self._update_game_states(memories)
 
         # 根据_think的结果，执行InstructSpeak还是ParseSpeak, 并将结果返回
         if isinstance(todo, InstructSpeak):

examples/werewolf_game/roles/seer.py

@@ -35,7 +35,7 @@ class Seer(BasePlayer):
             msg = Message(
                 content=rsp, role=self.profile, sent_from=self.name,
                 cause_by=Verify, send_to="",
-                restricted_to="Moderator",
+                restricted_to=f"Moderator,{self.profile}",
             )
 
         logger.info(f"{self._setting}: {rsp}")

examples/werewolf_game/start_game.py

@@ -3,10 +3,12 @@ import platform
 import fire
 import random
 
+from metagpt.logs import logger
 from examples.werewolf_game.werewolf_game import WerewolfGame
 from examples.werewolf_game.roles import Moderator, Villager, Werewolf, Guard, Seer, Witch
+from examples.werewolf_game.roles.human_player import prepare_human_player
 
-def init_game_setup(shuffle=False):
+def init_game_setup(shuffle=True, add_human=False):
     roles = [
         Villager,
         Villager,
@@ -17,29 +19,39 @@ def init_game_setup(shuffle=False):
         Witch
     ]
     if shuffle:
-        random.seed(2023)
+        # random.seed(2023)
         random.shuffle(roles)
+    if add_human:
+        assigned_role_idx = random.randint(0, len(roles) - 1)
+        assigned_role = roles[assigned_role_idx]
+        roles[assigned_role_idx] = prepare_human_player(assigned_role)
+
     players = [role(name=f"Player{i+1}") for i, role in enumerate(roles)]
+
+    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
 
-async def start_game(investment: float = 3.0, n_round: int = 5):
+async def start_game(investment: float = 3.0, n_round: int = 5, shuffle : bool = True, add_human: bool = False):
     game = WerewolfGame()
-    game_setup, players = init_game_setup(shuffle=True)
+    game_setup, players = init_game_setup(shuffle=shuffle, add_human=add_human)
     players = [Moderator()] + players
     game.hire(players)
     game.invest(investment)
     game.start_project(game_setup)
     await game.run(n_round=n_round)
 
-def main(investment: float = 3.0, n_round: int = 100):
+def main(investment: float = 3.0, n_round: int = 100, shuffle : bool = True, add_human: bool = False):
     """
     :param investment: contribute a certain dollar amount to watch the debate
     :param n_round: maximum rounds of the debate
     :return:
     """
-    asyncio.run(start_game(investment, n_round))
+    asyncio.run(start_game(investment, n_round, shuffle, add_human))
 
 
 if __name__ == '__main__':