Merge pull request #357 from garylin2099/werewolf_game

Werewolf game

examples/debate.py

@@ -49,7 +49,6 @@ class Trump(Role):
         super().__init__(name, profile, **kwargs)
         self._init_actions([ShoutOut])
         self._watch([ShoutOut])
-        self.name = "Trump"
         self.opponent_name = "Biden"
 
     async def _observe(self) -> int:
@@ -89,7 +88,6 @@ class Biden(Role):
         super().__init__(name, profile, **kwargs)
         self._init_actions([ShoutOut])
         self._watch([BossRequirement, ShoutOut])
-        self.name = "Biden"
         self.opponent_name = "Trump"
 
     async def _observe(self) -> int:

examples/werewolf_game/actions/__init__.py

@@ -0,0 +1,8 @@
+from examples.werewolf_game.actions.moderator_actions import InstructSpeak
+from examples.werewolf_game.actions.common_actions import Speak
+from examples.werewolf_game.actions.werewolf_actions import Hunt
+
+ACTIONS = {
+    "Speak": Speak,
+    "Hunt": Hunt,
+}

examples/werewolf_game/actions/common_actions.py

@@ -0,0 +1,27 @@
+from metagpt.actions import Action
+
+class Speak(Action):
+    """Action: Any speak action in a game"""
+
+    PROMPT_TEMPLATE = """
+    ## BACKGROUND
+    It's a Werewolf game, you are {profile}, say whatever possible to increase your chance of win,
+    ## HISTORY
+    You have knowledge to the following conversation:
+    {context}
+    ## YOUR TURN
+    It's daytime and it is your turn to speak, you will say (in 100 words):
+    """
+
+    def __init__(self, name="Speak", context=None, llm=None):
+        super().__init__(name, context, llm)
+
+    async def run(self, context: str, profile: str):
+
+        prompt = self.PROMPT_TEMPLATE.format(context=context, profile=profile)
+
+        rsp = await self._aask(prompt)
+
+        return rsp
+
+

examples/werewolf_game/actions/moderator_actions.py

@@ -0,0 +1,49 @@
+from metagpt.actions import Action
+
+STAGE_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": "Werewolves, please open your eyes!",
+        "send_to": "Moderator", # for moderator to continuen speaking
+        "restricted_to": ""},
+    2: {"content": """Werewolves, I secretly tell you that Player 3 and Player 4 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:
+                   [Player 1, Player2]. """, # send to werewolf restrictedly for a response
+        "send_to": "Werewolf",
+        "restricted_to": "Werewolf"},
+    3: {"content": "Werewolves, close your eyes",
+        "send_to": "Moderator", # for moderator to continuen speaking
+        "restricted_to": ""},
+    4: {"content": """It's daytime. No one dies last night. Now freely talk about roles of other players with each other based on your observation and reflection
+                   with few sentences. Decide whether to reveal your identity based on your reflection.""",
+        "send_to": "", # send to all to speak in daytime
+        "restricted_to": ""}
+}
+
+class InstructSpeak(Action):
+    def __init__(self, name="InstructSpeak", context=None, llm=None):
+        super().__init__(name, context, llm)
+
+    async def run(self, context, stage_idx):
+        return STAGE_INSTRUCTIONS[stage_idx]
+
+class ParseSpeak(Action):
+    async def run(self):
+        return ""
+
+class SummarizeNight(Action):
+    """consider all events at night, conclude which player dies (can be a peaceful night)"""
+    pass
+
+class SummarizeDay(Action):
+    """consider all votes at day, conclude which player dies"""
+    pass
+
+class AnnounceGameResult(Action):
+
+    async def run(self, winner: str):
+        return f"Game over! The winner is {winner}"

examples/werewolf_game/actions/werewolf_actions.py

@@ -0,0 +1,24 @@
+from metagpt.actions import Action
+
+class Hunt(Action):
+    """Action: choose a villager to kill"""
+
+    PROMPT_TEMPLATE = """
+    It's a werewolf game and you are a werewolf,
+    this is game history:
+    {context}.
+    Attention: if your previous werewolf have chosen, follow its choice.
+    Now, choose one to kill, you will:
+    """
+
+    def __init__(self, name="Speak", context=None, llm=None):
+        super().__init__(name, context, llm)
+
+    async def run(self, context: str):
+
+        prompt = self.PROMPT_TEMPLATE.format(context=context)
+
+        rsp = await self._aask(prompt)
+        # rsp = "Kill Player 1"
+
+        return rsp

examples/werewolf_game/roles/__init__.py

@@ -0,0 +1,4 @@
+from examples.werewolf_game.roles.base_player import BasePlayer
+from examples.werewolf_game.roles.moderator import Moderator
+from examples.werewolf_game.roles.villager import Villager
+from examples.werewolf_game.roles.werewolf import Werewolf

examples/werewolf_game/roles/base_player.py

@@ -0,0 +1,78 @@
+from metagpt.roles import Role
+from metagpt.schema import Message
+from metagpt.logs import logger
+from examples.werewolf_game.actions import ACTIONS, Speak, InstructSpeak
+
+ROLE_STATES = {
+    # 存活状态
+    0: "Alive", # 开场
+    1: "Dead", # 结束
+    2: "Protected", # 被保护
+    3: "Poisoned", # 被毒
+    4: "Saved", # 被救
+}
+
+class BasePlayer(Role):
+    def __init__(
+        self,
+        name: str = "PlayerXYZ",
+        profile: str = "BasePlayer",
+        team: str = "good guys",
+        special_action_names: list[str] = [],
+        **kwargs,
+    ):
+        super().__init__(name, profile, **kwargs)
+        self._init_actions([Speak])
+        self._watch([InstructSpeak])
+        self.team = team 
+        # 调用 get_status() 来检查存活状态,并通过 set_status() 更新状态。
+        self.status = 0 # 初始状态为活着
+        
+        # 技能和监听配置
+        self._watch([InstructSpeak]) # 监听Moderator的指令以做行动
+        special_actions = [ACTIONS[action_name] for action_name in special_action_names]
+        capable_actions = [Speak] + special_actions
+        self._init_actions(capable_actions) # 给角色赋予行动技能
+        self.special_actions = special_actions
+    
+    async def _observe(self) -> int:
+        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)
+       
+    async def _think(self):
+        news = self._rc.news[0]
+        assert news.cause_by == InstructSpeak # 消息为来自Moderator的指令时，才去做动作
+        if not news.restricted_to:
+            # 消息接收范围为全体角色的，做公开发言（发表投票观点也算发言）
+            self._rc.todo = Speak()
+        elif self.profile in news.restricted_to.split(","): # FIXME: hard code to split, restricted为"Moderator"或"Moderator,角色profile"
+            # Moderator加密发给自己的，意味着要执行角色的特殊动作
+            self._rc.todo = self.special_actions[0]()
+
+    async def _act(self):
+        """每个角色要改写此函数以实现该角色的动作"""
+        raise NotImplementedError
+    
+    def get_all_memories(self) -> str:
+        memories = self._rc.memory.get()
+        memories = [f"{m.sent_from}: {m.content}" for m in memories]
+        memories = "\n".join(memories)
+        return memories
+ 
+    def get_name(self):
+        return self.name
+
+    def get_profile(self):
+        return self.profile
+
+    def get_team(self):
+        return self.team
+
+    def get_status(self):
+        return self.status
+
+    def set_status(self, new_status):
+        self.status = new_status

examples/werewolf_game/roles/moderator.py

@@ -0,0 +1,92 @@
+import asyncio
+
+from metagpt.roles import Role
+from metagpt.schema import Message
+from metagpt.logs import logger
+from examples.werewolf_game.actions.moderator_actions import (
+    InstructSpeak, ParseSpeak, AnnounceGameResult, STAGE_INSTRUCTIONS
+)
+from metagpt.actions import BossRequirement as UserRequirement
+
+class Moderator(Role):
+
+    # 游戏状态属性
+    is_game_over = False
+    winner = None
+
+    def __init__(
+        self,
+        name: str = "Moderator",
+        profile: str = "Moderator",
+        **kwargs,
+    ):
+        super().__init__(name, profile, **kwargs)
+        self._watch([UserRequirement, InstructSpeak, ParseSpeak])
+        self._init_actions([InstructSpeak, ParseSpeak, AnnounceGameResult])
+        self.stage_idx = 0
+    
+    async def _instruct_speak(self):
+        stage_idx = self.stage_idx % len(STAGE_INSTRUCTIONS)
+
+        stage_info = await InstructSpeak().run(context="", stage_idx=stage_idx)
+
+        self.stage_idx += 1
+
+        return stage_info["content"], stage_info["send_to"], stage_info["restricted_to"]
+
+    async def _parse_speak(self):
+        # 解析玩家消息并返回结果
+        parse_result = await ParseSpeak().run()
+        
+        # 理解结果，更新各角色状态、游戏状态
+
+        return "Player message processed"
+
+    async def _think(self):
+
+        if self.is_game_over:
+            self._rc.todo = AnnounceGameResult()
+            return
+        
+        # 确定当前是需要InstructSpeak还是ParseSpeak. 通过判断当前流程状态变量，以及消息的cause_by属性
+        # 0: InstructSpeak, 1: ParseSpeak,且需要判断消息的cause_by属性
+        if self._rc.memory.get()[-1].role in ["User", self.profile]:
+            # 1. 上一轮消息是用户指令，解析用户指令，开始游戏
+            # 2. 上一轮消息是Moderator自己的指令，继续发出指令，一个事情可以分几条消息来说
+            # 3. 上一轮消息是Moderator自己的解析消息，一个阶段结束，发出新一个阶段的指令
+            self._rc.todo = InstructSpeak()
+        
+        else:
+            # 上一轮消息是游戏角色的发言，解析角色的发言
+            self._rc.todo = ParseSpeak()
+
+    async def _act(self):
+        todo = self._rc.todo
+        logger.info(f"{self._setting} ready to {todo}")
+
+        memories = self.get_all_memories()
+        print("*" * 10, f"{self._setting}'s current memories: {memories}", "*" * 10)
+
+        # 根据_think的结果，执行InstructSpeak还是ParseSpeak, 并将结果返回
+        if isinstance(todo, InstructSpeak):
+            msg_content, msg_to_send_to, msg_restriced_to = await self._instruct_speak()
+            msg = Message(content=msg_content, role=self.profile, sent_from=self.name,
+                cause_by=InstructSpeak, send_to=msg_to_send_to, restricted_to=msg_restriced_to)
+        
+        elif isinstance(todo, ParseSpeak):
+            msg_content = await self._parse_speak()
+            msg = Message(content=msg_content, role=self.profile, sent_from=self.name, cause_by=ParseSpeak)
+        
+        elif isinstance(todo, AnnounceGameResult):
+            msg_content = await AnnounceGameResult().run(winner=self.winner)
+            msg = Message(content=msg_content, role=self.profile, sent_from=self.name, cause_by=AnnounceGameResult)
+        
+        logger.info(f"{self._setting}: {msg_content}")
+        
+        return msg
+    
+    def get_all_memories(self) -> str:
+        memories = self._rc.memory.get()
+        memories = [str(m) for m in memories]
+        memories = "\n".join(memories)
+        return memories

examples/werewolf_game/roles/villager.py

@@ -0,0 +1,38 @@
+from examples.werewolf_game.roles.base_player import BasePlayer
+from examples.werewolf_game.actions import Speak
+from metagpt.schema import Message
+from metagpt.logs import logger
+
+class Villager(BasePlayer):
+    def __init__(
+        self,
+        name: str = "",
+        profile: str = "Villager",
+        team: str = "good guys",
+        special_action_names: list[str] = [],
+        **kwargs,
+    ):
+        super().__init__(name, profile, team, special_action_names, **kwargs)
+
+    async def _act(self):
+
+        # todo为_think时确定的，在村民这里，就只有一种todo，即Speak
+        todo = self._rc.todo
+        logger.info(f"{self._setting}: ready to {todo}")
+
+        # 可以用这个函数获取该角色的全部记忆
+        memories = self.get_all_memories()
+        print("*" * 10, f"{self._setting}'s current memories: {memories}", "*" * 10)
+
+        # 根据自己定义的角色Action，对应地去run
+        rsp = await todo.run(profile=self.profile, context=memories)
+
+        # 返回消息，注意给Moderator发送的加密消息需要用restricted_to="Moderator"
+        msg = Message(
+            content=rsp, role=self.profile, sent_from=self.name,
+            cause_by=Speak, send_to="", restricted_to="",
+        )
+
+        logger.info(f"{self._setting}: {rsp}")
+        
+        return msg

examples/werewolf_game/roles/werewolf.py

@@ -0,0 +1,44 @@
+from examples.werewolf_game.roles.base_player import BasePlayer
+from examples.werewolf_game.actions import Speak, Hunt
+from metagpt.schema import Message
+from metagpt.logs import logger
+
+class Werewolf(BasePlayer):
+    def __init__(
+        self,
+        name: str = "",
+        profile: str = "Werewolf",
+        team: str = "werewolves",
+        special_action_names: list[str] = ["Hunt"],
+        **kwargs,
+    ):
+        super().__init__(name, profile, team, special_action_names, **kwargs)
+    
+    async def _act(self):
+        # todo为_think时确定的，有两种情况，Speak或Hunt
+        todo = self._rc.todo
+        logger.info(f"{self._setting}: ready to {str(todo)}")
+
+        # 可以用这个函数获取该角色的全部记忆
+        memories = self.get_all_memories()
+        print("*" * 10, f"{self._setting}'s current memories: {memories}", "*" * 10)
+
+        # 根据自己定义的角色Action，对应地去run，run的入参可能不同
+        if isinstance(todo, Speak):
+            rsp = await todo.run(profile=self.profile, context=memories)
+            msg = Message(
+                content=rsp, role=self.profile, sent_from=self.name,
+                cause_by=Speak, send_to="", restricted_to="",
+            )
+        
+        elif isinstance(todo, Hunt):
+            rsp = await todo.run(context=memories)
+            msg = Message(
+                content=rsp, role=self.profile, sent_from=self.name,
+                cause_by=Hunt, send_to="",
+                restricted_to=f"Moderator,{self.profile}", # 给Moderator及狼阵营发送要杀的人的加密消息
+            )
+        
+        logger.info(f"{self._setting}: {rsp}")
+
+        return msg

examples/werewolf_game/start_game.py

@@ -0,0 +1,41 @@
+import asyncio
+import platform
+import fire
+
+from examples.werewolf_game.werewolf_game import WerewolfGame
+from examples.werewolf_game.roles import Moderator, Villager, Werewolf
+
+DEFAULT_PLAYER_SETUP = """
+Game setup:
+Player1: Villager,
+Player2: Villager,
+Player3: Werewolf,
+Player4: Werewolf.
+"""
+
+async def start_game(idea: str = DEFAULT_PLAYER_SETUP, investment: float = 3.0, n_round: int = 5):
+    game = WerewolfGame()
+    game.hire([
+        Moderator(), 
+        Villager(name="Player1"),
+        Villager(name="Player2"),
+        Werewolf(name="Player3"),
+        Werewolf(name="Player4"),
+    ])
+    game.invest(investment)
+    game.start_project(idea)
+    await game.run(n_round=n_round)
+
+
+def main(idea: str = DEFAULT_PLAYER_SETUP, investment: float = 3.0, n_round: int = 10):
+    """
+    :param idea: game config instructions
+    :param investment: contribute a certain dollar amount to watch the debate
+    :param n_round: maximum rounds of the debate
+    :return:
+    """
+    asyncio.run(start_game(idea, investment, n_round))
+
+
+if __name__ == '__main__':
+    fire.Fire(main)

examples/werewolf_game/werewolf_game.py

@@ -0,0 +1,27 @@
+from metagpt.software_company import SoftwareCompany
+from metagpt.environment import Environment
+from metagpt.actions import BossRequirement as UserRequirement
+from metagpt.schema import Message
+
+class WerewolfEnvironment(Environment):
+
+    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()
+
+class WerewolfGame(SoftwareCompany):
+    """Use the "software company paradigm" to hold a werewolf game"""
+
+    environment = WerewolfEnvironment()
+
+    def start_project(self, idea):
+        """Start a project from user instruction."""
+        self.idea = idea
+        self.environment.publish_message(
+            Message(role="User", content=idea, cause_by=UserRequirement, restricted_to="Moderator")
+        )
+        print("a")

metagpt/environment.py

@@ -33,7 +33,7 @@ class Environment(BaseModel):
            Add a role in the current environment
         """
         role.set_env(self)
-        self.roles[role.profile] = role
+        self.roles[str(role._setting)] = role
 
     def add_roles(self, roles: Iterable[Role]):
         """增加一批在当前环境的角色
@@ -72,8 +72,8 @@ class Environment(BaseModel):
         """
         return self.roles
 
-    def get_role(self, name: str) -> Role:
+    def get_role(self, role_setting: str) -> Role:
         """获得环境内的指定角色
            get all the environment roles
         """
-        return self.roles.get(name, None)
+        return self.roles.get(role_setting, None)

metagpt/memory/longterm_memory.py

@@ -42,21 +42,21 @@ class LongTermMemory(Memory):
                 # and ignore adding messages from recover repeatedly
                 self.memory_storage.add(message)
 
-    def remember(self, observed: list[Message], k=0) -> list[Message]:
+    def find_news(self, observed: list[Message], k=0) -> list[Message]:
         """
-        remember the most similar k memories from observed Messages, return all when k=0
-            1. remember the short-term memory(stm) news
-            2. integrate the stm news with ltm(long-term memory) news
+        find news (previously unseen messages) from the the most recent k memories, from all memories when k=0
+            1. find the short-term memory(stm) news
+            2. furthermore, filter out similar messages based on ltm(long-term memory), get the final news
         """
-        stm_news = super(LongTermMemory, self).remember(observed, k=k)  # shot-term memory news
+        stm_news = super(LongTermMemory, self).find_news(observed, k=k)  # shot-term memory news
         if not self.memory_storage.is_initialized:
-            # memory_storage hasn't initialized, use default `remember` to get stm_news
+            # memory_storage hasn't initialized, use default `find_news` to get stm_news
             return stm_news
 
         ltm_news: list[Message] = []
         for mem in stm_news:
-            # integrate stm & ltm
-            mem_searched = self.memory_storage.search(mem)
+            # filter out messages similar to those seen previously in ltm, only keep fresh news
+            mem_searched = self.memory_storage.search_dissimilar(mem)
             if len(mem_searched) > 0:
                 ltm_news.append(mem)
         return ltm_news[-k:]

metagpt/memory/memory.py

@@ -63,8 +63,8 @@ class Memory:
         """Return the most recent k memories, return all when k=0"""
         return self.storage[-k:]
 
-    def remember(self, observed: list[Message], k=0) -> list[Message]:
-        """remember the most recent k memories from observed Messages, return all when k=0"""
+    def find_news(self, observed: list[Message], k=0) -> list[Message]:
+        """find news (previously unseen messages) from the the most recent k memories, from all memories when k=0"""
         already_observed = self.get(k)
         news: list[Message] = []
         for i in observed:

metagpt/memory/memory_storage.py

@@ -74,7 +74,7 @@ class MemoryStorage(FaissStore):
         self.persist()
         logger.info(f"Agent {self.role_id}'s memory_storage add a message")
 
-    def search(self, message: Message, k=4) -> List[Message]:
+    def search_dissimilar(self, message: Message, k=4) -> List[Message]:
         """search for dissimilar messages"""
         if not self.store:
             return []

metagpt/roles/role.py

@@ -137,6 +137,11 @@ class Role:
         """Get the role description (position)"""
         return self._setting.profile
 
+    @property
+    def name(self):
+        """Get the role name"""
+        return self._setting.name
+
     def _get_prefix(self):
         """Get the role prefix"""
         if self._setting.desc:
@@ -185,9 +190,13 @@ class Role:
 
         observed = self._rc.env.memory.get_by_actions(self._rc.watch)
         
-        self._rc.news = self._rc.memory.remember(observed)  # remember recent exact or similar memories
+        self._rc.news = self._rc.memory.find_news(observed)  # find news (previously unseen messages) from observed messages
 
         for i in env_msgs:
+            if i.restricted_to != "" and self.profile not in i.restricted_to and self.name not in i.restricted_to:
+                # if the msg is not send to the whole audience ("") nor this role (self.profile or self.name),
+                # then this role should not be able to receive it and record it into its memory
+                continue
             self.recv(i)
 
         news_text = [f"{i.role}: {i.content[:20]}..." for i in self._rc.news]

metagpt/schema.py

@@ -29,6 +29,7 @@ class Message:
     cause_by: Type["Action"] = field(default="")
     sent_from: str = field(default="")
     send_to: str = field(default="")
+    restricted_to: str = field(default="")
 
     def __str__(self):
         # prefix = '-'.join([self.role, str(self.cause_by)])

tests/metagpt/memory/test_longterm_memory.py

@@ -21,35 +21,35 @@ def test_ltm_search():
 
     idea = 'Write a cli snake game'
     message = Message(role='BOSS', content=idea, cause_by=BossRequirement)
-    news = ltm.remember([message])
+    news = ltm.find_news([message])
     assert len(news) == 1
     ltm.add(message)
 
     sim_idea = 'Write a game of cli snake'
     sim_message = Message(role='BOSS', content=sim_idea, cause_by=BossRequirement)
-    news = ltm.remember([sim_message])
+    news = ltm.find_news([sim_message])
     assert len(news) == 0
     ltm.add(sim_message)
 
     new_idea = 'Write a 2048 web game'
     new_message = Message(role='BOSS', content=new_idea, cause_by=BossRequirement)
-    news = ltm.remember([new_message])
+    news = ltm.find_news([new_message])
     assert len(news) == 1
     ltm.add(new_message)
 
     # restore from local index
     ltm_new = LongTermMemory()
     ltm_new.recover_memory(role_id, rc)
-    news = ltm_new.remember([message])
+    news = ltm_new.find_news([message])
     assert len(news) == 0
 
     ltm_new.recover_memory(role_id, rc)
-    news = ltm_new.remember([sim_message])
+    news = ltm_new.find_news([sim_message])
     assert len(news) == 0
 
     new_idea = 'Write a Battle City'
     new_message = Message(role='BOSS', content=new_idea, cause_by=BossRequirement)
-    news = ltm_new.remember([new_message])
+    news = ltm_new.find_news([new_message])
     assert len(news) == 1
 
     ltm_new.clear()

tests/metagpt/test_environment.py

@@ -24,7 +24,7 @@ def env():
 def test_add_role(env: Environment):
     role = ProductManager("Alice", "product manager", "create a new product", "limited resources")
     env.add_role(role)
-    assert env.get_role(role.profile) == role
+    assert env.get_role(str(role._setting)) == role
 
 
 def test_get_roles(env: Environment):