diff --git a/CHANGELOG.md b/CHANGELOG.md index d1f903a7f..c74400860 100644 --- a/CHANGELOG.md +++ b/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 - `InterruptionConfig` to specify criteria required to interrupt the bot when - it's speaking. You can specify `min_words` to require the user to say at - least `min_words` words before their speech will interrupt the bot. If not +- 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 diff --git a/examples/foundational/42-interruption-config.py b/examples/foundational/42-interruption-config.py index e82d56303..3cb64c204 100644 --- a/examples/foundational/42-interruption-config.py +++ b/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)], ), ) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index bd92fb400..45c1fc6c2 100644 --- a/src/pipecat/frames/frames.py +++ b/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: - """Configuration for interruption behavior. +class InterruptionStrategy: + """Base class for interruption strategies.""" - When specified, the bot will not be interrupted immediately when the user speaks. - Instead, interruption will only occur when the configured conditions are met. + 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: 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 diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 0647d1c02..2b3036ecf 100644 --- a/src/pipecat/pipeline/task.py +++ b/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) diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index ba762a2f9..be9c6f77f 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/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: - should_interrupt = await self._should_interrupt_based_on_config() + if self.interruption_strategies and self._bot_speaking: + 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: - """Check if interruption should occur based on configured conditions.""" - assert self._interruption_config is not None - - if not self._aggregation or self._interruption_config.min_words is None: + def _should_interrupt_based_on_strategies(self) -> bool: + """Check if interruption should occur based on configured strategies.""" + if not self.interruption_strategies: 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): diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index b444b4c58..36055c7c0 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/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() diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 923f8a47b..20c09202f 100644 --- a/src/pipecat/transports/base_input.py +++ b/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: - logger.trace( + elif self.interruption_strategies and self._bot_speaking: + 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