redid controllers into AgentInterface class, unified message system

src/agent_interfaces.py

@@ -0,0 +1,148 @@
+from typing import Type, NewType
+import json
+
+from openai import OpenAI
+from colorama import Fore, Style
+from pydantic import BaseModel, ValidationError
+
+from message import Message, AgentMessage
+
+FORMAT_INSTRUCTIONS = """The output should be reformatted as a JSON instance that conforms to the JSON schema below.
+Here is the output schema:
+```
+{schema}
+```
+"""
+
+
+class BaseAgentInterface:
+    """
+    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.
+    """
+
+    is_human: bool = False
+
+    def __init__(
+            self,
+            agent_id: str = None
+    ):
+        self.id = agent_id
+        self.messages = []
+
+    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))
+        self.messages.append(bound_message)
+
+    def respond_to(self, message: Message) -> Message:
+        """Adds a message to the message history, and generates a response message."""
+        self.add_message(message)
+        response = Message(type="agent", content=self._generate_response())
+        self.add_message(response)
+        return response
+
+    def respond_to_formatted(self, message: Message, output_format: Type[BaseModel], max_retries = 3) -> Type[BaseModel]:
+        """Adds a message to the message history, and generates a response matching the provided format."""
+        initial_response = self.respond_to(message)
+
+        reformat_message = Message(type="format", content=self._get_format_instructions(output_format))
+
+        output = None
+        retries = 0
+
+        while not output and retries < max_retries:
+            try:
+                formatted_response = self.respond_to(reformat_message)
+                output = output_format.model_validate_json(formatted_response.content)
+            except ValidationError as e:
+                if retries > max_retries:
+                    raise e
+                reformat_message = Message(type="retry", content=f"Error formatting response: {e} \n\n Please try again.")
+                retries += 1
+
+        return output
+
+    def _generate_response(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
+
+    # This should probably be put on a theoretical output format class...
+    # Or maybe on the Message class with a from format constructor
+    @staticmethod
+    def _get_format_instructions(output_format: Type[BaseModel]):
+        schema = output_format.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)
+
+
+AgentInterface = NewType("AgentInterface", BaseAgentInterface)
+
+
+class OpenAIAgentInterface(BaseAgentInterface):
+    """An interface that uses the OpenAI API (or compatible 3rd parties) to generate responses."""
+    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()
+
+    def _generate_response(self) -> str:
+        """Generates a response using the message history"""
+        open_ai_messages = [message.to_openai() for message in self.messages]
+
+        completion = self.client.chat.completions.create(
+            model=self.model_name,
+            messages=open_ai_messages
+        )
+
+        return completion.choices[0].message.content
+
+
+class HumanAgentInterface(BaseAgentInterface):
+
+    is_human = True
+
+    def respond_to_formatted(self, message: Message, output_format: Type[BaseModel], **kwargs) -> Type[BaseModel]:
+        """For Human agents, we can trust them enough to format their own responses... for now"""
+        response = super().respond_to(message)
+        # only works because current outputs have only 1 field...
+        field_name = output_format.model_fields.copy().popitem()[0]
+        output = output_format.model_validate({field_name: response.content})
+
+        return output
+
+
+class HumanAgentCLI(HumanAgentInterface):
+    """A Human agent that uses the command line interface to generate responses."""
+
+    def __init__(self, agent_id: str):
+        super().__init__(agent_id)
+
+    def add_message(self, message: Message):
+        super().add_message(message)
+        if message.type == "verbose":
+            print(Fore.GREEN + message.content + Style.RESET_ALL)
+        elif message.type == "debug":
+            print(Fore.YELLOW + "DEBUG: " + message.content + Style.RESET_ALL)
+        elif message.type != "agent":
+            # Prevents the agent from seeing its own messages on the command line
+            print(message.content)
+
+    def _generate_response(self) -> str:
+        """Generates a response using the message history"""
+        response = input()
+        return response

src/controllers.py

@@ -1,21 +0,0 @@
-import os
-
-from langchain_core.runnables import Runnable
-from langchain_openai import ChatOpenAI
-from langchain_core.messages import AIMessage
-
-MAX_TOKENS = 50
-
-
-def controller_from_name(name: str) -> Runnable:
-    if name == "tgi":
-        return ChatOpenAI(
-            api_base=os.environ['HF_ENDPOINT_URL'] + "/v1/",
-            api_key=os.environ['HF_API_TOKEN']
-        )
-    elif name == "openai":
-        return ChatOpenAI(model="gpt-3.5-turbo", max_tokens=MAX_TOKENS)
-    elif name == "ollama":
-        return ChatOpenAI(model="mistral", openai_api_key="ollama", openai_api_base="http://localhost:11434/v1", max_tokens=MAX_TOKENS)
-    else:
-        raise ValueError(f"Unknown controller name: {name}")

src/game.py

@@ -8,15 +8,14 @@ from game_utils import *
 from models import *
 from player import Player
 from prompts import fetch_prompt, format_prompt
-
-from langchain_core.runnables import RunnableLambda
-from langchain_core.messages import AIMessage
-from controllers import controller_from_name
+from message import Message
+from agent_interfaces import HumanAgentCLI, OpenAIAgentInterface
 
 # Default Values
 NUMBER_OF_PLAYERS = 6
 WINNING_SCORE = 11
 
+
 class Game:
 
     log_dir = os.path.join(os.pardir, "experiments")
@@ -40,11 +39,6 @@ class Game:
         # Game ID
         self.game_id = game_id()
         self.start_time = datetime.now().strftime('%y%m%d-%H%M%S')
-        self.log_dir = os.path.join(self.log_dir, f"{self.start_time}-{self.game_id}")
-        os.makedirs(self.log_dir, exist_ok=True)
-
-        # Choose Chameleon
-        self.chameleon_index = random_index(number_of_players)
 
         # Gather Player Names
         if human_name:
@@ -62,28 +56,22 @@ class Game:
         # Add Players
         self.players = []
         for i in range(0, number_of_players):
+            player_id = f"{self.game_id}-{i + 1}"
+
             if self.human_index == i:
                 name = human_name
-                controller_name = "human"
-                controller = RunnableLambda(self.human_input)
+                interface = HumanAgentCLI(player_id)
             else:
                 name = ai_names.pop()
-                controller_name = "openai"
-                controller = controller_from_name(controller_name)
-
-            if self.chameleon_index == i:
-                role = "chameleon"
-            else:
-                role = "herd"
+                interface = OpenAIAgentInterface(player_id)
 
-            player_id = f"{self.game_id}-{i + 1}"
+            self.players.append(Player(name, player_id, interface))
 
-            player_log_path = os.path.join(
-                self.log_dir,
-                self.player_log_file_template.format(player_id=player_id)
-            )
-
-            self.players.append(Player(name, controller, controller_name, player_id, log_filepath=player_log_path))
+        # 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")
+        else:
+            self.observer = None
 
         # Game State
         self.player_responses = []
@@ -101,55 +89,43 @@ class Game:
 
             return formatted_responses
 
+    def observer_message(self, message: Message):
+        """Sends a message to the observer if there is one."""
+        if self.observer:
+            self.observer.add_message(message)
 
     def game_message(
-            self, message: str,
+            self, content: str,
             recipient: Optional[Player] = None,  # If None, message is broadcast to all players
             exclude: bool = False  # If True, the message is broadcast to all players except the chosen player
     ):
         """Sends a message to a player. No response is expected, however it will be included next time the player is prompted"""
+        message = Message(type="info", content=content)
+
         if exclude or not recipient:
             for player in self.players:
                 if player != recipient:
-                    player.prompt_queue.append(message)
-                    if player.controller_type == "human":
-                        self.human_message(message)
-            if self.verbose and not self.human_index:
-                self.human_message(message)
+                    player.interface.add_message(message)
+            self.observer_message(message)
         else:
-            recipient.prompt_queue.append(message)
-            if recipient.controller_type == "human":
-                self.human_message(message)
-
-    async def instructional_message(self, message: str, player: Player,  output_format: Type[BaseModel]):
-        """Sends a message to a specific player and gets their response."""
-        if player.controller_type == "human":
-            self.human_message(message)
-        response = await player.respond_to(message, output_format)
-        return response
-
-    # The following methods are used to broadcast messages to a human.
-    # They are design so that they can be overridden by a subclass for a different player interface.
-    @staticmethod
-    async def human_input(prompt: str) -> AIMessage:
-        """Gets input from the human player."""
-        response = AIMessage(content=input())
-        return response
-
-    @staticmethod
-    def human_message(message: str):
-        """Sends a message for the human player to read. No response is expected."""
-        print(message)
+            recipient.interface.add_message(message)
 
-    def verbose_message(self, message: str):
+    def verbose_message(self, content: str):
         """Sends a message for the human player to read. No response is expected."""
         if self.verbose:
-            print(Fore.GREEN + message + Style.RESET_ALL)
+            message = Message(type="verbose", content=content)
+            if self.human_index:
+                self.players[self.human_index].interface.add_message(message)
+            self.observer_message(message)
 
-    def debug_message(self, message: str):
+    def debug_message(self, content: str):
         """Sends a message for a human observer. These messages contain secret information about the players such as their role."""
         if self.debug:
-            print(Fore.YELLOW + "DEBUG: " + message + Style.RESET_ALL)
+            message = Message(type="debug", content=content)
+            if self.human_index:
+                self.players[self.human_index].interface.add_message(message)
+            self.observer_message(message)
+
 
     async def start(self):
         """Sets up the game. This includes assigning roles and gathering player names."""
@@ -168,8 +144,6 @@ class Game:
 
         log(game_log, game_log_path)
 
-
-
     async def run_round(self):
         """Starts the round."""
 
@@ -194,13 +168,14 @@ class Game:
 
         self.game_message(f"Each player will now take turns describing themselves:")
         for i, current_player in enumerate(self.players):
-            if current_player.controller_type != "human":
+            if current_player.interface.is_ai:
                 self.verbose_message(f"{current_player.name} is thinking...")
 
             prompt = fetch_prompt("player_describe_animal")
 
             # Get Player Animal Description
-            response = await self.instructional_message(prompt, current_player, AnimalDescriptionModel)
+            message = Message(type="prompt", content=prompt)
+            response = current_player.interface.respond_to_formatted(message, AnimalDescriptionModel)
 
             self.player_responses.append({"sender": current_player.name, "response": response.description})
 
@@ -209,12 +184,13 @@ class Game:
         # Phase III: Chameleon Guesses the Animal
 
         self.game_message("All players have spoken. The Chameleon will now guess the secret animal...")
-        if self.human_index != self.chameleon_index:
+        if chameleon.interface.is_ai or self.observer:
             self.verbose_message("The Chameleon is thinking...")
 
         prompt = fetch_prompt("chameleon_guess_animal")
 
-        response = await self.instructional_message(prompt, chameleon, ChameleonGuessAnimalModel)
+        message = Message(type="prompt", content=prompt)
+        response = chameleon.interface.respond_to_formatted(message, ChameleonGuessAnimalModel)
 
         chameleon_animal_guess = response.animal
 
@@ -224,19 +200,20 @@ class Game:
         player_votes = []
         for player in self.players:
             if player.role == "herd":
-                if player.is_ai():
+                if player.interface.is_ai:
                     self.verbose_message(f"{player.name} is thinking...")
 
                 prompt = format_prompt("vote", player_responses=self.format_responses(exclude=player.name))
 
                 # Get Player Vote
-                response = await self.instructional_message(prompt, player, VoteModel)
+                message = Message(type="prompt", content=prompt)
+                response = player.interface.respond_to_formatted(message, VoteModel)
 
                 # check if a valid player was voted for...
 
                 # Add Vote to Player Votes
                 player_votes.append({"voter": player, "vote": response.vote})
-                if player.is_ai():
+                if player.interface.is_ai:
                     self.debug_message(f"{player.name} voted for {response.vote}")
 
 
@@ -285,7 +262,7 @@ class Game:
         # Log Round Info
         round_log = {
             "herd_animal": herd_animal,
-            "chameleon_name": self.players[self.chameleon_index].name,
+            "chameleon_name": chameleon.name,
             "chameleon_guess": chameleon_animal_guess,
             "herd_votes": player_votes,
         }

src/message.py

@@ -1,32 +1,13 @@
 from typing import Literal
 from pydantic import BaseModel, computed_field
 
-"""
-Right now we have two separate systems that use the word "message":
-
-1. The Game class uses messages to communicate with the players
-     - "game" messages pile up in the queue and are responded to by the player once an "instructional" message is sent.
-     - "verbose", and "debug" currently for the human player only
-   This does **NOT** use the Message class defined below 
-
-2. The Player class uses messages to communicate with the controller (either the AI or the human)
-   - "prompt" type messages come from the Game and are responded to by the player.
-   - "retry", "error", and "format" are internal messages used by the player to ensure the correct format
-   - "player" is used to communicate with the AI or human player.
-   All of these messages are logged, and use the Message class defined below
-
-For the future we should investigate redesigning/merging these two systems to avoid confusion
-"""
-
-MessageType = Literal["prompt", "player", "retry", "error", "format"]
+MessageType = Literal["prompt", "info", "agent", "retry", "error", "format", "verbose", "debug"]
 
 class Message(BaseModel):
-    player_id: str
-    """The id of the player that the message was sent by/to."""
-    message_number: int
-    """The number of the message, indicating the order in which it was sent."""
+    """A generic message, these are used to communicate between the game and the players."""
+
     type: MessageType
-    """The type of the message. Can be "prompt", "player", "retry", "error", or "format"."""
+    """The type of the message."""
     content: str
     """The content of the message."""
 
@@ -44,13 +25,35 @@ class Message(BaseModel):
         else:
             return "assistant"
 
-    @computed_field
-    def message_id(self) -> str:
-        """Returns the message id in the format used by the LLM."""
-        return f"{self.player_id}-{self.message_number}"
+    @property
+    def requires_response(self) -> bool:
+        """Returns True if the message requires a response."""
+        return self.type in ["prompt", "retry", "format"]
+
+    def to_openai(self) -> dict[str, str]:
+        """Returns the message in an OpenAI API compatible format."""
+        return {"role": self.conversation_role, "content": self.content}
 
-    def to_controller(self) -> tuple[str, str]:
-        """Returns the message in a format that can be used by the controller."""
-        return self.conversation_role, self.content
 
+class AgentMessage(Message):
+    """A message bound to a specific agent, this happens when an agent receives a message from the game."""
+
+    agent_id: str
+    """The id of the controller that the message was sent by/to."""
+    message_number: int
+    """The number of the message, indicating the order in which it was sent."""
 
+    @computed_field
+    def message_id(self) -> str:
+        """Returns the message id in the format used by the LLM."""
+        return f"{self.agent_id}-{self.message_number}"
+
+    @classmethod
+    def from_message(cls, message: Message, agent_id: str, message_number: int) -> "AgentMessage":
+        """Creates an AgentMessage from a Message."""
+        return cls(
+            type=message.type,
+            content=message.content,
+            agent_id=agent_id,
+            message_number=message_number
+        )

src/player.py

@@ -1,18 +1,7 @@
-import os
-from typing import Type, Literal, List
+from typing import Literal
 import logging
 
-from langchain_core.runnables import Runnable, RunnableLambda
-
-from langchain.output_parsers import PydanticOutputParser
-from langchain_core.prompts import PromptTemplate
-
-from langchain_core.exceptions import OutputParserException
-
-from pydantic import BaseModel
-
-from game_utils import log
-from message import Message, MessageType
+from agent_interfaces import AgentInterface
 
 Role = Literal["chameleon", "herd"]
 
@@ -34,43 +23,12 @@ class Player:
     def __init__(
             self,
             name: str,
-            controller: Type[Runnable | RunnableLambda],
-            controller_name: str,
-            player_id: str = None,
-            log_filepath: str = None
+            player_id: str,
+            interface: AgentInterface
     ):
         self.name = name
         self.id = player_id
-
-        if controller_name == "human":
-            self.controller_type = "human"
-        else:
-            self.controller_type = "ai"
-
-        self.controller = controller
-        """The controller for the player."""
-        self.log_filepath = log_filepath
-        """The filepath to the log file. If None, no logs will be written."""
-        self.messages: list[Message] = []
-        """The messages the player has sent and received."""
-        self.prompt_queue: List[str] = []
-        """A queue of prompts to be added to the next prompt."""
-
-        if log_filepath:
-            player_info = {
-                "id": self.id,
-                "name": self.name,
-                "role": self.role,
-                "controller": {
-                    "name": controller_name,
-                    "type": self.controller_type
-                }
-            }
-            log(player_info, log_filepath)
-
-        # initialize the runnables
-        self.generate = RunnableLambda(self._generate)
-        self.format_output = RunnableLambda(self._output_formatter)
+        self.interface = interface
 
     def assign_role(self, role: Role):
         self.role = role
@@ -78,85 +36,3 @@ class Player:
             self.rounds_played_as_chameleon += 1
         elif role == "herd":
             self.rounds_played_as_herd += 1
-
-    async def respond_to(self, prompt: str, output_format: Type[BaseModel], max_retries=3):
-        """Makes the player respond to a prompt. Returns the response in the specified format."""
-        if self.prompt_queue:
-            # If there are prompts in the queue, add them to the current prompt
-            prompt = "\n".join(self.prompt_queue + [prompt])
-            # Clear the prompt queue
-            self.prompt_queue = []
-
-        message = self.player_message("prompt", prompt)
-        output = await self.generate.ainvoke(message)
-        if self.controller_type == "ai":
-            retries = 0
-            try:
-                output = await self.format_output.ainvoke({"output_format": output_format})
-            except OutputParserException as e:
-                if retries < max_retries:
-                    retries += 1
-                    logger.warning(f"Player {self.id} failed to format response: {output} due to an exception: {e} \n\n Retrying {retries}/{max_retries}")
-                    retry_message = self.player_message("retry", f"Error formatting response: {e} \n\n Please try again.")
-                    self.add_to_history(retry_message)
-                    output = await self.format_output.ainvoke({"output_format": output_format})
-
-                else:
-                    error_message = self.player_message("error", f"Error formatting response: {e} \n\n Max retries reached.")
-                    self.add_to_history(error_message)
-                    logging.error(f"Max retries reached due to Error: {e}")
-                    raise e
-        else:
-            # Convert the human message to the pydantic object format
-            field_name = output_format.model_fields.copy().popitem()[0]  # only works because current outputs have only 1 field
-            output = output_format.model_validate({field_name: output.content})
-
-        return output
-
-    def player_message(self, message_type: MessageType, content: str) -> Message:
-        """Creates a message assigned to the player."""
-        return Message(player_id=self.id, message_number=len(self.messages), type=message_type, content=content)
-
-
-    def add_to_history(self, message: Message):
-        self.messages.append(message)
-        log(message.model_dump(), self.log_filepath)
-
-    def is_human(self):
-        return self.controller_type == "human"
-
-    def is_ai(self):
-        return not self.is_human()
-
-    async def _generate(self, message: Message):
-        """Entry point for the Runnable generating responses, automatically logs the message."""
-        self.add_to_history(message)
-
-        # AI's need to be fed the whole message history, but humans can just go back and look at it
-        if self.controller_type == "human":
-            response = await self.controller.ainvoke(message.content)
-        else:
-            formatted_messages = [message.to_controller() for message in self.messages]
-            response = await self.controller.ainvoke(formatted_messages)
-
-        self.add_to_history(self.player_message("player", response.content))
-
-        return response
-
-    async def _output_formatter(self, inputs: dict):
-        """Formats the output of the response."""
-        output_format: BaseModel = inputs["output_format"]
-
-        prompt_template = PromptTemplate.from_template(
-            "Please rewrite your previous response using the following format: \n\n{format_instructions}"
-        )
-
-        parser = PydanticOutputParser(pydantic_object=output_format)
-
-        prompt = prompt_template.invoke({"format_instructions": parser.get_format_instructions()})
-
-        message = self.player_message("format", prompt.text)
-
-        response = await self.generate.ainvoke(message)
-
-        return await parser.ainvoke(response)