streamlit app version of game

{src/.streamlit → .streamlit}/config.toml

@@ -1,3 +1,6 @@
 [theme]
 base="dark"
 font="monospace"
+
+[runner]
+magicEnabled=false

src/agent_interfaces.py

@@ -26,64 +26,100 @@ class BaseAgentInterface:
         self.id = agent_id
         self.messages = []
 
+    @property
+    def is_ai(self):
+        return not self.is_human
+
     def add_message(self, message: Message):
         """Adds a message to the message history, without generating a response."""
         bound_message = AgentMessage.from_message(message, self.id, len(self.messages))
         save(bound_message)
         self.messages.append(bound_message)
 
+    # Respond To methods - These take a message as input and generate a response
+
     def respond_to(self, message: Message) -> Message:
-        """Adds a message to the message history, and generates a response message."""
+        """Take a message as input and return a response. Both the message and the response are added to history."""
         self.add_message(message)
-        response = Message(type="agent", content=self._generate_response())
-        self.add_message(response)
+        response = self.generate_response()
         return response
 
     def respond_to_formatted(
+            self, message: Message,
+            output_format: Type[OutputFormatModel],
+            additional_fields: dict = None,
+            **kwargs
+    ) -> OutputFormatModel:
+        """Responds to a message and logs the response."""
+        self.add_message(message)
+        output = self.generate_formatted_response(output_format, additional_fields, **kwargs)
+        return output
+
+    # Generate response methods - These do not take a message as input and only use the current message history
+
+    def generate_response(self) -> Message:
+        """Generates a response based on the current messages in the history."""
+        response = Message(type="agent", content=self._generate())
+        self.add_message(response)
+        return response
+
+    def generate_formatted_response(
             self,
-            message: Message,
             output_format: Type[OutputFormatModel],
+            additional_fields: dict = None,
             max_retries=3,
-            **kwargs
     ) -> OutputFormatModel:
-        """Adds a message to the message history, and generates a response matching the provided format."""
-        initial_response = self.respond_to(message)
+        """Generates a response matching the provided format."""
+        initial_response = self.generate_response()
 
         reformat_message = Message(type="format", content=output_format.get_format_instructions())
 
         output = None
         retries = 0
 
-        while not output and retries < max_retries:
+        while not output:
             try:
                 formatted_response = self.respond_to(reformat_message)
-                if kwargs:
-                    fields = json.loads(formatted_response.content)
-                    fields.update(kwargs)
-                    output = output_format.model_validate(fields)
-                else:
-                    output = output_format.model_validate_json(formatted_response.content)
-
-            except ValidationError or JSONDecodeError as e:
+
+                fields = json.loads(formatted_response.content)
+                if additional_fields:
+                    fields.update(additional_fields)
+
+                output = output_format.model_validate(fields)
+
+            except ValidationError as e:
+                # If the response doesn't match the format, we ask the agent to try again
                 if retries > max_retries:
                     raise e
+
                 retry_message = Message(type="retry", content=f"Error formatting response: {e} \n\n Please try again.")
                 reformat_message = retry_message
 
                 retries += 1
 
+            except JSONDecodeError as e:
+                # Occasionally models will output json as a code block, which will cause a JSONDecodeError
+                if retries > max_retries:
+                    raise e
+
+                retry_message = Message(type="retry",
+                                        content="There was an Error with your JSON format. Make sure you are not using code blocks."
+                                                "i.e. your response should be:\n{...}\n"
+                                                "Instead of:\n```json\n{...}\n```\n\n Please try again.")
+                reformat_message = retry_message
+
+                retries += 1
+
         return output
 
-    def _generate_response(self) -> str:
+    # How agents actually generate responses
+
+    def _generate(self) -> str:
         """Generates a response from the Agent."""
         # This is the BaseAgent class, and thus has no response logic
         # Subclasses should implement this method to generate a response using the message history
         raise NotImplementedError
 
-    @property
-    def is_ai(self):
-        return not self.is_human
-
 
 AgentInterface = NewType("AgentInterface", BaseAgentInterface)
 
@@ -96,7 +132,7 @@ class OpenAIAgentInterface(BaseAgentInterface):
         self.model_name = model_name
         self.client = OpenAI()
 
-    def _generate_response(self) -> str:
+    def _generate(self) -> str:
         """Generates a response using the message history"""
         open_ai_messages = [message.to_openai() for message in self.messages]
 
@@ -111,13 +147,18 @@ class OpenAIAgentInterface(BaseAgentInterface):
 class HumanAgentInterface(BaseAgentInterface):
     is_human = True
 
-    def respond_to_formatted(self, message: Message, output_format: Type[OutputFormatModel], **kwargs) -> OutputFormatModel:
+    def generate_formatted_response(
+            self,
+            output_format: Type[OutputFormatModel],
+            additional_fields: dict = None,
+            max_retries: int = 3
+    ) -> OutputFormatModel:
         """For Human agents, we can trust them enough to format their own responses... for now"""
-        response = super().respond_to(message)
+        response = self.generate_response()
         # only works because current outputs have only 1 field...
-        fields = {output_format.model_fields.copy().popitem()[0], response.content}
-        if kwargs:
-            fields.update(kwargs)
+        fields = {output_format.model_fields.copy().popitem()[0]: response.content}
+        if additional_fields:
+            fields.update(additional_fields)
         output = output_format.model_validate(fields)
 
         return output
@@ -139,7 +180,7 @@ class HumanAgentCLI(HumanAgentInterface):
             # Prevents the agent from seeing its own messages on the command line
             print(message.content)
 
-    def _generate_response(self) -> str:
+    def _generate(self) -> str:
         """Generates a response using the message history"""
         response = input()
         return response

src/app.py

@@ -1,63 +1,120 @@
-import asyncio
-from time import sleep
+from typing import Type
 
 import streamlit as st
-from langchain_core.messages import AIMessage
+from streamlit import session_state
 
 from game import Game
+from agent_interfaces import HumanAgentInterface
+from message import Message
+from prompts import fetch_prompt, format_prompt
 
 st.set_page_config(layout="wide", page_title="Chameleon", page_icon="img/logo.svg")
-human_turn = False
 
 
-def display_message(message):
-    if message["type"] == "game":
-        messages_container.markdown(f"{message['content']}")
-    elif message["type"] == "verbose":
-        messages_container.markdown(f":green[{message['content']}]")
-    elif message["type"] == "debug":
-        messages_container.markdown(f":orange[DEBUG: {message['content']}]")
+def display_message(message: Message):
+    if message.type == "verbose":
+        messages_container.markdown(f":green[{message.content}]")
+    elif message.type == "debug":
+        messages_container.markdown(f":orange[DEBUG: {message.content}]")
+    else:
+        messages_container.markdown(f"{message.content}")
 
 
-class StreamlitGame(Game):
-    @staticmethod
-    async def human_input(prompt: str) -> str:
-        _user_input = st.chat_input("Your message", key=f"user_input_{st.session_state.user_input_id}")
-        st.session_state.user_input_id += 1
-
-        while _user_input is None or _user_input == "":
-            sleep(0.1)
-
-        print(f"User input: {_user_input}")
+if "messages" not in session_state:
+    session_state.messages = []
+    session_state.awaiting_human_input = False
+    session_state.game_state = "game_start"
 
-        response = AIMessage(content=_user_input)
 
-        return response
-
-    def human_message(self, message: str):
-        message = {"type": "game", "content": message}
-        st.session_state["messages"].append(message)
+class StreamlitInterface(HumanAgentInterface):
+    def add_message(self, message: Message):
+        super().add_message(message)
+        session_state.messages.append(message)
         display_message(message)
 
-    def verbose_message(self, message: str):
-        if self.verbose:
-            message = {"type": "verbose", "content": message}
-            st.session_state["messages"].append(message)
-            display_message(message)
+    def _generate(self) -> str:
+        return session_state.user_input
 
-    def debug_message(self, message: str):
-        if self.debug:
-            message = {"type": "debug", "content": message}
-            st.session_state["messages"].append(message)
-            display_message(message)
 
-
-if "messages" not in st.session_state:
-    st.session_state["messages"] = []
-if "game_started" not in st.session_state:
-    st.session_state["game_started"] = False
-if "user_input_id" not in st.session_state:
-    st.session_state["user_input_id"] = 0
+class StreamlitGame(Game):
+    """A Streamlit version of the Game class that uses a state machine to manage the game state."""
+
+    def run_game(self):
+        """Starts the game."""
+        if session_state.game_state == "game_start":
+            self.game_message(fetch_prompt("game_rules"))
+            session_state.game_state = "setup_round"
+
+        if session_state.game_state == "setup_round":
+            self.setup_round()
+            session_state.game_state = "animal_description"
+        if session_state.game_state in ["animal_description", "chameleon_guess", "herd_vote"]:
+            self.run_round()
+        if session_state.game_state == "resolve_round":
+            self.resolve_round()
+            session_state.game_state = "setup_round"
+
+    def run_round(self):
+        """Starts the round."""
+
+        # Phase I: Collect Player Animal Descriptions
+        if session_state.game_state == "animal_description":
+            for current_player in self.players:
+                if current_player.id not in [animal_description['player_id'] for animal_description in self.round_animal_descriptions]:
+                    if current_player.interface.is_human:
+                        if not session_state.awaiting_human_input:
+                            self.game_message(fetch_prompt("player_describe_animal"), current_player)
+                            session_state.awaiting_human_input = True
+                            break
+                        else:
+                            self.player_turn_animal_description(current_player)
+                            session_state.awaiting_human_input = False
+                    else:
+                        self.game_message(fetch_prompt("player_describe_animal"), current_player)
+                        self.player_turn_animal_description(current_player)
+            if len(self.round_animal_descriptions) == len(self.players):
+                session_state.game_state = "chameleon_guess"
+                session_state.awaiting_human_input = False
+
+        # Phase II: Chameleon Guesses the Animal
+        if session_state.game_state == "chameleon_guess":
+            self.game_message("All players have spoken. The Chameleon will now guess the secret animal...")
+            player_responses = self.format_animal_descriptions(exclude=self.chameleon)
+            self.game_message(format_prompt("chameleon_guess_animal", player_responses=player_responses), self.chameleon)
+            if self.players[self.human_index].role == "chameleon":
+                if not session_state.awaiting_human_input:
+                    session_state.awaiting_human_input = True
+                else:
+                    self.player_turn_chameleon_guess(self.chameleon)
+                    session_state.awaiting_human_input = False
+            else:
+                self.player_turn_chameleon_guess(self.chameleon)
+                session_state.awaiting_human_input = False
+
+            session_state.game_state = "herd_vote"
+
+        # Phase III: The Herd Votes for who they think the Chameleon is
+        if session_state.game_state == "herd_vote":
+            for current_player in self.players:
+                if current_player.role == "herd" and current_player.id not in [vote['voter_id'] for vote in self.herd_vote_tally]:
+                    player_responses = self.format_animal_descriptions(exclude=current_player)
+                    if current_player.interface.is_human:
+                        if not session_state.awaiting_human_input:
+                            self.game_message(format_prompt("vote", player_responses=player_responses), current_player)
+                            session_state.awaiting_human_input = True
+                            break
+                        else:
+                            self.player_turn_herd_vote(current_player)
+                            session_state.awaiting_human_input = False
+                    else:
+                        self.game_message(format_prompt("vote", player_responses=player_responses), current_player)
+                        self.player_turn_herd_vote(current_player)
+
+            if len(self.herd_vote_tally) == len(self.players) - 1:
+                session_state.game_state = "resolve_round"
+
+
+# Streamlit App
 
 margin_size = 1
 center_size = 3
@@ -76,18 +133,16 @@ with center:
 
     messages_container.write("Enter your name to begin...")
 
+    user_input = st.chat_input("Your response:")
+
     if st.session_state.messages:
         for message in st.session_state.messages:
             display_message(message)
 
-    user_input = st.chat_input("Your message")
-    st.session_state.user_input_id += 1
-
-if not st.session_state.game_started and user_input:
-    st.session_state.game_started = True
+if user_input:
     if "game" not in st.session_state:
-        st.session_state.game = StreamlitGame(human_name=user_input, verbose=True)
-
-    asyncio.run(st.session_state.game.start())
+        st.session_state.game = StreamlitGame(human_name=user_input, verbose=True, human_interface=StreamlitInterface)
+    session_state.user_input = user_input
+    st.session_state.game.run_game()
 
 

src/game.py

@@ -19,18 +19,8 @@ WINNING_SCORE = 3
 class Game:
     """The main game class, handles the game logic and player interactions."""
 
-    number_of_players = NUMBER_OF_PLAYERS
-    """The number of players in the game."""
     winning_score = WINNING_SCORE
     """The Number of points required to win the game."""
-    chameleon_ids: List[str] = []
-    """Record of which player was the chameleon for each round."""
-    herd_animals: List[str] = []
-    """Record of what animal was the herd animal for each round."""
-    chameleon_guesses: List[str] = []
-    """Record of what animal the chameleon guessed for each round."""
-    herd_vote_tallies: List[List[dict]] = []
-    """Record of the votes of each herd member for the chameleon for each round."""
 
     def __init__(
             self,
@@ -40,9 +30,27 @@ class Game:
             verbose: bool = False,
             debug: bool = False
     ):
-        # Game ID
+        # Instance Variables
         self.game_id = game_id()
-        self.start_time = datetime.now().strftime('%y%m%d-%H%M%S')
+        """The unique id of the game."""
+        self.start_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+        """The time the game was started."""
+        self.verbose = verbose
+        """If True, the game will display verbose messages to the player."""
+        self.debug = debug
+        """If True, the game will display debug messages to the player."""
+        self.chameleon_ids: List[str] = []
+        """Record of which player was the chameleon for each round."""
+        self.herd_animals: List[str] = []
+        """Record of what animal was the herd animal for each round."""
+        self.all_animal_descriptions: List[List[dict]] = []
+        """Record of the animal descriptions each player has given for each round."""
+        self.chameleon_guesses: List[str] = []
+        """Record of what animal the chameleon guessed for each round."""
+        self.herd_vote_tallies: List[List[dict]] = []
+        """Record of the votes of each herd member for the chameleon for each round."""
+        self.winner_id: str | None = None
+        """The id of the player who has won the game."""
 
         # Gather Player Names
         if human_name:
@@ -67,14 +75,9 @@ class Game:
 
             self.players.append(Player(name, player_id, interface))
 
-        self.verbose = verbose
-        """If True, the game will display verbose messages to the player."""
-        self.debug = debug
-        """If True, the game will display debug messages to the player."""
-
         # Add Observer - an Agent who can see all the messages, but doesn't actually play
-        if (self.verbose or self.debug) and not self.human_index:
-            self.observer = HumanAgentCLI("{self.game_id}-observer")
+        if (self.verbose or self.debug) and self.human_index is None:
+            self.observer = human_interface(f"{self.game_id}-observer")
         else:
             self.observer = None
 
@@ -96,6 +99,11 @@ class Game:
         """Returns the current herd animal."""
         return self.herd_animals[-1]
 
+    @property
+    def round_animal_descriptions(self) -> List[dict]:
+        """Returns the current animal descriptions."""
+        return self.all_animal_descriptions[-1]
+
     @property
     def chameleon_guess(self) -> str:
         """Returns the current chameleon guess."""
@@ -106,6 +114,22 @@ class Game:
         """Returns the current herd vote tally."""
         return self.herd_vote_tallies[-1]
 
+    @property
+    def round_number(self) -> int:
+        """Returns the current round number."""
+        return len(self.herd_animals)
+
+    def format_animal_descriptions(self, exclude: Player = None) -> str:
+        """Formats the animal description responses of the players into a single string."""
+        formatted_responses = ""
+        for response in self.round_animal_descriptions:
+            # Used to exclude the player who is currently responding, so they don't vote for themselves like a fool
+            if response["player_id"] != exclude.id:
+                player = self.player_from_id(response["player_id"])
+                formatted_responses += f" - {player.name}: {response['description']}\n"
+
+        return formatted_responses
+
     def observer_message(self, message: Message):
         """Sends a message to the observer if there is one."""
         if self.observer:
@@ -144,22 +168,10 @@ class Game:
             self.observer_message(message)
 
 
-    async def start(self):
+    async def run_game(self):
         """Sets up the game. This includes assigning roles and gathering player names."""
         self.game_message(fetch_prompt("game_rules"))
 
-        winner = None
-        round_number = 0
-
-        # while not winner:
-        #     round_results = await self.run_round()
-        #     round_number += 1
-
-            # # Check for a Winner
-            # for player in self.players:
-            #     if player.points >= self.winning_score:
-            #         winner = player # ignoring the possibility of a tie for now
-
         self.setup_round()
 
         self.run_round()
@@ -194,29 +206,34 @@ class Game:
                 player.assign_role("herd")
                 self.game_message(format_prompt("assign_herd", herd_animal=herd_animal), player)
 
+        # Empty Animal Descriptions
+        self.all_animal_descriptions.append([])
+
+        # Empty Tally for Votes
+        self.herd_vote_tallies.append([])
+
+        self.game_message(f"Each player will now take turns describing themselves:")
+
     def run_round(self):
         """Starts the round."""
         # Phase I: Collect Player Animal Descriptions
-
-        self.game_message(f"Each player will now take turns describing themselves:")
-        for i, current_player in enumerate(self.players):
+        for current_player in self.players:
+            self.game_message(fetch_prompt("player_describe_animal"), current_player)
             self.player_turn_animal_description(current_player)
 
         # Phase II: Chameleon Guesses the Animal
-
+        self.game_message("All players have spoken. The Chameleon will now guess the secret animal...")
+        player_responses = self.format_animal_descriptions(exclude=self.chameleon)
+        self.game_message(format_prompt("chameleon_guess_animal", player_responses=player_responses), self.chameleon)
         self.player_turn_chameleon_guess(self.chameleon)
 
         # Phase III: The Herd Votes for who they think the Chameleon is
-        self.game_message("The Chameleon has guessed the animal. Now the Herd will vote on who they think the chameleon is.")
-
-        self.herd_vote_tallies.append([])
         for current_player in self.players:
             if current_player.role == "herd":
+                player_responses = self.format_animal_descriptions(exclude=current_player)
+                self.game_message(format_prompt("vote", player_responses=player_responses), current_player)
                 self.player_turn_herd_vote(current_player)
 
-
-
-
     def player_turn_animal_description(self, player: Player):
         """Handles a player's turn to describe themselves."""
         if player.interface.is_ai:
@@ -225,21 +242,21 @@ class Game:
         prompt = fetch_prompt("player_describe_animal")
 
         # Get Player Animal Description
-        message = Message(type="prompt", content=prompt)
-        response = player.interface.respond_to_formatted(message, AnimalDescriptionFormat)
+        response = player.interface.generate_formatted_response(AnimalDescriptionFormat)
+
+        self.round_animal_descriptions.append({"player_id": player.id, "description": response.description})
 
         self.game_message(f"{player.name}: {response.description}", player, exclude=True)
 
     def player_turn_chameleon_guess(self, chameleon: Player):
         """Handles the Chameleon's turn to guess the secret animal."""
-        self.game_message("All players have spoken. The Chameleon will now guess the secret animal...")
+
         if chameleon.interface.is_ai or self.observer:
             self.verbose_message("The Chameleon is thinking...")
 
-        prompt = fetch_prompt("chameleon_guess_animal")
+        response = chameleon.interface.generate_formatted_response(ChameleonGuessFormat)
 
-        message = Message(type="prompt", content=prompt)
-        response = chameleon.interface.respond_to_formatted(message, ChameleonGuessFormat)
+        self.game_message("The Chameleon has guessed the animal. Now the Herd will vote on who they think the chameleon is.")
 
         self.chameleon_guesses.append(response.animal)
 
@@ -248,12 +265,9 @@ class Game:
         if player.interface.is_ai:
             self.verbose_message(f"{player.name} is thinking...")
 
-        prompt = fetch_prompt("vote")
-
         # Get Player Vote
-        message = Message(type="prompt", content=prompt)
-        other_player_names = [p.name for p in self.players if p != player]
-        response = player.interface.respond_to_formatted(message, HerdVoteFormat, player_names=other_player_names)
+        additional_fields = {"player_names": [p.name for p in self.players if p != player]}
+        response = player.interface.generate_formatted_response(HerdVoteFormat, additional_fields=additional_fields)
 
         if player.interface.is_ai:
             self.debug_message(f"{player.name} voted for {response.vote}")
@@ -261,14 +275,14 @@ class Game:
         voted_for_player = self.player_from_name(response.vote)
 
         # Add Vote to Player Votes
-        self.herd_vote_tally.append({"voter": player.id, "voted_for": voted_for_player.id})
+        self.herd_vote_tally.append({"voter_id": player.id, "voted_for_id": voted_for_player.id})
 
     def resolve_round(self):
         """Resolves the round, assigns points, and prints the results."""
         self.game_message("All players have voted!")
         for vote in self.herd_vote_tally:
-            voter = self.player_from_id(vote["voter"])
-            voted_for = self.player_from_id(vote["voted_for"])
+            voter = self.player_from_id(vote["voter_id"])
+            voted_for = self.player_from_id(vote["voted_for_id"])
             self.game_message(f"{voter.name} voted for {voted_for.name}")
 
         accused_player_id = count_chameleon_votes(self.herd_vote_tally)
@@ -296,8 +310,8 @@ class Game:
                     player.points += 1
         # If a Herd player votes for the Chameleon       =   +1 Point to that player
         for vote in self.herd_vote_tally:
-            if vote["voted_for"] == self.chameleon.id:
-                self.player_from_id(vote['voter']).points += 1
+            if vote["voted_for_id"] == self.chameleon.id:
+                self.player_from_id(vote['voter_id']).points += 1
 
         # If the Herd fails to accuse the Chameleon      =   +1 Point to the Chameleon
         if not accused_player_id or accused_player_id != self.chameleon.id:

src/game_utils.py

@@ -37,7 +37,7 @@ def random_index(number_of_players : int) -> int:
 
 def count_chameleon_votes(player_votes: list[dict]) -> str | None:
     """Counts the votes for each player."""
-    votes = [vote['voted_for'] for vote in player_votes]
+    votes = [vote['voted_for_id'] for vote in player_votes]
 
     freq = Counter(votes)
     most_voted_player, number_of_votes = freq.most_common()[0]

src/main.py

@@ -12,7 +12,7 @@ def main():
     else:
         game = Game(verbose=True)
 
-    asyncio.run(game.start())
+    asyncio.run(game.run_game())
 
 
 if __name__ == "__main__":

src/output_formats.py

@@ -28,12 +28,12 @@ class AnimalDescriptionFormat(OutputFormatModel):
     description: str = Field(description="A brief description of the animal")
     """A brief description of the animal"""
 
-    @field_validator('description')
-    @classmethod
-    def check_starting_character(cls, v) -> str:
-        if not v[0].upper() == 'I':
-            raise ValueError("Description must begin with 'I'")
-        return v
+    # @field_validator('description')
+    # @classmethod
+    # def check_starting_character(cls, v) -> str:
+    #     if not v[0].upper() == 'I':
+    #         raise ValueError("Description must begin with 'I'")
+    #     return v
 
 
 class ChameleonGuessFormat(OutputFormatModel):

src/prompts.py

@@ -34,10 +34,16 @@ Your Response:"""
 
 _chameleon_guess_animal = """\
 What animal do you think the other players are pretending to be?
+Player Responses:
+{player_responses}
+Your Guess:
 """
 
 _vote_prompt = """\
 Now it is time to vote. Choose from the other players who you think the Chameleon is.
+Player Responses:
+{player_responses}
+Your Vote:
 """
 
 prompts = {