diff --git a/src/dailyai/services/ai_services.py b/src/dailyai/services/ai_services.py index 29b7d6488..7b6535048 100644 --- a/src/dailyai/services/ai_services.py +++ b/src/dailyai/services/ai_services.py @@ -6,7 +6,6 @@ import wave from dailyai.queue_frame import ( AudioQueueFrame, - ControlQueueFrame, EndStreamQueueFrame, ImageQueueFrame, LLMMessagesQueueFrame, @@ -18,12 +17,68 @@ from dailyai.queue_frame import ( from abc import abstractmethod from typing import AsyncGenerator, AsyncIterable, BinaryIO, Iterable -from dataclasses import dataclass + +class AbstractPipeService(): + def __init__(self, source:asyncio.Queue[QueueFrame], sink:asyncio.Queue[QueueFrame]): + self.source: asyncio.Queue[QueueFrame] = source + self.sink: asyncio.Queue[QueueFrame] = sink + pass + + @abstractmethod + async def process_queue(self): + pass + +class PipeService(AbstractPipeService): + + def __init__( + self, + source: asyncio.Queue[QueueFrame] | AbstractPipeService | None=None, + sink: asyncio.Queue[QueueFrame] | AbstractPipeService | None=None, + ): + self.logger: logging.Logger = logging.getLogger("dailyai") + + if not source: + source = asyncio.Queue() + elif isinstance(source, AbstractPipeService): + source = source.sink + + if not sink: + sink = asyncio.Queue() + elif isinstance(sink, AbstractPipeService): + sink = sink.source + + self.source = source + self.sink = sink + + async def process_queue(self): + if not self.source: + return + + while True: + frame = await self.source.get() + async for output_frame in self.process_frame(frame): + await self.sink.put(output_frame) + if isinstance(frame, EndStreamQueueFrame): + print("end of stream", type(self)) + async for output_frame in self.finalize(): + await self.sink.put(output_frame) + return + + @abstractmethod + async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]: + yield frame + + @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() -class AIService: +class AIService(PipeService): - def __init__(self): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) self.logger = logging.getLogger("dailyai") def stop(self): @@ -67,17 +122,6 @@ class AIService: self.logger.error("Exception occurred while running AI service", e) raise e - @abstractmethod - async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]: - if isinstance(frame, ControlQueueFrame): - yield frame - - @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() - class LLMService(AIService): @abstractmethod @@ -98,8 +142,8 @@ class LLMService(AIService): class TTSService(AIService): - def __init__(self, aggregate_sentences=True): - super().__init__() + def __init__(self, aggregate_sentences=True, *args, **kwargs): + super().__init__(*args, **kwargs) self.aggregate_sentences: bool = aggregate_sentences self.current_sentence: str = "" @@ -138,7 +182,10 @@ class TTSService(AIService): 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): + async def say(self, sentence, queue: asyncio.Queue|None=None): + queue = queue or self.output_queue + if not queue: + raise Exception("No queue to send audio to") await self.run_to_queue(queue, [TextQueueFrame(sentence)]) diff --git a/src/dailyai/services/azure_ai_services.py b/src/dailyai/services/azure_ai_services.py index 1f3e67cd2..cecc2623f 100644 --- a/src/dailyai/services/azure_ai_services.py +++ b/src/dailyai/services/azure_ai_services.py @@ -48,8 +48,17 @@ class AzureTTSService(TTSService): class AzureLLMService(LLMService): - def __init__(self, *, api_key, endpoint, api_version="2023-12-01-preview", model): - super().__init__() + + def __init__( + self, + api_key, + endpoint, + model, + api_version="2023-12-01-preview", + *args, + **kwargs, + ): + super().__init__(*args, **kwargs) self._model: str = model self._client = AsyncAzureOpenAI( diff --git a/src/dailyai/services/base_transport_service.py b/src/dailyai/services/base_transport_service.py index 530990cfa..1e1547b68 100644 --- a/src/dailyai/services/base_transport_service.py +++ b/src/dailyai/services/base_transport_service.py @@ -82,7 +82,7 @@ class BaseTransportService(): self._stop_threads.set() - await self.send_queue.put(EndStreamQueueFrame()) + #await self.send_queue.put(EndStreamQueueFrame()) await async_output_queue_marshal_task await self.send_queue.join() self._frame_consumer_thread.join() diff --git a/src/dailyai/services/elevenlabs_ai_service.py b/src/dailyai/services/elevenlabs_ai_service.py index e1795aab3..fdc39d6fb 100644 --- a/src/dailyai/services/elevenlabs_ai_service.py +++ b/src/dailyai/services/elevenlabs_ai_service.py @@ -12,12 +12,13 @@ class ElevenLabsTTSService(TTSService): def __init__( self, - *, aiohttp_session: aiohttp.ClientSession, api_key, voice_id, + *args, + **kwargs ): - super().__init__() + super().__init__(*args, **kwargs) self._api_key = api_key self._voice_id = voice_id diff --git a/src/dailyai/services/playht_ai_service.py b/src/dailyai/services/playht_ai_service.py index 350e5cb88..0bbfab3c9 100644 --- a/src/dailyai/services/playht_ai_service.py +++ b/src/dailyai/services/playht_ai_service.py @@ -11,12 +11,13 @@ class PlayHTAIService(TTSService): def __init__( self, - *, api_key, user_id, - voice_url + voice_url, + *args, + **kwargs, ): - super().__init__() + super().__init__(*args, **kwargs) self.speech_key = api_key self.user_id = user_id diff --git a/src/examples/foundational/01-say-one-thing.py b/src/examples/foundational/01-say-one-thing.py index 9df54d708..872e78efb 100644 --- a/src/examples/foundational/01-say-one-thing.py +++ b/src/examples/foundational/01-say-one-thing.py @@ -35,6 +35,8 @@ async def main(room_url): voice_id=os.getenv("ELEVENLABS_VOICE_ID")) """ tts = PlayHTAIService( + source=None, + sink=transport.send_queue, api_key=os.getenv("PLAY_HT_API_KEY"), user_id=os.getenv("PLAY_HT_USER_ID"), voice_url=os.getenv("PLAY_HT_VOICE_URL"), @@ -48,8 +50,7 @@ async def main(room_url): return await tts.say( - "Hello there, " + participant["info"]["userName"] + "!", - transport.send_queue, + "Hello there, " + participant["info"]["userName"] + "!" ) # wait for the output queue to be empty, then leave the meeting diff --git a/src/examples/foundational/02-llm-say-one-thing.py b/src/examples/foundational/02-llm-say-one-thing.py index b15023380..f3a75a579 100644 --- a/src/examples/foundational/02-llm-say-one-thing.py +++ b/src/examples/foundational/02-llm-say-one-thing.py @@ -3,7 +3,7 @@ import os import aiohttp -from dailyai.queue_frame import LLMMessagesQueueFrame +from dailyai.queue_frame import EndStreamQueueFrame, LLMMessagesQueueFrame from dailyai.services.daily_transport_service import DailyTransportService from dailyai.services.azure_ai_services import AzureLLMService, AzureTTSService from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService @@ -23,28 +23,28 @@ async def main(room_url): mic_enabled=True ) - tts = ElevenLabsTTSService( - aiohttp_session=session, - api_key=os.getenv("ELEVENLABS_API_KEY"), - voice_id=os.getenv("ELEVENLABS_VOICE_ID")) - # tts = AzureTTSService(api_key=os.getenv("AZURE_SPEECH_API_KEY"), region=os.getenv("AZURE_SPEECH_REGION")) - # tts = DeepgramTTSService(aiohttp_session=session, api_key=os.getenv("DEEPGRAM_API_KEY"), voice=os.getenv("DEEPGRAM_VOICE")) - llm = AzureLLMService( api_key=os.getenv("AZURE_CHATGPT_API_KEY"), endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"), - model=os.getenv("AZURE_CHATGPT_MODEL")) - # llm = OpenAILLMService(api_key=os.getenv("OPENAI_CHATGPT_API_KEY")) + model=os.getenv("AZURE_CHATGPT_MODEL"), + ) + + tts = ElevenLabsTTSService( + source=llm, + sink=transport.send_queue, + aiohttp_session=session, + api_key=os.getenv("ELEVENLABS_API_KEY"), + voice_id=os.getenv("ELEVENLABS_VOICE_ID"), + ) + messages = [{ "role": "system", "content": "You are an LLM in a WebRTC session, and this is a 'hello world' demo. Say hello to the world." }] - tts_task = asyncio.create_task( - tts.run_to_queue( - transport.send_queue, - llm.run([LLMMessagesQueueFrame(messages)]), - ) - ) + await llm.source.put(LLMMessagesQueueFrame(messages)) + await llm.source.put(EndStreamQueueFrame()) + + tts_task = asyncio.gather(llm.process_queue(), tts.process_queue()) @transport.event_handler("on_first_other_participant_joined") async def on_first_other_participant_joined(transport):