converted player classes to pydantic, adding player logging

src/agent_interfaces.py

@@ -1,35 +1,30 @@
 from json import JSONDecodeError
-from typing import Type, NewType
+from typing import Type, NewType, List, Any
 import json
 
 from openai import OpenAI
 from colorama import Fore, Style
-from pydantic import BaseModel, ValidationError
+from pydantic import BaseModel, ValidationError, Field, ConfigDict
 
 from output_formats import OutputFormatModel
 from message import Message, AgentMessage
 from data_collection import save
 
 
-class BaseAgentInterface:
+class BaseAgentInterface(BaseModel):
     """
     The interface that agents use to receive info from and interact with the game.
     This is the base class and should not be used directly.
     """
 
+    agent_id: str
+    """The id of the agent."""
+    log_messages: bool = True
+    """Whether to log messages or not."""
+    messages: List[Message] = []
+    """The message history of the agent."""
     is_human: bool = False
-
-    def __init__(
-            self,
-            agent_id: str = None,
-            log_messages: bool = True
-    ):
-        self.id = agent_id
-        """The id of the agent."""
-        self.log_messages = log_messages
-        """Whether to log messages or not."""
-        self.messages = []
-        """The message history of the agent."""
+    """Whether the agent is human or not."""
 
     @property
     def is_ai(self):
@@ -37,7 +32,7 @@ class BaseAgentInterface:
 
     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))
+        bound_message = AgentMessage.from_message(message, self.agent_id, len(self.messages))
         if self.log_messages:
             save(bound_message)
         self.messages.append(bound_message)
@@ -129,16 +124,14 @@ class BaseAgentInterface:
         raise NotImplementedError
 
 
-AgentInterface = NewType("AgentInterface", BaseAgentInterface)
-
-
 class OpenAIAgentInterface(BaseAgentInterface):
     """An interface that uses the OpenAI API (or compatible 3rd parties) to generate responses."""
+    model_config = ConfigDict(protected_namespaces=())
 
-    def __init__(self, agent_id: str, model_name: str = "gpt-3.5-turbo"):
-        super().__init__(agent_id)
-        self.model_name = model_name
-        self.client = OpenAI()
+    model_name: str ="gpt-3.5-turbo"
+    """The name of the model to use for generating responses."""
+    client: Any = Field(default_factory=OpenAI, exclude=True)
+    """The OpenAI client used to generate responses."""
 
     def _generate(self) -> str:
         """Generates a response using the message history"""
@@ -153,7 +146,7 @@ class OpenAIAgentInterface(BaseAgentInterface):
 
 
 class HumanAgentInterface(BaseAgentInterface):
-    is_human = True
+    is_human: bool = Field(default=True, frozen=True)
 
     def generate_formatted_response(
             self,

src/data_collection.py

@@ -26,11 +26,11 @@ def save(log_object: Model):
 
 
 def get_log_file(log_object: Model) -> str:
-    match type(log_object):
-        case message.AgentMessage:
-            log_file = "messages.jsonl"
-        # ...
-        case _:
-            raise ValueError(f"Unknown log object type: {type(log_object)}")
+    if isinstance(log_object, message.AgentMessage):
+        log_file = "messages.jsonl"
+    elif isinstance(log_object, player.Player):
+        log_file = "players.jsonl"
+    else:
+        raise ValueError(f"Unknown log object type: {type(log_object)}")
 
     return os.path.join(data_dir, log_file)

src/game.py

@@ -4,7 +4,7 @@ from game_utils import *
 from player import Player
 from message import Message, MessageType
 from agent_interfaces import HumanAgentCLI, OpenAIAgentInterface, HumanAgentInterface
-
+from data_collection import save
 
 # Abstracting the Game Class is a WIP so that future games can be added
 class Game:
@@ -34,16 +34,12 @@ class Game:
 
     def player_from_id(self, player_id: str) -> Player:
         """Returns a player from their ID."""
-        return next((player for player in self.players if player.id == player_id), None)
+        return next((player for player in self.players if player.player_id == player_id), None)
 
     def player_from_name(self, name: str) -> Player:
         """Returns a player from their name."""
         return next((player for player in self.players if player.name == name), None)
 
-    def human_player(self) -> Player:
-        """Returns the human player."""
-        return next((player for player in self.players if player.interface.is_human), None)
-
     def game_message(
             self,
             content: str,
@@ -88,6 +84,11 @@ class Game:
         """Runs the game."""
         raise NotImplementedError("The run_game method must be implemented by the subclass.")
 
+    def end_game(self):
+        """Ends the game and declares a winner."""
+        for player in self.players:
+            save(player)
+
     @classmethod
     def from_human_name(
             cls, human_name: str = None,
@@ -113,18 +114,19 @@ class Game:
 
         for i in range(0, cls.number_of_players):
             player_id = f"{game_id}-{i + 1}"
+            player_dict = {"game_id": game_id, "player_id": player_id}
 
             if human_index == i:
-                name = human_name
-                interface = human_interface(player_id)
-                message_level = human_message_level
+                player_dict["name"] = human_name
+                player_dict["interface"] = human_interface(agent_id=player_id)
+                player_dict["message_level"] = human_message_level
             else:
-                name = ai_names.pop()
+                player_dict["name"] = ai_names.pop()
                 # all AI players use the OpenAI interface for now - this can be changed in the future
-                interface = OpenAIAgentInterface(player_id)
-                message_level = "info"
+                player_dict["interface"] = OpenAIAgentInterface(agent_id=player_id)
+                player_dict["message_level"] = "info"
 
-            players.append(Player(name, player_id, interface, message_level))
+            players.append(Player(**player_dict))
 
         # Add Observer - an Agent who can see all the messages, but doesn't actually play
         if human_index is None:

src/game_chameleon.py

@@ -66,11 +66,6 @@ class ChameleonGame(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 run_game(self):
         """Starts the game."""
 
@@ -91,9 +86,10 @@ class ChameleonGame(Game):
 
                 if max(points) >= self.winning_score:
                     self.game_state = "game_end"
-                    self.winner_id = self.players[points.index(max(points))].id
+                    self.winner_id = self.players[points.index(max(points))].player_id
                     winner = self.player_from_id(self.winner_id)
                     self.game_message(f"The game is over {winner.name} has won!")
+                    self.end_game()
 
                 else:
                     self.game_state = "setup_round"
@@ -109,8 +105,8 @@ class ChameleonGame(Game):
         # Phase I: Collect Player Animal Descriptions
         if self.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.player_id not in [animal_description['player_id'] for animal_description in
+                                                    self.round_animal_descriptions]:
 
                     response = self.player_turn_animal_description(current_player)
 
@@ -127,8 +123,8 @@ class ChameleonGame(Game):
         # Phase III: The Herd Votes for who they think the Chameleon is
         if self.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]:
+                if current_player.role == "herd" and current_player.player_id not in [vote['voter_id'] for vote in
+                                                                                      self.herd_vote_tally]:
 
                     response = self.player_turn_herd_vote(current_player)
 
@@ -147,7 +143,7 @@ class ChameleonGame(Game):
 
         # Assign Roles
         chameleon_index = random_index(len(self.players))
-        self.chameleon_ids.append(self.players[chameleon_index].id)
+        self.chameleon_ids.append(self.players[chameleon_index].player_id)
 
         for i, player in enumerate(self.players):
             if i == chameleon_index:
@@ -176,7 +172,7 @@ class ChameleonGame(Game):
         response = player.interface.generate_formatted_response(AnimalDescriptionFormat)
 
         if response:
-            self.round_animal_descriptions.append({"player_id": player.id, "description": response.description})
+            self.round_animal_descriptions.append({"player_id": player.player_id, "description": response.description})
             self.game_message(f"{player.name}: {response.description}", player, exclude=True)
             self.awaiting_input = False
         else:
@@ -221,7 +217,7 @@ class ChameleonGame(Game):
 
             voted_for_player = self.player_from_name(response.vote)
 
-            player_vote = {"voter_id": player.id, "voted_for_id": voted_for_player.id}
+            player_vote = {"voter_id": player.player_id, "voted_for_id": voted_for_player.player_id}
 
             self.herd_vote_tally.append(player_vote)
             self.awaiting_input = False
@@ -263,11 +259,11 @@ class ChameleonGame(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_id"] == self.chameleon.id:
+            if vote["voted_for_id"] == self.chameleon.player_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:
+        if not accused_player_id or accused_player_id != self.chameleon.player_id:
             self.chameleon.points += 1
 
         # Print Scores
@@ -299,7 +295,7 @@ class ChameleonGame(Game):
         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:
+            if response["player_id"] != exclude.player_id:
                 player = self.player_from_id(response["player_id"])
                 formatted_responses += f" - {player.name}: {response['description']}\n"
 

src/player.py

@@ -1,29 +1,25 @@
 from typing import Literal, List, Type
 
-from agent_interfaces import AgentInterface, HumanAgentCLI, BaseAgentInterface
+from pydantic import BaseModel, Field
+
+from agent_interfaces import HumanAgentCLI, BaseAgentInterface
 from message import MessageType
 
 Role = Literal["chameleon", "herd"]
 
 
 # Abstraction is a WIP and a little premature, but I'd like to reuse this framework to create more Games in the future
-class Player:
+class Player(BaseModel):
     """Base class for a player"""
 
-    def __init__(self,
-                 name: str,
-                 player_id: str,
-                 interface: BaseAgentInterface,
-                 message_level: str = "info"
-                 ):
-        self.name = name
-        """The name of the player."""
-        self.id = player_id
-        """The id of the player."""
-        self.interface = interface
-        """The interface used by the agent controlling the player to communicate with the game."""
-        self.message_level = message_level
-        """The level of messages that the player will receive. Can be "info", "verbose", or "debug"."""
+    name: str
+    """The name of the player."""
+    player_id: str
+    """The id of the player."""
+    interface: BaseAgentInterface = Field(exclude=True)
+    """The interface used by the agent controlling the player to communicate with the game."""
+    message_level: str = "info"
+    """The level of messages that the player will receive. Can be "info", "verbose", or "debug"."""
 
     def can_receive_message(self, message_type: MessageType) -> bool:
         """Returns True if the player can receive a message of the type."""
@@ -44,29 +40,28 @@ class Player:
     ):
         """Creates an observer player."""
         name = "Observer"
-        player_id = f"{game_id}_observer"
-        interface = interface_type(player_id, log_messages)
+        player_id = f"{game_id}-observer"
+        interface = interface_type(agent_id=player_id)
 
-        return cls(name, player_id, interface, message_level)
+        return cls(name=name, player_id=player_id, interface=interface, message_level=message_level)
 
 
 class PlayerSubclass(Player):
     @classmethod
     def from_player(cls, player: Player):
         """Creates a new instance of the subclass from a player instance."""
-        return cls(player.name, player.id, player.interface, player.message_level)
+        fields = player.model_dump()
+        fields['interface'] = player.interface
+        return cls(**fields)
 
 
 class ChameleonPlayer(PlayerSubclass):
     """A player in the game Chameleon"""
 
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-
-        self.points: int = 0
-        """The number of points the player has."""
-        self.roles: List[Role] = []
-        """The role of the player in the game. Can be "chameleon" or "herd". This changes every round."""
+    points: int = 0
+    """The number of points the player has."""
+    roles: List[Role] = []
+    """The role of the player in the game. Can be "chameleon" or "herd". This changes every round."""
 
     def assign_role(self, role: Role):
         self.roles.append(role)