Replicate integration

app.py

@@ -1,14 +1,13 @@
 import gradio as gr
-from gradio import ChatMessage
 import re
 import time
 import json
 import os
 import jsonpickle
 import configparser
-
 import example_worlds
-from models import GeminiModel
+
+from models import get_llm
 from prompts import prompt_narrate_current_scene, prompt_world_update, prompt_describe_objective
 from huggingface_hub import HfApi
 from huggingface_hub import login
@@ -24,11 +23,11 @@ config.read('config.ini')
 # language of the game
 language = config['Options']['Language']
 
-# Initialize the model and disable the safety settings
+# Initialize the model 
 reasoning_model_name = config['Models']['ReasoningModel']
 narrative_model_name = config['Models']['NarrativeModel']
-reasoning_model = GeminiModel("API_key", reasoning_model_name)
-narrative_model = GeminiModel("API_key", narrative_model_name)
+reasoning_model = get_llm(reasoning_model_name)
+narrative_model = get_llm(narrative_model_name)
 
 # Create a name for the log file
 timestamp = time.time()
@@ -52,8 +51,8 @@ def game_loop(message, history):
     answer = ""
 
     # Get the changes in the world
-    prompt_update = prompt_world_update(world.render_world(language=language), message, language=language)
-    response_update = reasoning_model.prompt_model(prompt_update)
+    system_msg_update, user_msg_update = prompt_world_update(world.render_world(language=language), message, language=language)
+    response_update = reasoning_model.prompt_model(system_msg=system_msg_update, user_msg=user_msg_update)
 
     # Show the detected changes in the fictional world
     print("🛠️ Predicted outcomes of the player input 🛠️")
@@ -74,12 +73,13 @@ def game_loop(message, history):
     if last_player_position is not world.player.location:
     #Narrate new scene
         last_player_position = world.player.location
-        new_scene_narration = narrative_model.prompt_model(
-            prompt_narrate_current_scene(world.render_world(language=language), 
-                                         previous_narrations = world.player.visited_locations[world.player.location.name],
-                                         language=language)
+        system_msg_new_scene, user_msg_new_scene = prompt_narrate_current_scene(
+            world.render_world(language=language),
+            previous_narrations = world.player.visited_locations[world.player.location.name],
+            language=language
+            )
 
-        )
+        new_scene_narration = narrative_model.prompt_model(system_msg=system_msg_new_scene, user_msg=user_msg_new_scene)
         world.player.visited_locations[world.player.location.name]+=[new_scene_narration] 
         answer += f"\n{new_scene_narration}\n\n"
     else:
@@ -104,7 +104,7 @@ def game_loop(message, history):
     with open(os.path.join(PATH_GAMELOGS,log_filename), 'w', encoding='utf-8') as f:
         json.dump(game_log_dictionary, f, ensure_ascii=False, indent=4)
 
-    if world.check_objective():
+    if world.check_objective() or "EXIT NOW" in message:
         
         api = HfApi()
         api.upload_file(
@@ -116,6 +116,7 @@ def game_loop(message, history):
     
     return answer.replace("<",r"\<").replace(">", r"\>")
 
+
 # Instantiate the world
 world_id = config["Options"]["WorldID"]
 world = example_worlds.get_world(world_id, language=language)
@@ -137,15 +138,18 @@ game_log_dictionary[0]["initial_symbolic_world_state"] = jsonpickle.encode(world
 game_log_dictionary[0]["initial_rendered_world_state"] = world.render_world(language=language)
 
 #Generate a description of the starting scene
-starting_narration = narrative_model.prompt_model(
-    prompt_narrate_current_scene(world.render_world(language=language), 
-                                 previous_narrations = world.player.visited_locations[world.player.location.name],
-                                 language=language, 
-                                 starting_scene=True))
+system_msg_current_scene, user_msg_current_scene = prompt_narrate_current_scene(
+    world.render_world(language=language),
+    previous_narrations = world.player.visited_locations[world.player.location.name],
+    language=language, 
+    starting_scene=True
+    )
+starting_narration = narrative_model.prompt_model(system_msg=system_msg_current_scene, user_msg=user_msg_current_scene)
 world.player.visited_locations[world.player.location.name]+=[starting_narration]
 
 #Generate a description of the main objective
-narrated_objective = narrative_model.prompt_model(prompt_describe_objective(world.objective, language=language))
+system_msg_objective, user_msg_objective = prompt_describe_objective(world.objective, language=language)
+narrated_objective = narrative_model.prompt_model(system_msg=system_msg_objective, user_msg=user_msg_objective)
 starting_narration += f"\n\n🎯 {re.findall(r'#([^#]*?)#',narrated_objective)[0]}"
 game_log_dictionary[0]["starting_narration"] = starting_narration
 

models.py

@@ -1,11 +1,51 @@
 """Load models to use them as a narrator and a common-sense oracle in the PAYADOR pipeline."""
 import google.generativeai as genai
 import requests
+import replicate
 import os
+from dotenv import load_dotenv
 
+load_dotenv()
+
+def get_llm(model_name: str = "gemini-1.5-flash") -> object:
+
+    google_models = ["gemini-1.0-pro", "gemini-1.5-pro", "gemini-1.5-flash"]
+    replicate_models = ["meta/meta-llama-3-70b", "meta/meta-llama-3-70b-instruct"]
+
+    model = None
+
+    if model_name in google_models:
+        model = GeminiModel(API_key="GOOGLE_API_KEY", model_name=model_name)
+    elif model_name in replicate_models:
+        model = ReplicateModel(API_key="REPLICATE_API_TOKEN", model_name=model_name)
+
+    return model
+
+
+class ReplicateModel():
+    def __init__ (self, API_key:str, model_name:str = "meta/meta-llama-3-70b-instruct") -> None:
+        self.temperature = 0.7
+        self.model_name = model_name
+
+    def prompt_model(self,system_msg: str, user_msg:str) -> str:
+        """Prompt the Replicate model."""
+
+        system_instructions = f"<|begin_of_text|><|start_header_id|>system<|end_header_id|>\n\n{system_msg}<|eot_id|><|start_header_id|>user<|end_header_id|>"
+
+        input = {
+            "top_p": 0.1,
+            "min_tokens": 0,
+            "temperature": self.temperature,
+            "prompt": user_msg,
+            "prompt_template": system_instructions + "\n\n{prompt}<|eot_id|><|start_header_id|>assistant<|end_header_id|>\n\n",
+        }
+
+        output = replicate.run(self.model_name,input=input)
+        
+        return "".join(output)
 
 class GeminiModel():
-    def __init__ (self, api_key_file:str, model_name:str = "gemini-1.0-pro") -> None:
+    def __init__ (self, API_key:str, model_name:str = "gemini-1.0-pro") -> None:
         """"Initialize the Gemini model using an API key."""
         self.safety_settings = [
             {
@@ -29,27 +69,9 @@ class GeminiModel():
                 "threshold": "BLOCK_NONE",
             },
         ]
-        genai.configure(api_key=os.getenv(api_key_file))
+        genai.configure(api_key=os.getenv("GOOGLE_API_KEY"))
         self.model = genai.GenerativeModel(model_name)
 
-    def prompt_model(self,prompt: str) -> str:
+    def prompt_model(self,system_msg: str, user_msg:str) -> str:
         """Prompt the Gemini model."""
-        return self.model.generate_content(prompt, safety_settings=self.safety_settings).text
-
-
-def prompt_HF_API (prompt: str, model: str = "microsoft/Phi-3-mini-4k-instruct", api_key_file: str = "HF_API_key"):
-    API_URL = f"https://api-inference.huggingface.co/models/{model}"
-
-    headers = {"Authorization": f"Bearer {get_api_key(api_key_file)}"}
-    payload = {"inputs": prompt}
-
-    output = requests.post(API_URL, headers=headers, json=payload).json()
-    
-    return output[0]["generated_text"]
-
-def get_api_key(path: str) -> str:
-    """Load an API key from path."""
-    key = ""
-    with open(path) as f:
-        key = f.readline()
-    return key
+        return self.model.generate_content(system_msg + "\n\n" + user_msg, safety_settings=self.safety_settings).text

prompts.py

@@ -1,117 +1,124 @@
-def prompt_describe_objective (objective, language:str = 'en') -> str:
-    prompt = ""
+def prompt_describe_objective (objective, language:str = 'en'):
+    system_msg = ""
+    user_msg = ""
 
     if language == 'es':
-        prompt = prompt_describe_objective_spanish(objective)
+        system_msg, user_msg = prompt_describe_objective_spanish(objective)
     else:
-        prompt = prompt_describe_objective_english(objective)
+        system_msg, user_msg = prompt_describe_objective_english(objective)
     
-    return prompt
+    return system_msg, user_msg 
 
-def prompt_describe_objective_english (objective) -> str:
+def prompt_describe_objective_english (objective):
+
+    system_msg = """You have to provide an alternative way to narrate the objective given to you. You should always use simple language. 
+    
+    Always put your generated narration between # characters. For example: # You have to get the <key> # or # You have to reach the <castle> #"""
 
-    prompt = "Please, provide an alternative way to narrate the following objective, using simple language: "
     first_component_class = objective[0].__class__.__name__
     second_component_class = objective[1].__class__.__name__
+    user_msg = ""
 
     if first_component_class == "Character" and second_component_class == "Location":
-        prompt+= f'"You have to go to <{objective[1].name}>."'
+        user_msg =  f'The objective to narrate in an alterative way is "You have to go to <{objective[1].name}>."'
     elif first_component_class == "Character" and second_component_class == "Item":
-        prompt+= f'"<{objective[0].name}> has to get the item <{objective[1].name}>."'
+        user_msg = f'The objective to narrate in an alterative way is "<{objective[0].name}> has to get the item <{objective[1].name}>."'
     elif first_component_class == "Item" and second_component_class == "Location":
-        prompt+= f'"You have to leave item <{objective[0].name}> in place <{objective[1].name}>."'
+        user_msg = f'The objective to narrate in an alterative way is "You have to leave item <{objective[0].name}> in place <{objective[1].name}>."'
     elif first_component_class == "Character" and second_component_class == "Character":
-        prompt+= f'"<{objective[0].name}> has to find <{objective[1].name}>."'
+        user_msg = f'The objective to narrate in an alterative way is "<{objective[0].name}> has to find <{objective[1].name}>."'
 
-    prompt+="\nPut your generated narration between # characters. For example: # You have to get the <key> # or # You have to reach the <castle> #"
+    return system_msg, user_msg 
 
-    return prompt
+def prompt_describe_objective_spanish (objective):
 
-def prompt_describe_objective_spanish (objective) -> str:
+    system_msg = """Tienes que dar una forma alternativa de narrar el objetivo que se te dará. Siempre usa lenguaje simple. 
+    
+    Pon siempre tu narración generada entre caracteres #. Por ejemplo: # Tienes que conseguir la <llave> # o # Tienes que llegaral <Castillo> #"""
 
-    prompt = "Por favor, dame una forma alternativa de narrar el siguiente objetivo, usando lenguaje simple: "
     first_component_class = objective[0].__class__.__name__
     second_component_class = objective[1].__class__.__name__
+    user_msg = ""
 
     if first_component_class == "Character" and second_component_class == "Location":
-        prompt+= f'"Tienes que ir a <{objective[1].name}>."'
+        user_msg = f'El objetivo a decir de forma alternativa es "Tienes que ir a <{objective[1].name}>."'
     elif first_component_class == "Character" and second_component_class == "Item":
-        prompt+= f'"<{objective[0].name}> tiene que conseguir el objeto <{objective[1].name}>."'
+        user_msg = f'El objetivo a decir de forma alternativa es "<{objective[0].name}> tiene que conseguir el objeto <{objective[1].name}>."'
     elif first_component_class == "Item" and second_component_class == "Location":
-        prompt+= f'"Tienes que dejar el objeto <{objective[0].name}> en el lugar <{objective[1].name}>."'
+        user_msg = f'El objetivo a decir de forma alternativa es "Tienes que dejar el objeto <{objective[0].name}> en el lugar <{objective[1].name}>."'
     elif first_component_class == "Character" and second_component_class == "Character":
-        prompt+= f'"<{objective[0].name}> tiene que encontrar a <{objective[1].name}>."'
-
-    prompt+="\nPon tu narración generada entre caracteres #. Por ejemplo: # Tienes que conseguir la <llave> # o # Tienes que llegaral <Castillo> #"
+        user_msg = f'El objetivo a decir de forma alternativa es "<{objective[0].name}> tiene que encontrar a <{objective[1].name}>."'
 
-    return prompt
+    return system_msg, user_msg
 
-def prompt_narrate_current_scene (world_state: str, previous_narrations: 'list[str]', language: str = 'en', starting_scene: bool = False) -> str:
-    prompt = ""
+def prompt_narrate_current_scene (world_state: str, previous_narrations: 'list[str]', language: str = 'en', starting_scene: bool = False):
+    system_msg = ""
+    user_msg = ""
 
     if language == 'es':
-        prompt = prompt_narrate_current_scene_spanish(world_state, previous_narrations, starting_scene)
+        system_msg, user_msg = prompt_narrate_current_scene_spanish(world_state, previous_narrations, starting_scene)
     else:
-        prompt = prompt_narrate_current_scene_english(world_state, previous_narrations, starting_scene)
+        system_msg, user_msg = prompt_narrate_current_scene_english(world_state, previous_narrations, starting_scene)
 
 
-    return prompt
+    return system_msg, user_msg
 
-def prompt_narrate_current_scene_english (world_state: str, previous_narrations: 'list[str]', starting_scene: bool = False) -> str:
+def prompt_narrate_current_scene_english (world_state: str, previous_narrations: 'list[str]', starting_scene: bool = False):
 
-    prompt = "You are a storyteller. Take the following state of the world and narrate it in a few sentences. Be careful not to include details that contradict the current state of the world or that move the story forward. Also, try to use simple sentences and do not overuse poetic language.\n"
+    system_msg = "You are a storyteller. Take the state of the world given to you and narrate it in a few sentences. Be careful not to include details that contradict the current state of the world or that move the story forward. Also, try to use simple sentences and do not overuse poetic language"
     
     if starting_scene:
-        prompt += "\nTake into account that this is the first scene in the story: introduce the main character, creating a small background story and why that character is in that specific location.\n"
+        system_msg += "\nTake into account that this is the first scene in the story: introduce the main character, creating a small background story and why that character is in that specific location.\n"
     elif len(previous_narrations)==0:
-        prompt += "Take into account that the player already knows what the main character looks like, so do not mention anything about that. However, it is the first time the player visits this place, so make sure to describe it exhaustively."
+        system_msg += "Take into account that the player already knows what the main character looks like, so do not mention anything about that. However, it is the first time the player visits this place, so make sure to describe it exhaustively."
     else:
-        prompt += "Take into account that the player already knows what the main character looks like, so do not mention anything about that. Additionally, it is not the first time the player visits this place. Next I’ll give you some previous narrations of this same location (from oldest to newest) so you can be sure to not repeat the same details again:\n"
+        system_msg += "Take into account that the player already knows what the main character looks like, so do not mention anything about that. Additionally, it is not the first time the player visits this place. Next I’ll give you some previous narrations of this same location (from oldest to newest) so you can be sure to not repeat the same details again:\n"
         for narration in previous_narrations:
-            prompt+=f'- {narration}\n'
+            system_msg+=f'- {narration}\n'
 
-    prompt+= "\nRemember: you are talking to the player, describing what his or her character has and what he or she can see or feel."
+    system_msg+= "\nRemember: you are talking to the player, describing what his or her character has and what he or she can see or feel."
 
-    prompt +=f"""This is the state of the world at the moment:
+    user_msg =f"""This is the state of the world at the moment:
     {world_state}
     """
 
-    return prompt
+    return system_msg, user_msg
 
-def prompt_narrate_current_scene_spanish (world_state: str, previous_narrations: 'list[str]', starting_scene: bool = False) -> str:
+def prompt_narrate_current_scene_spanish (world_state: str, previous_narrations: 'list[str]', starting_scene: bool = False):
     
-    prompt = f"""Eres un narrador. Toma el siguiente estado del mundo y nárralo en unas pocas oraciones. Ten cuidado de no incluir detalles que contradigan el estado del mundo actual, o que hagan avanzar la historia. Además, intenta usar oraciones simples, sin abusar del lenguaje poético."""
+    system_msg = f"""Eres un narrador. Toma el estado del mundo que se te de y nárralo en unas pocas oraciones. Ten cuidado de no incluir detalles que contradigan el estado del mundo actual, o que hagan avanzar la historia. Además, intenta usar oraciones simples, sin abusar del lenguaje poético."""
     
     if starting_scene:
-        prompt += "\nTen en cuenta que esta es la primera escena en la historia narrada: presenta al personaje del jugador, creando un pequeño trasfondo y por qué este personaje está en ese lugar específicamente. Puede usar las pequeñas descripciones presentes en el estado del mundo. Es importante que menciones todos los componentes que hay en este lugar. Sin embargo, es mejor si no describes cada componente: basta con que los menciones con una mínima descripción poco específica. Es muy importante que nombres los lugares a los que puede acceder el jugador desde esta posición. \n"
+        system_msg += "\nTen en cuenta que esta es la primera escena en la historia narrada: presenta al personaje del jugador, creando un pequeño trasfondo y por qué este personaje está en ese lugar específicamente. Puede usar las pequeñas descripciones presentes en el estado del mundo. Es importante que menciones todos los componentes que hay en este lugar. Sin embargo, es mejor si no describes cada componente: basta con que los menciones con una mínima descripción poco específica. Es muy importante que nombres los lugares a los que puede acceder el jugador desde esta posición. \n"
     elif len(previous_narrations)==0:
-        prompt += "Ten en cuenta que el jugador ya conoce a su personaje, y cómo se ve, así que no menciones nada sobre esto. Sin embargo, es la primera vez que el jugador visita este lugar, así que describelo. Es importante que menciones todos los componentes que hay en este lugar. Sin embargo, es mejor si no describes cada componente: basta con que los menciones con una mínima descripción poco específica. Es muy importante que nombres los lugares a los que puede acceder el jugador desde esta posición. \n"
+        system_msg += "Ten en cuenta que el jugador ya conoce a su personaje, y cómo se ve, así que no menciones nada sobre esto. Sin embargo, es la primera vez que el jugador visita este lugar, así que describelo. Es importante que menciones todos los componentes que hay en este lugar. Sin embargo, es mejor si no describes cada componente: basta con que los menciones con una mínima descripción poco específica. Es muy importante que nombres los lugares a los que puede acceder el jugador desde esta posición. \n"
     else:
-        prompt += "Ten en cuenta que el jugador ya conoce a su personaje, y cómo se ve, así que no menciones nada sobre esto. Además, no es la primera vez que el jugador visita este lugar. A continuación te daré algunas narraciones previas de este mismo lugar (de la más antigua a la más nueva), así te puedes asegurar de no repetir los mismos detalles de nuevo:\n"
+        system_msg += "Ten en cuenta que el jugador ya conoce a su personaje, y cómo se ve, así que no menciones nada sobre esto. Además, no es la primera vez que el jugador visita este lugar. A continuación te daré algunas narraciones previas de este mismo lugar (de la más antigua a la más nueva), así te puedes asegurar de no repetir los mismos detalles de nuevo:\n"
         for narration in previous_narrations:
-            prompt+=f'- {narration}\n'
+            system_msg+=f'- {narration}\n'
 
-    prompt+= "\nRecuerda: le estás hablando al jugador, describiendo lo que su personaje tiene y lo que puede sentir o ver."
+    system_msg+= "\nRecuerda: le estás hablando al jugador, describiendo lo que su personaje tiene y lo que puede sentir o ver."
 
-    prompt +=f"""Este es el estado del mundo en este momnto:
+    user_msg = f"""Este es el estado del mundo en este momento:
     {world_state}
     """
 
-    return prompt
+    return system_msg, user_msg
 
-def prompt_world_update (world_state: str, input: str, language: str = 'en') -> str:
-    prompt = ""
+def prompt_world_update (world_state: str, input: str, language: str = 'en'):
+    system_msg = ""
+    user_msg = ""
 
     if language == 'es':
-        prompt = prompt_world_update_spanish(world_state, input)
+        system_msg, user_msg = prompt_world_update_spanish(world_state, input)
     else:
-        prompt = prompt_world_update_english(world_state, input)
+        system_msg, user_msg = prompt_world_update_english(world_state, input)
 
 
-    return prompt
+    return system_msg, user_msg
 
-def prompt_world_update_spanish (world_state: str, input: str) -> str:
-    prompt = f"""Eres un narrador. Estás manejando un mundo ficticio, y el jugador puede interactuar con él. Siguiendo un formato específico, que voy a explicarte más abajo, tu tarea es encontrar los cambios en el mundo a raíz de las acciones del jugador. En específico, tendrás que encontrar qué objetos cambiaron de lugar, qué pasajes entre lugares se desbloquearon y si el jugador se movió de lugar.
+def prompt_world_update_spanish (world_state: str, input: str):
+    system_msg = f"""Eres un narrador. Estás manejando un mundo ficticio, y el jugador puede interactuar con él. Siguiendo un formato específico, que voy a explicarte más abajo, tu tarea es encontrar los cambios en el mundo a raíz de las acciones del jugador. En específico, tendrás que encontrar qué objetos cambiaron de lugar, qué pasajes entre lugares se desbloquearon y si el jugador se movió de lugar.
     
     Aquí hay algunas aclaraciones:
     (A) Presta atención a a la descripción de los componentes y sus capacidades.
@@ -184,17 +191,17 @@ def prompt_world_update_spanish (world_state: str, input: str) -> str:
     - Moved object: None
     - Blocked passages now available: None
     - Your location changed:  None
-    # Respuesta a la pregunta del jugador #
+    # Respuesta a la pregunta del jugador #"""
     
     
-    Ahora, expresa los cambios en el mundo siguiendo el formato pedido, teniendo en cuenta que el jugador ingresó esta entrada "{input}" a partir de este estado del mundo:
+    user_msg = f"""Expresa los cambios en el mundo siguiendo el formato pedido, teniendo en cuenta que el jugador ingresó esta entrada "{input}" a partir de este estado del mundo:
     
     {world_state}"""
 
-    return prompt
+    return system_msg, user_msg
 
-def prompt_world_update_english (world_state: str, input: str) -> str:
-    prompt = f"""You are a storyteller. You are managing a fictional world, and the player can interact with it. Following a specific format, that I will specify below, your task is to find the changes in the world after the actions in the player input. Specifically, you will have to find what objects were moved, which previously blocked passages are now unblocked, and if the player moved to a new place.
+def prompt_world_update_english (world_state: str, input: str):
+    system_msg = f"""You are a storyteller. You are managing a fictional world, and the player can interact with it. Following a specific format, that I will specify below, your task is to find the changes in the world after the actions in the player input. Specifically, you will have to find what objects were moved, which previously blocked passages are now unblocked, and if the player moved to a new place.
        
     Here are some clarifications:
     (A) Pay attention  to the description of the components and their capabilities.
@@ -267,11 +274,11 @@ def prompt_world_update_english (world_state: str, input: str) -> str:
     - Moved object: None
     - Blocked passages now available: None
     - Your location changed:  None
-    # Answer to the player's question #
+    # Answer to the player's question #"""
     
     
-    Now, give the changes in the world following the specified format, after this player input "{input}" on this world state:
+    user_msg = f"""Give the changes in the world following the specified format, after this player input "{input}" on this world state:
     
     {world_state}"""
 
-    return prompt
+    return system_msg, user_msg