diff --git a/CHANGELOG.md b/CHANGELOG.md index 025731588..c74400860 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/examples/foundational/42-interruption-config.py b/examples/foundational/42-interruption-config.py new file mode 100644 index 000000000..3cb64c204 --- /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 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) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index b24ff7b19..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, ) @@ -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 diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 0fe330655..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,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) diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 55ac0e2d5..be9c6f77f 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, @@ -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() 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 2a2343883..20c09202f 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, @@ -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 #