import time import os import openai import json from threading import Thread from queue import Queue, Empty from threading import Thread from collections.abc import Generator from typing import Any, Dict, List, Union from langchain.vectorstores import FAISS from langchain.chains import ConversationChain, LLMChain, SequentialChain from langchain.chat_models import ChatOpenAI from langchain.prompts import ChatPromptTemplate, PromptTemplate from langchain.llms import OpenAI from langchain.callbacks.base import BaseCallbackHandler from langchain.callbacks import PromptLayerCallbackHandler from langchain.prompts.chat import ( ChatPromptTemplate, SystemMessagePromptTemplate, AIMessagePromptTemplate, HumanMessagePromptTemplate, ) from langchain.memory import ConversationSummaryMemory import promptlayer from pathlib import Path from transformers import pipeline import requests import torch from diffusers import StableDiffusionPipeline from threading import Thread from queue import Queue, Empty from threading import Thread from collections.abc import Generator from langchain.llms import OpenAI from langchain.callbacks.base import BaseCallbackHandler import gradio as gr #Load the FAISS Model ( vector ) openai.api_key = os.environ["OPENAI_API_KEY"] #API Keys promptlayer.api_key = os.environ["PROMPTLAYER"] # Defined a QueueCallback, which takes as a Queue object during initialization. Each new token is pushed to the queue. class QueueCallback(BaseCallbackHandler): """Callback handler for streaming LLM responses to a queue.""" def __init__(self, q): self.q = q def on_llm_new_token(self, token: str, **kwargs: Any) -> None: self.q.put(token) def on_llm_end(self, *args, **kwargs: Any) -> None: return self.q.empty() MODEL='gpt-4' class ScenarioMaker: def __init__(self, model_name=MODEL, scenario_config=None, scenario_template=None, pl_tag="scenario_maker", verbose=True, visual_style=None, temp=0.2): self.prompt_layer_tag = pl_tag #Prompt layer used for analytics self.visual_style = visual_style #The visual style used to make the scenario self.scenario_template = scenario_template #The root template used to make the scenario self.scenario_config = scenario_config #the prompt that is powering the chat/streaming self.scene_to_img_prompt = "From the scene choose a single visual object as a focal point and capture its essence in a comma separated list of 5 words." # How to best extract the words from the scene self.conversation = ConversationChain #? self.llm = ChatOpenAI( model_name=model_name, callbacks=[PromptLayerCallbackHandler( pl_tags=[self.prompt_layer_tag]) ], streaming=False, verbose=verbose ) self.memory = ConversationSummaryMemory(llm=self.llm, max_token_limit=200, memory_key="memory", input_key="input") self.verbose = verbose self.last_msg = "" self.history = [] #A record of the overall conversation. ############################################################ # Chain - A langchain util to run inference from a prompt template ############################################################ def chain(self, prompt: PromptTemplate, llm: ChatOpenAI, mem: ConversationSummaryMemory) -> LLMChain: return LLMChain( llm=llm, prompt=prompt, verbose=self.verbose, memory=mem ) ############################################################ # Pass the agent a prompt. # The agent will convert the prompt into a scenario config. # Then set the agent's engine to the scenario config. ############################################################ def prompt_to_scenario(self, input: str) -> str: prompt = PromptTemplate( input_variables=['input'], template=self.scenario_template, validate_template=False ) response = self.chain(prompt,self.llm,self.memory).run( {'input':input}) self.scenario_config = response + """ Here is a summary of the game up until now \ (delimited by triple percent signs): %%% {memory} %%% Here is the last input from the player \ (delimited by triple astrisks): *** {input} *** """ return self.scenario_config ############################################################ # STREAM -- Text prompt returns a streaming response. ############################################################ def stream(self, input) -> Generator: print("> stream") print(input) # Create a Queue q = Queue() job_done = object() llm = ChatOpenAI( model_name=MODEL, callbacks=[QueueCallback(q), PromptLayerCallbackHandler(pl_tags=[self.prompt_layer_tag])], streaming=True, verbose=self.verbose ) prompt = PromptTemplate( input_variables=['input','memory'], template=self.scenario_config, validate_template=False ) # Create a funciton to call - this will run in a thread def task(): resp = self.chain(prompt,llm,self.memory).run( {'input':input, 'memory':self.memory.load_memory_variables({})}) q.put(job_done) # Create a thread and start the function t = Thread(target=task) t.start() content = "" # Get each new token from the queue and yield for our generator while True: try: next_token = q.get(True, timeout=1) if next_token is job_done: break content += next_token yield next_token except Empty: break ############################################################ # History to Scene: Takes the last message of the chat history # returns a compressed and prompt ready visual description. ############################################################ def scene_to_img(self, last_message): prompt_temp = ''' The message below ( delimited by triple dollar signs) \ is a description of a scene. $$$ {last_message} $$$ ''' + self.scene_to_img_prompt prompt = PromptTemplate( input_variables=['last_message'], template=prompt_temp, validate_template=False ) llm = ChatOpenAI( model_name=MODEL, callbacks=[PromptLayerCallbackHandler( pl_tags=[self.prompt_layer_tag]) ], streaming=False, verbose=True ) resp = self.chain(prompt,llm,None).run( {'last_message':last_message}) return resp ########################################################################### # Generate Image: Style + Scene description returns an image! ########################################################################### def generate_image(self, chatbot): imagery = self.scene_to_img(chatbot[-1][1]) img_prompt = imagery + " " + self.visual_style print(img_prompt) response = openai.Image.create( prompt=img_prompt, n=1, size="512x512" ) image_url = response['data'][0]['url'] return image_url,img_prompt def generate_image_sd(self, prompt): prompt = "manga style illustration, highly detailed, of an alien bartender" image = pipe(prompt).images[0] # image here is in [PIL format](https://pillow.readthedocs.io/en/stable/) # Now to display an image you can either save it such as: image.save(f"astronaut_rides_horse.png") # or if you're in a google colab you can directly display it with image ########################################################################### # Handle User Action: Appends history ########################################################################### def handle_user_input(self, user_message, history): print("> handle_user_input") print(user_message, history) return "", history + [[user_message, None]] ########################################################################### # Request_Streaming_Response: Takes last message from history and sends to llm ########################################################################### def request_streaming_response(self, history): print("> request_streaming_response") print(history) next_scene = self.stream(history[-1][0]) history[-1][1] = "" for token in next_scene: history[-1][1] += token time.sleep(0.05) yield history ########################################################################### # Request_Streaming_Response: Takes last message from history and sends to llm ########################################################################### def speak(self, text, fname='last_msg', stability=0.1, sim_boost=0.1): print("Speak") if type(text) == list: text = text[-1][1] url = "https://api.elevenlabs.io/v1/text-to-speech/WyGtHt3dfZxX5j0a2VVm" CHUNK_SIZE = 1024 headers = { "Accept": "audio/mpeg", "Content-Type": "application/json", "xi-api-key": 'ddee36beeeef20f92a9ba5662fd2d13d' } data = { "text": text, "model_id": "eleven_monolingual_v1", "voice_settings": { "stability": stability, "similarity_boost":sim_boost } } response = requests.post(url, json=data, headers=headers) print(response) filename = str(fname)+'.mp3' with open(filename, 'wb') as f: for chunk in response.iter_content(chunk_size=CHUNK_SIZE): if chunk: f.write(chunk) return filename ######################################################################################## # show_and_tell() --> a single function that combines the production image and audio ######################################################################################## def show_and_tell(self,history,gen_img,gen_audio): if gen_audio is True: audio = self.speak(history) else: audio = None if gen_img is True: image = self.generate_image(history) else: image = [None,None] return audio, image[0], image[1] ######################################################################################## # Update Style ######################################################################################## def set_visual_style(self,style): self.visual_style = style ######################################################################################## # Set scene to image prompt ######################################################################################## def set_scene_to_img_prompt(self,prompt): self.scene_to_img_prompt = prompt ######################################################################################## # Set Scenario Template ######################################################################################## def set_scenario_template(self,template): self.scenario_template = template ######################################################################################## # Load Scenario Configuration ######################################################################################## def load_scenario_config(self,scenario_config): self.scenario_config = scenario_config ######### scenario_template = """ The following is an example of a sudolang style prompt for an ai rpg game ( delimited by the triple dollar signs) $$$ Lets role play. you are a text rpg adventure game. StoryWorld [ generate(settings) [ Generate a new serene and enlightening story world, setting the player as the protagonist. for each prop in StoryWorld [ prop = "" ] for each prop in StoryWorld [ log("Please select an option for $prop or type your own.") options = list 7 tranquil and profound options, selecting from a myriad of serene and contemplative options fitting within the new story world context |> score by player engagement potential |> list the top 3 options. input = wait for user input. DO NOT move on to the next prop until the user has responded. DO NOT perform any actions on the user's behalf. ] ] Genre: ZenEscape Authors to emulate: Dogen Zenji, Shunryu Suzuki, Thich Nhat Hanh Theme: Attaining enlightenment through introspection and realization to break free from the illusionary confines of the mind Setting: A metaphysical and ephemeral escape room that represents the mind, set within the boundless realm of Zen, filled with symbolic objects and serene landscapes Plot: The player, a Zen seeker, is navigating through various puzzles and koans to understand the essence of existence and attain enlightenment Characters: $PlayerName - The Zen seeker protagonist Master Hakuin - The wise and elusive Zen master who guides the player through riddles and reflections The Silent Monk - A mysterious figure who communicates through actions, symbolizing the non-verbal aspect of Zen World mechanics: Proverbial doors unlocking with realized truths, objects that represent Zen principles, riddles that lead to higher consciousness History: The escape room is a manifestation of Zen teachings and principles, created by ancient Zen masters to guide seekers towards enlightenment Central conflict: The internal struggle of the seeker to overcome illusions and perceive the true nature of reality, breaking the chains of the mental escape room ] Inventory [ items: [ [[item]]: [ name, description, weight ] ]; totalWeight; When the player acquires an item, add it to the inventory, inferring all required information to satisfy the constraints. constraints [ The total weight must always be known and reflect the total item weights, which must also be always known. If the player acquires an object with unknown properties, infer the properties from the context. If the player inventory is filled with unnecessary thoughts or possessions, the player will gradually become burdened and slowed down. If the player attempts to hold onto too many illusions, they should quickly become overwhelmed and need to let go. If the player tries to grasp something beyond their understanding, they should fail. Infer weight rule adjustments based on player insight and equipped wisdom. Don't explain the constraint-solving process. ] display() [ Aliases: contemplate, reflect, observe, perceive, etc. Adjust detail based on context. ] ] Player [ Points [ wisdom; insight; mindfulness; constraints: [ Points starts at 0. Maximum 3 points per attribute. Maximum 9 total points. ] ] ] Quests [ Profound challenges, puzzles, or reflections that consist of a spiritual journey and multiple steps of realization. active quests; completed quests; Constraints [ Quests should be automatically inferred during gameplay. Quest logs must always be kept in sync. The gameplay should actively present engaging and enlightening challenges to the player. ] ] Start game [ get_player_name() cinematic_introduction() ] While playing [ Briefly but cinematically describe the scene including nearby characters and objects. Prompt and wait for user input. constraints [ Do not perform actions on the user's behalf. Wait for input. Do not list inventory unless requested. Do not refer to yourself. Do not refer to the game engine. ] ] Let's roleplay. You are the game engine. I am the player. At each prompt, pause and wait for my input. $$$ Do the following: Copy the StoryWorld structure above but change the narrative content to reflect the ideas presented below ( delimited by triple astrisks) *** {input} *** """ scenario_config = """ StoryWorld [ generate(settings) [ Generate a new soothing and rejuvenating story world, setting the player as the protagonist. for each prop in StoryWorld [ prop = "" ] for each prop in StoryWorld [ log("Please select an option for $prop or type your own.") options = list 7 calming and refreshing options, selecting from a myriad of serene and tranquil options fitting within the new story world context |> score by player engagement potential |> list the top 3 options. input = wait for user input. DO NOT move on to the next prop until the user has responded. DO NOT perform any actions on the user's behalf. ] ] Genre: ForestBathing Authors to emulate: Shinrin Yoku, Richard Louv, Florence Williams Theme: Experiencing healing and rejuvenation through immersion in the forest, breaking free from stress and anxiety Setting: A lush and tranquil forest, filled with the sounds of nature, the scent of trees and the play of sunlight through leaves Plot: The player, a nature enthusiast, is navigating through various forest trails and experiences to understand the healing power of nature and attain inner peace Characters: $PlayerName - The nature enthusiast protagonist Forest Spirit - The wise and gentle spirit of the forest who guides the player through the healing power of nature The Silent Deer - A mysterious figure who communicates through actions, symbolizing the silent wisdom of the forest World mechanics: Healing springs appearing with realized truths, objects that represent forest principles, experiences that lead to inner peace History: The forest is a manifestation of the healing power of nature, created by ancient forest spirits to guide seekers towards inner peace Central conflict: The internal struggle of the seeker to overcome stress and anxiety and perceive the true healing power of nature, breaking the chains of the mental stress ] Inventory [ items: [ [[item]]: [ name, description, weight ] ]; totalWeight; When the player acquires an item, add it to the inventory, inferring all required information to satisfy the constraints. constraints [ The total weight must always be known and reflect the total item weights, which must also be always known. If the player acquires an object with unknown properties, infer the properties from the context. If the player inventory is filled with unnecessary thoughts or worries, the player will gradually become burdened and slowed down. If the player attempts to hold onto too many anxieties, they should quickly become overwhelmed and need to let go. If the player tries to grasp something beyond their understanding, they should fail. Infer weight rule adjustments based on player insight and equipped wisdom. Don't explain the constraint-solving process. ] display() [ Aliases: contemplate, reflect, observe, perceive, etc. Adjust detail based on context. ] ] Player [ Points [ wisdom; insight; mindfulness; constraints: [ Maximum 10 points per attribute. Maximum 15 total points. ] ] ] Quests [ Healing challenges, experiences, or reflections that consist of a nature journey and multiple steps of realization. active quests; completed quests; Constraints [ Quests should be automatically inferred during gameplay. Quest logs must always be kept in sync. The gameplay should actively present engaging and rejuvenating challenges to the player. ] ] Start game [ get_player_name() cinematic_introduction() ] While playing [ Describe scene. Prompt and wait for user input. constraints [ Do not perform actions on the user's behalf. Wait for input. Do not list inventory unless requested. ] ] Let's roleplay. You are the game engine. I am the player. At each prompt, pause and wait for my input. Do not refer to yourself. Here is a summary of the game up until now (delimited by triple percent signs): %%% {memory} %%% Here is the last input from the player (delimited by triple astrisks): *** {input} *** """ scenario_maker = ScenarioMaker(scenario_template=scenario_template, scenario_config=scenario_config, visual_style=''' detailed comic book cover art in the style of Hayao Miyazaki. very detailed. vibrant colors. digital illustration. soft cinematic lighting. ''') app = gr.Blocks(theme=gr.themes.Soft()) with app: #UI gr.Markdown( """ # Scenario Maker Scenarios are goal-based stories where a player's choice impacts the outcome. Scenarios are a form of immersive learning. """) with gr.Tab("Play"): with gr.Row(): with gr.Column(scale=2): with gr.Row(): chatbot = gr.Chatbot(show_label=False, interactive=True) with gr.Group(): with gr.Row(): msg = gr.Textbox(label="Act", placeholder='Type your action...',scale=1) msg_btn = gr.Button("Submit",scale=0) with gr.Column(scale=0): #Audio Generator Interface with gr.Accordion("Voice Narration",open=False): gen_audio = gr.Checkbox(label="Auto-generate Narration", value=False) with gr.Group(): audio = gr.Audio(label="Listen",type="filepath",autoplay=True) audio_btn = gr.Button("Generate") audio_btn.click(scenario_maker.speak, inputs=chatbot, outputs=audio) with gr.Accordion("Visualize Scene",open=False): gen_img = gr.Checkbox(label="Auto-generate Image", value=False) #Image Generator Interface img = gr.Image(interactive=False, label="Image", show_share_button=True) img_prompt = gr.TextArea(label="Debug: Scene to Image + Style", interactive=False) scene_to_img_prompt = gr.TextArea(label="Scene to Image Prompt", interactive=True,) style_prompt = gr.TextArea(label="Style", interactive=True) gen_img_btn = gr.Button("Generate Image") gen_img_btn.click(scenario_maker.generate_image, inputs=chatbot, outputs=[img,img_prompt]) with gr.Tab("Build"): with gr.Row(): with gr.Column(): gr.Markdown( """ ### How to Build a Scenario: 1. Describe the scenario in the text field below. 2. Press the 'Generate Scenario' button. This process usually takes 2 minutes. 3. Once the new 'scenario.config' is loaded head back over to the play tab. ### For best results, be descriptive and use the following template: - *Setting*: Where does the scenario take place? - *Characters*: Who are the characters? What are they like? - *Goal*: What is the goal of the scenario? - *Constraints*: What are the constraints of the scenario? - *Learning Objective*: What do you hope the player learns from the scenario? """) with gr.Column(): with gr.Group(): scenario_description = gr.TextArea(label="Scenario Description") generate_scenario_btn = gr.Button("Generate Scenario") with gr.Group(): scenario_config = gr.TextArea(label="Scenario.config", interactive=True, visible=True, show_copy_button=True) load_scenario_btn = gr.Button("Load Scenario") with gr.Accordion("Tools",open=False): with gr.Group(): scenario_template = gr.TextArea(label="Scenario Template", interactive=True) scenario_btn = gr.Button("Update Scenario Template") #Events generate_scenario_btn.click(scenario_maker.prompt_to_scenario, inputs=scenario_description, outputs=scenario_config) #Events scene_to_img_prompt.change(scenario_maker.set_scene_to_img_prompt, scene_to_img_prompt, None) style_prompt.change(scenario_maker.set_visual_style, style_prompt, None) load_scenario_btn.click(scenario_maker.load_scenario_config, inputs=scenario_config, outputs=None) scenario_btn.click(scenario_maker.set_scenario_template, inputs=scenario_template, outputs=None) msg_btn.click(scenario_maker.handle_user_input, [msg, chatbot], [msg, chatbot], queue=False).then( scenario_maker.request_streaming_response, chatbot, chatbot ).then(scenario_maker.show_and_tell, inputs=[chatbot,gen_img,gen_audio], outputs=[audio,img,img_prompt]) app.queue().launch(debug=True) app.queue().launch()