diff --git a/.flake8 b/.flake8 index 1931c9c..867ac11 100644 --- a/.flake8 +++ b/.flake8 @@ -1,4 +1,4 @@ [flake8] -ignore = W293 +ignore = W293,W291 max-line-length = 120 max-complexity = 12 diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 8441e04..d0c8ff3 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -4,14 +4,12 @@ repos: - id: isort args: ["--profile", "black"] name: isort - exclude: scripts entry: poetry run isort language: system types: [python] - id: black name: black - exclude: scripts entry: poetry run black language: system types: [python] @@ -19,6 +17,5 @@ repos: - id: flake8 name: flake8 entry: poetry run flake8 - exclude: scripts language: system types: [python] diff --git a/README.md b/README.md index e995883..bf70ab9 100644 --- a/README.md +++ b/README.md @@ -44,7 +44,7 @@ git clone https://github.com/dirkbrnd/Resistance-Coup-Autogen.git poetry install ``` -3. Launch! +3. Launch and watch the AI agents play the game! ```sh python coup.py diff --git a/coup.py b/coup.py index 2d48644..18d44e3 100755 --- a/coup.py +++ b/coup.py @@ -1,34 +1,53 @@ -import random import sys -from autogen import GroupChat, GroupChatManager, UserProxyAgent, config_list_from_dotenv, AssistantAgent -from autogen.agentchat.contrib.gpt_assistant_agent import GPTAssistantAgent +from autogen import ( + AssistantAgent, + GroupChat, + GroupChatManager, + UserProxyAgent, + config_list_from_dotenv, +) -from src.ai.agents import create_player_agent, create_game_master_agent, create_user_proxy +from src.ai.agents import ( + create_game_master_agent, + create_player_agent, + create_user_proxy, +) from src.handler.game_handler import ResistanceCoupGameHandler -# SEED = 42 config_list = config_list_from_dotenv( - dotenv_file_path='.env', - filter_dict={ - "model": { - "gpt-4", - # "gpt-3.5-turbo", - } - } - ) + dotenv_file_path=".env", + filter_dict={ + "model": { + "gpt-4", + # "gpt-3.5-turbo", + } + }, +) def main(): - # Create game handler with 5 players - handler = ResistanceCoupGameHandler(5) + # Create game handler with 3 players + handler = ResistanceCoupGameHandler(3) print(f"First player is {handler.current_player}") # Create AI players agent_players = [] for ind, player in enumerate(handler.players): - agent_players.append(create_player_agent(name=player.name, other_player_names=[other_player.name for other_player in handler.players if other_player.name != player.name], - cards=player.cards, strategy=player.strategy, handler=handler, config_list=config_list)) + agent_players.append( + create_player_agent( + name=player.name, + other_player_names=[ + other_player.name + for other_player in handler.players + if other_player.name != player.name + ], + cards=player.cards, + strategy=player.strategy, + handler=handler, + config_list=config_list, + ) + ) # Game master game_master: AssistantAgent = create_game_master_agent(handler, config_list) @@ -37,7 +56,12 @@ def main(): user_proxy: UserProxyAgent = create_user_proxy(config_list) # Define group chat - group_chat = GroupChat(agents=[user_proxy, game_master, *agent_players], messages=[], admin_name=game_master.name, max_round=100) + group_chat = GroupChat( + agents=[user_proxy, game_master, *agent_players], + messages=[], + admin_name=game_master.name, + max_round=1000, + ) manager = GroupChatManager(groupchat=group_chat, llm_config={"config_list": config_list}) task = """ @@ -45,7 +69,6 @@ def main(): """ game_master.initiate_chat(manager, message=task) - # handler.perform_action(ActionType.income) print("GAME OVER") diff --git a/src/ai/agents.py b/src/ai/agents.py index 6c91d16..c6bd8af 100644 --- a/src/ai/agents.py +++ b/src/ai/agents.py @@ -1,18 +1,15 @@ -from autogen import UserProxyAgent, AssistantAgent -from autogen.agentchat.contrib.gpt_assistant_agent import GPTAssistantAgent +from autogen import AssistantAgent, UserProxyAgent from src.handler.game_handler import ResistanceCoupGameHandler from src.models.action import ActionType from src.models.card import Card +from src.models.player import PlayerStrategy -# SEED = 42 - def create_user_proxy(config_list: list) -> UserProxyAgent: llm_config = { "config_list": config_list, - # "seed": SEED, "temperature": 0, } @@ -21,7 +18,8 @@ def create_user_proxy(config_list: list) -> UserProxyAgent: llm_config=llm_config, is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"), system_message=""" - You are facilitating a game of The Resistance: Coup between five players. Respond with TERMINATE once the game has a winner. + You are facilitating a game of The Resistance: Coup between five players. + Respond with TERMINATE once the game has a winner. At the start of the game, you will inform the starting player that it is their turn. In between each player's turn you have to retrieve the game state and provide it to the players. """, @@ -33,10 +31,11 @@ def create_user_proxy(config_list: list) -> UserProxyAgent: return user_proxy -def create_game_master_agent(handler: ResistanceCoupGameHandler, config_list: list) -> AssistantAgent: +def create_game_master_agent( + handler: ResistanceCoupGameHandler, config_list: list +) -> AssistantAgent: llm_config = { "config_list": config_list, - # "seed": SEED, "temperature": 1, "functions": [ { @@ -44,23 +43,27 @@ def create_game_master_agent(handler: ResistanceCoupGameHandler, config_list: li "description": "Get the current state of the game", "parameters": { "type": "object", - "properties": { - }, - } + "properties": {}, + }, }, - ] + ], } instructions = f""" - You are the game master in a game of The Resistance: Coup between {handler.number_of_players} players. + You are the game master in a game of The Resistance: Coup between {len(handler.players)} players. - At the start of the game, you will inform the starting player that it is their turn and announce to everyone the starting game state. + At the start of the game, you will inform the starting player that it is their turn and announce to + everyone the starting game state. In between each player's turn you have to retrieve the game state and provide it to the current player. Make sure the correct player is taking their next turn based on the game state. + Players can counter another player's action if action_can_be_countered is "True" after they + tried to perform an action. + Once there is only one active player left in the game, you can declare the game is over and we have a winner. + Don't start another game after it has ended. Don't offer to the other players to play another game. """ game_master = AssistantAgent( @@ -71,17 +74,22 @@ def create_game_master_agent(handler: ResistanceCoupGameHandler, config_list: li "get_game_state": handler.get_game_state, }, # max_consecutive_auto_reply=100, - description="The game master in a game of The Resistance Coup" + description="The game master in a game of The Resistance Coup", ) return game_master -def create_player_agent(name: str, other_player_names: list[str], cards: list[Card], strategy: str, - handler: ResistanceCoupGameHandler, config_list: list) -> AssistantAgent: +def create_player_agent( + name: str, + other_player_names: list[str], + cards: list[Card], + strategy: PlayerStrategy, + handler: ResistanceCoupGameHandler, + config_list: list, +) -> AssistantAgent: llm_config = { "config_list": config_list, - # "seed": SEED, - "temperature": 0.3, + "temperature": 0.5, "functions": [ { "name": "perform_action", @@ -92,46 +100,115 @@ def create_player_agent(name: str, other_player_names: list[str], cards: list[Ca "player_name": { "type": "string", "description": "Send your own name.", - "enum": [name] + "enum": [name], }, "action_name": { "type": "string", "description": "The name of the action to perform.", - "enum": [ActionType.income, ActionType.foreign_aid, ActionType.tax, ActionType.coup, - ActionType.steal, ActionType.assassinate, ActionType.exchange] + "enum": [ + ActionType.income, + ActionType.foreign_aid, + ActionType.tax, + ActionType.coup, + ActionType.steal, + ActionType.assassinate, + ActionType.exchange, + ], }, "target_player_name": { "type": "string", "description": "The player name to target.", }, }, - "required": [ - "player_name", - "action_name" - ] - } - } - ] + "required": ["player_name", "action_name"], + }, + }, + { + "name": "counter_action", + "description": "Counter the previous action that was performed", + "parameters": { + "type": "object", + "properties": { + "countering_player_name": { + "type": "string", + "description": "Send your own name.", + "enum": [name], + }, + }, + "required": ["countering_player_name"], + }, + }, + { + "name": "execute_action", + "description": "Execute the action that was performed and complete the turn.", + "parameters": { + "type": "object", + "properties": { + "player_name": { + "type": "string", + "description": "Send your own name.", + "enum": [name], + }, + "action_name": { + "type": "string", + "description": "The name of the action to perform.", + "enum": [ + ActionType.income, + ActionType.foreign_aid, + ActionType.tax, + ActionType.coup, + ActionType.steal, + ActionType.assassinate, + ActionType.exchange, + ], + }, + "target_player_name": { + "type": "string", + "description": "The player name to target.", + }, + }, + "required": ["player_name", "action_name"], + }, + }, + ], } + if strategy == PlayerStrategy.aggressive: + strategy_str = ("Your strategy is to play aggressive. Try to assassinate, coup, or steal as soon as you can. " + "Feel free to bluff, but be careful because it can be challenged. If you keep getting blocked," + "rather get income on your next turn, before playing aggressive again.") + elif strategy == PlayerStrategy.conservative: + strategy_str = ("Your strategy is to play conservative. " + "Build up your money, avoid bluffing, and wait for the opportunity to perform a coup.") + else: + strategy_str = ("Your strategy is to perform a coup as soon as you can and gather money as " + "quickly as possible.") + instructions = f"""Your name is {name} and you are a player in the game The Resistance: Coup. You are playing against {", ".join(other_player_names)}. You start with a {str(cards[0])} card and a {str(cards[1])} card, as well as 2 coins. - On your turn you have to pick a valid action based on your current available cards and coins. Also provide your own name to the function. + On your turn you have to pick a valid action based on your current available cards and coins. + Also provide your own name to the function. Never announce what cards you have, they are secret. - If your action was invalid, you have to pick another action. However feel free to bluff and perform an action even if you don't have the card. + If your action was invalid, you have to pick another action. However feel free to bluff and perform an + action even if you don't have the card, but be careful because it could be challenged. - The possible actions are {[ActionType.income, ActionType.foreign_aid, ActionType.tax, ActionType.coup, ActionType.steal, ActionType.assassinate, ActionType.exchange]} + You can counter another player's action if action_can_be_countered is "True", + after they tried to perform their action. + If no one counters your action, you have the call "execute_action" to complete the turn. + If after perform_action you find that turn_complete is "True", you don't have to execute your action. + You also chit-chat with your opponent when you communicate an action to light up the mood. - You should ensure both you and your opponents are making valid actions. Also that everyone is only taking actions when it is their turn. + You should ensure both you and your opponents are making valid actions. + Also that everyone is only taking actions when it is their turn. - Your strategy should be to play {strategy}. + {strategy_str} Don't hoard up coins, but rather try the assassinate or coup actions when you have a chance. @@ -145,9 +222,11 @@ def create_player_agent(name: str, other_player_names: list[str], cards: list[Ca llm_config=llm_config, function_map={ "perform_action": handler.perform_action, + "counter_action": handler.counter_action, + "execute_action": handler.execute_action, }, max_consecutive_auto_reply=100, - description=f"The player named {name} the game of The Resistance Coup" + description=f"The player named {name} the game of The Resistance Coup", ) return player diff --git a/src/handler/game_handler.py b/src/handler/game_handler.py index fe06910..f539a3d 100644 --- a/src/handler/game_handler.py +++ b/src/handler/game_handler.py @@ -2,10 +2,20 @@ from enum import Enum from typing import List, Optional -from src.models.action import Action, ActionType, TaxAction, ForeignAidAction, StealAction, AssassinateAction, \ - ExchangeAction, IncomeAction, CoupAction +from src.models.action import ( + Action, + ActionType, + AssassinateAction, + CoupAction, + ExchangeAction, + ForeignAidAction, + IncomeAction, + StealAction, + TaxAction, + get_counter_action, +) from src.models.card import Card, CardType -from src.models.player import Player +from src.models.player import Player, PlayerStrategy class ChallengeResult(Enum): @@ -53,26 +63,26 @@ def _create_card(card_type: CardType): class ResistanceCoupGameHandler: _players: dict[str, Player] = {} _player_names: list[str] = [] - _current_player_index = 0 + _deck: List[Card] = [] - _number_of_players: int = 0 _treasury: int = 0 + # Turn state + _current_player_index = 0 + _current_action: Optional[Action] + _current_action_is_countered: bool = False + _current_action_target_player_name: Optional[str] + def __init__(self, number_of_players: int): - self._number_of_players = number_of_players for i in range(number_of_players): player_name = f"Player_{str(i + 1)}" - strategy = random.choice(["conservative", "aggressive"]) + strategy = random.choice([PlayerStrategy.conservative, PlayerStrategy.aggressive, PlayerStrategy.coup_freak]) self._players[player_name] = Player(name=player_name, strategy=strategy) self._player_names.append(player_name) self.initialize_game() - @property - def number_of_players(self): - return self._number_of_players - @property def current_player(self) -> Player: return self._players[self._player_names[self._current_player_index]] @@ -85,21 +95,29 @@ def get_game_state(self) -> dict: players_str = "" for player_name, player in self._players.items(): if player.is_active: - players_str += f" - {player_name} with {len(player.cards)} cards and {player.coins} coins\n" + players_str += ( + f" - {player_name} with {len(player.cards)} cards and {player.coins} coins\n" + ) return { - "active_players": [{"name": player_name, - "coins": player.coins, - "cards": len(player.cards)} for player_name, player in self._players.items() if player.is_active], + "active_players": [ + {"name": player_name, "coins": player.coins, "cards": len(player.cards)} + for player_name, player in self._players.items() + if player.is_active + ], "treasury_coin": self._treasury, - "next_player": self.current_player.name + "next_player": self.current_player.name, } def get_game_state_str(self) -> str: players_str = "" for player_name, player in self._players.items(): if player.is_active: - players_str += f" - {player_name} {len(player.cards)} cards | {player.coins} coins\n" + players_str += ( + f" - {player_name} [{player.strategy.value}] " + f"{len(player.cards)} cards | " + f"{player.coins} coins\n" + ) return f""" The remaining players are: @@ -107,9 +125,6 @@ def get_game_state_str(self) -> str: The number of coins in the treasury: {self._treasury} """ - def _shuffle_deck(self) -> None: - random.shuffle(self._deck) - def initialize_game(self) -> None: self._deck = build_deck() self._shuffle_deck() @@ -130,7 +145,10 @@ def initialize_game(self) -> None: player.is_active = True # Random starting player - self._current_player_index = random.randint(0, self._number_of_players - 1) + self._current_player_index = random.randint(0, len(self._players) - 1) + + def _shuffle_deck(self) -> None: + random.shuffle(self._deck) def _swap_card(self, player: Player, card: Card) -> None: self._deck.append(card) @@ -167,38 +185,33 @@ def _deactivate_player(self) -> Optional[Player]: def _determine_win_state(self) -> bool: return sum(player.is_active for player in self._players.values()) == 1 -<<<<<<< Updated upstream - def validate_action(self, action: Action, current_player: Player, target_player: Optional[Player]) -> bool: - if action.action_type in [ActionType.coup, ActionType.steal, ActionType.assassinate] and not target_player: - return False - - # Can't take coin if the treasury has none -======= def _validate_action( self, action: Action, current_player: Player, target_player: Optional[Player] ): if current_player.coins >= 10 and action.action_type != ActionType.coup: - raise Exception(f"Invalid action: You have more than 10 coins and have to perform {ActionType.coup} action.") + raise Exception(f"Invalid action: You have more than 10 coins and have to perform " + f"{ActionType.coup.value} action.") if ( action.action_type in [ActionType.coup, ActionType.steal, ActionType.assassinate] and not target_player ): - raise Exception(f"Invalid action: You need a `target_player` for the action {action.action_type}") + raise Exception(f"Invalid action: You need a `target_player` for the action {action.action_type.value}") # Can't take coin if the treasury has none if ( action.action_type in [ActionType.income, ActionType.foreign_aid, ActionType.tax] and self._treasury == 0 ): raise Exception(f"Invalid action: The treasury has no coin to give") ->>>>>>> Stashed changes # You can only do a coup if you have at least 7 coins. if action.action_type == ActionType.coup and current_player.coins < 7: - raise Exception(f"Invalid action: You need more coins to be able to perform the {ActionType.coup} action.") + raise Exception(f"Invalid action: You need more coins to be able to perform the " + f"{ActionType.coup.value} action.") # You can only do an assassination if you have at least 3 coins. if action.action_type == ActionType.assassinate and current_player.coins < 3: - raise Exception(f"Invalid action: You need more coins to be able to perform the {ActionType.assassinate} action.") + raise Exception(f"Invalid action: You need more coins to be able to perform the " + f"{ActionType.assassinate.value} action.") # Can't steal from player with 0 coins if action.action_type == ActionType.steal and target_player.coins == 0: @@ -206,37 +219,79 @@ def _validate_action( return True - def perform_action(self, player_name: str, action_name: ActionType, target_player_name: Optional[str] = "", - countered: bool = False) -> dict: + def perform_action( + self, player_name: str, action_name: ActionType, target_player_name: Optional[str] = "" + ) -> dict: + if self._determine_win_state(): + raise Exception(f"You can't play anymore, the game has already ended. {self.current_player} won already.") + + # Reset current action + self._current_action = None + self._current_action_target_player_name = None + self._current_action_is_countered = False action = ACTIONS_MAP[action_name] target_player = None if target_player_name: target_player = self._players[target_player_name] + if not self._players[player_name].is_active: + raise Exception(f"You have been defeated and can't play anymore! " + f"It is currently {self.current_player.name}'s turn.") + if player_name != self.current_player.name: raise Exception(f"Wrong player, it is currently {self.current_player.name}'s turn.") -<<<<<<< Updated upstream - if not self.validate_action(action, self.current_player, target_player): - raise Exception("Invalid action") -======= self._validate_action(action, self.current_player, target_player) ->>>>>>> Stashed changes + # Keep track of the currently played action + self._current_action = action + self._current_action_target_player_name = target_player_name + + if action.can_be_countered: + return {"turn_complete": False, "action_can_be_countered": True, "game_over": False} + else: + return self.execute_action(self.current_player.name, action.action_type, target_player_name) + + def counter_action(self, countering_player_name: str): + countering_player = self._players[countering_player_name] + + self._current_action_is_countered = True + + print(f"{countering_player} is countering the previous action: {self._current_action.action_type.value}") + + return self.execute_action( + player_name=self.current_player.name, + action_name=self._current_action.action_type, + target_player_name=self._current_action_target_player_name, + ) + + def execute_action( + self, player_name: str, action_name: ActionType, target_player_name: Optional[str] = "" + ) -> dict: result_action_str = "" + action = ACTIONS_MAP[action_name] + target_player = None + if target_player_name: + target_player = self._players[target_player_name] + + if player_name != self.current_player.name: + raise Exception(f"Wrong player, it is currently {self.current_player.name}'s turn.") + match action.action_type: case ActionType.income: # Player gets 1 coin self.current_player.coins += self._take_coin_from_treasury(1) result_action_str = f"{self.current_player}'s coins are increased by 1" case ActionType.foreign_aid: - if not countered: + if not self._current_action_is_countered: # Player gets 2 coin taken_coin = self._take_coin_from_treasury(2) self.current_player.coins += taken_coin - result_action_str = f"{self.current_player}'s coins are increased by {taken_coin}" + result_action_str = ( + f"{self.current_player}'s coins are increased by {taken_coin}" + ) case ActionType.coup: # Player pays 7 coin self.current_player.coins -= self._give_coin_to_treasury(7) @@ -253,16 +308,18 @@ def perform_action(self, player_name: str, action_name: ActionType, target_playe case ActionType.assassinate: # Player pays 3 coin self.current_player.coins -= self._give_coin_to_treasury(3) - if not countered and target_player.cards: + if not self._current_action_is_countered and target_player.cards: result_action_str = f"{self.current_player} assassinates {target_player}" target_player.remove_card() case ActionType.steal: - if not countered: + if not self._current_action_is_countered: # Take 2 (or all) coins from a player steal_amount = min(target_player.coins, 2) target_player.coins -= steal_amount self.current_player.coins += steal_amount - result_action_str = f"{self.current_player} steals {steal_amount} coins from {target_player}" + result_action_str = ( + f"{self.current_player} steals {steal_amount} coins from {target_player}" + ) case ActionType.exchange: # Get 2 random cards from deck @@ -272,29 +329,30 @@ def perform_action(self, player_name: str, action_name: ActionType, target_playe self.current_player.cards += cards random.shuffle(self.current_player.cards) - first_card, second_card = self.current_player.cards.pop(), self.current_player.cards.pop() + first_card, second_card = ( + self.current_player.cards.pop(), + self.current_player.cards.pop(), + ) self._deck.append(first_card) self._deck.append(second_card) + print(result_action_str + "\n" + self.get_game_state_str()) + # Is any player out of the game? while player := self._deactivate_player(): - result_action_str += f"\n{player} was defeated! They can no longer play" + print(f"{player} was defeated! They can no longer play") # Have we reached a winner? if self._determine_win_state(): - print(f"The game is over! {self.current_player} has won!") - return { - "success": True, - "game_over": True - } + print("\n" + f"The game is over! {self.current_player} has won!") + return {"turn_complete": True, "game_over": True} # Next player self._next_player() - print(result_action_str + "\n" + self.get_game_state_str()) - return { - "success": True, + "turn_complete": True, + "action_can_be_countered": False, "next_player": self.current_player.name, - "game_over": False + "game_over": False, } diff --git a/src/models/action.py b/src/models/action.py index e0bc9e7..0382ca1 100644 --- a/src/models/action.py +++ b/src/models/action.py @@ -104,4 +104,3 @@ def get_counter_action(action_type: ActionType) -> CounterAction: ActionType.steal: BlockStealCounterAction(), ActionType.assassinate: BlockAssassinationCounterAction(), }[action_type] - diff --git a/src/models/player.py b/src/models/player.py index 6b4f18b..ce4f953 100644 --- a/src/models/player.py +++ b/src/models/player.py @@ -1,5 +1,6 @@ import random from abc import ABC +from enum import Enum from typing import List, Optional from pydantic import BaseModel @@ -7,11 +8,16 @@ from src.models.card import Card, CardType +class PlayerStrategy(str, Enum): + aggressive = "aggressive" + conservative = "conservative" + coup_freak = "coup_freak" + class Player(BaseModel, ABC): name: str coins: int = 0 cards: List[Card] = [] - strategy: str + strategy: PlayerStrategy is_active: bool = False def __str__(self):