Merge pull request #1931 from pipecat-ai/mb/num-words

Add support for interruption strategies

CHANGELOG.md

@@ -9,6 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added `interruption_strategies` to `PipelineParams` using
+  `MinWordsInterruptionStrategy` to specify minimum words required to interrupt
+  the bot when it's speaking. Use
+  `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
   the transport will pause sending frames downstream until a new `StartFrame` is
   received. This allows the transport to be reused (keeping the same connection)

examples/foundational/42-interruption-config.py

@@ -0,0 +1,125 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+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
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.transcript_processor import TranscriptProcessor
+from pipecat.services.cartesia.tts import CartesiaTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams
+from pipecat.transports.services.daily import DailyParams
+
+load_dotenv(override=True)
+
+# We store functions so objects (e.g. SileroVADAnalyzer) don't get
+# instantiated. The function will be called when the desired transport gets
+# selected.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+}
+
+
+async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_sigint: bool):
+    logger.info(f"Starting bot")
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = CartesiaTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY"),
+        voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+    )
+
+    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+
+    transcript = TranscriptProcessor()
+
+    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 so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+        },
+    ]
+
+    context = OpenAILLMContext(messages)
+    context_aggregator = llm.create_context_aggregator(context)
+
+    pipeline = Pipeline(
+        [
+            transport.input(),  # Transport user input
+            stt,
+            transcript.user(),  # User transcripts
+            context_aggregator.user(),  # User responses
+            llm,  # LLM
+            tts,  # TTS
+            transport.output(),  # Transport bot output
+            context_aggregator.assistant(),  # Assistant spoken responses
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            allow_interruptions=True,
+            enable_metrics=True,
+            enable_usage_metrics=True,
+            report_only_initial_ttfb=True,
+            interruption_strategies=[MinWordsInterruptionStrategy(min_words=3)],
+        ),
+    )
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info(f"Client connected")
+        # Kick off the conversation.
+        messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+        await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info(f"Client disconnected")
+        await task.cancel()
+
+    # Register event handler for transcript updates
+    @transcript.event_handler("on_transcript_update")
+    async def on_transcript_update(processor, frame):
+        for message in frame.messages:
+            logger.info(f"Transcription [{message.role}]: {message.content}")
+
+    runner = PipelineRunner(handle_sigint=handle_sigint)
+
+    await runner.run(task)
+
+
+if __name__ == "__main__":
+    from pipecat.examples.run import main
+
+    main(run_example, transport_params=transport_params)

src/pipecat/frames/frames.py

@@ -15,6 +15,7 @@ from typing import (
     Literal,
     Mapping,
     Optional,
+    Sequence,
     Tuple,
 )
 
@@ -438,6 +439,28 @@ class OutputDTMFFrame(DTMFFrame, DataFrame):
 #
 
 
+@dataclass
+class InterruptionStrategy:
+    """Base class for interruption strategies."""
+
+    pass
+
+
+@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: int
+
+    def __post_init__(self):
+        if self.min_words <= 0:
+            raise ValueError("min_words must be greater than 0")
+
+
 @dataclass
 class StartFrame(SystemFrame):
     """This is the first frame that should be pushed down a pipeline."""
@@ -448,6 +471,7 @@ class StartFrame(SystemFrame):
     enable_metrics: bool = False
     enable_usage_metrics: bool = False
     report_only_initial_ttfb: bool = False
+    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,6 +22,7 @@ from pipecat.frames.frames import (
     ErrorFrame,
     Frame,
     HeartbeatFrame,
+    InterruptionStrategy,
     LLMFullResponseEndFrame,
     MetricsFrame,
     StartFrame,
@@ -58,6 +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_strategies: Strategies for bot interruption behavior.
     """
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
@@ -73,6 +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_strategies: Optional[Sequence[InterruptionStrategy]] = None
 
 
 class PipelineTaskSource(FrameProcessor):
@@ -518,6 +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_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

@@ -12,6 +12,7 @@ from typing import Dict, List, Literal, Optional, Set
 from loguru import logger
 
 from pipecat.frames.frames import (
+    BotInterruptionFrame,
     BotStartedSpeakingFrame,
     BotStoppedSpeakingFrame,
     CancelFrame,
@@ -31,6 +32,7 @@ from pipecat.frames.frames import (
     LLMSetToolChoiceFrame,
     LLMSetToolsFrame,
     LLMTextFrame,
+    MinWordsInterruptionStrategy,
     OpenAILLMContextAssistantTimestampFrame,
     StartFrame,
     StartInterruptionFrame,
@@ -193,7 +195,7 @@ class LLMContextResponseAggregator(BaseLLMResponseAggregator):
         self._context = context
         self._role = role
 
-        self._aggregation = ""
+        self._aggregation: str = ""
 
     @property
     def messages(self) -> List[dict]:
@@ -320,18 +322,51 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
         else:
             await self.push_frame(frame, direction)
 
+    async def _process_aggregation(self):
+        """Process the current aggregation and push it downstream."""
+        aggregation = self._aggregation
+        self.reset()
+        await self.handle_aggregation(aggregation)
+        frame = OpenAILLMContextFrame(self._context)
+        await self.push_frame(frame)
+
     async def push_aggregation(self):
+        """Pushes the current aggregation based on interruption configuration and conditions."""
         if len(self._aggregation) > 0:
-            aggregation = self._aggregation
+            if self.interruption_strategies and self._bot_speaking:
+                should_interrupt = self._should_interrupt_based_on_strategies()
 
-            # Reset the aggregation. Reset it before pushing it down, otherwise
-            # if the tasks gets cancelled we won't be able to clear things up.
-            self.reset()
+                if should_interrupt:
+                    logger.debug(
+                        "Interruption conditions met - pushing BotInterruptionFrame and aggregation"
+                    )
+                    await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM)
+                    await self._process_aggregation()
+                else:
+                    logger.debug("Interruption conditions not met - not pushing aggregation")
+                    # Don't process aggregation, just reset it
+                    self.reset()
+            else:
+                # No interruption config - normal behavior (always push aggregation)
+                await self._process_aggregation()
 
-            await self.handle_aggregation(aggregation)
+    def _should_interrupt_based_on_strategies(self) -> bool:
+        """Check if interruption should occur based on configured strategies."""
+        if not self.interruption_strategies:
+            return False
 
-            frame = OpenAILLMContextFrame(self._context)
-            await self.push_frame(frame)
+        # 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 >= strategy.min_words
 
     async def _start(self, frame: StartFrame):
         self._create_aggregation_task()

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

@@ -17,6 +17,8 @@ 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,
     EmulateUserStartedSpeakingFrame,
     EmulateUserStoppedSpeakingFrame,
@@ -51,6 +53,9 @@ class BaseInputTransport(FrameProcessor):
         # Input sample rate. It will be initialized on StartFrame.
         self._sample_rate = 0
 
+        # Track bot speaking state for interruption logic
+        self._bot_speaking = False
+
         # We read audio from a single queue one at a time and we then run VAD in
         # a thread. Therefore, only one thread should be necessary.
         self._executor = ThreadPoolExecutor(max_workers=1)
@@ -189,6 +194,12 @@ class BaseInputTransport(FrameProcessor):
             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)
+        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))
@@ -230,13 +241,26 @@ class BaseInputTransport(FrameProcessor):
         if isinstance(frame, UserStartedSpeakingFrame):
             logger.debug("User started speaking")
             await self.push_frame(frame)
+
+            # Only push StartInterruptionFrame if:
+            # 1. No interruption config is set, OR
+            # 2. Interruption config is set but bot is not speaking
+            should_push_immediate_interruption = (
+                self.interruption_strategies is None or not self._bot_speaking
+            )
+
             # Make sure we notify about interruptions quickly out-of-band.
-            if self.interruptions_allowed:
+            if should_push_immediate_interruption and self.interruptions_allowed:
                 await self._start_interruption()
                 # Push an out-of-band frame (i.e. not using the ordered push
                 # frame task) to stop everything, specially at the output
                 # transport.
                 await self.push_frame(StartInterruptionFrame())
+            elif self.interruption_strategies and self._bot_speaking:
+                logger.debug(
+                    "User started speaking while bot is speaking with interruption config - "
+                    "deferring interruption to aggregator"
+                )
         elif isinstance(frame, UserStoppedSpeakingFrame):
             logger.debug("User stopped speaking")
             await self.push_frame(frame)
@@ -244,6 +268,16 @@ class BaseInputTransport(FrameProcessor):
                 await self._stop_interruption()
                 await self.push_frame(StopInterruptionFrame())
 
+    #
+    # Handle bot speaking state
+    #
+
+    async def _handle_bot_started_speaking(self, frame: BotStartedSpeakingFrame):
+        self._bot_speaking = True
+
+    async def _handle_bot_stopped_speaking(self, frame: BotStoppedSpeakingFrame):
+        self._bot_speaking = False
+
     #
     # Audio input
     #