Merge pull request #60 from daily-co/remove-ai-service-methods

Remove run_to_queue and run from AIService class

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "dailyai"
-version = "0.0.1"
+version = "0.0.3"
 description = "An open source framework for real-time, multi-modal, conversational AI applications"
 license = { text = "BSD 2-Clause License" }
 readme = "README.md"

src/dailyai/pipeline/merge_pipeline.py

@@ -0,0 +1,21 @@
+from typing import List
+from dailyai.pipeline.frames import EndFrame, EndPipeFrame
+from dailyai.pipeline.pipeline import Pipeline
+
+
+class SequentialMergePipeline(Pipeline):
+    """This class merges the sink queues from a list of pipelines. Frames from
+    each pipeline's sink are merged in the order of pipelines in the list."""
+    def __init__(self, pipelines:List[Pipeline]):
+        super().__init__([])
+        self.pipelines = pipelines
+
+    async def run_pipeline(self):
+        for pipeline in self.pipelines:
+            while True:
+                frame = await pipeline.sink.get()
+                if isinstance(frame, EndFrame) or isinstance(frame, EndPipeFrame):
+                    break
+                await self.sink.put(frame)
+
+        await self.sink.put(EndFrame())

src/dailyai/pipeline/pipeline.py

@@ -1,5 +1,5 @@
 import asyncio
-from typing import AsyncGenerator, List
+from typing import AsyncGenerator, AsyncIterable, Iterable, List
 from dailyai.pipeline.frame_processor import FrameProcessor
 
 from dailyai.pipeline.frames import EndPipeFrame, EndFrame, Frame
@@ -17,17 +17,17 @@ class Pipeline:
         self,
         processors: List[FrameProcessor],
         source: asyncio.Queue | None = None,
-        sink: asyncio.Queue[Frame] | None = None,
+        sink: asyncio.Queue[Frame] | None = None
     ):
-        """Create a new pipeline. By default neither the source nor sink
+        """Create a new pipeline. By default we create the sink and source queues
-        queues are set, so you'll need to pass them to this constructor or
+        if they're not provided, but these can be overridden to point to other
-        call set_source and set_sink before using the pipeline. Note that
+        queues. If this pipeline is run by a transport, its sink and source queues
-        the transport's run_*_pipeline methods will set the source and sink
+        will be overridden.
-        queues on the pipeline for you.
         """
-        self.processors = processors
+        self.processors: List[FrameProcessor] = processors
-        self.source: asyncio.Queue[Frame] | None = source
+
-        self.sink: asyncio.Queue[Frame] | None = sink
+        self.source: asyncio.Queue[Frame] = source or asyncio.Queue()
+        self.sink: asyncio.Queue[Frame] = sink or asyncio.Queue()
 
     def set_source(self, source: asyncio.Queue[Frame]):
         """Set the source queue for this pipeline. Frames from this queue
@@ -44,21 +44,24 @@ class Pipeline:
         """Convenience function to get the next frame from the source queue. This
         lets us consistently have an AsyncGenerator yield frames, from either the
         source queue or a frame_processor."""
-        if self.source is None:
+
-            raise ValueError("Source queue not set")
         yield await self.source.get()
 
-    async def run_pipeline_recursively(
+    async def queue_frames(
-        self, initial_frame: Frame, processors: List[FrameProcessor]
+        self,
-    ) -> AsyncGenerator[Frame, None]:
+        frames: Iterable[Frame] | AsyncIterable[Frame],
-        if processors:
+    ) -> None:
-            async for frame in processors[0].process_frame(initial_frame):
+        """Insert frames directly into a pipeline. This is typically used inside a transport
-                async for final_frame in self.run_pipeline_recursively(
+        participant_joined callback to prompt a bot to start a conversation, for example."""
-                    frame, processors[1:]
+
-                ):
+        if isinstance(frames, AsyncIterable):
-                    yield final_frame
+            async for frame in frames:
+                await self.source.put(frame)
+        elif isinstance(frames, Iterable):
+            for frame in frames:
+                await self.source.put(frame)
         else:
-            yield initial_frame
+            raise Exception("Frames must be an iterable or async iterable")
 
     async def run_pipeline(self):
         """Run the pipeline. Take each frame from the source queue, pass it to
@@ -73,13 +76,10 @@ class Pipeline:
         if it's not the last frame yielded by the last frame_processor in the pipeline..
         """
 
-        if self.source is None or self.sink is None:
-            raise ValueError("Source or sink queue not set")
-
         try:
             while True:
                 initial_frame = await self.source.get()
-                async for frame in self.run_pipeline_recursively(
+                async for frame in self._run_pipeline_recursively(
                     initial_frame, self.processors
                 ):
                     await self.sink.put(frame)
@@ -94,11 +94,17 @@ class Pipeline:
                 await processor.interrupted()
             pass
 
-    async def queue_frames(self, frames: Frame | List[Frame]):
+    async def _run_pipeline_recursively(
-        """Insert frames directly into a pipeline. This is typically used inside a transport
+        self, initial_frame: Frame, processors: List[FrameProcessor]
-        participant_joined callback to prompt a bot to start a conversation, for example.
+    ) -> AsyncGenerator[Frame, None]:
-        """
+        """Internal function to add frames to the pipeline as they're yielded
-        if not isinstance(frames, List):
+        by each processor."""
-            frames = [frames]
+        if processors:
-        for f in frames:
+            async for frame in processors[0].process_frame(initial_frame):
-            await self.source.put(f)
+                async for final_frame in self._run_pipeline_recursively(
+                    frame, processors[1:]
+                ):
+                    yield final_frame
+        else:
+            yield initial_frame
+

src/dailyai/services/ai_services.py

@@ -8,6 +8,7 @@ from dailyai.pipeline.frame_processor import FrameProcessor
 from dailyai.pipeline.frames import (
     AudioFrame,
     EndFrame,
+    EndPipeFrame,
     ImageFrame,
     LLMMessagesQueueFrame,
     LLMResponseEndFrame,
@@ -20,53 +21,13 @@ from dailyai.pipeline.frames import (
 )
 
 from abc import abstractmethod
-from typing import AsyncGenerator, AsyncIterable, BinaryIO, Iterable, List
+from typing import AsyncGenerator, BinaryIO
 
 
 class AIService(FrameProcessor):
-
     def __init__(self):
         self.logger = logging.getLogger("dailyai")
 
-    def stop(self):
-        pass
-
-    async def run_to_queue(
-        self, queue: asyncio.Queue, frames, add_end_of_stream=False
-    ) -> None:
-        async for frame in self.run(frames):
-            await queue.put(frame)
-
-        if add_end_of_stream:
-            await queue.put(EndFrame())
-
-    async def run(
-        self,
-        frames: Iterable[Frame] | AsyncIterable[Frame] | asyncio.Queue[Frame],
-    ) -> AsyncGenerator[Frame, None]:
-        try:
-            if isinstance(frames, AsyncIterable):
-                async for frame in frames:
-                    async for output_frame in self.process_frame(frame):
-                        yield output_frame
-            elif isinstance(frames, Iterable):
-                for frame in frames:
-                    async for output_frame in self.process_frame(frame):
-                        yield output_frame
-            elif isinstance(frames, asyncio.Queue):
-                while True:
-                    frame = await frames.get()
-                    async for output_frame in self.process_frame(frame):
-                        yield output_frame
-                    if isinstance(frame, EndFrame):
-                        break
-            else:
-                raise Exception("Frames must be an iterable or async iterable")
-        except Exception as e:
-            self.logger.error("Exception occurred while running AI service", e)
-            raise e
-
-
 class LLMService(AIService):
     """This class is a no-op but serves as a base class for LLM services."""
 
@@ -92,7 +53,7 @@ class TTSService(AIService):
         yield bytes()
 
     async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]:
-        if isinstance(frame, EndFrame):
+        if isinstance(frame, EndFrame) or isinstance(frame, EndPipeFrame):
             if self.current_sentence:
                 async for audio_chunk in self.run_tts(self.current_sentence):
                     yield AudioFrame(audio_chunk)
@@ -118,12 +79,6 @@ class TTSService(AIService):
             # note we pass along the text frame *after* the audio, so the text frame is completed after the audio is processed.
             yield TextFrame(text)
 
-    # 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, [LLMResponseStartFrame(), TextFrame(sentence), LLMResponseEndFrame()]
-        )
-
 
 class ImageGenService(AIService):
     def __init__(self, image_size, **kwargs):

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

@@ -20,11 +20,12 @@ from dailyai.pipeline.frames import (
     PipelineStartedFrame,
     SpriteFrame,
     StartFrame,
-    TranscriptionQueueFrame,
+    TextFrame,
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
 )
 from dailyai.pipeline.pipeline import Pipeline
+from dailyai.services.ai_services import TTSService
 
 torch.set_num_threads(1)
 
@@ -125,7 +126,7 @@ class BaseTransportService:
 
         self._logger: logging.Logger = logging.getLogger()
 
-    async def run(self):
+    async def run(self, pipeline:Pipeline | None=None, override_pipeline_source_queue=True):
         self._prerun()
 
         async_output_queue_marshal_task = asyncio.create_task(self._marshal_frames())
@@ -148,6 +149,12 @@ class BaseTransportService:
             self._vad_thread = threading.Thread(target=self._vad, daemon=True)
             self._vad_thread.start()
 
+        pipeline_task = None
+        if pipeline:
+            pipeline_task = asyncio.create_task(
+                self.run_pipeline(pipeline, override_pipeline_source_queue)
+            )
+
         try:
             while time.time() < self._expiration and not self._stop_threads.is_set():
                 await asyncio.sleep(1)
@@ -160,9 +167,12 @@ class BaseTransportService:
 
         self._stop_threads.set()
 
+        if pipeline_task:
+            pipeline_task.cancel()
+
         await self.send_queue.put(EndFrame())
+
         await async_output_queue_marshal_task
-        await self.send_queue.join()
         self._frame_consumer_thread.join()
 
         if self._speaker_enabled:
@@ -171,9 +181,10 @@ class BaseTransportService:
         if self._vad_enabled:
             self._vad_thread.join()
 
-    async def run_uninterruptible_pipeline(self, pipeline: Pipeline):
+    async def run_pipeline(self, pipeline:Pipeline, override_pipeline_source_queue=True):
         pipeline.set_sink(self.send_queue)
-        pipeline.set_source(self.receive_queue)
+        if override_pipeline_source_queue:
+            pipeline.set_source(self.receive_queue)
         await pipeline.run_pipeline()
 
     async def run_interruptible_pipeline(
@@ -232,6 +243,11 @@ class BaseTransportService:
 
         await asyncio.gather(pipeline_task, post_process_task)
 
+    async def say(self, text:str, tts:TTSService):
+        """Say a phrase. Use with caution; this bypasses any running pipelines."""
+        async for frame in tts.process_frame(TextFrame(text)):
+            await self.send_queue.put(frame)
+
     def _post_run(self):
         # Note that this function must be idempotent! It can be called multiple times
         # if, for example, a keyboard interrupt occurs.
@@ -399,6 +415,7 @@ class BaseTransportService:
                 for frame in frames:
                     if isinstance(frame, EndFrame):
                         self._logger.info("Stopping frame consumer thread")
+                        self._stop_threads.set()
                         self._threadsafe_send_queue.task_done()
                         if self._loop:
                             asyncio.run_coroutine_threadsafe(

src/dailyai/services/daily_transport_service.py

@@ -219,7 +219,8 @@ class DailyTransportService(BaseTransportService, EventHandler):
         pass
 
     def call_joined(self, join_data, client_error):
-        self._logger.info(f"Call_joined: {join_data}, {client_error}")
+        #self._logger.info(f"Call_joined: {join_data}, {client_error}")
+        pass
 
     def dialout(self, number):
         self.client.start_dialout({"phoneNumber": number})

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/dailyai/tests/test_ai_services.py

@@ -12,7 +12,7 @@ class SimpleAIService(AIService):
 
 
 class TestBaseAIService(unittest.IsolatedAsyncioTestCase):
-    async def test_async_input(self):
+    async def test_simple_processing(self):
         service = SimpleAIService()
 
         input_frames = [
@@ -20,28 +20,10 @@ class TestBaseAIService(unittest.IsolatedAsyncioTestCase):
             EndFrame()
         ]
 
-        async def iterate_frames() -> AsyncGenerator[Frame, None]:
-            for frame in input_frames:
-                yield frame
-
         output_frames = []
-        async for frame in service.run(iterate_frames()):
+        for input_frame in input_frames:
-            output_frames.append(frame)
+            async for output_frame in service.process_frame(input_frame):
-
+                output_frames.append(output_frame)
-        self.assertEqual(input_frames, output_frames)
-
-    async def test_nonasync_input(self):
-        service = SimpleAIService()
-
-        input_frames = [TextFrame("hello"), EndFrame()]
-
-        def iterate_frames() -> Generator[Frame, None, None]:
-            for frame in input_frames:
-                yield frame
-
-        output_frames = []
-        async for frame in service.run(iterate_frames()):
-            output_frames.append(frame)
 
         self.assertEqual(input_frames, output_frames)
 

src/examples/foundational/01-say-one-thing.py

@@ -2,6 +2,8 @@ import asyncio
 import aiohttp
 import logging
 import os
+from dailyai.pipeline.frames import EndFrame, TextFrame
+from dailyai.pipeline.pipeline import Pipeline
 
 from dailyai.services.daily_transport_service import DailyTransportService
 from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService
@@ -28,19 +30,7 @@ async def main(room_url):
             voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
         )
 
-        other_joined_event = asyncio.Event()
+        pipeline = Pipeline([tts])
-        participant_name = ''
-
-        async def say_hello():
-            nonlocal tts
-            nonlocal participant_name
-
-            await other_joined_event.wait()
-            await tts.say(
-                "Hello there, " + participant_name + "!",
-                transport.send_queue,
-            )
-            await transport.stop_when_done()
 
         # Register an event handler so we can play the audio when the participant joins.
         @transport.event_handler("on_participant_joined")
@@ -48,11 +38,10 @@ async def main(room_url):
             if participant["info"]["isLocal"]:
                 return
 
-            nonlocal participant_name
             participant_name = participant["info"]["userName"] or ''
-            other_joined_event.set()
+            await pipeline.queue_frames([TextFrame("Hello there, " + participant_name + "!"), EndFrame()])
 
-        await asyncio.gather(transport.run(), say_hello())
+        await transport.run(pipeline)
         del tts
 
 

src/examples/foundational/02-llm-say-one-thing.py

@@ -4,7 +4,8 @@ import logging
 
 import aiohttp
 
-from dailyai.pipeline.frames import LLMMessagesQueueFrame
+from dailyai.pipeline.frames import EndFrame, LLMMessagesQueueFrame
+from dailyai.pipeline.pipeline import Pipeline
 from dailyai.services.daily_transport_service import DailyTransportService
 from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService
 from dailyai.services.open_ai_services import OpenAILLMService
@@ -42,20 +43,13 @@ async def main(room_url):
             }
         ]
 
-        other_joined_event = asyncio.Event()
+        pipeline= Pipeline([llm, tts])
-        async def speak_from_llm():
-            await other_joined_event.wait()
-            await tts.run_to_queue(
-                transport.send_queue,
-                llm.run([LLMMessagesQueueFrame(messages)])
-            )
-            await transport.stop_when_done()
 
         @transport.event_handler("on_first_other_participant_joined")
         async def on_first_other_participant_joined(transport):
-            other_joined_event.set()
+            await pipeline.queue_frames([LLMMessagesQueueFrame(messages), EndFrame()])
 
-        await asyncio.gather(transport.run(), speak_from_llm())
+        await transport.run(pipeline)
 
 
 if __name__ == "__main__":

src/examples/foundational/03-still-frame.py

@@ -3,7 +3,8 @@ import aiohttp
 import logging
 import os
 
-from dailyai.pipeline.frames import TextFrame
+from dailyai.pipeline.frames import EndFrame, TextFrame
+from dailyai.pipeline.pipeline import Pipeline
 from dailyai.services.daily_transport_service import DailyTransportService
 from dailyai.services.fal_ai_services import FalImageGenService
 
@@ -33,19 +34,18 @@ async def main(room_url):
             key_secret=os.getenv("FAL_KEY_SECRET"),
         )
 
-        other_joined_event = asyncio.Event()
+        pipeline = Pipeline([imagegen])
-
-        async def show_image():
-            await other_joined_event.wait()
-            await imagegen.run_to_queue(
-                transport.send_queue, [TextFrame("a cat in the style of picasso")]
-            )
 
         @transport.event_handler("on_first_other_participant_joined")
         async def on_first_other_participant_joined(transport):
-            other_joined_event.set()
+            # Note that we do not put an EndFrame() item in the pipeline for this demo.
+            # This means that the bot will stay in the channel until it times out.
+            # An EndFrame() in the pipeline would cause the transport to shut down.
+            await pipeline.queue_frames(
+                [TextFrame("a cat in the style of picasso")]
+            )
 
-        await asyncio.gather(transport.run(), show_image())
+        await transport.run(pipeline)
 
 
 if __name__ == "__main__":

src/examples/foundational/04-utterance-and-speech.py

@@ -3,12 +3,13 @@ import logging
 import os
 
 import aiohttp
+from dailyai.pipeline.merge_pipeline import SequentialMergePipeline
 from dailyai.pipeline.pipeline import Pipeline
 
 from dailyai.services.daily_transport_service import DailyTransportService
 from dailyai.services.azure_ai_services import AzureLLMService, AzureTTSService
 from dailyai.services.deepgram_ai_services import DeepgramTTSService
-from dailyai.pipeline.frames import EndFrame, LLMMessagesQueueFrame
+from dailyai.pipeline.frames import EndFrame, EndPipeFrame, LLMMessagesQueueFrame, TextFrame
 from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService
 from examples.support.runner import configure
 
@@ -53,49 +54,24 @@ async def main(room_url: str):
         # Start a task to run the LLM to create a joke, and convert the LLM output to audio frames. This task
         # will run in parallel with generating and speaking the audio for static text, so there's no delay to
         # speak the LLM response.
-        buffer_queue = asyncio.Queue()
+        llm_pipeline = Pipeline([llm, elevenlabs_tts])
-        source_queue = asyncio.Queue()
+        await llm_pipeline.queue_frames([LLMMessagesQueueFrame(messages), EndPipeFrame()])
-        pipeline = Pipeline(
+
-            source=source_queue, sink=buffer_queue, processors=[llm, elevenlabs_tts]
+        simple_tts_pipeline = Pipeline([azure_tts])
+        await simple_tts_pipeline.queue_frames(
+            [
+                TextFrame("My friend the LLM is going to tell a joke about llamas"),
+                EndPipeFrame(),
+            ]
         )
-        await source_queue.put(LLMMessagesQueueFrame(messages))
-        await source_queue.put(EndFrame())
-        pipeline_run_task = pipeline.run_pipeline()
 
-        other_participant_joined = asyncio.Event()
+        merge_pipeline = SequentialMergePipeline([simple_tts_pipeline, llm_pipeline])
 
-        @transport.event_handler("on_first_other_participant_joined")
+        await asyncio.gather(
-        async def on_first_other_participant_joined(transport):
+            transport.run(merge_pipeline),
-            other_participant_joined.set()
+            simple_tts_pipeline.run_pipeline(),
-
+            llm_pipeline.run_pipeline(),
-        async def say_something():
+        )
-            await other_participant_joined.wait()
-
-            await azure_tts.say(
-                "My friend the LLM is now going to tell a joke about llamas.",
-                transport.send_queue,
-            )
-
-            # khk: deepgram_tts.say() doesn't seem to put bytes in the transport
-            # queue. I get a debug log line that indicates we're set up okay, but
-            # no further log lines or audio bytes. debug this later:
-            #   20 2024-03-10 13:24:46,235 Running deepgram tts for My friend the LLM is now going to tell a joke about llamas.
-            # await deepgram_tts.say(
-            #    "My friend the LLM is now going to tell a joke about llamas.",
-            #    transport.send_queue,
-            # )
-
-            async def buffer_to_send_queue():
-                while True:
-                    frame = await buffer_queue.get()
-                    await transport.send_queue.put(frame)
-                    buffer_queue.task_done()
-                    if isinstance(frame, EndFrame):
-                        break
-
-            await asyncio.gather(pipeline_run_task, buffer_to_send_queue())
-
-        await asyncio.gather(transport.run(), say_something())
 
 
 if __name__ == "__main__":

src/examples/foundational/05-sync-speech-and-image.py

@@ -89,8 +89,28 @@ async def main(room_url):
             key_secret=os.getenv("FAL_KEY_SECRET"),
         )
 
-        source_queue = asyncio.Queue()
+        gated_aggregator = GatedAggregator(
+            gate_open_fn=lambda frame: isinstance(frame, ImageFrame),
+            gate_close_fn=lambda frame: isinstance(frame, LLMResponseStartFrame),
+            start_open=False,
+        )
 
+        sentence_aggregator = SentenceAggregator()
+        month_prepender = MonthPrepender()
+        llm_full_response_aggregator = LLMFullResponseAggregator()
+
+        pipeline = Pipeline(
+            processors=[
+                llm,
+                sentence_aggregator,
+                ParallelPipeline(
+                    [[month_prepender, tts], [llm_full_response_aggregator, imagegen]]
+                ),
+                gated_aggregator,
+            ],
+        )
+
+        frames = []
         for month in [
             "January",
             "February",
@@ -111,47 +131,13 @@ async def main(room_url):
                     "content": f"Describe a nature photograph suitable for use in a calendar, for the month of {month}. Include only the image description with no preamble. Limit the description to one sentence, please.",
                 }
             ]
-            await source_queue.put(MonthFrame(month))
+            frames.append(MonthFrame(month))
-            await source_queue.put(LLMMessagesQueueFrame(messages))
+            frames.append(LLMMessagesQueueFrame(messages))
 
-        await source_queue.put(EndFrame())
+        frames.append(EndFrame())
+        await pipeline.queue_frames(frames)
 
-        gated_aggregator = GatedAggregator(
+        await transport.run(pipeline, override_pipeline_source_queue=False)
-            gate_open_fn=lambda frame: isinstance(frame, ImageFrame),
-            gate_close_fn=lambda frame: isinstance(frame, LLMResponseStartFrame),
-            start_open=False,
-        )
-
-        sentence_aggregator = SentenceAggregator()
-        month_prepender = MonthPrepender()
-        llm_full_response_aggregator = LLMFullResponseAggregator()
-
-        pipeline = Pipeline(
-            source=source_queue,
-            sink=transport.send_queue,
-            processors=[
-                llm,
-                sentence_aggregator,
-                ParallelPipeline(
-                    [[month_prepender, tts], [llm_full_response_aggregator, imagegen]]
-                ),
-                gated_aggregator,
-            ],
-        )
-        pipeline_task = pipeline.run_pipeline()
-
-        other_joined = asyncio.Event()
-
-        @transport.event_handler("on_first_other_participant_joined")
-        async def on_first_other_participant_joined(transport):
-            other_joined.set()
-
-        async def show_calendar():
-            await other_joined.wait()
-            await pipeline_task
-            await transport.stop_when_done()
-
-        await asyncio.gather(transport.run(), show_calendar())
 
 
 if __name__ == "__main__":

src/examples/foundational/06-listen-and-respond.py

@@ -2,6 +2,7 @@ import asyncio
 import aiohttp
 import logging
 import os
+from dailyai.pipeline.frames import LLMMessagesQueueFrame
 from dailyai.pipeline.pipeline import Pipeline
 
 from dailyai.services.daily_transport_service import DailyTransportService
@@ -44,38 +45,37 @@ async def main(room_url: str, token):
         )
         fl = FrameLogger("Inner")
         fl2 = FrameLogger("Outer")
+        messages = [
+            {
+                "role": "system",
+                "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio. Respond to what the user said in a creative and helpful way.",
+            },
+        ]
+
+        tma_in = LLMUserContextAggregator(messages, transport._my_participant_id)
+        tma_out = LLMAssistantContextAggregator(
+            messages, transport._my_participant_id
+        )
+        pipeline = Pipeline(
+            processors=[
+                fl,
+                tma_in,
+                llm,
+                fl2,
+                tts,
+                tma_out,
+            ],
+        )
 
         @transport.event_handler("on_first_other_participant_joined")
         async def on_first_other_participant_joined(transport):
-            await tts.say("Hi, I'm listening!", transport.send_queue)
+            # Kick off the conversation.
-
+            messages.append({"role": "system", "content": "Please introduce yourself to the user."})
-        async def have_conversation():
+            await pipeline.queue_frames([LLMMessagesQueueFrame(messages)])
-            messages = [
-                {
-                    "role": "system",
-                    "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio. Respond to what the user said in a creative and helpful way.",
-                },
-            ]
-
-            tma_in = LLMUserContextAggregator(messages, transport._my_participant_id)
-            tma_out = LLMAssistantContextAggregator(
-                messages, transport._my_participant_id
-            )
-            pipeline = Pipeline(
-                processors=[
-                    fl,
-                    tma_in,
-                    llm,
-                    fl2,
-                    tts,
-                    tma_out,
-                ],
-            )
-            await transport.run_uninterruptible_pipeline(pipeline)
 
         transport.transcription_settings["extra"]["endpointing"] = True
         transport.transcription_settings["extra"]["punctuate"] = True
-        await asyncio.gather(transport.run(), have_conversation())
+        await transport.run(pipeline)
 
 
 if __name__ == "__main__":

src/examples/foundational/07-interruptible.py

@@ -49,7 +49,7 @@ async def main(room_url: str, token):
 
         @transport.event_handler("on_first_other_participant_joined")
         async def on_first_other_participant_joined(transport):
-            await tts.say("Hi, I'm listening!", transport.send_queue)
+            await transport.say("Hi, I'm listening!", tts)
 
         async def run_conversation():
             messages = [

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=[
@@ -273,7 +274,7 @@ async def main(room_url: str, token):
                     lra,
                 ]
             )
-            await transport.run_uninterruptible_pipeline(
+            await transport.run_pipeline(
                 pipeline,
             )