Update apps

src/dailyai/services/azure_ai_services.py

@@ -64,13 +64,17 @@ class AzureTTSService(TTSService):
 
 class AzureLLMService(BaseOpenAILLMService):
     def __init__(self, *, api_key, endpoint, api_version="2023-12-01-preview", model):
-        super().__init__(model)
+        self._endpoint = endpoint
+        self._api_version = api_version
 
-        # This overrides the client created by the super class init
+        super().__init__(api_key=api_key, model=model)
+        self._model: str = model
+
+    def create_client(self, api_key=None, base_url=None):
         self._client = AsyncAzureOpenAI(
             api_key=api_key,
-            azure_endpoint=endpoint,
+            azure_endpoint=self._endpoint,
-            api_version=api_version,
+            api_version=self._api_version,
         )
 
 

src/dailyai/services/base_transport_service.py

@@ -151,10 +151,9 @@ class BaseTransportService:
 
         pipeline_task = None
         if pipeline:
-            pipeline.set_sink(self.send_queue)
+            pipeline_task = asyncio.create_task(
-            if override_pipeline_source_queue:
+                self.run_pipeline(pipeline, override_pipeline_source_queue)
-                pipeline.set_source(self.receive_queue)
+            )
-            pipeline_task =  asyncio.create_task(pipeline.run_pipeline())
 
         try:
             while time.time() < self._expiration and not self._stop_threads.is_set():
@@ -182,6 +181,12 @@ class BaseTransportService:
         if self._vad_enabled:
             self._vad_thread.join()
 
+    async def run_pipeline(self, pipeline:Pipeline, override_pipeline_source_queue=True):
+        pipeline.set_sink(self.send_queue)
+        if override_pipeline_source_queue:
+            pipeline.set_source(self.receive_queue)
+        await pipeline.run_pipeline()
+
     async def run_interruptible_pipeline(
         self,
         pipeline: Pipeline,

src/dailyai/services/openai_api_llm_service.py

@@ -35,6 +35,9 @@ class BaseOpenAILLMService(LLMService):
     def __init__(self, model: str, api_key=None, base_url=None):
         super().__init__()
         self._model: str = model
+        self.create_client(api_key=api_key, base_url=base_url)
+
+    def create_client(self, api_key=None, base_url=None):
         self._client = AsyncOpenAI(api_key=api_key, base_url=base_url)
 
     async def _stream_chat_completions(

src/examples/starter-apps/chatbot.py

@@ -6,9 +6,7 @@ from PIL import Image
 from typing import AsyncGenerator
 
 from dailyai.pipeline.aggregators import (
-    LLMAssistantContextAggregator,
     LLMResponseAggregator,
-    LLMUserContextAggregator,
     UserResponseAggregator,
 )
 from dailyai.pipeline.frames import (
@@ -16,15 +14,12 @@ from dailyai.pipeline.frames import (
     SpriteFrame,
     Frame,
     LLMResponseEndFrame,
-    LLMResponseStartFrame,
     LLMMessagesQueueFrame,
-    UserStartedSpeakingFrame,
     AudioFrame,
     PipelineStartedFrame,
 )
 from dailyai.services.ai_services import AIService
 from dailyai.pipeline.pipeline import Pipeline
-from dailyai.services.ai_services import FrameLogger
 from dailyai.services.daily_transport_service import DailyTransportService
 from dailyai.services.open_ai_services import OpenAILLMService
 from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService
@@ -130,7 +125,7 @@ async def main(room_url: str, token):
         @transport.event_handler("on_first_other_participant_joined")
         async def on_first_other_participant_joined(transport):
             print(f"!!! in here, pipeline.source is {pipeline.source}")
-            await pipeline.queue_frames(LLMMessagesQueueFrame(messages))
+            await pipeline.queue_frames([LLMMessagesQueueFrame(messages)])
 
         async def run_conversation():
 

src/examples/starter-apps/storybot.py

@@ -27,6 +27,7 @@ from dailyai.pipeline.aggregators import (
 )
 from examples.support.runner import configure
 from dailyai.pipeline.frames import (
+    EndPipeFrame,
     LLMMessagesQueueFrame,
     TranscriptionQueueFrame,
     Frame,
@@ -187,10 +188,6 @@ class StoryImageGenerator(FrameProcessor):
 
 async def main(room_url: str, token):
     async with aiohttp.ClientSession() as session:
-        global transport
-        global llm
-        global tts
-
         messages = [
             {
                 "role": "system",
@@ -235,8 +232,15 @@ async def main(room_url: str, token):
             vad_stop_s=1.5,
         )
 
+        start_story_event = asyncio.Event()
+
         @transport.event_handler("on_first_other_participant_joined")
         async def on_first_other_participant_joined(transport):
+            start_story_event.set()
+
+        async def storytime():
+            await start_story_event.wait()
+
             # We're being a bit tricky here by using a special system prompt to
             # ask the user for a story topic. After their intial response, we'll
             # use a different system prompt to create story pages.
@@ -247,20 +251,17 @@ async def main(room_url: str, token):
                 }
             ]
             lca = LLMAssistantContextAggregator(messages)
-            await tts.run_to_queue(
+            local_pipeline = Pipeline([llm, lca, tts], sink=transport.send_queue)
-                transport.send_queue,
+            await local_pipeline.queue_frames(
-                lca.run(
+                [
-                    llm.run(
+                    ImageFrame(None, images["grandma-listening.png"]),
-                        [
+                    LLMMessagesQueueFrame(intro_messages),
-                            ImageFrame(None, images["grandma-listening.png"]),
+                    AudioFrame(sounds["listening.wav"]),
-                            LLMMessagesQueueFrame(intro_messages),
+                    EndPipeFrame(),
-                            AudioFrame(sounds["listening.wav"]),
+                ]
-                        ]
-                    ),
-                ),
             )
+            await local_pipeline.run_pipeline()
 
-        async def storytime():
             fl = FrameLogger("### After Image Generation")
             pipeline = Pipeline(
                 processors=[