starting on interruptions

src/dailyai/services/base_transport_service.py

@@ -18,6 +18,7 @@ from dailyai.pipeline.frames import (
     QueueFrame,
     SpriteQueueFrame,
     StartStreamQueueFrame,
+    TranscriptionQueueFrame,
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame
 )
@@ -108,6 +109,8 @@ class BaseTransportService():
         self.send_queue = asyncio.Queue()
         self.receive_queue = asyncio.Queue()
 
+        self.completed_queue = asyncio.Queue()
+
         self._threadsafe_send_queue = queue.Queue()
 
         self._images = None
@@ -167,21 +170,54 @@ class BaseTransportService():
         if self._vad_enabled:
             self._vad_thread.join()
 
-    async def run_pipeline(self, pipeline:Pipeline, allow_interruptions=True):
+    async def run_uninterruptible_pipeline(self, pipeline:Pipeline):
         pipeline.set_sink(self.send_queue)
-        if not allow_interruptions:
-            pipeline.set_source(self.receive_queue)
-            await pipeline.run_pipeline()
-        else:
-            source_queue = asyncio.Queue()
-            pipeline.set_source(source_queue)
-            pipeline.set_sink(self.send_queue)
-            pipeline_task = asyncio.create_task(pipeline.run_pipeline())
+        pipeline.set_source(self.receive_queue)
+        await pipeline.run_pipeline()
 
-            async for frame in self.get_receive_frames():
+    async def run_interruptible_pipeline(self, pipeline:Pipeline, allow_interruptions=True, pre_processor=None, post_processor=None):
+        pipeline.set_sink(self.send_queue)
+        source_queue = asyncio.Queue()
+        pipeline.set_source(source_queue)
+        pipeline.set_sink(self.send_queue)
+        pipeline_task = asyncio.create_task(pipeline.run_pipeline())
+
+        async def yield_frame(frame:QueueFrame) -> AsyncGenerator[QueueFrame, None]:
+            yield frame
+
+        async def post_process(post_processor):
+            if not post_processor:
+                return
+
+            while True:
+                frame = await self.completed_queue.get()
+                print("post-processing frame: ", frame.__class__.__name__)
+                await post_processor.process_frame(frame)
+
+                if isinstance(frame, EndStreamQueueFrame):
+                    break
+
+        post_process_task = asyncio.create_task(post_process(post_processor))
+
+        async for frame in self.get_receive_frames():
+            print("Got frame: ", frame.__class__.__name__)
+            if isinstance(frame, UserStartedSpeakingFrame):
+                pipeline_task.cancel()
+                self.interrupt()
+                pipeline_task = asyncio.create_task(pipeline.run_pipeline())
+
+            if pre_processor:
+                frame_generator = pre_processor.process_frame(frame)
+            else:
+                frame_generator = yield_frame(frame)
+
+            async for frame in frame_generator:
                 await source_queue.put(frame)
 
+            if isinstance(frame, EndStreamQueueFrame):
+                break
 
+        await asyncio.gather(pipeline_task, post_process_task)
 
     def _post_run(self):
         # Note that this function must be idempotent! It can be called multiple times
@@ -341,6 +377,10 @@ class BaseTransportService():
                     if isinstance(frame, EndStreamQueueFrame):
                         self._logger.info("Stopping frame consumer thread")
                         self._threadsafe_send_queue.task_done()
+                        if self._loop:
+                            asyncio.run_coroutine_threadsafe(
+                                self.completed_queue.put(frame), self._loop
+                            )
                         return
 
                     # if interrupted, we just pull frames off the queue and discard them
@@ -365,6 +405,11 @@ class BaseTransportService():
                         elif len(b):
                             self.write_frame_to_mic(bytes(b))
                             b = bytearray()
+
+                        if self._loop:
+                            asyncio.run_coroutine_threadsafe(
+                                self.completed_queue.put(frame), self._loop
+                            )
                     else:
                         # if there are leftover audio bytes, write them now; failing to do so
                         # can cause static in the audio stream.

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

@@ -5,8 +5,8 @@ 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.ai_services import FrameLogger
-from dailyai.pipeline.aggregators import LLMAssistantContextAggregator, LLMContextAggregator, LLMUserContextAggregator
-from support.runner import configure
+from dailyai.pipeline.aggregators import LLMAssistantContextAggregator, LLMUserContextAggregator
+from examples.foundational.support.runner import configure
 
 
 async def main(room_url: str, token):
@@ -46,8 +46,6 @@ async def main(room_url: str, token):
         tma_in = LLMUserContextAggregator(messages, transport._my_participant_id)
         tma_out = LLMAssistantContextAggregator(messages, transport._my_participant_id)
         pipeline = Pipeline(
-            source=transport.receive_queue,
-            sink=transport.send_queue,
             processors=[
                 fl,
                 tma_in,
@@ -57,7 +55,7 @@ async def main(room_url: str, token):
                 tts
             ],
         )
-        await pipeline.run_pipeline()
+        await transport.run_pipeline(pipeline)
 
     transport.transcription_settings["extra"]["endpointing"] = True
     transport.transcription_settings["extra"]["punctuate"] = True

src/examples/foundational/07-interruptible.py

@@ -2,8 +2,11 @@ import asyncio
 import aiohttp
 import os
 from dailyai.conversation_wrappers import InterruptibleConversationWrapper
+from dailyai.pipeline.aggregators import LLMAssistantContextAggregator, LLMUserContextAggregator
 
 from dailyai.pipeline.frames import StartStreamQueueFrame, TextQueueFrame
+from dailyai.pipeline.pipeline import Pipeline
+from dailyai.services.ai_services import FrameLogger
 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
@@ -32,17 +35,7 @@ async def main(room_url: str, token):
             api_key=os.getenv("AZURE_SPEECH_API_KEY"),
             region=os.getenv("AZURE_SPEECH_REGION"))
 
-        async def run_response(user_speech, tma_in, tma_out):
-            await tts.run_to_queue(
-                transport.send_queue,
-                tma_out.run(
-                    llm.run(
-                        tma_in.run(
-                            [StartStreamQueueFrame(), TextQueueFrame(user_speech)]
-                        )
-                    )
-                ),
-            )
+        pipeline = Pipeline([FrameLogger(), llm, FrameLogger(), tts])
 
         @transport.event_handler("on_first_other_participant_joined")
         async def on_first_other_participant_joined(transport):
@@ -53,14 +46,15 @@ async def main(room_url: str, token):
                 {"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."},
             ]
 
-            conversation_wrapper = InterruptibleConversationWrapper(
-                frame_generator=transport.get_receive_frames,
-                runner=run_response,
-                interrupt=transport.interrupt,
-                my_participant_id=transport._my_participant_id,
-                llm_messages=messages,
+            await transport.run_interruptible_pipeline(
+                pipeline,
+                post_processor=LLMAssistantContextAggregator(
+                    messages, transport._my_participant_id
+                ),
+                pre_processor=LLMUserContextAggregator(
+                    messages, transport._my_participant_id, complete_sentences=False
+                ),
             )
-            await conversation_wrapper.run_conversation()
 
         transport.transcription_settings["extra"]["punctuate"] = False
         await asyncio.gather(transport.run(), run_conversation())