playing around with this

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):
+    def __init__(self, aggregate_sentences=True, *args, **kwargs):
-        super().__init__()
+        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)])
 
 

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(

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()

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

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

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"] + "!",
+                "Hello there, " + participant["info"]["userName"] + "!"
-                transport.send_queue,
             )
 
             # wait for the output queue to be empty, then leave the meeting

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"))
+            model=os.getenv("AZURE_CHATGPT_MODEL"),
-        # llm = OpenAILLMService(api_key=os.getenv("OPENAI_CHATGPT_API_KEY"))
+        )
+
+        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(
+        await llm.source.put(LLMMessagesQueueFrame(messages))
-            tts.run_to_queue(
+        await llm.source.put(EndStreamQueueFrame())
-                transport.send_queue,
+
-                llm.run([LLMMessagesQueueFrame(messages)]),
+        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):