diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 02cf6ce7b..fe9e5b90f 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -48,6 +48,7 @@ from pipecat.observers.base_observer import BaseObserver, FrameProcessed, FrameP from pipecat.processors.metrics.frame_processor_metrics import FrameProcessorMetrics from pipecat.utils.asyncio.task_manager import BaseTaskManager from pipecat.utils.base_object import BaseObject +from pipecat.utils.frame_queue import FrameQueue class FrameDirection(Enum): @@ -228,7 +229,7 @@ class FrameProcessor(BaseObject): # called. To resume processing frames we need to call # `resume_processing_frames()` which will wake up the event. self.__should_block_frames = False - self.__process_queue = asyncio.Queue() + self.__process_queue = FrameQueue(frame_getter=lambda item: item[0]) self.__process_event: Optional[asyncio.Event] = None self.__process_frame_task: Optional[asyncio.Task] = None self.__process_current_frame: Optional[Frame] = None @@ -818,9 +819,14 @@ class FrameProcessor(BaseObject): async def _start_interruption(self): """Start handling an interruption by cancelling current tasks.""" try: - if isinstance(self.__process_current_frame, UninterruptibleFrame): - # We don't want to cancel UninterruptibleFrame, so we simply - # cleanup the queue. + current_is_uninterruptible = isinstance( + self.__process_current_frame, UninterruptibleFrame + ) + if current_is_uninterruptible or self.__process_queue.has_uninterruptible: + # We don't want to cancel an UninterruptibleFrame (either the + # one currently being processed or one waiting in the queue), + # so we simply cleanup the queue keeping only + # UninterruptibleFrames. self.__reset_process_queue() else: # Cancel and re-create the process task. @@ -920,22 +926,7 @@ class FrameProcessor(BaseObject): def __reset_process_queue(self): """Reset non-system frame processing queue.""" - # Create a new queue to insert UninterruptibleFrame frames. - new_queue = asyncio.Queue() - - # Process current queue and keep UninterruptibleFrame frames. - while not self.__process_queue.empty(): - item = self.__process_queue.get_nowait() - frame = item[0] - if isinstance(frame, UninterruptibleFrame): - new_queue.put_nowait(item) - self.__process_queue.task_done() - - # Put back UninterruptibleFrame frames into our process queue. - while not new_queue.empty(): - item = new_queue.get_nowait() - self.__process_queue.put_nowait(item) - new_queue.task_done() + self.__process_queue.reset() async def __cancel_process_task(self): """Cancel the non-system frame processing task.""" diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 1d26d5e53..ee4571425 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -48,6 +48,7 @@ from pipecat.frames.frames import ( ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.transports.base_transport import TransportParams +from pipecat.utils.frame_queue import FrameQueue from pipecat.utils.time import nanoseconds_to_seconds BOT_VAD_STOP_SECS = 0.35 @@ -518,14 +519,20 @@ class BaseOutputTransport(FrameProcessor): _: The start interruption frame (unused). """ # Cancel tasks. - await self._cancel_audio_task() await self._cancel_clock_task() await self._cancel_video_task() + if self._audio_queue.has_uninterruptible: + # Keep the audio task running but drain all interruptible frames + # so the pending UninterruptibleFrames are still delivered. + self._audio_queue.reset() + else: + await self._cancel_audio_task() + self._create_audio_task() + # Create tasks. self._create_video_task() self._create_clock_task() - self._create_audio_task() # Let's send a bot stopped speaking if we have to. await self._bot_stopped_speaking() @@ -609,7 +616,7 @@ class BaseOutputTransport(FrameProcessor): def _create_audio_task(self): """Create the audio processing task.""" if not self._audio_task: - self._audio_queue = asyncio.Queue() + self._audio_queue = FrameQueue() self._audio_task = self._transport.create_task(self._audio_task_handler()) async def _cancel_audio_task(self):