From 1e552958aa54d4cb80931e6a1b0bbd9a464566a4 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Sun, 25 Feb 2024 21:41:55 -0800 Subject: [PATCH] hackathon code --- src/dailyai/queue_frame.py | 8 + .../services/base_transport_service.py | 22 ++- .../services/daily_transport_service.py | 24 ++- src/dailyai/services/fal_ai_services.py | 3 +- src/dailyai/services/fireworks_ai_services.py | 122 ++++++++++++ src/khk-hackathon/06d-listen.py | 178 ++++++++++++++++++ 6 files changed, 350 insertions(+), 7 deletions(-) create mode 100644 src/dailyai/services/fireworks_ai_services.py create mode 100644 src/khk-hackathon/06d-listen.py diff --git a/src/dailyai/queue_frame.py b/src/dailyai/queue_frame.py index cb0078a89..b285e1568 100644 --- a/src/dailyai/queue_frame.py +++ b/src/dailyai/queue_frame.py @@ -51,9 +51,17 @@ class SpriteQueueFrame(QueueFrame): class TextQueueFrame(QueueFrame): text: str + +@dataclass() +class TextQueueOutOfBandFrame(TextQueueFrame): + outOfBand: bool = True + + @dataclass() class TTSCompletedFrame(QueueFrame): text: str + outOfBand: bool = False + @dataclass() class TranscriptionQueueFrame(TextQueueFrame): diff --git a/src/dailyai/services/base_transport_service.py b/src/dailyai/services/base_transport_service.py index f3dab7132..24ebbbbe4 100644 --- a/src/dailyai/services/base_transport_service.py +++ b/src/dailyai/services/base_transport_service.py @@ -14,6 +14,7 @@ import torch import torchaudio from enum import Enum import datetime +import traceback from typing import AsyncGenerator, AsyncIterable, BinaryIO, Iterable from dailyai.queue_aggregators import LLMAssistantContextAggregator, LLMUserContextAggregator @@ -133,8 +134,23 @@ class BaseTransportService(): self._current_phrase = "" self._context = new_context - def append_to_context(self, role, text): + def append_to_context(self, role, chunk_or_text): + print("IN APPEND", chunk_or_text) + # if we get a non-string, append it to the context without further error checking + # unless the outOfBand property is True + if not isinstance(chunk_or_text, str): + + if not chunk_or_text.get("outOfBand") == True: + self._context.append(chunk_or_text) + return + + text = chunk_or_text last_context_item = self._context[-1] + + print("TEXT", text) + print("LAST CONTEXT ITEM", last_context_item) + traceback.print_stack() + if last_context_item and last_context_item['role'] == role: last_context_item['content'] += f" {text}" else: @@ -153,7 +169,7 @@ class BaseTransportService(): self._runner = runner async for frame in self.get_receive_frames(): - print(f"got frame of type: {type(frame)}") + print(f"got frame of type: {type(frame)}, {frame}") if isinstance(frame, EndStreamQueueFrame): break # elif not isinstance(frame, TranscriptionQueueFrame): @@ -409,7 +425,7 @@ class BaseTransportService(): self._set_image(frame.image) elif isinstance(frame, SpriteQueueFrame): self._set_images(frame.images) - elif isinstance(frame, TTSCompletedFrame): + elif isinstance(frame, TTSCompletedFrame) and not frame.outOfBand: self.append_to_context( "assistant", frame.text) elif len(b): diff --git a/src/dailyai/services/daily_transport_service.py b/src/dailyai/services/daily_transport_service.py index 9c613d7ab..e012801bc 100644 --- a/src/dailyai/services/daily_transport_service.py +++ b/src/dailyai/services/daily_transport_service.py @@ -9,7 +9,7 @@ from daily import ( ) from threading import Event from dailyai.queue_frame import ( - TranscriptionQueueFrame, + TranscriptionQueueFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame ) from functools import partial import types @@ -27,7 +27,7 @@ torch.set_num_threads(1) model, utils = torch.hub.load(repo_or_dir='snakers4/silero-vad', model='silero_vad', - force_reload=True) + force_reload=False) (get_speech_timestamps, save_audio, @@ -276,8 +276,26 @@ class DailyTransportService(BaseTransportService, EventHandler): if len(self.client.participants()) < self._min_others_count + 1: self._stop_threads.set() + async def insert_speech(self, text, sender, date): + await self.receive_queue.put(UserStartedSpeakingFrame()) + await asyncio.sleep(0.3) + + # frame = TranscriptionQueueFrame(text, sender, date) + # await self.receive_queue.put(frame) + self.on_transcription_message({ + "text": text, + "participantId": "cb65b845-aac0-4fc8-987d-2e7ce3c7d8f0", + "timestamp": date + }) + + await asyncio.sleep(0.3) + await self.receive_queue.put(UserStoppedSpeakingFrame()) + def on_app_message(self, message, sender): - pass + if self._loop: + print("APP MESSAGE", message) + asyncio.run_coroutine_threadsafe( + self.insert_speech(message["message"], sender, message["date"]), self._loop) def on_transcription_message(self, message: dict): if self._loop: diff --git a/src/dailyai/services/fal_ai_services.py b/src/dailyai/services/fal_ai_services.py index 9464b46dd..c9435da43 100644 --- a/src/dailyai/services/fal_ai_services.py +++ b/src/dailyai/services/fal_ai_services.py @@ -32,7 +32,8 @@ class FalImageGenService(ImageGenService): handler = fal.apps.submit( "110602490-fast-sdxl", arguments={ - "prompt": sentence + "prompt": sentence, + "seed": 23 }, ) for event in handler.iter_events(): diff --git a/src/dailyai/services/fireworks_ai_services.py b/src/dailyai/services/fireworks_ai_services.py new file mode 100644 index 000000000..77d48f545 --- /dev/null +++ b/src/dailyai/services/fireworks_ai_services.py @@ -0,0 +1,122 @@ +import aiohttp +from PIL import Image +import io +from openai import AsyncOpenAI + +import asyncio +import json +from collections.abc import AsyncGenerator + +from dailyai.services.ai_services import LLMService, ImageGenService + +from dailyai.queue_frame import (TextQueueFrame, TextQueueOutOfBandFrame) + + +class FireworksLLMService(LLMService): + def __init__(self, *, api_key, model="", tools=[], context, change_appearance, transport=""): + super().__init__(context) + self._model = model + self._tools = tools + self._change_appearance = change_appearance + self._transport = transport + self._client = AsyncOpenAI( + api_key=api_key, + base_url="https://api.fireworks.ai/inference/v1" + ) + + async def get_response(self, messages, stream): + print("GET RESPONSE ... WHEN DO WE EXPECT THIS TO BE CALLED?") + return await self._client.chat.completions.create( + stream=stream, + messages=messages, + model=self._model, + temperature=0.1, + tools=self._tools + ) + + async def run_llm_async(self, messages) -> AsyncGenerator[str, None]: + print("IN ASYNC") + messages_for_log = json.dumps(messages) + self.logger.debug(f"Generating chat via openai: {messages_for_log}") + + chunks = await self._client.chat.completions.create( + model=self._model, + stream=True, # BLARGH + messages=messages, + temperature=0.1, + tools=self._tools + ) + + tool_call = {} + + async for chunk in chunks: + print(f"CHUNK: {chunk}") + if len(chunk.choices) == 0: + continue + + if chunk.choices[0].delta.content: + yield chunk.choices[0].delta.content + + if chunk.choices[0].delta.tool_calls: + print(f"TOOL CALLS: {chunk.choices[0].delta.tool_calls[0]}") + if chunk.choices[0].delta.tool_calls[0].function.name: + tool_call["id"] = chunk.choices[0].delta.tool_calls[0].id + tool_call["name"] = chunk.choices[0].delta.tool_calls[0].function.name + tool_call["arguments"] = '' + if chunk.choices[0].delta.tool_calls[0].function.arguments: + tool_call["arguments"] += chunk.choices[0].delta.tool_calls[0].function.arguments + + if chunk.choices[0].finish_reason: + print(f"TOOL CALLS ACCUM -- {tool_call}") + if tool_call.get("name"): + # hard coding tool call action for now. we should assemble the tool call + # from the streaming response, then yield it to the pipeline. + # this approach works for the first few change appearance requests but + # then the model starts refusing. need to read more about function + # calling, try this with the OpenAI APIs, and talk to the Fireworks people. + self._transport.append_to_context("assistant", { + # pipeline will append the content to this context after it goes + # through tts. we need to manually append the tool call, though + "content": "", + "role": "assistant", + "tool_calls": [ + { + "id": tool_call["id"], + "type": "function", + "index": 0, + "function": { + "name": tool_call["name"], + "arguments": tool_call["arguments"] + }, + } + ], + }) + self._transport.append_to_context("tool", { + "content": "image generated by prompt arguments: " + tool_call["arguments"], + "role": "tool", + "tool_call_id": tool_call["id"] + }) + self._transport.append_to_context("assistant", { + "content": f"call to {tool_call['name']} function succeeded", + "role": "assistant", + }) + print("APPENDED TO CONTEXT") + image_prompt = json.loads( + tool_call["arguments"]).get("appearance") + print("IMAGE PROMPT", image_prompt) + asyncio.create_task( + self._change_appearance(image_prompt)) + yield TextQueueOutOfBandFrame("Sure, let me work on that for you!") + # yield {"content": "Sure, let me work on that for you!"} + # yield "Sure, let me work on that for you!" + + async def run_llm(self, messages) -> str | None: + print("--> IN SYNC ... WHEN DO WE EXPECT THIS TO BE CALLED?") + messages_for_log = json.dumps(messages) + self.logger.debug(f"Generating chat via openai: {messages_for_log}") + + response = await self._client.chat.completions.create(model=self._model, stream=False, messages=messages) + if response and len(response.choices) > 0: + return response.choices[0].message.content + else: + return None diff --git a/src/khk-hackathon/06d-listen.py b/src/khk-hackathon/06d-listen.py new file mode 100644 index 000000000..2ab01ad50 --- /dev/null +++ b/src/khk-hackathon/06d-listen.py @@ -0,0 +1,178 @@ +from datetime import datetime +import asyncio +import aiohttp +import os +import sys +from dailyai.conversation_wrappers import InterruptibleConversationWrapper + +from dailyai.queue_frame import StartStreamQueueFrame, TranscriptionQueueFrame, TextQueueFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame +from dailyai.services.daily_transport_service import DailyTransportService +from dailyai.services.fireworks_ai_services import FireworksLLMService +from dailyai.services.deepgram_ai_services import DeepgramTTSService +from dailyai.services.ai_services import FrameLogger + +from dailyai.services.fal_ai_services import FalImageGenService + +from examples.foundational.support.runner import configure + + +command_line_prompt = ' '.join(sys.argv[1:]) + +system_prompt = """ +You are a friendly robot character with a cartoon body with head, torso, arms, feet, +and legs. + +You can change your appearance using the `change_appearance` function call. +You can add or remove items from your body, change +your color, and more. You can use function calling to change your appearance. + +When changing your appearance, please create a prompt as an argument to the function. +The prompt will help the image generation model +create a new appearance for you. Include as much detail as possible. Include the +keywords "robot", "friendly", "cartoon", "smiling", "happy", "animated". +The initial image prompt you are adding to or changing is +"A friendly cartoon robot, smiling and happy, animated." + +Do not include the image model prompt in your response. The prompt must be passed to the function +as a parameter. +""" + +do_not_respond_function = { + "name": "do_not_respond", + "description": "Call this function when the users are not talking to the robot.", + "parameters": { + "type": "object", + "properties": { + "transcribed_text": { + "type": "string", + "description": "The transcribed text from the users." + } + } + } +} + +change_appearance_function = { + "name": "change_appearance", + "description": "Call this function when the users want you to change your appearance.", + "parameters": { + "type": "object", + "properties": { + "appearance": { + "type": "string", + "description": "The new appearance for the robot, in the form of a prompt for an generative AI diffusion model." + } + } + } +} + +tools = [ + { + "type": "function", + "function": do_not_respond_function + }, + { + "type": "function", + "function": change_appearance_function + } +] + + +async def main(room_url: str, token): + async with aiohttp.ClientSession() as session: + context = [ + { + "role": "system", + "content": system_prompt, + }, + ] + transport = DailyTransportService( + room_url, + token, + "Respond bot", + duration_minutes=30, + start_transcription=True, + mic_enabled=True, + mic_sample_rate=16000, + camera_enabled=True, + camera_width=1024, + camera_height=1024, + # TODO-CB: Should this be VAD enabled or something? + speaker_enabled=True, + context=context + ) + + imagegen = FalImageGenService( + image_size="512x512", + aiohttp_session=session, + key_id=os.getenv("FAL_KEY_ID"), + key_secret=os.getenv("FAL_KEY_SECRET")) + + async def change_appearance(appearance): + await asyncio.create_task( + imagegen.run_to_queue( + transport.send_queue, [ + TextQueueFrame(appearance)])) + + llm = FireworksLLMService( + context=context, + api_key=os.getenv("FIREWORKS_API_KEY"), + model="accounts/fireworks/models/firefunction-v1", + # TODO - how can we modify tools list on the fly? + tools=tools, + change_appearance=change_appearance, + transport=transport + ) + tts = DeepgramTTSService(aiohttp_session=session, api_key=os.getenv( + "DEEPGRAM_API_KEY"), voice=os.getenv("DEEPGRAM_VOICE")) + fl = FrameLogger("just outside the innermost layer") + + async def run_response(in_frame): + await tts.run_to_queue( + transport.send_queue, + # tma_out.run( + llm.run( + # tma_in.run( + fl.run( + [StartStreamQueueFrame(), in_frame] + ) + # ) + ) + # ), + ) + + @transport.event_handler("on_first_other_participant_joined") + async def on_first_other_participant_joined(transport): + await change_appearance("A friendly cartoon robot, smiling and happy, animated.") + return + + await tts.say("Hi, I'm listening!", transport.send_queue) + await asyncio.sleep(1) + + await transport.receive_queue.put(UserStartedSpeakingFrame()) + await asyncio.sleep(0.1) + + transport.on_transcription_message({ + "text": command_line_prompt, + "participantId": "cb65b845-aac0-4fc8-987d-2e7ce3c7d8f0", + "timestamp": datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z' + }) +# putting the frame into the queue directly doesn't seem to work +# await transport.receive_queue.put( +# TranscriptionQueueFrame( +# "tell me a joke.", +# "cb65b845-aac0-4fc8-987d-2e7ce3c7d8f0", +# datetime.utcnow().strftime( +# '%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z' +# )) + await asyncio.sleep(0.1) + await transport.receive_queue.put(UserStoppedSpeakingFrame()) + + transport.transcription_settings["extra"]["endpointing"] = True + transport.transcription_settings["extra"]["punctuate"] = True + + await asyncio.gather(transport.run(), transport.run_conversation(run_response)) + + +if __name__ == "__main__": + (url, token) = configure() + asyncio.run(main(url, token))