added output parsing for LLMs

src/game.py

@@ -1,6 +1,11 @@
-from game_utils import fetch_prompt, random_animal, random_names, random_index
+import asyncio
+
+from game_utils import *
+from models import *
 from player import Player
 
+from parser import ParserKani
+
 # Default Values
 NUMBER_OF_PLAYERS = 5
 
@@ -37,18 +42,11 @@ class Game:
 
             self.players.append(Player(name, controller, role))
 
-
+        # Game State
         self.player_responses = []
 
-        print("Game Created")
-
-
-    def broadcast(self, message):
-        """Sends a message to all the players, no response required."""
-        for player_index in range(0, len(self.players)):
-            self.players[player_index].add_message(message)
-            if self.human_index == player_index:
-                print(message)
+        # Parser
+        self.parser = ParserKani.default()
 
     def format_responses(self) -> str:
         """Formats the responses of the players into a single string."""
@@ -58,34 +56,48 @@ class Game:
         """Returns the names of the players."""
         return [player.name for player in self.players]
 
-    def start(self):
+    @staticmethod
+    def player_action(prompt: str, player, validator: callable = None):
+        """Prompts the player to take an action and validates the response."""
+        max_attempts = 3
+        response = player.respond_to(prompt)
+
+        if validator:
+            attempts = 0
+            while not validator(response):
+                attempts += 1
+                if attempts >= max_attempts:
+                    raise ValueError(f"Player {player.name} did not provide a valid response to the following prompt:\n{prompt} Response: {response}")
+                response = player.respond_to(prompt)
+
+        return response
+
+    async def start(self):
         """Starts the game."""
-        print("Welcome to Chameleon! This is a social deduction game powered by LLMs.")
+        # print("Welcome to Chameleon! This is a social deduction game powered by LLMs.")
 
         self.player_responses = []
         herd_animal = random_animal()
 
         # Collect Player Animal Descriptions
         for player in self.players:
-            match player.role:
-                case "herd":
-                    prompt_template = fetch_prompt("herd_animal")
-                    prompt = prompt_template.format(animal=herd_animal, player_responses=self.format_responses())
-
-                case "chameleon":
-                    prompt_template = fetch_prompt("chameleon_animal")
-                    prompt = prompt_template.format(player_responses=self.format_responses())
+            if player.role == "chameleon":
+                prompt_template = fetch_prompt("chameleon_animal")
+                prompt = prompt_template.format(player_responses=self.format_responses())
+            else:
+                prompt_template = fetch_prompt("herd_animal")
+                prompt = prompt_template.format(animal=herd_animal, player_responses=self.format_responses())
 
-            response = player.collect_input(prompt)
-            self.player_responses.append({"sender": player.name, "response": response})
+            # Get Player Animal Description
+            response = await player.respond_to(prompt)
+            # Parse Animal Description
+            output = await self.parser.parse(prompt, response, AnimalDescriptionModel)
 
-        self.player_votes = []
+            self.player_responses.append({"sender": player.name, "response": output.description})
 
         # Show All Player Responses
-        self.broadcast(self.format_responses())
 
         # Chameleon Decides if they want to guess the animal
-        # TODO: Add Chameleon Guess Decision Logic
         chameleon_will_guess = False
 
         if chameleon_will_guess:
@@ -94,30 +106,25 @@ class Game:
             pass
         else:
             # All Players Vote for Chameleon
+            player_votes = []
             for player in self.players:
                 prompt_template = fetch_prompt("vote")
-                prompt = prompt_template.format(players=self.get_player_names())
+                prompt = prompt_template.format(player_responses=self.format_responses())
+
+                # Get Player Vote
+                response = await player.respond_to(prompt)
+                # Parse Vote
+                output = await self.parser.parse(prompt, response, VoteModel)
 
-                response = player.collect_input(prompt)
-                self.player_responses.append(response)
+                # Add Vote to Player Votes
+                player_votes.append(output.vote)
+
+            print(player_votes)
 
         # Assign Points
         # Chameleon Wins - 3 Points
         # Herd Wins by Failed Chameleon Guess - 1 Point (each)
         # Herd Wins by Correctly Guessing Chameleon - 2 points (each)
 
-    @staticmethod
-    def validate_animal_description(self, description: str) -> bool:
-        """Validates that the description starts with I and is less than 10 words."""
-        if not description.startswith("I"):
-            return False
-
-        if len(description.split(" ")) > 10:
-            return False
 
-        return True
 
-    def validate_vote(self, vote: str) -> bool:
-        """Validates that the vote is for a valid player."""
-        player_names = [player.name.lower() for player in self.players]
-        return vote.lower() in player_names

src/main.py

@@ -1,4 +1,5 @@
 from game import Game
+import asyncio
 from player import Player
 
 def main():
@@ -7,7 +8,7 @@ def main():
 
     game = Game(human_name=name)
 
-    game.start()
+    asyncio.run(game.start())
 
 
 if __name__ == "__main__":

src/models.py

@@ -0,0 +1,43 @@
+from pydantic import BaseModel, field_validator
+
+MAX_DESCRIPTION_LEN = 10
+
+
+class AnimalDescriptionModel(BaseModel):
+    # Define fields of our class here
+    description: str
+
+    # @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 wordcount(cls, v) -> str:
+    #     count = len(v.split())
+    #     if count > MAX_DESCRIPTION_LEN:
+    #         raise ValueError(f"Animal Description must be {MAX_DESCRIPTION_LEN} words or less")
+    #     return v
+
+
+class ChameleonDecisionModel(BaseModel):
+    will_guess: bool
+
+
+class AnimalGuessModel(BaseModel):
+    animal_name: str
+
+
+class VoteModel(BaseModel):
+    vote: str
+
+    # @field_validator('vote')
+    # @classmethod
+    # def check_player_exists(cls, v) -> str:
+    #     if v.lower() not in [player.lower() for player in players]:
+    #         raise ValueError(f"Player {v} does not exist")
+    #     return v
+

src/parser.py

@@ -0,0 +1,95 @@
+from typing import Type
+import asyncio
+import json
+
+from kani import Kani
+from kani.engines.openai import OpenAIEngine
+from pydantic import BaseModel, ValidationError
+
+FORMAT_INSTRUCTIONS = """The output should be reformatted as a JSON instance that conforms to the JSON schema below.
+Here is the output schema:
+```
+{schema}
+```
+"""
+
+parser_prompt = """\
+The user gave the following output to the prompt:
+Prompt: 
+{prompt}
+Output:
+{message}
+
+{format_instructions}
+"""
+
+
+class ParserKani(Kani):
+    def __init__(self, engine, *args, **kwargs):
+        super().__init__(engine, *args, **kwargs)
+
+    async def parse(self, prompt: str, message: str, format_model: Type[BaseModel], max_retries: int = 3, **kwargs):
+        format_instructions = self.get_format_instructions(format_model)
+
+        parser_instructions = parser_prompt.format(
+            prompt=prompt,
+            message=message,
+            format_instructions=format_instructions
+        )
+
+        response = await self.chat_round_str(parser_instructions, **kwargs)
+
+        try:
+            output = format_model.model_validate_json(response)
+        except ValidationError as e:
+            print(f"Output did not conform to the expected format: {e}")
+            raise e
+
+        # Clear the Chat History after successful parse
+        self.chat_history = []
+
+        return output
+
+    @staticmethod
+    def get_format_instructions(format_model: Type[BaseModel]):
+        schema = format_model.model_json_schema()
+
+        reduced_schema = schema
+        if "title" in reduced_schema:
+            del reduced_schema["title"]
+        if "type" in reduced_schema:
+            del reduced_schema["type"]
+
+        schema_str = json.dumps(reduced_schema, indent=4)
+
+        return FORMAT_INSTRUCTIONS.format(schema=schema_str)
+
+    @classmethod
+    def default(cls):
+        """Default ParserKani with OpenAIEngine."""
+        engine = OpenAIEngine(model="gpt-3.5-turbo")
+        return cls(engine)
+
+
+
+# Testing
+# parser = ParserKani(engine=OpenAIEngine(model="gpt-3.5-turbo"))
+#
+# sample_prompt = """\
+# Below are the responses from all players. Now it is time to vote. Choose from the players below who you think the Chameleon is.
+# - Mallory: I am tall and have a long neck.
+# - Jack: I am a herbivore and have a long neck.
+# - Jill: I am a herbivore and have a long neck.
+# - Bob: I am tall and have a long neck.
+# - Courtney: I am tall and have a long neck.
+# """
+#
+# sample_message = """\
+# I think the Chameleon is Mallory.
+# """
+#
+# test_output = asyncio.run(parser.parse(prompt=sample_prompt, message=sample_message, format_model=VoteModel))
+#
+# print(test_output)
+#
+# print(VoteModel.model_validate_json(test_output))