mirror of
https://github.com/FoundationAgents/MetaGPT.git
synced 2026-05-21 14:05:17 +02:00
update werewolf to meet WerewolfEnv
This commit is contained in:
better629
parent
d692d9fb7b
commit
b568b8f4a0
18 changed files with 451 additions and 348 deletions
|
|
@ -29,7 +29,7 @@ async def start_game(
|
|||
use_memory_selection=use_memory_selection,
|
||||
new_experience_version=new_experience_version,
|
||||
)
|
||||
logger.info(f"game_setup\n{game_setup}")
|
||||
logger.info(f"{game_setup}")
|
||||
|
||||
players = [Moderator()] + players
|
||||
game.hire(players)
|
||||
|
|
|
|||
106
metagpt/environment/werewolf/const.py
Normal file
106
metagpt/environment/werewolf/const.py
Normal file
|
|
@ -0,0 +1,106 @@
|
|||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
# @Desc :
|
||||
|
||||
from enum import Enum
|
||||
from metagpt.const import MESSAGE_ROUTE_TO_ALL
|
||||
|
||||
|
||||
class RoleType(Enum):
|
||||
VILLAGER = "Villager"
|
||||
WEREWOLF = "Werewolf"
|
||||
GUARD = "Guard"
|
||||
SEER = "Seer"
|
||||
WITCH = "Witch"
|
||||
MODERATOR = "Moderator"
|
||||
|
||||
|
||||
class RoleState(Enum):
|
||||
ALIVE = "alive" # the role is alive
|
||||
DEAD = "dead" # killed or poisoned
|
||||
KILLED = "killed" # killed by werewolf or voting
|
||||
POISONED = "poisoned" # killed by poison
|
||||
SAVED = "saved" # saved by antidote
|
||||
PROTECTED = "projected" # projected by guard
|
||||
|
||||
|
||||
class RoleActionRes(Enum):
|
||||
SAVE = "save"
|
||||
PASS = "pass" # ignore current action output
|
||||
|
||||
|
||||
# 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 continue speaking
|
||||
"restricted_to": {},
|
||||
},
|
||||
1: {
|
||||
"content": "Guard, please open your eyes!",
|
||||
"send_to": {"Moderator"}, # for moderator to continue 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": {MESSAGE_ROUTE_TO_ALL}, # 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": {MESSAGE_ROUTE_TO_ALL},
|
||||
"restricted_to": {},
|
||||
},
|
||||
18: {"content": """{player_current_dead} was eliminated.""", "send_to": {"Moderator"}, "restricted_to": {}},
|
||||
}
|
||||
49
metagpt/environment/werewolf/env_space.py
Normal file
49
metagpt/environment/werewolf/env_space.py
Normal file
|
|
@ -0,0 +1,49 @@
|
|||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
# @Desc : werewolf observation/action space and its action definition
|
||||
|
||||
from pydantic import ConfigDict, Field
|
||||
from gymnasium import spaces
|
||||
|
||||
from metagpt.environment.base_env_space import BaseEnvAction, BaseEnvActionType
|
||||
from metagpt.environment.werewolf.const import STEP_INSTRUCTIONS, RoleState
|
||||
|
||||
|
||||
class EnvActionType(BaseEnvActionType):
|
||||
NONE = 0 # no action to run, just get observation
|
||||
WOLF_KILL = 1 # wolf kill someone
|
||||
VOTE_KILL = 2 # vote kill someone
|
||||
WITCH_POISON = 3 # witch poison someone
|
||||
WITCH_SAVE = 4 # witch save someone
|
||||
GUARD_PROTECT = 5 # guard protect someone
|
||||
PROGRESS_STEP = 6 # step increment
|
||||
|
||||
|
||||
class EnvAction(BaseEnvAction):
|
||||
model_config = ConfigDict(arbitrary_types_allowed=True)
|
||||
|
||||
action_type: int = Field(default=EnvActionType.NONE, description="action type")
|
||||
player_name: str = Field(default="", description="the name of the player to do the action")
|
||||
target_player_name: str = Field(default="", description="the name of the player who take the action")
|
||||
|
||||
|
||||
def get_observation_space(player_num: int) -> spaces.Dict:
|
||||
space = spaces.Dict({
|
||||
"step_idx": spaces.Discrete(len(STEP_INSTRUCTIONS)),
|
||||
"player_states": spaces.MultiDiscrete([len(RoleState)] * player_num),
|
||||
"vote_counts": spaces.MultiDiscrete([player_num - 1] * player_num),
|
||||
|
||||
"player_current_dead": None, # TODO
|
||||
"winner": spaces.Text(16),
|
||||
"win_reason": spaces.Text(64)
|
||||
})
|
||||
return space
|
||||
|
||||
|
||||
def get_action_space() -> spaces.Dict:
|
||||
space = spaces.Dict({
|
||||
"action_type": spaces.Discrete(len(EnvActionType)),
|
||||
"player_name": spaces.Text(16), # the player to do the action
|
||||
"target_player_name": spaces.Text(16) # the target player who take the action
|
||||
})
|
||||
return space
|
||||
|
|
@ -6,7 +6,6 @@ from pydantic import Field
|
|||
|
||||
from metagpt.environment.base_env import Environment
|
||||
from metagpt.environment.werewolf.werewolf_ext_env import WerewolfExtEnv
|
||||
from metagpt.logs import logger
|
||||
from metagpt.schema import Message
|
||||
|
||||
|
||||
|
|
@ -15,13 +14,11 @@ class WerewolfEnv(Environment, WerewolfExtEnv):
|
|||
|
||||
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}"
|
||||
super().publish_message(message)
|
||||
|
||||
async def run(self, k=1):
|
||||
"""Process all Role runs by order"""
|
||||
|
|
|
|||
|
|
@ -4,98 +4,15 @@
|
|||
|
||||
import random
|
||||
from collections import Counter
|
||||
from enum import Enum
|
||||
from typing import Any, Callable, Optional
|
||||
|
||||
from pydantic import ConfigDict, Field
|
||||
|
||||
from metagpt.environment.base_env import ExtEnv, mark_as_readable, mark_as_writeable
|
||||
from metagpt.environment.base_env_space import BaseEnvAction, BaseEnvObsParams
|
||||
from metagpt.environment.base_env_space import BaseEnvObsParams
|
||||
from metagpt.environment.werewolf.env_space import EnvAction, EnvActionType
|
||||
from metagpt.logs import logger
|
||||
|
||||
|
||||
class RoleState(Enum):
|
||||
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": "",
|
||||
},
|
||||
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.const import STEP_INSTRUCTIONS, RoleState, RoleType
|
||||
|
||||
|
||||
class WerewolfExtEnv(ExtEnv):
|
||||
|
|
@ -115,13 +32,13 @@ class WerewolfExtEnv(ExtEnv):
|
|||
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)
|
||||
witch_poison_left: int = Field(default=1, description="should be 1 or 0")
|
||||
witch_antidote_left: int = Field(default=1, description="should be 1 or 0")
|
||||
|
||||
# game current round states, a round is from closing your eyes to the next time you close your eyes
|
||||
round_hunts: dict[str, str] = Field(default=dict(), description="nighttime wolf hunt result")
|
||||
round_votes: dict[str, str] = Field(
|
||||
default=dict(), description="daytime all players vote result, key=voteer, value=voted one"
|
||||
default=dict(), description="daytime all players vote result, key=voter, value=voted one"
|
||||
)
|
||||
player_hunted: Optional[str] = Field(default=None)
|
||||
player_protected: Optional[str] = Field(default=None)
|
||||
|
|
@ -140,8 +57,63 @@ class WerewolfExtEnv(ExtEnv):
|
|||
def observe(self, obs_params: Optional[BaseEnvObsParams] = None) -> Any:
|
||||
pass
|
||||
|
||||
def step(self, action: BaseEnvAction) -> tuple[dict[str, Any], float, bool, bool, dict[str, Any]]:
|
||||
pass
|
||||
def _get_obs(self):
|
||||
return {
|
||||
"game_setup": self.game_setup,
|
||||
"step_idx": self.step_idx,
|
||||
|
||||
"living_players": self.living_players,
|
||||
"werewolf_players": self.werewolf_players,
|
||||
"player_hunted": self.player_hunted,
|
||||
"player_current_dead": self.player_current_dead,
|
||||
"witch_poison_left": self.witch_poison_left,
|
||||
"witch_antidote_left": self.witch_antidote_left,
|
||||
"winner": self.winner,
|
||||
"win_reason": self.win_reason
|
||||
}
|
||||
|
||||
def step(self, action: EnvAction) -> tuple[dict[str, Any], float, bool, bool, dict[str, Any]]:
|
||||
action_type = action.action_type
|
||||
player_name = action.player_name
|
||||
target_player_name = action.target_player_name
|
||||
if action_type == EnvActionType.WOLF_KILL:
|
||||
self.wolf_kill_someone(wolf_name=player_name, player_name=target_player_name)
|
||||
elif action_type == EnvActionType.VOTE_KILL:
|
||||
self.vote_kill_someone(voter_name=player_name, player_name=target_player_name)
|
||||
elif action_type == EnvActionType.WITCH_POISON:
|
||||
self.witch_poison_someone(witch_name=player_name, player_name=target_player_name)
|
||||
elif action_type == EnvActionType.WITCH_SAVE:
|
||||
self.witch_save_someone(witch_name=player_name, player_name=target_player_name)
|
||||
elif action_type == EnvActionType.GUARD_PROTECT:
|
||||
self.guard_protect_someone(guard_name=player_name, player_name=target_player_name)
|
||||
elif action_type == EnvActionType.PROGRESS_STEP:
|
||||
self.progress_step()
|
||||
elif action_type == EnvActionType.NONE:
|
||||
pass
|
||||
else:
|
||||
raise ValueError(f"not supported action_type: {action_type}")
|
||||
|
||||
self.update_game_states()
|
||||
terminated = self._check_game_finish()
|
||||
obs = self._get_obs()
|
||||
return obs, 1.0, terminated, False, {}
|
||||
|
||||
def _check_game_finish(self) -> bool:
|
||||
"""return True if game finished else False"""
|
||||
# game's termination condition
|
||||
terminated = False
|
||||
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"
|
||||
terminated = True
|
||||
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"
|
||||
terminated = True
|
||||
return terminated
|
||||
|
||||
@property
|
||||
def living_players(self) -> list[str]:
|
||||
|
|
@ -161,12 +133,12 @@ class WerewolfExtEnv(ExtEnv):
|
|||
|
||||
@property
|
||||
def werewolf_players(self) -> list[str]:
|
||||
player_names = self._role_type_players(role_type="Werewolf")
|
||||
player_names = self._role_type_players(role_type=RoleType.WEREWOLF.value)
|
||||
return player_names
|
||||
|
||||
@property
|
||||
def villager_players(self) -> list[str]:
|
||||
player_names = self._role_type_players(role_type="Villager")
|
||||
player_names = self._role_type_players(role_type=RoleType.VILLAGER.value)
|
||||
return player_names
|
||||
|
||||
def _init_players_state(self, players: list["Role"]):
|
||||
|
|
@ -193,14 +165,14 @@ class WerewolfExtEnv(ExtEnv):
|
|||
"""init players using different roles' num"""
|
||||
role_objs = []
|
||||
for role_obj in role_uniq_objs:
|
||||
if str(role_obj) == "Villager":
|
||||
if "Villager" in str(role_obj):
|
||||
role_objs.extend([role_obj] * num_villager)
|
||||
elif str(role_obj) == "Werewolf":
|
||||
elif "Werewolf" in str(role_obj):
|
||||
role_objs.extend([role_obj] * num_werewolf)
|
||||
else:
|
||||
role_objs.append(role_obj)
|
||||
if shuffle:
|
||||
random.shuffle(len(role_objs))
|
||||
random.shuffle(role_objs)
|
||||
if add_human:
|
||||
assigned_role_idx = random.randint(0, len(role_objs) - 1)
|
||||
assigned_role = role_objs[assigned_role_idx]
|
||||
|
|
@ -233,10 +205,12 @@ class WerewolfExtEnv(ExtEnv):
|
|||
roletype_state = self.players_state[player_name]
|
||||
self.players_state[player_name] = (roletype_state[0], state)
|
||||
|
||||
def _check_valid_role(self, player: "Role", role_type: str) -> bool:
|
||||
return True if role_type in str(player) else False
|
||||
def _check_valid_role(self, player_name: str, role_type: str) -> bool:
|
||||
roletype_state = self.players_state.get(player_name)
|
||||
return True if roletype_state and role_type in roletype_state[0] else False
|
||||
|
||||
def _check_player_continue(self, player_name: str, particular_step: int = -1) -> bool:
|
||||
"""to check if can do the operation to the player"""
|
||||
step_idx = self.step_idx % self.per_round_steps
|
||||
if particular_step > 0 and step_idx != particular_step: # step no
|
||||
# particular_step = 18, not daytime vote time, ignore
|
||||
|
|
@ -253,6 +227,10 @@ class WerewolfExtEnv(ExtEnv):
|
|||
self.step_idx += 1
|
||||
return instruction
|
||||
|
||||
@mark_as_writeable
|
||||
def progress_step(self):
|
||||
self.step_idx += 1
|
||||
|
||||
@mark_as_readable
|
||||
def get_players_state(self, player_names: list[str]) -> dict[str, RoleState]:
|
||||
players_state = {
|
||||
|
|
@ -263,14 +241,14 @@ class WerewolfExtEnv(ExtEnv):
|
|||
return players_state
|
||||
|
||||
@mark_as_writeable
|
||||
def vote_kill_someone(self, voteer: "Role", player_name: str = None):
|
||||
def vote_kill_someone(self, voter_name: str, player_name: str = None):
|
||||
"""player vote result at daytime
|
||||
player_name: if it's None, regard as abstaining from voting
|
||||
"""
|
||||
if not self._check_player_continue(voteer.name, particular_step=18): # 18=step no
|
||||
if not self._check_player_continue(voter_name, particular_step=18): # 18=step no
|
||||
return
|
||||
|
||||
self.round_votes[voteer.name] = player_name
|
||||
self.round_votes[voter_name] = player_name
|
||||
# check if all living players finish voting, then get the dead one
|
||||
if list(self.round_votes.keys()) == self.living_players:
|
||||
voted_all = list(self.round_votes.values()) # TODO in case of tie vote, check who was voted first
|
||||
|
|
@ -279,41 +257,55 @@ class WerewolfExtEnv(ExtEnv):
|
|||
self._update_players_state([self.player_current_dead])
|
||||
|
||||
@mark_as_writeable
|
||||
def wolf_kill_someone(self, wolf: "Role", player_name: str):
|
||||
if not self._check_valid_role(wolf, "Werewolf"):
|
||||
def wolf_kill_someone(self, wolf_name: str, player_name: str):
|
||||
if not self._check_valid_role(wolf_name, RoleType.WEREWOLF.value):
|
||||
return
|
||||
if not self._check_player_continue(wolf.name, particular_step=5): # 5=step no
|
||||
if not self._check_player_continue(wolf_name, particular_step=5): # 5=step no
|
||||
return
|
||||
|
||||
self.round_hunts[wolf.name] = player_name
|
||||
living_werewolf = [p for p in self.werewolf_players if p in self.living_players]
|
||||
self.round_hunts[wolf_name] = player_name
|
||||
# living_werewolf = [p for p in self.werewolf_players if p in self.living_players]
|
||||
# check if all living wolfs finish hunting, then get the hunted one
|
||||
if list(self.round_hunts.keys()) == living_werewolf:
|
||||
hunted_all = list(self.round_hunts.values())
|
||||
self.player_hunted = Counter(hunted_all).most_common()[0][0]
|
||||
# if list(self.round_hunts.keys()) == living_werewolf:
|
||||
# hunted_all = list(self.round_hunts.values())
|
||||
# self.player_hunted = Counter(hunted_all).most_common()[0][0]
|
||||
self.player_hunted = player_name
|
||||
|
||||
@mark_as_writeable
|
||||
def witch_poison_someone(self, witch: "Role", player_name: str = None):
|
||||
if not self._check_valid_role(witch, "Witch"):
|
||||
def _witch_poison_or_save_someone(self, witch_name: str, player_name: str = None,
|
||||
state: RoleState = RoleState.POISONED):
|
||||
if not self._check_valid_role(witch_name, RoleType.WITCH.value):
|
||||
return
|
||||
if not self._check_player_continue(player_name):
|
||||
return
|
||||
|
||||
self._update_players_state([player_name], RoleState.POISONED)
|
||||
self.player_poisoned = player_name
|
||||
assert state in [RoleState.POISONED, RoleState.SAVED]
|
||||
self._update_players_state([player_name], state)
|
||||
if state == RoleState.POISONED:
|
||||
self.player_poisoned = player_name
|
||||
self.witch_poison_left -= 1
|
||||
else:
|
||||
# self.player_protected = player_name
|
||||
self.is_hunted_player_saved = True
|
||||
self.witch_antidote_left -= 1
|
||||
|
||||
@mark_as_writeable
|
||||
def witch_save_someone(self, witch: "Role", player_name: str = None):
|
||||
if not self._check_valid_role(witch, "Witch"):
|
||||
def witch_poison_someone(self, witch_name: str, player_name: str = None):
|
||||
self._witch_poison_or_save_someone(witch_name, player_name, RoleState.POISONED)
|
||||
|
||||
@mark_as_writeable
|
||||
def witch_save_someone(self, witch_name: str, player_name: str = None):
|
||||
self._witch_poison_or_save_someone(witch_name, player_name, RoleState.SAVED)
|
||||
|
||||
@mark_as_writeable
|
||||
def guard_protect_someone(self, guard_name: str, player_name: str = None):
|
||||
if not self._check_valid_role(guard_name, RoleType.GUARD.value):
|
||||
return
|
||||
if not self._check_player_continue(player_name):
|
||||
return
|
||||
|
||||
self._update_players_state([player_name], RoleState.SAVED)
|
||||
self.player_protected = player_name
|
||||
|
||||
@mark_as_writeable
|
||||
def update_game_states(self, memories: list):
|
||||
def update_game_states(self):
|
||||
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
|
||||
|
|
@ -335,16 +327,6 @@ class WerewolfExtEnv(ExtEnv):
|
|||
self.player_protected = None
|
||||
self.is_hunted_player_saved = False
|
||||
self.player_poisoned = None
|
||||
|
||||
# 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
|
||||
elif step_idx == 18:
|
||||
# updated use vote_kill_someone
|
||||
pass
|
||||
|
|
|
|||
|
|
@ -1,12 +1,13 @@
|
|||
from metagpt.actions import Action
|
||||
from metagpt.environment.werewolf.werewolf_ext_env import STEP_INSTRUCTIONS
|
||||
from metagpt.environment.werewolf.const 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": {}})
|
||||
instruction_info = STEP_INSTRUCTIONS.get(step_idx, {"content": "Unknown instruction.", "send_to": {},
|
||||
"restricted_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)
|
||||
|
|
@ -20,7 +21,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"]
|
||||
return content, instruction_info["send_to"], instruction_info["restricted_to"]
|
||||
|
||||
|
||||
class ParseSpeak(Action):
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
from metagpt.ext.werewolf.actions.common_actions import NighttimeWhispers
|
||||
from metagpt.environment.werewolf.const import RoleActionRes
|
||||
|
||||
|
||||
class Save(NighttimeWhispers):
|
||||
|
|
@ -41,6 +42,6 @@ class Poison(NighttimeWhispers):
|
|||
|
||||
async def run(self, *args, **kwargs):
|
||||
rsp = await super().run(*args, **kwargs)
|
||||
if "pass" in rsp.lower():
|
||||
if RoleActionRes.PASS.value in rsp.lower():
|
||||
action_name, rsp = rsp.split() # 带PASS，只需回复PASS，不需要带上action名，否则是Poison PlayerX，无需改动
|
||||
return rsp
|
||||
|
|
|
|||
|
|
@ -1,5 +1,7 @@
|
|||
import re
|
||||
|
||||
from pydantic import SerializeAsAny, Field
|
||||
|
||||
from metagpt.ext.werewolf.actions import (
|
||||
ACTIONS,
|
||||
AddNewExperiences,
|
||||
|
|
@ -9,65 +11,72 @@ from metagpt.ext.werewolf.actions import (
|
|||
RetrieveExperiences,
|
||||
Speak,
|
||||
)
|
||||
from metagpt.ext.werewolf.schema import RoleExperience
|
||||
from metagpt.ext.werewolf.schema import RoleExperience, WwMessage
|
||||
from metagpt.environment.werewolf.const import RoleType, RoleState
|
||||
from metagpt.logs import logger
|
||||
from metagpt.roles import Role
|
||||
from metagpt.schema import Message
|
||||
from metagpt.actions.action import Action
|
||||
|
||||
|
||||
class BasePlayer(Role):
|
||||
def __init__(
|
||||
self,
|
||||
name: str = "PlayerXYZ",
|
||||
profile: str = "BasePlayer",
|
||||
special_action_names: list[str] = [],
|
||||
use_reflection: bool = True,
|
||||
use_experience: bool = False,
|
||||
use_memory_selection: bool = False,
|
||||
new_experience_version: str = "",
|
||||
**kwargs,
|
||||
):
|
||||
super().__init__(name, profile, **kwargs)
|
||||
# 通过 set_status() 更新状态。
|
||||
self.status = 0 # 0代表活着，1代表死亡
|
||||
name: str = "PlayerXYZ"
|
||||
profile: str = "BasePlayer"
|
||||
special_action_names: list[str] = []
|
||||
use_reflection: bool = True
|
||||
use_experience: bool = False
|
||||
use_memory_selection: bool = False
|
||||
new_experience_version: str = ""
|
||||
status: RoleState = RoleState.ALIVE
|
||||
|
||||
special_actions: list[SerializeAsAny[Action]] = Field(default=[], validate_default=True)
|
||||
experiences: list[RoleExperience] = []
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
super().__init__(**kwargs)
|
||||
# 技能和监听配置
|
||||
self._watch([InstructSpeak]) # 监听Moderator的指令以做行动
|
||||
special_actions = [ACTIONS[action_name] for action_name in special_action_names]
|
||||
special_actions = [ACTIONS[action_name] for action_name in self.special_action_names]
|
||||
capable_actions = [Speak] + special_actions
|
||||
self.set_actions(capable_actions) # 给角色赋予行动技能
|
||||
self.special_actions = special_actions
|
||||
|
||||
self.use_reflection = use_reflection
|
||||
if not self.use_reflection and use_experience:
|
||||
if not self.use_reflection and self.use_experience:
|
||||
logger.warning("You must enable use_reflection before using experience")
|
||||
self.use_experience = False
|
||||
else:
|
||||
self.use_experience = use_experience
|
||||
self.new_experience_version = new_experience_version
|
||||
self.use_memory_selection = use_memory_selection
|
||||
|
||||
self.experiences = []
|
||||
|
||||
async def _observe(self) -> int:
|
||||
if self.status == 1:
|
||||
async def _observe(self, ignore_memory=False) -> int:
|
||||
if self.status != RoleState.ALIVE:
|
||||
# 死者不再参与游戏
|
||||
return 0
|
||||
news = []
|
||||
if not news:
|
||||
news = self.rc.msg_buffer.pop_all()
|
||||
old_messages = [] if ignore_memory else self.rc.memory.get()
|
||||
for m in news:
|
||||
if len(m.restricted_to) and self.profile not in m.restricted_to and self.name not in m.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.rc.memory.add(m)
|
||||
self.rc.news = [
|
||||
n for n in news if (n.cause_by in self.rc.watch or self.profile in n.send_to) and n not in old_messages
|
||||
]
|
||||
|
||||
await super()._observe()
|
||||
# 只有发给全体的（""）或发给自己的（self.profile）消息需要走下面的_react流程，
|
||||
# 其他的收听到即可，不用做动作
|
||||
self.rc.news = [msg for msg in self.rc.news if msg.send_to in ["", self.profile]]
|
||||
# TODO to delete
|
||||
# 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.send_to:
|
||||
if not news.restricted_to:
|
||||
# 消息接收范围为全体角色的，做公开发言（发表投票观点也算发言）
|
||||
self.rc.todo = Speak()
|
||||
elif self.profile in news.send_to:
|
||||
# FIXME: hard code to split, restricted为"Moderator"或"Moderator,角色profile"
|
||||
elif self.profile in news.restricted_to:
|
||||
# FIXME: hard code to split, restricted为"Moderator"或"Moderator, 角色profile"
|
||||
# Moderator加密发给自己的，意味着要执行角色的特殊动作
|
||||
self.rc.todo = self.special_actions[0]()
|
||||
|
||||
|
|
@ -79,7 +88,6 @@ class BasePlayer(Role):
|
|||
# 可以用这个函数获取该角色的全部记忆和最新的instruction
|
||||
memories = self.get_all_memories()
|
||||
latest_instruction = self.get_latest_instruction()
|
||||
# print("*" * 10, f"{self._setting}'s current memories: {memories}", "*" * 10)
|
||||
|
||||
reflection = (
|
||||
await Reflect().run(
|
||||
|
|
@ -107,20 +115,21 @@ class BasePlayer(Role):
|
|||
reflection=reflection,
|
||||
experiences=experiences,
|
||||
)
|
||||
send_to = ""
|
||||
restricted_to = {}
|
||||
|
||||
elif isinstance(todo, NighttimeWhispers):
|
||||
rsp = await todo.run(
|
||||
profile=self.profile, name=self.name, context=memories, reflection=reflection, experiences=experiences
|
||||
)
|
||||
send_to = {"Moderator", self.profile} # 给Moderator发送使用特殊技能的加密消息
|
||||
restricted_to = {RoleType.MODERATOR.value, self.profile} # 给Moderator发送使用特殊技能的加密消息
|
||||
|
||||
msg = Message(
|
||||
msg = WwMessage(
|
||||
content=rsp,
|
||||
role=self.profile,
|
||||
sent_from=self.name,
|
||||
cause_by=type(todo),
|
||||
send_to=send_to,
|
||||
send_to={},
|
||||
restricted_to=restricted_to
|
||||
)
|
||||
|
||||
self.experiences.append(
|
||||
|
|
@ -149,7 +158,7 @@ class BasePlayer(Role):
|
|||
def get_latest_instruction(self) -> str:
|
||||
return self.rc.important_memory[-1].content # 角色监听着Moderator的InstructSpeak，是其重要记忆，直接获取即可
|
||||
|
||||
def set_status(self, new_status):
|
||||
def set_status(self, new_status: RoleState):
|
||||
self.status = new_status
|
||||
|
||||
def record_experiences(self, round_id: str, outcome: str, game_setup: str):
|
||||
|
|
|
|||
|
|
@ -1,7 +1,8 @@
|
|||
from metagpt.ext.werewolf.roles.base_player import BasePlayer
|
||||
from metagpt.environment.werewolf.const import RoleType
|
||||
|
||||
|
||||
class Guard(BasePlayer):
|
||||
name: str = "Guard"
|
||||
profile: str = "Guard"
|
||||
name: str = RoleType.GUARD.value
|
||||
profile: str = RoleType.GUARD.value
|
||||
special_action_names: list[str] = ["Protect"]
|
||||
|
|
|
|||
|
|
@ -1,7 +1,8 @@
|
|||
from metagpt.ext.werewolf.actions import Speak
|
||||
from metagpt.ext.werewolf.roles import BasePlayer
|
||||
from metagpt.logs import logger
|
||||
from metagpt.schema import Message
|
||||
from metagpt.ext.werewolf.schema import WwMessage
|
||||
from metagpt.environment.werewolf.const import RoleType
|
||||
|
||||
|
||||
async def _act(self):
|
||||
|
|
@ -22,14 +23,15 @@ async def _act(self):
|
|||
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_restricted_to = {} if isinstance(todo, Speak) else {RoleType.MODERATOR.value, self.profile}
|
||||
|
||||
msg = Message(
|
||||
msg = WwMessage(
|
||||
content=rsp,
|
||||
role=self.profile,
|
||||
sent_from=self.name,
|
||||
cause_by=msg_cause_by,
|
||||
send_to=msg_restricted_to, # 给Moderator及自身阵营发送加密消息
|
||||
send_to={},
|
||||
restricted_to=msg_restricted_to, # 给Moderator及自身阵营发送加密消息
|
||||
)
|
||||
|
||||
logger.info(f"{self._setting}: {rsp}")
|
||||
|
|
|
|||
|
|
@ -1,10 +1,11 @@
|
|||
import re
|
||||
from collections import Counter
|
||||
from typing import Union
|
||||
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.environment.werewolf.const import STEP_INSTRUCTIONS, RoleType
|
||||
from metagpt.environment.werewolf.env_space import EnvAction, EnvActionType
|
||||
from metagpt.ext.werewolf.actions import Hunt, Poison, Protect, Save, Verify
|
||||
from metagpt.ext.werewolf.actions.moderator_actions import (
|
||||
AnnounceGameResult,
|
||||
|
|
@ -12,55 +13,29 @@ from metagpt.ext.werewolf.actions.moderator_actions import (
|
|||
ParseSpeak,
|
||||
)
|
||||
from metagpt.logs import logger
|
||||
from metagpt.roles import Role
|
||||
from metagpt.schema import Message
|
||||
from metagpt.ext.werewolf.roles.base_player import BasePlayer
|
||||
from metagpt.ext.werewolf.schema import WwMessage
|
||||
from metagpt.environment.werewolf.const import RoleState, RoleActionRes
|
||||
|
||||
|
||||
class Moderator(Role):
|
||||
name: str = "Moderator"
|
||||
profile: str = "Moderator"
|
||||
class Moderator(BasePlayer):
|
||||
name: str = RoleType.MODERATOR.value
|
||||
profile: str = RoleType.MODERATOR.value
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
name: str = "Moderator",
|
||||
profile: str = "Moderator",
|
||||
**kwargs,
|
||||
):
|
||||
super().__init__(name, profile, **kwargs)
|
||||
def __init__(self, **kwargs):
|
||||
super().__init__(**kwargs)
|
||||
self._watch([UserRequirement, InstructSpeak, ParseSpeak])
|
||||
self.set_actions([InstructSpeak, ParseSpeak, AnnounceGameResult])
|
||||
self.step_idx = 0
|
||||
self.eval_step_idx = []
|
||||
|
||||
# game states
|
||||
self.step_idx = 0
|
||||
self.game_setup = ""
|
||||
self.living_players = []
|
||||
self.werewolf_players = []
|
||||
self.villager_players = []
|
||||
self.special_role_players = []
|
||||
self.winner = None
|
||||
self.win_reason = None
|
||||
self.witch_poison_left = 1
|
||||
self.witch_antidote_left = 1
|
||||
|
||||
# player states of current night
|
||||
self.player_hunted = None
|
||||
self.player_protected = None
|
||||
self.is_hunted_player_saved = False
|
||||
self.player_poisoned = None
|
||||
self.player_current_dead = []
|
||||
|
||||
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
|
||||
]
|
||||
|
||||
def update_player_status(self, player_names: list[str]):
|
||||
if not player_names:
|
||||
return
|
||||
|
|
@ -68,33 +43,23 @@ class Moderator(Role):
|
|||
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) # 更新为死亡
|
||||
role.set_status(new_status=RoleState.DEAD) # 更新为死亡
|
||||
|
||||
def _record_all_experiences(self):
|
||||
logger.info(f"The winner of the game: {self.winner}, start to record roles' experiences")
|
||||
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:
|
||||
continue
|
||||
if self.winner == "werewolf":
|
||||
outcome = "won" if role.name in self.werewolf_players else "lost"
|
||||
outcome = "won" if role.profile in RoleType.WEREWOLF.value else "lost"
|
||||
else:
|
||||
outcome = "won" if role.name not in self.werewolf_players else "lost"
|
||||
outcome = "won" if role.profile not in RoleType.WEREWOLF.value else "lost"
|
||||
role.record_experiences(round_id=timestamp, outcome=outcome, game_setup=self.game_setup)
|
||||
|
||||
async def _instruct_speak(self):
|
||||
step_idx = self.step_idx % len(STEP_INSTRUCTIONS)
|
||||
self.step_idx += 1
|
||||
return await InstructSpeak().run(
|
||||
step_idx,
|
||||
living_players=self.living_players,
|
||||
werewolf_players=self.werewolf_players,
|
||||
player_hunted=self.player_hunted,
|
||||
player_current_dead=self.player_current_dead,
|
||||
)
|
||||
|
||||
async def _parse_speak(self, memories):
|
||||
logger.info(self.step_idx)
|
||||
logger.info(f"step_idx: {self.step_idx}")
|
||||
|
||||
latest_msg = memories[-1]
|
||||
latest_msg_content = latest_msg.content
|
||||
|
|
@ -104,157 +69,135 @@ class Moderator(Role):
|
|||
|
||||
# default return
|
||||
msg_content = "Understood"
|
||||
send_to = set()
|
||||
restricted_to = set()
|
||||
|
||||
msg_cause_by = latest_msg.cause_by
|
||||
if msg_cause_by == Hunt:
|
||||
self.player_hunted = target
|
||||
self.rc.env.step(EnvAction(action_type=EnvActionType.WOLF_KILL, player_name=latest_msg.send_from,
|
||||
target_player_name=target))
|
||||
elif msg_cause_by == Protect:
|
||||
self.player_protected = target
|
||||
self.rc.env.step(EnvAction(action_type=EnvActionType.GUARD_PROTECT, player_name=latest_msg.send_from,
|
||||
target_player_name=target))
|
||||
elif msg_cause_by == Verify:
|
||||
if target in self.werewolf_players:
|
||||
msg_content = f"{target} is a werewolf"
|
||||
else:
|
||||
msg_content = f"{target} is a good guy"
|
||||
send_to = {"Moderator", "Seer"}
|
||||
restricted_to = {RoleType.MODERATOR.value, RoleType.SEER.value}
|
||||
elif msg_cause_by == Save:
|
||||
if "pass" in latest_msg_content.lower():
|
||||
if RoleActionRes.PASS.value in latest_msg_content.lower():
|
||||
# the role ignore to response, answer `pass`
|
||||
pass
|
||||
elif not self.witch_antidote_left:
|
||||
msg_content = "You have no antidote left and thus can not save the player"
|
||||
send_to = {"Moderator", "Witch"}
|
||||
restricted_to = {RoleType.MODERATOR.value, RoleType.WITCH.value}
|
||||
else:
|
||||
self.witch_antidote_left -= 1
|
||||
self.is_hunted_player_saved = True
|
||||
self.rc.env.step(EnvAction(action_type=EnvActionType.WITCH_SAVE, player_name=latest_msg.send_from,
|
||||
target_player_name=target))
|
||||
elif msg_cause_by == Poison:
|
||||
if "pass" in latest_msg_content.lower():
|
||||
if RoleActionRes.PASS.value in latest_msg_content.lower():
|
||||
pass
|
||||
elif not self.witch_poison_left:
|
||||
msg_content = "You have no poison left and thus can not poison the player"
|
||||
send_to = {"Moderator", "Witch"}
|
||||
restricted_to = {RoleType.MODERATOR.value, RoleType.WITCH.value}
|
||||
else:
|
||||
self.witch_poison_left -= 1
|
||||
self.player_poisoned = target # "" if not poisoned and "PlayerX" if poisoned
|
||||
self.rc.env.step(EnvAction(action_type=EnvActionType.WITCH_POISON, player_name=latest_msg.send_from,
|
||||
target_player_name=target))
|
||||
|
||||
return msg_content, send_to
|
||||
return msg_content, restricted_to
|
||||
|
||||
def _update_game_states(self, memories):
|
||||
step_idx = self.step_idx % len(STEP_INSTRUCTIONS)
|
||||
if step_idx not in [15, 18] or self.step_idx in self.eval_step_idx: # FIXME: hard code
|
||||
return
|
||||
else:
|
||||
self.eval_step_idx.append(self.step_idx) # record evaluation, avoid repetitive evaluation at the same step
|
||||
def _update_player_status(self, step_idx: int, player_current_dead: list[str]):
|
||||
"""update dead player's status"""
|
||||
if step_idx in [15, 18]:
|
||||
self.update_player_status(player_current_dead)
|
||||
|
||||
if step_idx == 15: # FIXME: hard code
|
||||
# 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: # FIXME: hard code
|
||||
# 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()
|
||||
|
||||
def _record_game_history(self):
|
||||
if self.step_idx % len(STEP_INSTRUCTIONS) == 0 or self.winner is not None:
|
||||
def _record_game_history(self, step_idx: int):
|
||||
if step_idx % len(STEP_INSTRUCTIONS) == 0 or self.winner:
|
||||
logger.info("a night and day cycle completed, examine all history")
|
||||
print(self.get_all_memories())
|
||||
logger.debug(f"all_memories: {self.get_all_memories()}")
|
||||
with open(DEFAULT_WORKSPACE_ROOT / "werewolf_transcript.txt", "w") as f:
|
||||
f.write(self.get_all_memories())
|
||||
|
||||
async def _think(self):
|
||||
if self.winner is not None:
|
||||
if self.winner:
|
||||
self.rc.todo = AnnounceGameResult()
|
||||
return
|
||||
|
||||
latest_msg = self.rc.memory.get()[-1]
|
||||
if latest_msg.role in ["User"]:
|
||||
# 上一轮消息是用户指令，解析用户指令，开始游戏
|
||||
game_setup = latest_msg.content
|
||||
self._parse_game_setup(game_setup)
|
||||
if latest_msg.role in ["User", "Human", self.profile]:
|
||||
# 1. 上一轮消息是用户指令，解析用户指令，开始游戏
|
||||
# 2.1. 上一轮消息是Moderator自己的指令，继续发出指令，一个事情可以分几条消息来说
|
||||
# 2.2. 上一轮消息是Moderator自己的解析消息，一个阶段结束，发出新一个阶段的指令
|
||||
self.rc.todo = InstructSpeak()
|
||||
|
||||
elif latest_msg.role in [self.profile]:
|
||||
# 1. 上一轮消息是Moderator自己的指令，继续发出指令，一个事情可以分几条消息来说
|
||||
# 2. 上一轮消息是Moderator自己的解析消息，一个阶段结束，发出新一个阶段的指令
|
||||
self.rc.todo = InstructSpeak()
|
||||
|
||||
else:
|
||||
# 上一轮消息是游戏角色的发言，解析角色的发言
|
||||
self.rc.todo = ParseSpeak()
|
||||
|
||||
def _init_fields_from_obj(self, obs: dict[str, Union[int, str, list[str]]]):
|
||||
self.game_setup = obs.get("game_setup", "")
|
||||
self.winner = obs.get("winner")
|
||||
self.win_reason = obs.get("win_reason")
|
||||
self.werewolf_players = obs.get("werewolf_players", [])
|
||||
self.witch_poison_left = obs.get("witch_poison_left", 0)
|
||||
self.witch_antidote_left = obs.get("witch_antidote_left", 0)
|
||||
|
||||
async def _act(self):
|
||||
todo = self.rc.todo
|
||||
logger.info(f"{self._setting} ready to {todo}")
|
||||
|
||||
memories = self.get_all_memories(mode="msg")
|
||||
|
||||
# 若进行完一夜一日的循环，打印和记录一次完整发言历史
|
||||
self._record_game_history()
|
||||
obs, _, _, _, _ = self.rc.env.step(action=EnvAction(action_type=EnvActionType.NONE))
|
||||
step_idx = obs["step_idx"]
|
||||
living_players = obs["living_players"]
|
||||
werewolf_players = obs["werewolf_players"]
|
||||
player_hunted = obs["player_hunted"]
|
||||
player_current_dead = obs["player_current_dead"]
|
||||
self._init_fields_from_obj(obs)
|
||||
|
||||
# 若一晚或一日周期结束，对当晚或当日的死者进行总结，并更新游戏状态
|
||||
self._update_game_states(memories)
|
||||
# 若进行完一夜一日的循环，打印和记录一次完整发言历史
|
||||
self._record_game_history(step_idx)
|
||||
|
||||
# 若一晚或一日周期结束，对当晚或当日的死者进行总结，并更新玩家状态
|
||||
self._update_player_status(step_idx, player_current_dead)
|
||||
if self.winner:
|
||||
self._record_all_experiences()
|
||||
|
||||
# 根据_think的结果，执行InstructSpeak还是ParseSpeak, 并将结果返回
|
||||
if isinstance(todo, InstructSpeak):
|
||||
msg_content, msg_to_send_to = await self._instruct_speak()
|
||||
msg_content, msg_to_send_to, msg_restricted_to = await InstructSpeak().run(
|
||||
step_idx,
|
||||
living_players=living_players,
|
||||
werewolf_players=werewolf_players,
|
||||
player_hunted=player_hunted,
|
||||
player_current_dead=player_current_dead,
|
||||
)
|
||||
# msg_content = f"Step {self.step_idx}: {msg_content}" # HACK: 加一个unique的step_idx避免记忆的自动去重
|
||||
msg = Message(
|
||||
msg = WwMessage(
|
||||
content=msg_content,
|
||||
role=self.profile,
|
||||
sent_from=self.name,
|
||||
cause_by=InstructSpeak,
|
||||
send_to=msg_to_send_to,
|
||||
restricted_to=msg_restricted_to
|
||||
)
|
||||
self.rc.env.step(EnvAction(action_type=EnvActionType.PROGRESS_STEP)) # to update step_idx
|
||||
|
||||
elif isinstance(todo, ParseSpeak):
|
||||
msg_content, msg_to_send_to = await self._parse_speak(memories)
|
||||
msg_content, msg_restricted_to = await self._parse_speak(memories)
|
||||
# msg_content = f"Step {self.step_idx}: {msg_content}" # HACK: 加一个unique的step_idx避免记忆的自动去重
|
||||
msg = Message(
|
||||
msg = WwMessage(
|
||||
content=msg_content,
|
||||
role=self.profile,
|
||||
sent_from=self.name,
|
||||
cause_by=ParseSpeak,
|
||||
send_to=msg_to_send_to,
|
||||
send_to={},
|
||||
restricted_to=msg_restricted_to
|
||||
)
|
||||
|
||||
elif isinstance(todo, AnnounceGameResult):
|
||||
msg_content = await AnnounceGameResult().run(winner=self.winner, win_reason=self.win_reason)
|
||||
msg = Message(content=msg_content, role=self.profile, sent_from=self.name, cause_by=AnnounceGameResult)
|
||||
msg = WwMessage(content=msg_content, role=self.profile, sent_from=self.name, cause_by=AnnounceGameResult)
|
||||
|
||||
logger.info(f"{self._setting}: {msg_content}")
|
||||
|
||||
|
|
|
|||
|
|
@ -1,7 +1,8 @@
|
|||
from metagpt.ext.werewolf.roles.base_player import BasePlayer
|
||||
from metagpt.environment.werewolf.const import RoleType
|
||||
|
||||
|
||||
class Seer(BasePlayer):
|
||||
name: str = "Seer"
|
||||
profile: str = "Seer"
|
||||
name: str = RoleType.SEER.value
|
||||
profile: str = RoleType.SEER.value
|
||||
special_action_names: list[str] = ["Verify"]
|
||||
|
|
|
|||
|
|
@ -1,7 +1,8 @@
|
|||
from metagpt.ext.werewolf.roles.base_player import BasePlayer
|
||||
from metagpt.environment.werewolf.const import RoleType
|
||||
|
||||
|
||||
class Villager(BasePlayer):
|
||||
name: str = "Villager"
|
||||
profile: str = "Villager"
|
||||
name: str = RoleType.VILLAGER.value
|
||||
profile: str = RoleType.VILLAGER.value
|
||||
special_action_names: list[str] = []
|
||||
|
|
|
|||
|
|
@ -1,10 +1,11 @@
|
|||
from metagpt.ext.werewolf.actions import Impersonate, Speak
|
||||
from metagpt.ext.werewolf.roles.base_player import BasePlayer
|
||||
from metagpt.environment.werewolf.const import RoleType
|
||||
|
||||
|
||||
class Werewolf(BasePlayer):
|
||||
name: str = "Werewolf"
|
||||
profile: str = "Werewolf"
|
||||
name: str = RoleType.WEREWOLF.value
|
||||
profile: str = RoleType.WEREWOLF.value
|
||||
special_action_names: list[str] = ["Hunt"]
|
||||
|
||||
async def _think(self):
|
||||
|
|
|
|||
|
|
@ -1,20 +1,21 @@
|
|||
from metagpt.ext.werewolf.actions import InstructSpeak, Poison, Save, Speak
|
||||
from metagpt.ext.werewolf.roles.base_player import BasePlayer
|
||||
from metagpt.environment.werewolf.const import RoleType
|
||||
|
||||
|
||||
class Witch(BasePlayer):
|
||||
name: str = "Witch"
|
||||
profile: str = "Witch"
|
||||
name: str = RoleType.WITCH.value
|
||||
profile: str = RoleType.WITCH.value
|
||||
special_action_names: list[str] = ["Save", "Poison"]
|
||||
|
||||
async def _think(self):
|
||||
"""女巫涉及两个特殊技能，因此在此需要改写_think进行路由"""
|
||||
news = self.rc.news[0]
|
||||
assert news.cause_by == InstructSpeak # 消息为来自Moderator的指令时，才去做动作
|
||||
if not news.send_to:
|
||||
if not news.restricted_to:
|
||||
# 消息接收范围为全体角色的，做公开发言（发表投票观点也算发言）
|
||||
self.rc.todo = Speak()
|
||||
elif self.profile in news.send_to:
|
||||
elif self.profile in news.restricted_to:
|
||||
# FIXME: hard code to split, restricted为"Moderator"或"Moderator,角色profile"
|
||||
# Moderator加密发给自己的，意味着要执行角色的特殊动作
|
||||
# 这里用关键词进行动作的选择，需要Moderator侧的指令进行配合
|
||||
|
|
|
|||
|
|
@ -1,4 +1,6 @@
|
|||
from pydantic import BaseModel
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from metagpt.schema import Message
|
||||
|
||||
|
||||
class RoleExperience(BaseModel):
|
||||
|
|
@ -12,3 +14,8 @@ class RoleExperience(BaseModel):
|
|||
round_id: str = ""
|
||||
game_setup: str = ""
|
||||
version: str = ""
|
||||
|
||||
|
||||
class WwMessage(Message):
|
||||
# Werewolf Message
|
||||
restricted_to: set[str] = Field(default={}, validate_default=True)
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@ from typing import Any, Optional
|
|||
from metagpt.actions.add_requirement import UserRequirement
|
||||
from metagpt.context import Context
|
||||
from metagpt.environment.werewolf.werewolf_env import WerewolfEnv
|
||||
from metagpt.schema import Message
|
||||
from metagpt.ext.werewolf.schema import WwMessage
|
||||
from metagpt.team import Team
|
||||
|
||||
|
||||
|
|
@ -23,4 +23,4 @@ class WerewolfGame(Team):
|
|||
def run_project(self, idea):
|
||||
"""Run a project from user instruction."""
|
||||
self.idea = idea
|
||||
self.env.publish_message(Message(role="User", content=idea, cause_by=UserRequirement, send_to={"Moderator"}))
|
||||
self.env.publish_message(WwMessage(role="User", content=idea, cause_by=UserRequirement, restricted_to={"Moderator"}))
|
||||
|
|
|
|||
|
|
@ -2,7 +2,8 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# @Desc : the unittest of WerewolfExtEnv
|
||||
|
||||
from metagpt.environment.werewolf.werewolf_ext_env import RoleState, WerewolfExtEnv
|
||||
from metagpt.environment.werewolf.werewolf_ext_env import WerewolfExtEnv
|
||||
from metagpt.environment.werewolf.const import RoleState
|
||||
from metagpt.roles.role import Role
|
||||
|
||||
|
||||
|
|
@ -41,9 +42,9 @@ def test_werewolf_ext_env():
|
|||
assert "Werewolves, please open your eyes" in curr_instr["content"]
|
||||
|
||||
# current step_idx = 5
|
||||
ext_env.wolf_kill_someone(wolf=Role(name="Player10"), player_name="Player4")
|
||||
ext_env.wolf_kill_someone(wolf=Werewolf(name="Player0"), player_name="Player4")
|
||||
ext_env.wolf_kill_someone(wolf=Werewolf(name="Player1"), player_name="Player4")
|
||||
ext_env.wolf_kill_someone(wolf_name="Player10", player_name="Player4")
|
||||
ext_env.wolf_kill_someone(wolf_name="Player0", player_name="Player4")
|
||||
ext_env.wolf_kill_someone(wolf_name="Player1", player_name="Player4")
|
||||
assert ext_env.player_hunted == "Player4"
|
||||
assert len(ext_env.living_players) == 5 # hunted but can be saved by witch
|
||||
|
||||
|
|
@ -52,11 +53,11 @@ def test_werewolf_ext_env():
|
|||
|
||||
# current step_idx = 18
|
||||
assert ext_env.step_idx == 18
|
||||
ext_env.vote_kill_someone(voteer=Werewolf(name="Player0"), player_name="Player2")
|
||||
ext_env.vote_kill_someone(voteer=Werewolf(name="Player1"), player_name="Player3")
|
||||
ext_env.vote_kill_someone(voteer=Villager(name="Player2"), player_name="Player3")
|
||||
ext_env.vote_kill_someone(voteer=Witch(name="Player3"), player_name="Player4")
|
||||
ext_env.vote_kill_someone(voteer=Guard(name="Player4"), player_name="Player2")
|
||||
ext_env.vote_kill_someone(voter_name="Player0", player_name="Player2")
|
||||
ext_env.vote_kill_someone(voter_name="Player1", player_name="Player3")
|
||||
ext_env.vote_kill_someone(voter_name="Player2", player_name="Player3")
|
||||
ext_env.vote_kill_someone(voter_name="Player3", player_name="Player4")
|
||||
ext_env.vote_kill_someone(voter_name="Player4", player_name="Player2")
|
||||
assert ext_env.player_current_dead == "Player2"
|
||||
assert len(ext_env.living_players) == 4
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue