Code review feedback

CHANGELOG.md

@@ -9,10 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
-- Added `interruption_config` to `PipelineParams` which uses an
+- Added `interruption_strategies` to `PipelineParams` using
-  `InterruptionConfig` to specify criteria required to interrupt the bot when
+  `MinWordsInterruptionStrategy` to specify minimum words required to interrupt
-  it's speaking. You can specify `min_words` to require the user to say at
+  the bot when it's speaking. Use
-  least `min_words` words before their speech will interrupt the bot. If not
+  `interruption_strategies=[MinWordsInterruptionStrategy(min_words=N)]` to
+  require users to speak at least N words before interrupting. If not
   specified, the normal interruption behavior applies.
 
 - `BaseInputTransport` now handles `StopFrame`. When a `StopFrame` is received

examples/foundational/42-interruption-config.py

@@ -11,7 +11,7 @@ from dotenv import load_dotenv
 from loguru import logger
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
-from pipecat.frames.frames import InterruptionConfig
+from pipecat.frames.frames import MinWordsInterruptionStrategy
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
@@ -92,7 +92,7 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si
             enable_metrics=True,
             enable_usage_metrics=True,
             report_only_initial_ttfb=True,
-            interruption_config=InterruptionConfig(min_words=3),
+            interruption_strategies=[MinWordsInterruptionStrategy(min_words=3)],
         ),
     )
 

src/pipecat/frames/frames.py

@@ -15,6 +15,7 @@ from typing import (
     Literal,
     Mapping,
     Optional,
+    Sequence,
     Tuple,
 )
 
@@ -439,17 +440,25 @@ class OutputDTMFFrame(DTMFFrame, DataFrame):
 
 
 @dataclass
-class InterruptionConfig:
+class InterruptionStrategy:
-    """Configuration for interruption behavior.
+    """Base class for interruption strategies."""
 
-    When specified, the bot will not be interrupted immediately when the user speaks.
+    pass
-    Instead, interruption will only occur when the configured conditions are met.
+
+
+@dataclass
+class MinWordsInterruptionStrategy(InterruptionStrategy):
+    """Strategy for interruption behavior based on a minimum number of words spoken by the user.
 
     Args:
         min_words: If set, user must speak at least this many words to interrupt
     """
 
-    min_words: Optional[int] = None
+    min_words: int
+
+    def __post_init__(self):
+        if self.min_words <= 0:
+            raise ValueError("min_words must be greater than 0")
 
 
 @dataclass
@@ -462,7 +471,7 @@ class StartFrame(SystemFrame):
     enable_metrics: bool = False
     enable_usage_metrics: bool = False
     report_only_initial_ttfb: bool = False
-    interruption_config: Optional[InterruptionConfig] = None
+    interruption_strategies: Optional[Sequence[InterruptionStrategy]] = None
 
 
 @dataclass

src/pipecat/pipeline/task.py

@@ -6,7 +6,7 @@
 
 import asyncio
 import time
-from typing import Any, AsyncIterable, Dict, Iterable, List, Optional, Tuple, Type
+from typing import Any, AsyncIterable, Dict, Iterable, List, Optional, Sequence, Tuple, Type
 
 from loguru import logger
 from pydantic import BaseModel, ConfigDict, Field
@@ -22,7 +22,7 @@ from pipecat.frames.frames import (
     ErrorFrame,
     Frame,
     HeartbeatFrame,
-    InterruptionConfig,
+    InterruptionStrategy,
     LLMFullResponseEndFrame,
     MetricsFrame,
     StartFrame,
@@ -59,7 +59,7 @@ class PipelineParams(BaseModel):
         report_only_initial_ttfb: Whether to report only initial time to first byte.
         send_initial_empty_metrics: Whether to send initial empty metrics.
         start_metadata: Additional metadata for pipeline start.
-        interruption_config: Configuration for bot interruption behavior.
+        interruption_strategies: Strategies for bot interruption behavior.
     """
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
@@ -75,7 +75,7 @@ class PipelineParams(BaseModel):
     report_only_initial_ttfb: bool = False
     send_initial_empty_metrics: bool = True
     start_metadata: Dict[str, Any] = Field(default_factory=dict)
-    interruption_config: Optional[InterruptionConfig] = None
+    interruption_strategies: Optional[Sequence[InterruptionStrategy]] = None
 
 
 class PipelineTaskSource(FrameProcessor):
@@ -521,7 +521,7 @@ class PipelineTask(BaseTask):
             enable_metrics=self._params.enable_metrics,
             enable_usage_metrics=self._params.enable_usage_metrics,
             report_only_initial_ttfb=self._params.report_only_initial_ttfb,
-            interruption_config=self._params.interruption_config,
+            interruption_strategies=self._params.interruption_strategies,
         )
         start_frame.metadata = self._params.start_metadata
         await self._source.queue_frame(start_frame, FrameDirection.DOWNSTREAM)

src/pipecat/processors/aggregators/llm_response.py

@@ -24,7 +24,6 @@ from pipecat.frames.frames import (
     FunctionCallInProgressFrame,
     FunctionCallResultFrame,
     InterimTranscriptionFrame,
-    InterruptionConfig,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesAppendFrame,
@@ -33,6 +32,7 @@ from pipecat.frames.frames import (
     LLMSetToolChoiceFrame,
     LLMSetToolsFrame,
     LLMTextFrame,
+    MinWordsInterruptionStrategy,
     OpenAILLMContextAssistantTimestampFrame,
     StartFrame,
     StartInterruptionFrame,
@@ -195,7 +195,7 @@ class LLMContextResponseAggregator(BaseLLMResponseAggregator):
         self._context = context
         self._role = role
 
-        self._aggregation = ""
+        self._aggregation: str = ""
 
     @property
     def messages(self) -> List[dict]:
@@ -268,8 +268,6 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
         self._seen_interim_results = False
         self._waiting_for_aggregation = False
 
-        self._interruption_config: Optional[InterruptionConfig] = None
-
         self._aggregation_event = asyncio.Event()
         self._aggregation_task = None
 
@@ -335,8 +333,8 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
     async def push_aggregation(self):
         """Pushes the current aggregation based on interruption configuration and conditions."""
         if len(self._aggregation) > 0:
-            if self._interruption_config and self._bot_speaking:
+            if self.interruption_strategies and self._bot_speaking:
-                should_interrupt = await self._should_interrupt_based_on_config()
+                should_interrupt = self._should_interrupt_based_on_strategies()
 
                 if should_interrupt:
                     logger.debug(
@@ -352,18 +350,25 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
                 # No interruption config - normal behavior (always push aggregation)
                 await self._process_aggregation()
 
-    async def _should_interrupt_based_on_config(self) -> bool:
+    def _should_interrupt_based_on_strategies(self) -> bool:
-        """Check if interruption should occur based on configured conditions."""
+        """Check if interruption should occur based on configured strategies."""
-        assert self._interruption_config is not None
+        if not self.interruption_strategies:
-
-        if not self._aggregation or self._interruption_config.min_words is None:
             return False
 
+        # Check strategies one by one until first match
+        for strategy in self.interruption_strategies:
+            if isinstance(strategy, MinWordsInterruptionStrategy):
+                if self._should_interrupt_min_words(strategy):
+                    return True
+
+        return False
+
+    def _should_interrupt_min_words(self, strategy: MinWordsInterruptionStrategy) -> bool:
+        """Check if word count threshold is met."""
         word_count = len(self._aggregation.split())
-        return word_count >= self._interruption_config.min_words
+        return word_count >= strategy.min_words
 
     async def _start(self, frame: StartFrame):
-        self._interruption_config = frame.interruption_config
         self._create_aggregation_task()
 
     async def _stop(self, frame: EndFrame):

src/pipecat/processors/frame_processor.py

@@ -7,7 +7,7 @@
 import asyncio
 from dataclasses import dataclass
 from enum import Enum
-from typing import Awaitable, Callable, Coroutine, Optional
+from typing import Awaitable, Callable, Coroutine, Optional, Sequence
 
 from loguru import logger
 
@@ -16,6 +16,7 @@ from pipecat.frames.frames import (
     CancelFrame,
     ErrorFrame,
     Frame,
+    InterruptionStrategy,
     StartFrame,
     StartInterruptionFrame,
     StopInterruptionFrame,
@@ -67,6 +68,7 @@ class FrameProcessor(BaseObject):
         self._enable_metrics = False
         self._enable_usage_metrics = False
         self._report_only_initial_ttfb = False
+        self._interruption_strategies: Optional[Sequence[InterruptionStrategy]] = None
 
         # Indicates whether we have received the StartFrame.
         self.__started = False
@@ -119,6 +121,10 @@ class FrameProcessor(BaseObject):
     def report_only_initial_ttfb(self):
         return self._report_only_initial_ttfb
 
+    @property
+    def interruption_strategies(self) -> Optional[Sequence[InterruptionStrategy]]:
+        return self._interruption_strategies
+
     def can_generate_metrics(self) -> bool:
         return False
 
@@ -272,6 +278,7 @@ class FrameProcessor(BaseObject):
         self._enable_metrics = frame.enable_metrics
         self._enable_usage_metrics = frame.enable_usage_metrics
         self._report_only_initial_ttfb = frame.report_only_initial_ttfb
+        self._interruption_strategies = frame.interruption_strategies
         self.__create_input_task()
         self.__create_push_task()
 

src/pipecat/transports/base_input.py

@@ -27,7 +27,6 @@ from pipecat.frames.frames import (
     Frame,
     InputAudioRawFrame,
     InputImageRawFrame,
-    InterruptionConfig,
     MetricsFrame,
     StartFrame,
     StartInterruptionFrame,
@@ -54,9 +53,6 @@ class BaseInputTransport(FrameProcessor):
         # Input sample rate. It will be initialized on StartFrame.
         self._sample_rate = 0
 
-        # Interruption configuration from StartFrame
-        self._interruption_config: Optional[InterruptionConfig] = None
-
         # Track bot speaking state for interruption logic
         self._bot_speaking = False
 
@@ -137,9 +133,6 @@ class BaseInputTransport(FrameProcessor):
 
         self._sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate
 
-        # Store interruption configuration
-        self._interruption_config = frame.interruption_config
-
         # Configure VAD analyzer.
         if self._params.vad_analyzer:
             self._params.vad_analyzer.set_sample_rate(self._sample_rate)
@@ -203,8 +196,10 @@ class BaseInputTransport(FrameProcessor):
             await self._handle_bot_interruption(frame)
         elif isinstance(frame, BotStartedSpeakingFrame):
             await self._handle_bot_started_speaking(frame)
+            await self.push_frame(frame)
         elif isinstance(frame, BotStoppedSpeakingFrame):
             await self._handle_bot_stopped_speaking(frame)
+            await self.push_frame(frame)
         elif isinstance(frame, EmulateUserStartedSpeakingFrame):
             logger.debug("Emulating user started speaking")
             await self._handle_user_interruption(UserStartedSpeakingFrame(emulated=True))
@@ -251,7 +246,7 @@ class BaseInputTransport(FrameProcessor):
             # 1. No interruption config is set, OR
             # 2. Interruption config is set but bot is not speaking
             should_push_immediate_interruption = (
-                self._interruption_config is None or not self._bot_speaking
+                self.interruption_strategies is None or not self._bot_speaking
             )
 
             # Make sure we notify about interruptions quickly out-of-band.
@@ -261,8 +256,8 @@ class BaseInputTransport(FrameProcessor):
                 # frame task) to stop everything, specially at the output
                 # transport.
                 await self.push_frame(StartInterruptionFrame())
-            elif self._interruption_config and self._bot_speaking:
+            elif self.interruption_strategies and self._bot_speaking:
-                logger.trace(
+                logger.debug(
                     "User started speaking while bot is speaking with interruption config - "
                     "deferring interruption to aggregator"
                 )
@@ -279,11 +274,9 @@ class BaseInputTransport(FrameProcessor):
 
     async def _handle_bot_started_speaking(self, frame: BotStartedSpeakingFrame):
         self._bot_speaking = True
-        await self.push_frame(frame)
 
     async def _handle_bot_stopped_speaking(self, frame: BotStoppedSpeakingFrame):
         self._bot_speaking = False
-        await self.push_frame(frame)
 
     #
     # Audio input