From b489e52080a103ad48953c78b8d506cf1c8c9d33 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 30 May 2025 15:20:42 -0400 Subject: [PATCH 1/6] Add InterruptionConfig --- src/pipecat/frames/frames.py | 15 +++++++++++++++ src/pipecat/pipeline/task.py | 4 ++++ 2 files changed, 19 insertions(+) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index b24ff7b19..bd92fb400 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -438,6 +438,20 @@ class OutputDTMFFrame(DTMFFrame, DataFrame): # +@dataclass +class InterruptionConfig: + """Configuration for interruption behavior. + + When specified, the bot will not be interrupted immediately when the user speaks. + Instead, interruption will only occur when the configured conditions are met. + + Args: + min_words: If set, user must speak at least this many words to interrupt + """ + + min_words: Optional[int] = None + + @dataclass class StartFrame(SystemFrame): """This is the first frame that should be pushed down a pipeline.""" @@ -448,6 +462,7 @@ class StartFrame(SystemFrame): enable_metrics: bool = False enable_usage_metrics: bool = False report_only_initial_ttfb: bool = False + interruption_config: Optional[InterruptionConfig] = None @dataclass diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 0fe330655..0647d1c02 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -22,6 +22,7 @@ from pipecat.frames.frames import ( ErrorFrame, Frame, HeartbeatFrame, + InterruptionConfig, 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_config: Configuration 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_config: Optional[InterruptionConfig] = 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_config=self._params.interruption_config, ) start_frame.metadata = self._params.start_metadata await self._source.queue_frame(start_frame, FrameDirection.DOWNSTREAM) From 6bc4b4a17f11715dd61c25dadaba97fb476d3d08 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 30 May 2025 15:29:03 -0400 Subject: [PATCH 2/6] Update BaseInputTransport to not push StartInterruptionFrame when InterruptionConfig is set and bot is speaking --- src/pipecat/transports/base_input.py | 43 +++++++++++++++++++++++++++- 1 file changed, 42 insertions(+), 1 deletion(-) diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 2a2343883..923f8a47b 100644 --- a/src/pipecat/transports/base_input.py +++ b/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, @@ -25,6 +27,7 @@ from pipecat.frames.frames import ( Frame, InputAudioRawFrame, InputImageRawFrame, + InterruptionConfig, MetricsFrame, StartFrame, StartInterruptionFrame, @@ -51,6 +54,12 @@ 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 + # 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) @@ -128,6 +137,9 @@ 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) @@ -189,6 +201,10 @@ 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) + elif isinstance(frame, BotStoppedSpeakingFrame): + await self._handle_bot_stopped_speaking(frame) elif isinstance(frame, EmulateUserStartedSpeakingFrame): logger.debug("Emulating user started speaking") await self._handle_user_interruption(UserStartedSpeakingFrame(emulated=True)) @@ -230,13 +246,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_config 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_config and self._bot_speaking: + logger.trace( + "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 +273,18 @@ 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 + 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 # From 2d609a0bde57c5c83907a49b3c76077f342455d3 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 30 May 2025 15:45:05 -0400 Subject: [PATCH 3/6] Update LLmUserContextAggregator to conditionally push_aggregation --- .../processors/aggregators/llm_response.py | 44 ++++++++++++++++--- 1 file changed, 37 insertions(+), 7 deletions(-) diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 55ac0e2d5..ba762a2f9 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/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, @@ -23,6 +24,7 @@ from pipecat.frames.frames import ( FunctionCallInProgressFrame, FunctionCallResultFrame, InterimTranscriptionFrame, + InterruptionConfig, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesAppendFrame, @@ -266,6 +268,8 @@ 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 @@ -320,20 +324,46 @@ 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_config and self._bot_speaking: + should_interrupt = await self._should_interrupt_based_on_config() - # 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) + 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 - frame = OpenAILLMContextFrame(self._context) - await self.push_frame(frame) + if not self._aggregation or self._interruption_config.min_words is None: + return False + + word_count = len(self._aggregation.split()) + return word_count >= self._interruption_config.min_words async def _start(self, frame: StartFrame): + self._interruption_config = frame.interruption_config self._create_aggregation_task() async def _stop(self, frame: EndFrame): From 62efbc334258b3f8c8ae69f930f28990d49f7a1a Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 30 May 2025 15:59:10 -0400 Subject: [PATCH 4/6] Add foundational example 42 --- .../foundational/42-interruption-config.py | 125 ++++++++++++++++++ 1 file changed, 125 insertions(+) create mode 100644 examples/foundational/42-interruption-config.py diff --git a/examples/foundational/42-interruption-config.py b/examples/foundational/42-interruption-config.py new file mode 100644 index 000000000..e82d56303 --- /dev/null +++ b/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 InterruptionConfig +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_config=InterruptionConfig(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) From b34c593c54ed8817a8a003647c3066898bcd0784 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 30 May 2025 16:07:17 -0400 Subject: [PATCH 5/6] Add changelog entry --- CHANGELOG.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 025731588..d1f903a7f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,12 @@ 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 + 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) From 7a4efc621201b98ebdb9ec18efb51e25784eb901 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 30 May 2025 17:27:00 -0400 Subject: [PATCH 6/6] Code review feedback --- CHANGELOG.md | 9 +++--- .../foundational/42-interruption-config.py | 4 +-- src/pipecat/frames/frames.py | 21 +++++++++---- src/pipecat/pipeline/task.py | 10 +++--- .../processors/aggregators/llm_response.py | 31 +++++++++++-------- src/pipecat/processors/frame_processor.py | 9 +++++- src/pipecat/transports/base_input.py | 17 +++------- 7 files changed, 58 insertions(+), 43 deletions(-) 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