diff --git a/src/dailyai/queue_aggregators.py b/src/dailyai/queue_aggregators.py index 2681b622d..ee6ca34bb 100644 --- a/src/dailyai/queue_aggregators.py +++ b/src/dailyai/queue_aggregators.py @@ -1,6 +1,6 @@ import asyncio -from dailyai.queue_frame import QueueFrame, FrameType +from dailyai.queue_frame import LLMMessagesQueueFrame, QueueFrame, TextQueueFrame from dailyai.services.ai_services import AIService from typing import AsyncGenerator, List @@ -34,26 +34,14 @@ class LLMContextAggregator(AIService): async def process_frame(self, frame:QueueFrame) -> AsyncGenerator[QueueFrame, None]: content: str = "" - if frame.frame_type == FrameType.TRANSCRIPTION: - message = frame.frame_data - if not isinstance(message, dict): - return + # TODO: split up transcription by participant + if isinstance(frame, TextQueueFrame): + content = frame.text - if message["session_id"] == self.bot_participant_id: - return - - content = message["text"] - elif frame.frame_type == FrameType.TEXT: - if not isinstance(frame.frame_data, str): - return - - content = frame.frame_data - - # todo: we should differentiate between transcriptions from different participants self.sentence += content if self.sentence.endswith((".", "?", "!")): self.messages.append({"role": self.role, "content": self.sentence}) self.sentence = "" - yield QueueFrame(FrameType.LLM_MESSAGE, self.messages) + yield LLMMessagesQueueFrame(self.messages) yield frame diff --git a/src/dailyai/queue_frame.py b/src/dailyai/queue_frame.py index 9d6ce6407..d72345aaf 100644 --- a/src/dailyai/queue_frame.py +++ b/src/dailyai/queue_frame.py @@ -1,18 +1,38 @@ from enum import Enum from dataclasses import dataclass +from typing import Any -class FrameType(Enum): - NOOP = -1 - START_STREAM = 0 - END_STREAM = 1 - AUDIO = 2 - IMAGE = 3 - TEXT = 4 - TRANSCRIPTION = 5 - LLM_MESSAGE = 6 - APP_MESSAGE = 7 - -@dataclass(frozen=True) class QueueFrame: - frame_type: FrameType - frame_data: str | dict | bytes | list | None + pass + +class StartStreamQueueFrame(QueueFrame): + pass + +class EndStreamQueueFrame(QueueFrame): + pass + +@dataclass() +class AudioQueueFrame(QueueFrame): + data: bytes + +@dataclass() +class ImageQueueFrame(QueueFrame): + url: str | None + image: bytes + +@dataclass() +class TextQueueFrame(QueueFrame): + text: str + +@dataclass() +class TranscriptionQueueFrame(TextQueueFrame): + participantId: str + timestamp: str + +@dataclass() +class LLMMessagesQueueFrame(QueueFrame): + messages: list[dict[str,str]] # TODO: define this more concretely! + +class AppMessageQueueFrame(QueueFrame): + message: Any + participantId: str diff --git a/src/dailyai/services/ai_services.py b/src/dailyai/services/ai_services.py index ab7652dde..684912989 100644 --- a/src/dailyai/services/ai_services.py +++ b/src/dailyai/services/ai_services.py @@ -1,10 +1,14 @@ import asyncio import logging -import re -from httpx import request - -from dailyai.queue_frame import QueueFrame, FrameType +from dailyai.queue_frame import ( + AudioQueueFrame, + EndStreamQueueFrame, + ImageQueueFrame, + LLMMessagesQueueFrame, + QueueFrame, + TextQueueFrame, +) from abc import abstractmethod from typing import AsyncGenerator, AsyncIterable, Iterable @@ -24,7 +28,7 @@ class AIService: await queue.put(frame) if add_end_of_stream: - await queue.put(QueueFrame(FrameType.END_STREAM, None)) + await queue.put(EndStreamQueueFrame()) async def run( self, @@ -46,7 +50,7 @@ class AIService: frame = await frames.get() async for output_frame in self.process_frame(frame): yield output_frame - if frame.frame_type == FrameType.END_STREAM: + if isinstance(frame, EndStreamQueueFrame): break else: raise Exception("Frames must be an iterable or async iterable") @@ -61,21 +65,15 @@ class AIService: async def process_frame(self, frame:QueueFrame) -> AsyncGenerator[QueueFrame, None]: # This is a trick for the interpreter (and linter) to know that this is a generator. if False: - yield QueueFrame(FrameType.NOOP, None) + yield QueueFrame() @abstractmethod async def finalize(self) -> AsyncGenerator[QueueFrame, None]: # This is a trick for the interpreter (and linter) to know that this is a generator. if False: - yield QueueFrame(FrameType.NOOP, None) + yield QueueFrame() class LLMService(AIService): - def allowed_input_frame_types(self) -> set[FrameType]: - return set([FrameType.LLM_MESSAGE]) - - def allowed_output_frame_types(self) -> set[FrameType]: - return set([FrameType.TEXT]) - @abstractmethod async def run_llm_async(self, messages) -> AsyncGenerator[str, None]: yield "" @@ -85,13 +83,9 @@ class LLMService(AIService): pass async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]: - if frame.frame_type == FrameType.LLM_MESSAGE: - if type(frame.frame_data) != list: - raise Exception("LLM service requires a dict for the data field") - - messages: list[dict[str, str]] = frame.frame_data - async for text_chunk in self.run_llm_async(messages): - yield QueueFrame(FrameType.TEXT, text_chunk) + if isinstance(frame, LLMMessagesQueueFrame): + async for text_chunk in self.run_llm_async(frame.messages): + yield TextQueueFrame(text_chunk) class TTSService(AIService): @@ -112,34 +106,31 @@ class TTSService(AIService): yield bytes() async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]: - if frame.frame_type != FrameType.TEXT: + if not isinstance(frame, TextQueueFrame): yield frame return - if not isinstance(frame.frame_data, str): - raise(Exception(f"Invalid data type in frame type: {frame.frame_type}, type: {type(frame.frame_data)}")) - text: str | None = None if not self.aggregate_sentences: - text = frame.frame_data + text = frame.text else: - self.current_sentence += frame.frame_data + self.current_sentence += frame.text if self.current_sentence.endswith((".", "?", "!")): text = self.current_sentence self.current_sentence = "" if text: async for audio_chunk in self.run_tts(text): - yield QueueFrame(FrameType.AUDIO, audio_chunk) + yield AudioQueueFrame(audio_chunk) async def finalize(self): if self.current_sentence: async for audio_chunk in self.run_tts(self.current_sentence): - yield QueueFrame(FrameType.AUDIO, audio_chunk) + yield AudioQueueFrame(audio_chunk) # Convenience function to send the audio for a sentence to the given queue async def say(self, sentence, queue: asyncio.Queue): - await self.run_to_queue(queue, [QueueFrame(FrameType.TEXT, sentence)]) + await self.run_to_queue(queue, [TextQueueFrame(sentence)]) class ImageGenService(AIService): @@ -149,15 +140,15 @@ class ImageGenService(AIService): # Renders the image. Returns an Image object. @abstractmethod - async def run_image_gen(self, sentence) -> tuple[str, bytes]: + async def run_image_gen(self, sentence:str) -> tuple[str, bytes]: pass async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]: - if type(frame.frame_data) != str: - raise Exception("Image service requires a string for the data field") + if not isinstance(frame, TextQueueFrame): + return - (_, image_data) = await self.run_image_gen(frame.frame_data) - yield QueueFrame(FrameType.IMAGE, image_data) + (url, image_data) = await self.run_image_gen(frame.text) + yield ImageQueueFrame(url, image_data) @dataclass diff --git a/src/dailyai/services/daily_transport_service.py b/src/dailyai/services/daily_transport_service.py index 77756a5c2..f1eac2bbd 100644 --- a/src/dailyai/services/daily_transport_service.py +++ b/src/dailyai/services/daily_transport_service.py @@ -8,9 +8,16 @@ import types from functools import partial from queue import Queue, Empty -from dailyai.queue_frame import QueueFrame, FrameType +from dailyai.queue_frame import ( + AudioQueueFrame, + EndStreamQueueFrame, + ImageQueueFrame, + QueueFrame, + StartStreamQueueFrame, + TranscriptionQueueFrame, +) -from threading import Thread, Event, Timer +from threading import Thread, Event from daily import ( EventHandler, @@ -201,7 +208,7 @@ class DailyTransportService(EventHandler): while True: frame = await self.receive_queue.get() yield frame - if frame.frame_type == FrameType.END_STREAM: + if isinstance(frame, EndStreamQueueFrame): break def get_async_send_queue(self): @@ -212,7 +219,7 @@ class DailyTransportService(EventHandler): frame: QueueFrame | list = await self.send_queue.get() self.threadsafe_send_queue.put(frame) self.send_queue.task_done() - if type(frame) == QueueFrame and frame.frame_type == FrameType.END_STREAM: + if isinstance(frame, EndStreamQueueFrame): break async def wait_for_send_queue_to_empty(self): @@ -242,8 +249,8 @@ class DailyTransportService(EventHandler): self.stop_threads.set() - await self.receive_queue.put(QueueFrame(FrameType.END_STREAM, None)) - await self.send_queue.put(QueueFrame(FrameType.END_STREAM, None)) + await self.receive_queue.put(EndStreamQueueFrame()) + await self.send_queue.put(EndStreamQueueFrame()) await async_output_queue_marshal_task if self.camera_thread and self.camera_thread.is_alive(): @@ -281,7 +288,12 @@ class DailyTransportService(EventHandler): def on_transcription_message(self, message:dict): if self.loop: - frame = QueueFrame(FrameType.TRANSCRIPTION, message) + participantId = "" + if "participantId" in message: + participantId = message["participantId"] + elif "session_id" in message: + participantId = message["session_id"] + frame = TranscriptionQueueFrame(message["text"], participantId, message["timestamp"]) asyncio.run_coroutine_threadsafe(self.receive_queue.put(frame), self.loop) def on_transcription_stopped(self, stopped_by, stopped_by_error): @@ -314,15 +326,15 @@ class DailyTransportService(EventHandler): while True: try: frames_or_frame: QueueFrame | list[QueueFrame] = self.threadsafe_send_queue.get() - if type(frames_or_frame) == QueueFrame: + if isinstance(frames_or_frame, QueueFrame): frames: list[QueueFrame] = [frames_or_frame] - elif type(frames_or_frame) == list: + elif isinstance(frames_or_frame, list): frames: list[QueueFrame] = frames_or_frame else: raise Exception("Unknown type in output queue") for frame in frames: - if frame.frame_type == FrameType.END_STREAM: + if isinstance(frame, EndStreamQueueFrame): self.logger.info("Stopping frame consumer thread") self.threadsafe_send_queue.task_done() return @@ -330,8 +342,8 @@ class DailyTransportService(EventHandler): # if interrupted, we just pull frames off the queue and discard them if not self.is_interrupted.is_set(): if frame: - if frame.frame_type == FrameType.AUDIO: - chunk = frame.frame_data + if isinstance(frame, AudioQueueFrame): + chunk = frame.data all_audio_frames.extend(chunk) @@ -340,8 +352,8 @@ class DailyTransportService(EventHandler): if l: self.mic.write_frames(bytes(b[:l])) b = b[l:] - elif frame.frame_type == FrameType.IMAGE: - self.set_image(frame.frame_data) + elif isinstance(frame, ImageQueueFrame): + self.set_image(frame.image) elif len(b): self.mic.write_frames(bytes(b)) b = bytearray() @@ -352,7 +364,7 @@ class DailyTransportService(EventHandler): ) self.interrupt_time = None - if frame.frame_type == FrameType.START_STREAM: + if isinstance(frame, StartStreamQueueFrame): self.is_interrupted.clear() self.threadsafe_send_queue.task_done() diff --git a/src/dailyai/tests/test_ai_services.py b/src/dailyai/tests/test_ai_services.py index bfe8dbde8..8bb1e2b00 100644 --- a/src/dailyai/tests/test_ai_services.py +++ b/src/dailyai/tests/test_ai_services.py @@ -4,7 +4,7 @@ import unittest from typing import AsyncGenerator, Generator from dailyai.services.ai_services import AIService -from dailyai.queue_frame import QueueFrame, FrameType +from dailyai.queue_frame import EndStreamQueueFrame, QueueFrame, TextQueueFrame class SimpleAIService(AIService): async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]: @@ -15,8 +15,8 @@ class TestBaseAIService(unittest.IsolatedAsyncioTestCase): service = SimpleAIService() input_frames = [ - QueueFrame(FrameType.TEXT, "hello"), - QueueFrame(FrameType.END_STREAM, None), + TextQueueFrame("hello"), + EndStreamQueueFrame() ] async def iterate_frames() -> AsyncGenerator[QueueFrame, None]: for frame in input_frames: @@ -31,10 +31,7 @@ class TestBaseAIService(unittest.IsolatedAsyncioTestCase): async def test_nonasync_input(self): service = SimpleAIService() - input_frames = [ - QueueFrame(FrameType.TEXT, "hello"), - QueueFrame(FrameType.END_STREAM, None), - ] + input_frames = [TextQueueFrame("hello"), EndStreamQueueFrame()] def iterate_frames() -> Generator[QueueFrame, None, None]: for frame in input_frames: diff --git a/src/samples/foundational/01-say-one-thing.py b/src/samples/foundational/01-say-one-thing.py index 27b6a3b3d..ac0778fbc 100644 --- a/src/samples/foundational/01-say-one-thing.py +++ b/src/samples/foundational/01-say-one-thing.py @@ -1,10 +1,7 @@ import argparse import asyncio -from typing import AsyncGenerator -from dailyai.queue_frame import QueueFrame, FrameType from dailyai.services.daily_transport_service import DailyTransportService -from dailyai.services.azure_ai_services import AzureTTSService from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService async def main(room_url): diff --git a/src/samples/foundational/02-llm-say-one-thing.py b/src/samples/foundational/02-llm-say-one-thing.py index 3ea0f2714..a54c0ecb7 100644 --- a/src/samples/foundational/02-llm-say-one-thing.py +++ b/src/samples/foundational/02-llm-say-one-thing.py @@ -1,8 +1,7 @@ import argparse import asyncio -from typing import AsyncGenerator -from dailyai.queue_frame import QueueFrame, FrameType +from dailyai.queue_frame import LLMMessagesQueueFrame from dailyai.services.daily_transport_service import DailyTransportService from dailyai.services.azure_ai_services import AzureLLMService from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService @@ -27,7 +26,7 @@ async def main(room_url): tts_task = asyncio.create_task( tts.run_to_queue( transport.send_queue, - llm.run([QueueFrame(FrameType.LLM_MESSAGE, messages)]) + llm.run([LLMMessagesQueueFrame(messages)]), ) ) diff --git a/src/samples/foundational/03-still-frame.py b/src/samples/foundational/03-still-frame.py index c69b12547..5ffdcc5ac 100644 --- a/src/samples/foundational/03-still-frame.py +++ b/src/samples/foundational/03-still-frame.py @@ -1,7 +1,7 @@ import argparse import asyncio -from dailyai.queue_frame import QueueFrame, FrameType +from dailyai.queue_frame import TextQueueFrame from dailyai.services.daily_transport_service import DailyTransportService from dailyai.services.open_ai_services import OpenAIImageGenService @@ -23,7 +23,7 @@ async def main(room_url): imagegen = OpenAIImageGenService(image_size="1024x1024") image_task = asyncio.create_task( - imagegen.run_to_queue(transport.send_queue, [QueueFrame(FrameType.TEXT, "a cat in the style of picasso")]) + imagegen.run_to_queue(transport.send_queue, [TextQueueFrame("a cat in the style of picasso")]) ) @transport.event_handler("on_participant_joined") diff --git a/src/samples/foundational/04-utterance-and-speech.py b/src/samples/foundational/04-utterance-and-speech.py index 296d17e6d..32844cb6f 100644 --- a/src/samples/foundational/04-utterance-and-speech.py +++ b/src/samples/foundational/04-utterance-and-speech.py @@ -4,7 +4,7 @@ import re from dailyai.services.daily_transport_service import DailyTransportService from dailyai.services.azure_ai_services import AzureLLMService, AzureTTSService -from dailyai.queue_frame import QueueFrame, FrameType +from dailyai.queue_frame import EndStreamQueueFrame, LLMMessagesQueueFrame from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService async def main(room_url:str): @@ -35,7 +35,7 @@ async def main(room_url:str): llm_response_task = asyncio.create_task( elevenlabs_tts.run_to_queue( buffer_queue, - llm.run([QueueFrame(FrameType.LLM_MESSAGE, messages)]), + llm.run([LLMMessagesQueueFrame(messages)]), True, ) ) @@ -52,7 +52,7 @@ async def main(room_url:str): frame = await buffer_queue.get() await transport.send_queue.put(frame) buffer_queue.task_done() - if frame.frame_type == FrameType.END_STREAM: + if isinstance(frame, EndStreamQueueFrame): break await asyncio.gather(llm_response_task, buffer_to_send_queue()) diff --git a/src/samples/foundational/05-sync-speech-and-text.py b/src/samples/foundational/05-sync-speech-and-text.py index d589e778a..7e6239288 100644 --- a/src/samples/foundational/05-sync-speech-and-text.py +++ b/src/samples/foundational/05-sync-speech-and-text.py @@ -1,13 +1,9 @@ import argparse import asyncio -from asyncio.queues import Queue -import re - -from dailyai.queue_frame import QueueFrame, FrameType +from dailyai.queue_frame import AudioQueueFrame, ImageQueueFrame from dailyai.services.azure_ai_services import AzureLLMService from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService -from dailyai.services.open_ai_services import OpenAIImageGenService from dailyai.services.daily_transport_service import DailyTransportService from dailyai.services.fal_ai_services import FalImageGenService @@ -48,14 +44,20 @@ async def main(room_url): ] image_description = await llm.run_llm(messages) + if not image_description: + return + to_speak = f"{month}: {image_description}" + audio_task = asyncio.create_task(get_all_audio(to_speak)) + image_task = asyncio.create_task(dalle.run_image_gen(image_description)) (audio, image_data) = await asyncio.gather( - get_all_audio(to_speak), dalle.run_image_gen(image_description) + audio_task, image_task ) return { "month": month, "text": image_description, + "image_url": image_data[0], "image": image_data[1], "audio": audio, } @@ -84,8 +86,8 @@ async def main(room_url): data = await month_data_task await transport.send_queue.put( [ - QueueFrame(FrameType.IMAGE, data["image"]), - QueueFrame(FrameType.AUDIO, data["audio"]), + ImageQueueFrame(data["image_url"], data["image"]), + AudioQueueFrame(data["audio"]), ] )