reconstructed players using langchain

src/agents.py

@@ -1,15 +0,0 @@
-from kani import Kani
-
-class LogMessagesKani(Kani):
-    def __init__(self, engine, log_filepath: str = None,  *args, **kwargs):
-        super().__init__(engine, *args, **kwargs)
-        self.log_filepath = log_filepath
-
-    async def add_to_history(self, message, *args, **kwargs):
-        await super().add_to_history(message, *args, **kwargs)
-
-        # Logs Message to File
-        if self.log_filepath:
-            with open(self.log_filepath, "a+") as log_file:
-                log_file.write(message.model_dump_json())
-                log_file.write("\n")

src/controllers.py

@@ -0,0 +1,28 @@
+import os
+
+from langchain_core.runnables import RunnableLambda
+from langchain_openai import ChatOpenAI
+from langchain_core.messages import AIMessage
+
+
+def player_input(prompt):
+    print(prompt)
+    # even though they are human, we still need to return an AIMessage, since the HumanMessages are from the GameMaster
+    response = AIMessage(content=input())
+    return response
+
+
+def controller_from_name(name: str):
+    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")
+    # elif name == "ollama":
+    #     return ollama_controller
+    elif name == "human":
+        return RunnableLambda(player_input)
+    else:
+        raise ValueError(f"Unknown controller name: {name}")

src/game.py

@@ -1,9 +1,9 @@
 import os
+from datetime import datetime
 
 from game_utils import *
 from models import *
 from player import Player
-from parser import ParserKani
 from prompts import fetch_prompt, format_prompt
 
 # Default Values
@@ -13,7 +13,6 @@ NUMBER_OF_PLAYERS = 5
 class Game:
     log_dir = os.path.join(os.pardir, "experiments")
     player_log_file = "{player_id}.jsonl"
-    parser_log_file = "{game_id}-parser.jsonl"
     game_log_file = "{game_id}-game.jsonl"
 
     def __init__(
@@ -24,6 +23,9 @@ 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)
 
         # Gather Player Names
         if human_name:
@@ -44,7 +46,7 @@ class Game:
                 controller = "human"
             else:
                 name = ai_names.pop()
-                controller = "ai"
+                controller = "openai"
 
             if self.chameleon_index == i:
                 role = "chameleon"
@@ -63,13 +65,18 @@ class Game:
         # Game State
         self.player_responses = []
 
-        # Parser
-        parser_log_path = os.path.join(self.log_dir, self.parser_log_file.format(game_id=self.game_id))
-        self.parser = ParserKani.default(parser_log_path)
-
-    def format_responses(self) -> str:
+    def format_responses(self, exclude: str = None) -> str:
         """Formats the responses of the players into a single string."""
-        return "\n".join([f" - {response['sender']}: {response['response']}" for response in self.player_responses])
+        if len(self.player_responses) == 0:
+            return "None, you are the first player!"
+        else:
+            formatted_responses = ""
+            for response in self.player_responses:
+                # Used to exclude the player who is currently responding, so they don't vote for themselves like a fool
+                if response["sender"] != exclude:
+                    formatted_responses += f" - {response['sender']}: {response['response']}\n"
+
+            return formatted_responses
 
     def get_player_names(self) -> list[str]:
         """Returns the names of the players."""
@@ -93,19 +100,16 @@ class Game:
                     prompt = format_prompt("herd_animal", animal=herd_animal, player_responses=self.format_responses())
 
                 # Get Player Animal Description
-                response = await player.respond_to(prompt)
-                # Parse Animal Description
-                output = await self.parser.parse(prompt, response, AnimalDescriptionModel)
+                response = await player.respond_to(prompt, AnimalDescriptionModel)
 
-                self.player_responses.append({"sender": player.name, "response": output.description})
+                self.player_responses.append({"sender": player.name, "response": response.description})
 
             # Phase II: Chameleon Decides if they want to guess the animal (secretly)
             prompt = format_prompt("chameleon_guess_decision", player_responses=self.format_responses())
 
-            response = await self.players[self.chameleon_index].respond_to(prompt)
-            output = await self.parser.parse(prompt, response, ChameleonGuessDecisionModel)
+            response = await self.players[self.chameleon_index].respond_to(prompt, ChameleonGuessDecisionModel)
 
-            if output.decision == "guess":
+            if response.decision == "guess":
                 chameleon_will_guess = True
             else:
                 chameleon_will_guess = False
@@ -115,10 +119,9 @@ class Game:
                 # Chameleon Guesses Animal
                 prompt = fetch_prompt("chameleon_guess_animal")
 
-                response = await self.players[self.chameleon_index].respond_to(prompt)
-                output = await self.parser.parse(prompt, response, ChameleonGuessAnimalModel)
+                response = await self.players[self.chameleon_index].respond_to(prompt, ChameleonGuessAnimalModel)
 
-                if output.animal == herd_animal:
+                if response.animal == herd_animal:
                     winner = "chameleon"
                 else:
                     winner = "herd"
@@ -130,14 +133,12 @@ class Game:
                     prompt = format_prompt("vote", 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 = await player.respond_to(prompt, VoteModel)
 
                     # check if a valid player was voted for...
 
                     # Add Vote to Player Votes
-                    player_votes.append(output.vote)
+                    player_votes.append(response.vote)
 
                 print(player_votes)
 
@@ -160,6 +161,7 @@ class Game:
         # Log Game Info
         game_log = {
             "game_id": self.game_id,
+            "start_time": self.start_time,
             "herd_animal": herd_animal,
             "number_of_players": len(self.players),
             "human_player": self.players[self.human_index].id if self.human_index else "None",

src/models.py

@@ -1,4 +1,6 @@
-from typing import Annotated
+import random
+from typing import Annotated, Type, List
+import json
 
 from pydantic import BaseModel, field_validator
 
@@ -7,7 +9,7 @@ MAX_DESCRIPTION_LEN = 10
 
 class AnimalDescriptionModel(BaseModel):
     # Define fields of our class here
-    description: str
+    description: Annotated[str, "A brief description of the animal"]
 
     # @field_validator('description')
     # @classmethod
@@ -45,11 +47,18 @@ class ChameleonGuessDecisionModel(BaseModel):
 
 
 class ChameleonGuessAnimalModel(BaseModel):
-    animal: str
+    animal: Annotated[str, "The name of the animal the chameleon is guessing"]
+
+    @field_validator('animal')
+    @classmethod
+    def is_one_word(cls, v) -> str:
+        if len(v.split()) > 1:
+            raise ValueError("Animal's name must be one word")
+        return v
 
 
 class VoteModel(BaseModel):
-    vote: str
+    vote: Annotated[str, "The name of the player you are voting for"]
 
     # @field_validator('vote')
     # @classmethod

src/parser.py

@@ -1,96 +0,0 @@
-from typing import Type
-import asyncio
-import json
-
-from kani.engines.openai import OpenAIEngine
-from pydantic import BaseModel, ValidationError
-
-from agents import LogMessagesKani
-
-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(LogMessagesKani):
-    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, log_filepath: str = None):
-        """Default ParserKani with OpenAIEngine."""
-        engine = OpenAIEngine(model="gpt-3.5-turbo")
-        return cls(engine, log_filepath=log_filepath)
-
-
-
-# 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))

src/player.py

@@ -1,44 +1,35 @@
 import os
-import json
-import asyncio
+from typing import Type
 
-import openai
-from agents import LogMessagesKani
-from kani import ChatMessage
-from kani.engines.openai import OpenAIEngine
+from langchain_core.runnables import Runnable, RunnableParallel, RunnableLambda, chain
 
-from game_utils import log
+from langchain.output_parsers import PydanticOutputParser
+from langchain_core.prompts import PromptTemplate
+from langchain_core.messages import HumanMessage
 
-# api_type = "tgi"
-api_type = "openai"
-# api_type = "ollama"
-
-match api_type:
-    case "tgi":
-        # Using TGI Inference Endpoints from Hugging Face
-        default_engine = OpenAIEngine(  # type: ignore
-            api_base=os.environ['HF_ENDPOINT_URL'] + "/v1/",
-            api_key=os.environ['HF_API_TOKEN']
-        )
-    case "openai":
-        # Using OpenAI GPT-3.5 Turbo
-        default_engine = OpenAIEngine(model="gpt-3.5-turbo")  # type: ignore
-    case "ollama":
-        # Using Ollama
-        default_engine = OpenAIEngine(
-            api_base="http://localhost:11434/v1",
-            api_key="ollama",
-            model="mistral"
-        )
+from pydantic import BaseModel
 
+from game_utils import log
+from controllers import controller_from_name
 
 class Player:
-    def __init__(self, name: str, controller_type: str, role: str, id: str = None, log_filepath: str = None):
+    def __init__(
+            self,
+            name: str,
+            controller: str,
+            role: str,
+            id: str = None,
+            log_filepath: str = None
+    ):
         self.name = name
         self.id = id
-        self.controller = controller_type
-        if controller_type == "ai":
-            self.kani = LogMessagesKani(default_engine, log_filepath=log_filepath)
+
+        if controller == "human":
+            self.controller_type = "human"
+        else:
+            self.controller_type = "ai"
+
+        self.controller = controller_from_name(controller)
 
         self.role = role
         self.messages = []
@@ -50,26 +41,71 @@ class Player:
                 "id": self.id,
                 "name": self.name,
                 "role": self.role,
-                "controller": controller_type,
+                "controller": {
+                    "name": controller,
+                    "type": self.controller_type
+                }
             }
             log(player_info, log_filepath)
 
-    async def respond_to(self, prompt: str) -> str:
-        """Makes the player respond to a prompt. Returns the response."""
-        if self.controller == "human":
-            # We're pretending the human is an ai for logging purposes... I don't love this but it's fine for now
-            log(ChatMessage.user(prompt).model_dump_json(), self.log_filepath)
-            print(prompt)
-            output = input()
-            log(ChatMessage.assistant(output).model_dump_json(), self.log_filepath)
-
-            return output
-
-        elif self.controller == "ai":
-            output = await self.kani.chat_round_str(prompt)
+        # initialize the runnables
+        self.generate = RunnableLambda(self._generate)
+        self.format_output = RunnableLambda(self._output_formatter)
+
+    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."""
+        message = HumanMessage(content=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 ValueError as e:
+                if retries < max_retries:
+                    self.add_to_history(HumanMessage(content=f"Error formatting response: {e} \n\n Please try again."))
+                    output = await self.format_output.ainvoke({"output_format": output_format})
+                    retries += 1
+                else:
+                    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 add_to_history(self, message):
+        self.messages.append(message)
+        # log(message.model_dump_json(), self.log_filepath)
+
+    def _generate(self, message: HumanMessage):
+        """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 = self.controller.invoke(message.content)
+        else:
+            response = self.controller.invoke(self.messages)
+
+        self.add_to_history(response)
+
+        return response
+
+    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}"
+        )
 
-            return output
+        parser = PydanticOutputParser(pydantic_object=output_format)
 
+        prompt = prompt_template.invoke({"format_instructions": parser.get_format_instructions()})
 
+        message = HumanMessage(content=prompt.text)
 
+        response = self.generate.invoke(message)
 
+        return parser.invoke(response)

src/prompts.py

@@ -1,12 +1,66 @@
+from models import *
+from langchain.prompts.few_shot import FewShotPromptTemplate
+from langchain.prompts.prompt import PromptTemplate
+
+
 def fetch_prompt(prompt_name):
     """Fetches a static prompt."""
     return prompts[prompt_name]
-
 def format_prompt(prompt_name, **kwargs):
     """Fetches a template prompt and populates it."""
     return fetch_prompt(prompt_name).format(**kwargs)
 
 
+class Task:
+    def __init__(self, prompt: str, response_format: Type[BaseModel], few_shot_examples: List[dict] = None):
+        self.prompt = prompt
+        self.response_format = response_format
+        self.few_shot_examples = few_shot_examples
+
+    def full_prompt(self, **kwargs):
+        prompt = self.prompt.format(**kwargs)
+
+        format_instructions = self.get_format_instructions()
+        if self.few_shot_examples:
+            few_shot = self.get_few_shot()
+
+    def get_format_instructions(self):
+        schema = self.get_input_schema()
+        format_instructions = FORMAT_INSTRUCTIONS.format(schema=schema)
+
+        return format_instructions
+
+    def get_input_schema(self):
+        schema = self.response_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 schema_str
+
+    def get_few_shot(self, max_examples=3):
+        if len(self.few_shot_examples) <= max_examples:
+            examples = self.few_shot_examples
+        else:
+            examples = random.sample(self.few_shot_examples, max_examples)
+
+        few_shot = "\n\n".join([self.format_example(ex) for ex in examples])
+
+        return few_shot
+
+    def format_example(self, example):
+        ex_prompt = self.prompt.format(**example['inputs'])
+        ex_response = example['response']
+
+        return f"Prompt: {ex_prompt}\nResponse: {ex_response}"
+
+
+
 _game_rules = '''\
 GAME RULES: You are playing a social deduction game where every player pretends the be the same animal. 
 During the round players go around the room and make an "I"-statement as if they were the animal. 
@@ -17,11 +71,11 @@ The other players must be careful not to give away too much information with the
 
 _herd_animal = """\
 You are a {animal}, keep this a secret at all costs. 
-In 10 words or less give a description of yourself starting with "I". The description should not give away too much information about the {animal} as you do not want the Chameleon to be able to guess what animal you are. Do not repeat responses from other players.
-If the Chameleon can guess what animal you really are you will LOSE.
+In as few words as possible describe of yourself starting with "I". Your description should be vague but true, \
+since if the Chameleon can guess animal you are, you will LOSE. Do not repeat responses from other players.
 Previously Mentioned Descriptions: 
 {player_responses}
-"""
+Your Response: """
 
 _chameleon_animal = """\
 You are the Chameleon, keep this a secret at all costs. 
@@ -31,7 +85,7 @@ If no one else has said anything try to say something generic that could be true
 If the other players realize you are the Chameleon you will LOSE. 
 Previously Mentioned Descriptions:
 {player_responses}
-"""
+Your Response: """
 
 _all_responses = """\
 Below are the responses from all the other players. 
@@ -42,6 +96,8 @@ _chameleon_guess_decision = """\
 You now have the opportunity to guess what animal the other players are pretending to be.
 If you guess correctly you will WIN, if you guess incorrectly you will LOSE.
 If you believe you know what animal the other players are pretending to be make choose to GUESS, otherwise choose to PASS.
+Your response should be one of ("GUESS", "PASS")
+Your Response: 
 """
 
 _chameleon_guess_animal = """\
@@ -53,10 +109,9 @@ Now it is time to vote. Choose from the players above who you think the Chameleo
 """
 
 prompts = {
-    "herd_animal": _game_rules + _herd_animal,
+    "herd_animal":  _game_rules + _herd_animal,
     "chameleon_animal": _game_rules + _chameleon_animal,
     "chameleon_guess_decision": _all_responses + _chameleon_guess_decision,
     "chameleon_guess_animal": _chameleon_guess_animal,
     "vote": _all_responses + _vote_prompt
 }
-