Talk to GPT-4 and create a story together.
import time
import urllib

from dotenv import load_dotenv
import requests
from flask import jsonify
import os

load_dotenv()

def get_meeting_token(room_name, daily_api_key, token_expiry):
    api_path = os.getenv('DAILY_API_PATH') or 'https://api.daily.co/v1'

    if not token_expiry:
        token_expiry = time.time() + 600
    res = requests.post(f'{api_path}/meeting-tokens',
                        headers={'Authorization': f'Bearer {daily_api_key}'},
                        json={'properties': {'room_name': room_name, 'is_owner': True, 'exp': token_expiry}})
    if res.status_code != 200:
        return jsonify({'error': 'Unable to create meeting token', 'detail': res.text}), 500
    meeting_token = res.json()['token']
    return meeting_token


def get_room_name(room_url):
    return urllib.parse.urlparse(room_url).path[1:]
auth.py
daily-demos/llm-talk
from services.azure_ai_service import AzureAIService
from services.google_ai_service import GoogleAIService
from services.open_ai_service import OpenAIService
from services.huggingface_ai_service import HuggingFaceAIService
from services.mock_ai_service import MockAIService

services = {
    "azure": AzureAIService,
    "google": GoogleAIService,
    "openai": OpenAIService,
    "huggingface": HuggingFaceAIService,
    "mock": MockAIService
}

config.py
daily-demos/llm-talk
import os
import requests
import subprocess
import time

from flask import Flask, jsonify
from flask_cors import CORS
from auth import get_meeting_token

app = Flask(__name__)
CORS(app)

@app.route('/spin-up-bot', methods=['POST'])
def spin_up_bot():
    daily_api_key = os.getenv('DAILY_API_KEY')
    api_path = os.getenv('DAILY_API_PATH') or 'https://api.daily.co/v1'

    timeout = int(os.getenv("BOT_MAX_DURATION") or 300)
    exp = time.time() + timeout
    res = requests.post(f'{api_path}/rooms',
                        headers={'Authorization': f'Bearer {daily_api_key}'},
                        json={
                            'properties': {
                                'exp': exp,
                                'enable_chat':True,
                                'enable_emoji_reactions': True,
                                'eject_at_room_exp': True,
                                'enable_prejoin_ui': False,
                            }
                        })
    if res.status_code != 200:
        return jsonify({'error': 'Unable to create room', 'status_code': res.status_code, 'text': res.text}), 500
    room_url = res.json()['url']
    room_name = res.json()['name']

    meeting_token = get_meeting_token(room_name, daily_api_key, exp)

    proc = subprocess.Popen([f'python ./daily-llm.py -u {room_url} -t {meeting_token}'], shell=True, bufsize=1)

    # Don't return until the bot has joined the room, but wait for at most 2 seconds.
    attempts = 0
    while attempts < 20:
        time.sleep(0.1)
        attempts += 1
        res = requests.get(f"{api_path}/rooms/{room_name}/get-session-data", headers={'Authorization': f'Bearer {daily_api_key}'})
        if res.status_code == 200:
            break
    print(f"Took {attempts} attempts to join room {room_name}")

    return jsonify({'room_url': room_url, 'token': meeting_token}), 200

daily-bot-manager.py
daily-demos/llm-talk
import argparse
import os
import random
import re
import time
from threading import Thread

from daily import EventHandler, CallClient, Daily
from datetime import datetime
from dotenv import load_dotenv

import config

from orchestrator import Orchestrator
from scenes.story_intro_scene import StoryIntroScene
from scenes.start_listening_scene import StartListeningScene
from auth import get_meeting_token, get_room_name

load_dotenv()

class DailyLLM(EventHandler):
    def __init__(
            self,
            room_url=os.getenv("DAILY_URL"),
            token=os.getenv('DAILY_TOKEN'),
            bot_name="Storybot",
            image_style="watercolor illustrated children's book", # keep the generated images to a certain theme, like "golden age of illustration," "talented kid's crayon drawing", etc.
        ):

        # room + bot details
        self.room_url = room_url
        room_name = get_room_name(room_url)
        if token:
            self.token = token
        else:
            self.token = get_meeting_token(room_name, os.getenv("DAILY_API_KEY"))
        self.bot_name = bot_name
        self.image_style = image_style

        duration = os.getenv("BOT_MAX_DURATION")
        if not duration:
            duration = 300
        else:
            duration = int(duration)
        self.expiration = time.time() + duration
        self.story_started = False

        self.finished_talking_at = None


        print(f"{room_url} Joining", self.room_url, "as", self.bot_name, "leaving at", self.expiration, "current time is", time.time())

        self.print_debug(f"expiration: {datetime.utcfromtimestamp(self.expiration).strftime('%Y-%m-%d %H:%M:%S')}")
        self.print_debug(f"now: {datetime.utcfromtimestamp(time.time()).strftime('%Y-%m-%d %H:%M:%S')}")

        self.my_participant_id = None

        self.print_debug("configuring services")
        self.configure_ai_services()
        self.print_debug("configuring daily")
        self.configure_daily()

        self.stop_threads = False
        self.image = None

        self.print_debug("starting camera thread")
        self.camera_thread = Thread(target=self.run_camera)
        self.camera_thread.start()

        self.print_debug("starting orchestrator")
        self.orchestrator = Orchestrator(self, self.mic, self.tts, self.image_gen, self.llm, self.story_id)
        self.orchestrator.enqueue(StoryIntroScene)
        self.orchestrator.enqueue(StartListeningScene)

        self.participant_left = False
        self.transcription = ""
        self.last_fragment_at = None

        try:
            participant_count = len(self.client.participants())
            self.print_debug(f"{participant_count} participants in room")
            while time.time() < self.expiration and not self.participant_left:
                # all handling of incoming transcriptions happens in on_transcription_message
                if self.last_fragment_at is not None and (time.time() > self.last_fragment_at + 5):
                    # They probably stopped talking, but Deepgram didn't give us a complete sentence
                    print(f"Sending transcript due to timeout: {self.transcription}")
                    self.send_transcription()
                time.sleep(1)
        except Exception as e:
            self.print_debug(f"Exception {e}")
        finally:
            self.client.leave()

        self.stop_threads = True
        self.print_debug("Shutting down")
        self.camera_thread.join()
        self.print_debug("camera thread stopped")

    def print_debug(self, s):
        print(f"{self.room_url} {s}", flush=True)

    def configure_ai_services(self):
        self.story_id = hex(random.getrandbits(128))[2:]

        self.tts = config.services[os.getenv("TTS_SERVICE")]()
        self.image_gen = config.services[os.getenv("IMAGE_GEN_SERVICE")]()
        self.llm = config.services[os.getenv("LLM_SERVICE")]()

    def configure_daily(self):
        Daily.init()
        self.client = CallClient(event_handler = self)

        self.mic = Daily.create_microphone_device("mic", sample_rate = 16000, channels = 1)
        self.speaker = Daily.create_speaker_device("speaker", sample_rate = 16000, channels = 1)
        self.camera = Daily.create_camera_device("camera", width = 512, height = 512, color_format="RGB")

        self.image_gen.set_image_style(self.image_style)

        Daily.select_speaker_device("speaker")

        self.client.set_user_name(self.bot_name)
        self.client.join(self.room_url, self.token, completion=self.call_joined)

        self.client.update_inputs({
            "camera": {
                "isEnabled": True,
                "settings": {
                    "deviceId": "camera"
                }
            },
            "microphone": {
                "isEnabled": True,
                "settings": {
                    "deviceId": "mic"
                }
            }
        })

        self.my_participant_id = self.client.participants()['local']['id']

    def call_joined(self, join_data, client_error):
        self.print_debug(f"call_joined: {join_data}, {client_error}")
        self.client.start_transcription()

    def on_participant_updated(self, participant):
        pass

    def on_call_state_updated(self, call_state):
        pass

    def on_inputs_updated(self, inputs):
        pass

    def on_participant_counts_updated(self, participant_counts):
        pass

    def on_active_speaker_changed(self, active_speaker):
        pass

    def on_network_stats_updated(self, network_stats):
        pass

    def on_participant_joined(self, participant):
        self.print_debug(f"on_participant_joined: {participant}")
        self.client.send_app_message({ "event": "story-id", "storyID": self.story_id})
        self.wave()
        time.sleep(2)

        # don't run intro question for newcomers
        if not self.story_started:
            #self.orchestrator.request_intro()
            self.orchestrator.action()
            self.story_started = True

    def on_participant_left(self, participant, reason):
        if len(self.client.participants()) < 2:
            self.print_debug("participant left")
            self.participant_left = True

    def send_transcription(self):
        self.orchestrator.handle_user_speech(self.transcription)
        self.transcription = ""
        self.last_fragment_at = None

    def on_transcription_message(self, message):
        # TODO: This should maybe match on my own participant id instead but I'm in a hurry
        if message['session_id'] != self.my_participant_id:
            if self.orchestrator.started_listening_at:
                # TODO: Check actual transcription timestamp against started_listening_at
                self.transcription += f" {message['text']}"
                # Deepgram says we should look for ending punctuation
                if re.search(r'[\.\!\?]$', self.transcription):
                    print(f"✏️ Sending reply: {self.transcription}")
                    self.send_transcription()
                else:
                    print(f"✏️ Got a transcription fragment: \"{self.transcription}\"")
                    self.last_fragment_at = time.time()


    def set_image(self, image):
        self.image = image

    def run_camera(self):
        try:
            while not self.stop_threads:
                if self.image:
                    self.camera.write_frame(self.image.tobytes())
                time.sleep(1.0 / 15.0) # 15 fps
        except Exception as e:
            self.print_debug(f"Exception {e} in camera thread.")

    def wave(self):
        self.client.send_app_message({
            "event": "sync-emoji-reaction",
            "reaction": {
                "emoji": "👋",
                "room": "main-room",
                "sessionId": "bot",
                "id": time.time(),
            }
        })

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Daily LLM bot")
    parser.add_argument("-u", "--url", type=str, help="URL of the Daily room")
    parser.add_argument("-t", "--token", type=str, help="Token for Daily API")
    parser.add_argument("-b", "--bot-name", type=str, help="Name of the bot")

    args = parser.parse_args()

    url = args.url or os.getenv("DAILY_URL")
    bot_name = args.bot_name or "Storybot"
    token = args.token or None

    app = DailyLLM(url, token, bot_name)

daily-llm.py
daily-demos/llm-talk
import json
import io
import random
import re
import time
from datetime import datetime
from queue import Queue
from queue import Empty

from search import SearchIndexer
from services.mock_ai_service import MockAIService
from scenes.story_page_scene import StoryPageScene
from scenes.story_page_async_scene import StoryPageAsyncScene

from scenes.story_grandma_scene import StoryGrandmaScene
from scenes.stop_listening_scene import StopListeningScene
from scenes.start_listening_scene import StartListeningScene

from threading import Thread

class Orchestrator():
    def __init__(self, image_setter, microphone, ai_tts_service, ai_image_gen_service, ai_llm_service, story_id):
        self.image_setter = image_setter
        self.microphone = microphone

        self.ai_tts_service = ai_tts_service
        self.ai_image_gen_service = ai_image_gen_service
        self.ai_llm_service = ai_llm_service

        self.search_indexer = SearchIndexer(story_id)

        self.tts_getter = None
        self.image_getter = None

        self.messages = [{"role": "system", "content": "You are a storyteller who loves to make up fantastic, fun, and educational stories for children between the ages of 5 and 10 years old. Your stories are full of friendly, magical creatures. Your stories are never scary. Each sentence of your story will become a page in a storybook. Stop after 4-6 sentences and give the child a choice to make that will influence the next part of the story. Once the child responds, start by saying something nice about the choice they made, then include [start] in your response. Include [break] after each sentence of the story. Include [prompt] between the story and the prompt."}]
        self.intro_messages = [{"role": "system", "content": "You are a storyteller who loves to make up fantastic, fun, and educational stories for children between the ages of 5 and 10 years old. Your stories are full of friendly, magical creatures. Your stories are never scary. Begin by asking what a child wants you to tell a story about."}]

        self.llm_response_thread = None

        self.scene_queue = Queue()
        self.stop_threads = False
        self.started_listening_at = None
        self.image_this_time = True
        self.story_sentences = []



    def handle_user_speech(self, user_speech):
        print(f"👅 Handling user speech: {user_speech}")
        if not self.llm_response_thread or not self.llm_response_thread.is_alive():
            self.enqueue(StopListeningScene)
            self.llm_response_thread = Thread(target=self.request_llm_response, args=(user_speech,))
            self.llm_response_thread.start()
        else:
            print("discarding overlapping speech, TODO barge-in")

    def start_listening(self):
        self.started_listening_at = datetime.now()

    def stop_listening(self):
        self.started_listening_at = None

    def listening_since(self):
        return self.started_listening_at

    def request_llm_response(self, user_speech):
        try:
            self.messages.append({"role": "user", "content": user_speech})
            response = self.ai_llm_service.run_llm(self.messages)
            self.handle_llm_response(response)
        except Exception as e:
            print(f"Exception in request_llm_response: {e}")

    def request_intro(self):
        response = self.ai_llm_service.run_llm(self.intro_messages)
        return self.handle_intro(response)

    def handle_intro(self, llm_response):
        # Do this all as one piece, at least for now
        out = ''
        for chunk in llm_response:
            if len(chunk["choices"]) == 0:
                continue
            if "content" in chunk["choices"][0]["delta"]:
                if chunk["choices"][0]["delta"]["content"] != {}: #streaming a content chunk
                    next_chunk = chunk["choices"][0]["delta"]["content"]
                    out += next_chunk
        return self.ai_tts_service.run_tts(out)

    def handle_llm_response(self, llm_response):
        out = ''
        full_response = ''
        prompt_started = False
        for chunk in llm_response:
            if len(chunk["choices"]) == 0:
                continue
            if "content" in chunk["choices"][0]["delta"]:
                if chunk["choices"][0]["delta"]["content"] != {}: #streaming a content chunk
                    next_chunk = chunk["choices"][0]["delta"]["content"]
                    out += next_chunk
                    full_response += next_chunk
                    #print(f"🎬 Out: {out}")

                    #if re.match(r'^.*[.!?]$', out): # it looks like a sentence
                    if prompt_started == False:
                        if re.findall(r'.*\[[sS]tart\].*', out):
                            # Then we have the intro. Send it to speech ASAP
                            out = out.replace("[Start]", "")
                            out = out.replace("[start]", "")

                            out = out.replace("\n", " ")
                            if len(out) > 2:
                                self.enqueue(StoryGrandmaScene, sentence=out)
                            out = ''

                        elif re.findall(r'.*\[[bB]reak\].*', out):
                            # Then it's a page of the story. Get an image too
                            out = out.replace("[Break]", "")
                            out = out.replace("[break]", "")
                            out = out.replace("\n", " ")
                            if len(out) > 2:
                                self.story_sentences.append(out)
                                self.enqueue(StoryPageAsyncScene, sentence=out, image=self.image_this_time, story_sentences=self.story_sentences.copy())

                                self.image_this_time = not self.image_this_time

                            out = ''
                        elif re.findall(r'.*\[[pP]rompt\].*', out):
                            # Then it's question time. Flush any
                            # text here as a story page, then set
                            # the var to get to prompt mode
                            #cb: trying scene now
                            # self.handle_chunk(out)
                            out = out.replace("[Prompt]", "")
                            out = out.replace("[prompt]", "")

                            out = out.replace("\n", " ")
                            if len(out) > 2:
                                self.story_sentences.append(out)
                                self.enqueue(StoryPageAsyncScene, sentence=out, image=self.image_this_time, story_sentences=self.story_sentences.copy())
                                self.image_this_time =  not self.image_this_time

                            out = ''
                    else:
                        # Just get the rest of the output as the prompt
                        pass


                        out = ''
        # get the last one too; it should be the prompt
        print(f"🎬 FINAL Out: {out}")
        self.enqueue(StoryGrandmaScene, sentence=out)
        self.enqueue(StartListeningScene)
        print(f"🎬 FULL MESSAGE: {full_response}")
        self.messages.append({"role": "assistant", "content": full_response})
        self.image_this_time = False

    def handle_chunk(self, chunk):
        #self.llm_history.append(chunk)

        self.tts_getter = Thread(target=self.request_tts, args=(chunk,))
        self.tts_getter.start()
        self.image_getter = Thread(target=self.request_image, args=(chunk,))
        self.image_getter.start()

        self.tts_getter.join()
        self.image_getter.join()

    def request_tts(self, text):
        try:
            audio = self.ai_tts_service.run_tts(text)
            return audio
        except Exception as e:
            print(f"Exception in request_tts: {e}")

    def request_image(self, text):
        try:
            (url, image) = self.ai_image_gen_service.run_image_gen(text)
            return (url, image)
        except Exception as e:
            print(f"Exception in request_image: {e}")
    
    def request_image_description(self, story_sentences):
        if len(self.story_sentences) == 1:
            prompt = f"You are an illustrator for a children's story book. Generate a prompt for DALL-E to create an illustration for the first page of the book, which reads: \"{self.story_sentences[0]}\"\n\n Your response should start with the phrase \"Children's book illustration of\"."
        else:
            prompt = f"You are an illustrator for a children's story book. Here is the story so far:\n\n\"{' '.join(self.story_sentences[:-1])}\"\n\nGenerate a prompt for DALL-E to create an illustration for the next page. Here's the sentence for the next page:\n\n\"{self.story_sentences[-1:][0]}\"\n\n Your response should start with the phrase \"Children's book illustration of\"."
        
        #prompt += " Children's story book illustration"
        print(f"🎆 Prompt: {prompt}")
        msgs = [{"role": "system", "content": prompt}]
        img_response = self.ai_llm_service.run_llm(msgs, stream = False)
        image_prompt = img_response['choices'][0]['message']['content']
        # It comes back wrapped in quotes for some reason
        image_prompt = re.sub(r'^"', '', image_prompt)
        image_prompt = re.sub(r'"$', '', image_prompt)
        print(f"🎆 Resulting image prompt: {image_prompt}")
        return image_prompt
            

    def handle_audio(self, audio):
        print("🏙️ orchestrator handle audio")
        stream = io.BytesIO(audio)

        # Skip RIFF header
        stream.read(44)
        self.microphone.write_frames(stream.read())

    def display_image(self, image):
        if self.image_setter:
            self.image_setter.set_image(image)

    def enqueue(self, scene_type, **kwargs):
        # Take the newly created scene object, call its prepare function, and queue it to perform
        kwargs['orchestrator'] = self
        self.scene_queue.put(scene_type(**kwargs))
        pass

    def action(self):
        # start playing scenes. This allows us to prepare scenes
        # before we actually start playback
        self.playback_thread = Thread(target=self.playback)
        self.playback_thread.start()

    def index_scene_async(self, scene):
        self.search_indexer.index_text(scene.sentence)
        if 'url' in scene.scene_data:
            self.search_indexer.index_image(scene.scene_data['url'])

    def index_scene(self, scene):
        if 'sentence' in scene.__dict__:
            thread = Thread(target=self.index_scene_async, args=(scene,))
            thread.start()
            # we let this thread run unattended

    def playback(self):
        while True:
            try:
                if self.stop_threads:
                    print("🎬 Shutting down playback thread")
                    break
                scene = self.scene_queue.get(block=False)
                if 'sentence' in scene.__dict__:
                    print(f"🎬 Performing scene: {type(scene).__name__}, {scene.sentence}")
                else:
                    print(f"🎬 Performing sentenceless scene: {type(scene).__name__}")
                scene.perform()
                time.sleep(0)
            except Empty:
                time.sleep(0.1)
                continue


if __name__ == "__main__":
    mock_ai_service = MockAIService()
    class MockImageSetter():
        def set_image(self, image):
            print("setting image", image)

    class MockMicrophone():
        def write_frames(self, frames):
            print("writing frames")

    uim = Orchestrator("context", MockImageSetter(), MockMicrophone(), mock_ai_service, mock_ai_service, mock_ai_service, 0)
    uim.handle_user_speech("hello world")


orchestrator.py
daily-demos/llm-talk
from abc import abstractmethod
import threading
from PIL import Image
import os

class Scene:
	def __init__(self, **kwargs):
		# call this last in your scene subclass inits. It will
		# handle all the threading and data storage for you.
		# Your prepare and perform methods can pass data through
		# the scene_data dictionary

		# each subclass should use its prepare method to prep
		# an audio thread and image thread. You can join either
		# of those threads in the prepare method to block
		# starting this scene until they're ready, and/or join
		# them in the perform method before you call super().perform()
		# or just leave it alone to do one or both async
		self.orchestrator = kwargs.get('orchestrator', None)
		self.scene_data = {}		
		self.image_thread = None
		self.audio_thread = None
		
		# everybody needs a grandma now and then
		script_dir = os.path.dirname(__file__)
		listening_path = "../grandma-listening.png"
		writing_path = "../grandma-writing.png"
		listening_path = os.path.join(script_dir, listening_path)
		writing_path = os.path.join(script_dir, writing_path)
		self.grandma_listening = Image.open(listening_path)
		self.grandma_writing = Image.open(writing_path)

		self.prepare_thread = threading.Thread(target=self.prepare)
		self.prepare_thread.start()

		pass

	# Speak the sentence. Returns None
	@abstractmethod
	def prepare(self):
		pass

	def play_image(self):
		try:
			if self.image_thread:
				self.image_thread.join()
			if self.scene_data.get('image', None):
				print("🖼️ setting image")
				self.orchestrator.display_image(self.scene_data['image'])
		except Exception as e:
			print(f"Exception in play_image: {e}")

	def play_audio(self):
		try:
			if self.audio_thread:
				self.audio_thread.join()
			if self.scene_data.get('audio', None):
				print("🖼️ playing audio")
				self.orchestrator.handle_audio(self.scene_data['audio'])
		except Exception as e:
			print(f"Exception in play_audio: {e}")

	def perform(self):
		self.prepare_thread.join()
		print("🖼️ in parent perform, past prepare join")
		img = threading.Thread(target=self.play_image)
		img.start()
		aud = threading.Thread(target=self.play_audio)
		aud.start()
		# Don't join the image thread, so we can skip
		# images that arrive too late
		# img.join()
		aud.join()




scenes/scene.py
daily-demos/llm-talk
from scenes.scene import Scene
from threading import Thread
import os
import wave
from PIL import Image

class StartListeningScene(Scene):
	def __init__(self, **kwargs):
		print("📣 StartListeningScene init")
		script_dir = os.path.dirname(__file__)
		rel_path = "../ai.wav"
		abs_file_path = os.path.join(script_dir, rel_path)
		with wave.open(abs_file_path) as audio_file:
			# Get audio data as a bytestream
			self.ai_complete_sound = audio_file.readframes(-1)

		super().__init__(**kwargs)
		
	def prepare(self):
		print("📣 StartListeningScene prepare")
		# don't need threads here because image
		# is effectively instant
		self.scene_data['image'] = self.grandma_listening
		self.scene_data['audio'] = self.ai_complete_sound
		print("📣 StartListeningScene prepare complete")
		
	
	def perform(self):
		print("📣 StartListeningScene perform")
		self.orchestrator.start_listening()
		super().perform()
scenes/start_listening_scene.py
daily-demos/llm-talk
from scenes.scene import Scene
from threading import Thread
import os
import wave

class StopListeningScene(Scene):
	def __init__(self, **kwargs):
		print("🤫 StopListeningScene init")
		script_dir = os.path.dirname(__file__)
		rel_path = "../human.wav"
		abs_file_path = os.path.join(script_dir, rel_path)
		with wave.open(abs_file_path) as audio_file:
			# Get audio data as a bytestream
			self.human_complete_sound = audio_file.readframes(-1)
		super().__init__(**kwargs)
		
	def prepare(self):
		print("🤫 StopListeningScene prepare")
		# don't need threads here because image
		# is effectively instant
		self.scene_data['image'] = self.grandma_writing
		self.scene_data['audio'] = self.human_complete_sound
		print("🤫 StopListeningScene prepare complete")
		
	
	def perform(self):
		print("🤫 StopListeningScene perform")
		self.orchestrator.stop_listening()
		super().perform()
scenes/stop_listening_scene.py
daily-demos/llm-talk
from scenes.scene import Scene
from threading import Thread

class StoryGrandmaScene(Scene):
	def __init__(self, **kwargs):
		print("StoryPageScene init")
		self.sentence = kwargs.get('sentence', None)
		super().__init__(**kwargs)
		
	def prepare(self):
		print(f"👩‍💼 StoryStartScene prepare sentence: {self.sentence}")
		# don't need threads here because image
		# is effectively instant
		self.scene_data['image'] = (self.grandma_writing)
		self.scene_data['audio'] = self.orchestrator.request_tts(self.sentence)
		print(f"👩‍💼 StoryStartScene prepare complete for: {self.sentence}")
		
	
	def perform(self):
		print(f"👩‍💼 StoryStartScene perform sentence: {self.sentence}")
		super().perform()
scenes/story_grandma_scene.py
daily-demos/llm-talk
from scenes.scene import Scene

class StoryIntroScene(Scene):
	def __init__(self, **kwargs):
		print("StoryIntroScene init")
		super().__init__(**kwargs)
		pass
		
	def prepare(self):
		print("StoryIntroScene prepare")
		# Get intro prompt, then
		self.scene_data['image'] = self.grandma_listening
		self.scene_data['audio'] = self.orchestrator.request_intro()
		
	
	def perform(self):
		print("StoryIntroScene perform")
		super().perform()
		pass
scenes/story_intro_scene.py
daily-demos/llm-talk
from scenes.scene import Scene
from threading import Thread

# This is the async version that will try to display the image
# as soon as it's ready.
class StoryPageAsyncScene(Scene):
	def __init__(self, **kwargs):
		print("StoryPageScene init")
		self.sentence = kwargs.get('sentence', None)
		self.image = kwargs.get('image', False)
		self.story_sentences = kwargs.get('story_sentences', None)
		super().__init__(**kwargs)

	def fetch_audio(self):
		try:
			self.scene_data['audio'] = self.orchestrator.request_tts(self.sentence)
			print(f"🌆 fetch audio complete for {self.sentence}")
		except Exception as e:
			print(f"Exception in fetch_audio {e}")

	def fetch_image(self):
		try:
			if self.image:
				desc = self.orchestrator.request_image_description(self.story_sentences)
				
				(url, image) = self.orchestrator.request_image(desc)
				(self.scene_data['url'], self.scene_data['image']) = (url, image)
				print(f"🌆 fetch image complete for {self.sentence}")
			else:
				print(f"🌆 Skipping image for {self.sentence}")
		except Exception as e:
			print(f"Exception in fetch_image {e}")


	def prepare(self):
		print(f"🌆 StoryPageScene prepare sentence: {self.sentence}")
		# Get intro prompt, then
		self.image_thread = Thread(target=self.fetch_image)
		self.image_thread.start()
		self.audio_thread = Thread(target=self.fetch_audio)
		self.audio_thread.start()

		#self.image_getter.join()
		if self.audio_thread:
			self.audio_thread.join()
		print(f"🌆 StoryPageScene prepare complete for: {self.sentence}")


	def perform(self):
		print(f"🌆 StoryPageScene perform sentence: {self.sentence}")
		super().perform()
		self.orchestrator.index_scene(self)

scenes/story_page_async_scene.py
daily-demos/llm-talk
from scenes.scene import Scene
from threading import Thread

# This is the synchronous version
class StoryPageScene(Scene):
	def __init__(self, **kwargs):
		print("StoryPageScene init")
		self.sentence = kwargs.get('sentence', None)
		super().__init__(**kwargs)

	def fetch_audio(self):
		try:
			self.scene_data['audio'] = self.orchestrator.request_tts(self.sentence)
			print(f"🌆 fetch audio complete for {self.sentence}")
		except Exception as e:
			print(f"Exception in fetch_audio {e}")

	def fetch_image(self):
		try:
			(url, image) = self.orchestrator.request_image(self.sentence)
			(self.scene_data['url'], self.scene_data['image']) = (url, image)
			print(f"🌆 fetch image complete for {self.sentence}")
		except Exception as e:
			print(f"Exception in fetch_image {e}")

	def prepare(self):
		print(f"🌆 StoryPageScene prepare sentence: {self.sentence}")
		# Get intro prompt, then
		self.image_getter = Thread(target=self.fetch_image)
		self.image_getter.start()
		self.audio_getter = Thread(target=self.fetch_audio)
		self.audio_getter.start()

		self.image_getter.join()
		self.audio_getter.join()
		print(f"🌆 StoryPageScene prepare complete for: {self.sentence}")


	def perform(self):
		print(f"🌆 StoryPageScene perform sentence: {self.sentence}")
		super().perform()
		self.orchestrator.index_scene(self)


scenes/story_page_scene.py
daily-demos/llm-talk
import os
import random
import time

from algoliasearch.configs import SearchConfig
from algoliasearch.search_client import SearchClient

class SearchIndexer():
    def __init__(self, story_id):
        self.index = None
        self.story_id = story_id

        self.search_enabled = os.getenv('ALGOLIA_APP_ID') and os.getenv('ALGOLIA_API_KEY')
        if self.search_enabled:
            config = SearchConfig(os.getenv('ALGOLIA_APP_ID'), os.getenv('ALGOLIA_API_KEY'))
            self.algolia = SearchClient.create_with_config(config)
            self.index = self.algolia.init_index('daily-llm-conversations')

    def index_text(self, text):
        # this should be async, probably
        if self.index:
            res = self.index.save_object({
                "objectID": hex(random.getrandbits(128))[2:],
                "storyID": self.story_id,
                "type": "text",
                "text": text,
                "createdAt": int(time.time())
            }).wait()

    def index_image(self, url):
        if self.index:
            self.index.save_object({
                "objectID": hex(random.getrandbits(128))[2:],
                "storyID": self.story_id,
                "type": "image",
                "image": url,
                "createdAt": int(time.time())
            }).wait()

search.py
daily-demos/llm-talk
from abc import abstractmethod

class AIService:
    def __init__(self):
        pass

    def set_image_style(self, image_style):
        self.image_style = image_style

    # Speak the sentence. Returns None
    @abstractmethod
    def run_tts(self, sentence):
        pass

    # Generate an image from the sentence. Returns an image
    @abstractmethod
    def run_image_gen(self, sentence):
        pass

    # Generate a set of responses to a prompt. Yields a list of responses.
    @abstractmethod
    def run_llm(self, context, human_histories, llm_histories, prompt):
        pass

services/ai_service.py
daily-demos/llm-talk
from services.ai_service import AIService
from PIL import Image
import openai
import requests
import os
import io

# See .env.example for Azure configuration needed
from azure.cognitiveservices.speech import SpeechSynthesizer, SpeechConfig, ResultReason, CancellationReason

class AzureAIService(AIService):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)

        self.speech_key = os.getenv("AZURE_SPEECH_SERVICE_KEY")
        self.speech_region = os.getenv("AZURE_SPEECH_SERVICE_REGION")

        self.speech_config = SpeechConfig(subscription=self.speech_key, region=self.speech_region)
        # self.speech_config.speech_synthesis_voice_name='en-US-JennyMultilingualV2Neural'

        self.speech_synthesizer = SpeechSynthesizer(speech_config=self.speech_config, audio_config=None)

    def run_tts(self, sentence):
        print("⌨️ running azure tts async")
        ssml = "<speak version='1.0' xml:lang='en-US' xmlns='http://www.w3.org/2001/10/synthesis' " \
           "xmlns:mstts='http://www.w3.org/2001/mstts'>" \
           "<voice name='en-US-SaraNeural'>" \
           "<mstts:silence type='Sentenceboundary' value='20ms' />" \
           "<mstts:express-as style='lyrical' styledegree='2' role='SeniorFemale'>" \
           "<prosody rate='1.05'>" \
           f"{sentence}" \
           "</prosody></mstts:express-as></voice></speak> "
        result = self.speech_synthesizer.speak_ssml(ssml)
        print("⌨️ got azure tts result")
        if result.reason == ResultReason.SynthesizingAudioCompleted:
            print("⌨️ returning result")
            return result.audio_data
        elif result.reason == ResultReason.Canceled:
            cancellation_details = result.cancellation_details
            print("Speech synthesis canceled: {}".format(cancellation_details.reason))
            if cancellation_details.reason == CancellationReason.Error:
                print("Error details: {}".format(cancellation_details.error_details))

    # generate a chat using Azure OpenAI based on the participant's most recent speech
    def run_llm(self, messages, stream = True):
        print("generating chat")

        response = openai.ChatCompletion.create(
            api_type = 'azure',
            api_version = '2023-06-01-preview',
            api_key = os.getenv("AZURE_CHATGPT_KEY"),
            api_base = os.getenv("AZURE_CHATGPT_ENDPOINT"),
            deployment_id=os.getenv("AZURE_CHATGPT_DEPLOYMENT_ID"),
            stream=stream,
            messages=messages
        )
        return response

    def run_image_gen(self, sentence):
        print("generating azure image", sentence)

        image = openai.Image.create(
            api_type = 'azure',
            api_version = '2023-06-01-preview',
            api_key = os.getenv('AZURE_DALLE_KEY'),
            api_base = os.getenv('AZURE_DALLE_ENDPOINT'),
            deployment_id = os.getenv("AZURE_DALLE_DEPLOYMENT_ID"),
            prompt=f'{sentence} in the style of {self.image_style}',
            n=1,
            size=f"512x512",
        )

        url = image["data"][0]["url"]
        response = requests.get(url)

        dalle_stream = io.BytesIO(response.content)
        dalle_im = Image.open(dalle_stream)

        return (url, dalle_im)

services/azure_ai_service.py
daily-demos/llm-talk
import requests
import os
from services.ai_service import AIService

# Note that Cloudflare's AI workers are still in beta.
# https://developers.cloudflare.com/workers-ai/
class CloudflareAIService(AIService):
    def __init__(self):
        super().__init__()
        self.cloudflare_account_id = os.getenv("CLOUDFLARE_ACCOUNT_ID")
        self.cloudflare_api_token = os.getenv("CLOUDFLARE_API_TOKEN")

        self.api_base_url = f'https://api.cloudflare.com/client/v4/accounts/{self.cloudflare_account_id}/ai/run/'
        self.headers = {"Authorization": f'Bearer {self.cloudflare_api_token}'}

    # base endpoint, used by the others
    def run(self, model, input):
        response = requests.post(f"{self.api_base_url}{model}", headers=self.headers, json=input)
        return response.json()
    
    # https://developers.cloudflare.com/workers-ai/models/llm/
    def run_llm(self, sentence):
        input = {
            "messages": [
                { "role": "system", "content": "You are a friendly assistant" },
                { "role": "user", "content": sentence }
            ]
        }

        return self.run("@cf/meta/llama-2-7b-chat-int8", input)

    # https://developers.cloudflare.com/workers-ai/models/translation/
    def run_text_translation(self, sentence, source_language, target_language):
        return self.run('@cf/meta/m2m100-1.2b', {
            "text": sentence,
            "source_lang": source_language,
            "target_lang": target_language
        })
    
    # https://developers.cloudflare.com/workers-ai/models/sentiment-analysis/
    def run_text_sentiment(self, sentence):
        return self.run("@cf/huggingface/distilbert-sst-2-int8", {"text": sentence})

    # https://developers.cloudflare.com/workers-ai/models/image-classification/
    def run_image_classification(self, image_url):
        response = requests.get(image_url)

        if response.status_code != 200:
            return {"error": "There was a problem downloading the image."}

        if response.status_code == 200:
            data = response.content
            inputs = {"image": list(data)}

        return self.run("@cf/microsoft/resnet-50", inputs)
    
    # https://developers.cloudflare.com/workers-ai/models/embedding/
    def run_embeddings(self, texts, size="medium"):
        models = {
            "small": "@cf/baai/bge-small-en-v1.5", # 384 output dimensions
            "medium": "@cf/baai/bge-base-en-v1.5", # 768 output dimensions
            "large": "@cf/baai/bge-large-en-v1.5" #1024 output dimensions
        }

        return self.run(models[size], {"text": texts})
services/cloudflare_ai_service.py
daily-demos/llm-talk
from services.ai_service import AIService
import openai
import os

# To use Google Cloud's AI products, you'll need to install Google Cloud CLI and enable the TTS and in your project: https://cloud.google.com/sdk/docs/install
from google.cloud import texttospeech

class GoogleAIService(AIService):
    def __init__(self):
        super().__init__()

        self.client = texttospeech.TextToSpeechClient()
        self.voice = texttospeech.VoiceSelectionParams(
            language_code="en-GB", name="en-GB-Neural2-F"
        )

        self.audio_config = texttospeech.AudioConfig(
            audio_encoding = texttospeech.AudioEncoding.LINEAR16,
            sample_rate_hertz = 16000
        )

    def run_tts(self, sentence):
        print("running google tts")
        synthesis_input = texttospeech.SynthesisInput(text = sentence.strip())
        result = self.client.synthesize_speech(input=synthesis_input, voice=self.voice, audio_config=self.audio_config)
        return result

services/google_ai_service.py
daily-demos/llm-talk
from services.ai_service import AIService
from transformers import pipeline

# These functions are just intended for testing, not production use. If you'd like to use HuggingFace, you should use your own models, or do some research into the specific models that will work best for your use case.
class HuggingFaceAIService(AIService):
    def __init__(self):
        super().__init__()

    def run_text_sentiment(self, sentence):
        classifier = pipeline("sentiment-analysis")
        return classifier(sentence)

    # available models at https://huggingface.co/Helsinki-NLP (**not all models use 2-character language codes**)
    def run_text_translation(self, sentence, source_language, target_language):
        translator = pipeline(f"translation", model=f"Helsinki-NLP/opus-mt-{source_language}-{target_language}")
        print(translator(sentence))

        return translator(sentence)[0]["translation_text"]

    def run_text_summarization(self, sentence):
        summarizer = pipeline("summarization")
        return summarizer(sentence)

    def run_image_classification(self, image_path):
        classifier = pipeline("image-classification")
        return classifier(image_path)

services/huggingface_ai_service.py
daily-demos/llm-talk
import io
import requests
import time
from PIL import Image
from services.ai_service import AIService

class MockAIService(AIService):
    def __init__(self):
        super().__init__()

    def run_tts(self, sentence):
        print("running tts", sentence)
        time.sleep(2)

    def run_image_gen(self, sentence):
        image_url = "https://d3d00swyhr67nd.cloudfront.net/w800h800/collection/ASH/ASHM/ASH_ASHM_WA1940_2_22-001.jpg"
        response = requests.get(image_url)
        image_stream = io.BytesIO(response.content)
        image = Image.open(image_stream)
        time.sleep(1)
        return (image_url, image)

    def run_llm(self, messages):
        for i in range(5):
            time.sleep(1)
            yield({"choices": [{"delta": {"content": f"hello {i}!"}}]})


services/mock_ai_service.py
daily-demos/llm-talk
from services.ai_service import AIService
import requests
from PIL import Image
import io
import openai
import os

class OpenAIService(AIService):
    def __init__(self):
        # we handle all the api config directly in the calls below
        pass

    def run_llm(self, messages, stream = True):
        model = os.getenv("OPEN_AI_MODEL")
        if not model:
            model = "gpt-4"
        response = openai.ChatCompletion.create(
            api_type = 'openai',
            api_version = '2020-11-07',
            api_base = "https://api.openai.com/v1",
            api_key = os.getenv("OPEN_AI_KEY"),
            model=model,
            stream=stream,
            messages=messages
        )

        return response

    def run_image_gen(self, sentence):
        print("🖌️ generating openai image async for ", sentence)

        image = openai.Image.create(
            api_type = 'openai',
            api_version = '2020-11-07',
            api_base = "https://api.openai.com/v1",
            api_key = os.getenv("OPEN_AI_KEY"),
            prompt=f'{sentence} in the style of {self.image_style}',
            n=1,
            size=f"512x512",
        )
        image_url = image["data"][0]["url"]
        print("🖌️ generated image from url", image["data"][0]["url"])
        response = requests.get(image_url)
        print("🖌️ got image from url", response)
        dalle_stream = io.BytesIO(response.content)
        dalle_im = Image.open(dalle_stream)
        return (image_url, dalle_im)

services/open_ai_service.py
daily-demos/llm-talk