PipelineTask: handle BotInterruptionFrame in PipelineTask

This allows having processors before the input transport or even not having an
input transport.

CHANGELOG.md

@@ -97,6 +97,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- `BotInterruptionFrame` frame handling has been moved from `BaseInputTransport`
+  to `PipelineTask`. This allows any type of pipeline to handle
+  `BotInterruptionFrame` without the need of an input transport.
+
 - `pipeline.tests.utils.run_test()` now allows passing `PipelineParams` instead
   of individual parameters.
 

src/pipecat/pipeline/task.py

@@ -23,6 +23,7 @@ from pipecat.audio.interruptions.base_interruption_strategy import BaseInterrupt
 from pipecat.clocks.base_clock import BaseClock
 from pipecat.clocks.system_clock import SystemClock
 from pipecat.frames.frames import (
+    BotInterruptionFrame,
     BotSpeakingFrame,
     CancelFrame,
     CancelTaskFrame,
@@ -36,6 +37,7 @@ from pipecat.frames.frames import (
     LLMFullResponseEndFrame,
     MetricsFrame,
     StartFrame,
+    StartInterruptionFrame,
     StopFrame,
     StopTaskFrame,
     TranscriptionFrame,
@@ -632,7 +634,11 @@ class PipelineTask(BasePipelineTask):
         if isinstance(frame, self._reached_upstream_types):
             await self._call_event_handler("on_frame_reached_upstream", frame)
 
-        if isinstance(frame, EndTaskFrame):
+        if isinstance(frame, BotInterruptionFrame) and self.params.allow_interruptions:
+            # Tell the pipeline we should interrupt.
+            logger.debug("Bot interruption")
+            await self.queue_frame(StartInterruptionFrame())
+        elif isinstance(frame, EndTaskFrame):
             # Tell the task we should end nicely.
             await self.queue_frame(EndFrame())
         elif isinstance(frame, CancelTaskFrame):

src/pipecat/transports/base_input.py

@@ -22,7 +22,6 @@ from pipecat.audio.turn.base_turn_analyzer import (
 )
 from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADState
 from pipecat.frames.frames import (
-    BotInterruptionFrame,
     BotStartedSpeakingFrame,
     BotStoppedSpeakingFrame,
     CancelFrame,
@@ -289,8 +288,6 @@ class BaseInputTransport(FrameProcessor):
         elif isinstance(frame, CancelFrame):
             await self.cancel(frame)
             await self.push_frame(frame, direction)
-        elif isinstance(frame, BotInterruptionFrame):
-            await self._handle_bot_interruption(frame)
         elif isinstance(frame, BotStartedSpeakingFrame):
             await self._handle_bot_started_speaking(frame)
             await self.push_frame(frame, direction)
@@ -335,13 +332,6 @@ class BaseInputTransport(FrameProcessor):
     # Handle interruptions
     #
 
-    async def _handle_bot_interruption(self, frame: BotInterruptionFrame):
-        """Handle bot interruption frames."""
-        logger.debug("Bot interruption")
-        if self.interruptions_allowed:
-            await self._start_interruption()
-            await self.push_frame(StartInterruptionFrame())
-
     async def _handle_user_interruption(self, frame: Frame):
         """Handle user interruption events based on speaking state."""
         if isinstance(frame, UserStartedSpeakingFrame):

tests/test_dtmf_aggregator.py

@@ -12,6 +12,7 @@ from pipecat.frames.frames import (
     KeypadEntry,
     TranscriptionFrame,
 )
+from pipecat.pipeline.task import PipelineParams
 from pipecat.processors.aggregators.dtmf_aggregator import DTMFAggregator
 from pipecat.tests.utils import SleepFrame, run_test
 
@@ -69,6 +70,8 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase):
             aggregator,
             frames_to_send=frames_to_send,
             expected_down_frames=expected_down_frames,
+            # TODO(aleix): we should handle StartInterruptionFrame
+            pipeline_params=PipelineParams(allow_interruptions=False),
         )
 
         # Find the TranscriptionFrames
@@ -105,6 +108,8 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase):
             aggregator,
             frames_to_send=frames_to_send,
             expected_down_frames=expected_down_frames,
+            # TODO(aleix): we should handle StartInterruptionFrame
+            pipeline_params=PipelineParams(allow_interruptions=False),
         )
 
         transcription_frames = [
@@ -134,6 +139,8 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase):
             aggregator,
             frames_to_send=frames_to_send,
             expected_down_frames=expected_down_frames,
+            # TODO(aleix): we should handle StartInterruptionFrame
+            pipeline_params=PipelineParams(allow_interruptions=False),
             send_end_frame=False,  # We're sending one in the test to test EndFrame logic
         )