diff --git a/CHANGELOG.md b/CHANGELOG.md index e912e5239..f17c11ab0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Deprecated +- `BotInterruptionFrame` is now deprecated, use `InterruptionTaskFrame` instead. + - `StartInterruptionFrame` is now deprected, use `InterruptionFrame` instead. - Deprecate `VisionImageFrameAggregator` because `VisionImageRawFrame` has been diff --git a/examples/foundational/04b-transports-livekit.py b/examples/foundational/04b-transports-livekit.py index 5a80d149f..0907bb53e 100644 --- a/examples/foundational/04b-transports-livekit.py +++ b/examples/foundational/04b-transports-livekit.py @@ -14,7 +14,7 @@ from loguru import logger from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import ( - BotInterruptionFrame, + InterruptionFrame, TextFrame, TranscriptionFrame, UserStartedSpeakingFrame, @@ -115,7 +115,7 @@ async def main(): await task.queue_frames( [ - BotInterruptionFrame(), + InterruptionFrame(), UserStartedSpeakingFrame(), TranscriptionFrame( user_id=participant_id, diff --git a/examples/foundational/30-observer.py b/examples/foundational/30-observer.py index 62efd0d3e..2f0ee11ca 100644 --- a/examples/foundational/30-observer.py +++ b/examples/foundational/30-observer.py @@ -48,7 +48,7 @@ class CustomObserver(BaseObserver): """Observer to log interruptions and bot speaking events to the console. Logs all frame instances of: - - StartInterruptionFrame + - InterruptionFrame - BotStartedSpeakingFrame - BotStoppedSpeakingFrame diff --git a/src/pipecat/extensions/voicemail/voicemail_detector.py b/src/pipecat/extensions/voicemail/voicemail_detector.py index 12c429a3f..1b5e5ac62 100644 --- a/src/pipecat/extensions/voicemail/voicemail_detector.py +++ b/src/pipecat/extensions/voicemail/voicemail_detector.py @@ -21,9 +21,9 @@ from typing import List, Optional from loguru import logger from pipecat.frames.frames import ( - BotInterruptionFrame, EndFrame, Frame, + InterruptionTaskFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMTextFrame, @@ -360,7 +360,7 @@ class ClassificationProcessor(FrameProcessor): await self._voicemail_notifier.notify() # Clear buffered TTS frames # Interrupt the current pipeline to stop any ongoing processing - await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM) + await self.push_frame(InterruptionTaskFrame(), FrameDirection.UPSTREAM) # Set the voicemail event to trigger the voicemail handler self._voicemail_event.clear() diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 309fa153e..012ab9af7 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -935,20 +935,6 @@ class VADUserStoppedSpeakingFrame(SystemFrame): pass -@dataclass -class BotInterruptionFrame(SystemFrame): - """Frame indicating the bot should be interrupted. - - Emitted when the bot should be interrupted. This will mainly cause the - same actions as if the user interrupted except that the - UserStartedSpeakingFrame and UserStoppedSpeakingFrame won't be generated. - This frame should be pushed upstreams. It results in the BaseInputTransport - starting an interruption by pushing a StartInterruptionFrame downstream. - """ - - pass - - @dataclass class BotStartedSpeakingFrame(SystemFrame): """Frame indicating the bot started speaking. @@ -1336,6 +1322,47 @@ class StopTaskFrame(TaskFrame): pass +@dataclass +class InterruptionTaskFrame(TaskFrame): + """Frame indicating the bot should be interrupted. + + Emitted when the bot should be interrupted. This will mainly cause the + same actions as if the user interrupted except that the + UserStartedSpeakingFrame and UserStoppedSpeakingFrame won't be generated. + This frame should be pushed upstream. + """ + + pass + + +@dataclass +class BotInterruptionFrame(InterruptionTaskFrame): + """Frame indicating the bot should be interrupted. + + .. deprecated:: 0.0.85 + This frame is deprecated and will be removed in a future version. + Instead, use `InterruptionTaskFrame`. + + Emitted when the bot should be interrupted. This will mainly cause the + same actions as if the user interrupted except that the + UserStartedSpeakingFrame and UserStoppedSpeakingFrame won't be generated. + This frame should be pushed upstream. + """ + + def __post_init__(self): + super().__post_init__() + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "BotInterruptionFrame is deprecated and will be removed in a future version. " + "Instead, use InterruptionTaskFrame.", + DeprecationWarning, + stacklevel=2, + ) + + # # Control frames # diff --git a/src/pipecat/observers/loggers/debug_log_observer.py b/src/pipecat/observers/loggers/debug_log_observer.py index abb0db8d7..7de6fe6ce 100644 --- a/src/pipecat/observers/loggers/debug_log_observer.py +++ b/src/pipecat/observers/loggers/debug_log_observer.py @@ -54,7 +54,7 @@ class DebugLogObserver(BaseObserver): Log frames with specific source/destination filters:: - from pipecat.frames.frames import StartInterruptionFrame, UserStartedSpeakingFrame, LLMTextFrame + from pipecat.frames.frames import InterruptionFrame, UserStartedSpeakingFrame, LLMTextFrame from pipecat.observers.loggers.debug_log_observer import DebugLogObserver, FrameEndpoint from pipecat.transports.base_output import BaseOutputTransport from pipecat.services.stt_service import STTService @@ -62,8 +62,8 @@ class DebugLogObserver(BaseObserver): observers=[ DebugLogObserver( frame_types={ - # Only log StartInterruptionFrame when source is BaseOutputTransport - StartInterruptionFrame: (BaseOutputTransport, FrameEndpoint.SOURCE), + # Only log InterruptionFrame when source is BaseOutputTransport + InterruptionFrame: (BaseOutputTransport, FrameEndpoint.SOURCE), # Only log UserStartedSpeakingFrame when destination is STTService UserStartedSpeakingFrame: (STTService, FrameEndpoint.DESTINATION), # Log LLMTextFrame regardless of source or destination type diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 8c1e4c044..975c70cb9 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -32,6 +32,8 @@ from pipecat.frames.frames import ( Frame, HeartbeatFrame, InputAudioRawFrame, + InterruptionFrame, + InterruptionTaskFrame, MetricsFrame, StartFrame, StopFrame, @@ -627,13 +629,20 @@ class PipelineTask(BasePipelineTask): if isinstance(frame, EndTaskFrame): # Tell the task we should end nicely. + logger.debug(f"{self}: received end task frame {frame}") await self.queue_frame(EndFrame()) elif isinstance(frame, CancelTaskFrame): # Tell the task we should end right away. + logger.debug(f"{self}: received cancel task frame {frame}") await self.queue_frame(CancelFrame()) elif isinstance(frame, StopTaskFrame): # Tell the task we should stop nicely. + logger.debug(f"{self}: received stop task frame {frame}") await self.queue_frame(StopFrame()) + elif isinstance(frame, InterruptionTaskFrame): + # Tell the task we should interrupt the pipeline. + logger.debug(f"{self}: received interruption task frame {frame}") + await self.queue_frame(InterruptionFrame()) elif isinstance(frame, ErrorFrame): if frame.fatal: logger.error(f"A fatal error occurred: {frame}") @@ -642,7 +651,7 @@ class PipelineTask(BasePipelineTask): # Tell the task we should stop. await self.queue_frame(StopTaskFrame()) else: - logger.warning(f"Something went wrong: {frame}") + logger.warning(f"{self}: Something went wrong: {frame}") async def _sink_push_frame(self, frame: Frame, direction: FrameDirection): """Process frames coming downstream from the pipeline. diff --git a/src/pipecat/processors/aggregators/dtmf_aggregator.py b/src/pipecat/processors/aggregators/dtmf_aggregator.py index c50b0d8a8..b8b947272 100644 --- a/src/pipecat/processors/aggregators/dtmf_aggregator.py +++ b/src/pipecat/processors/aggregators/dtmf_aggregator.py @@ -16,15 +16,15 @@ from typing import Optional from pipecat.audio.dtmf.types import KeypadEntry from pipecat.frames.frames import ( - BotInterruptionFrame, CancelFrame, EndFrame, Frame, InputDTMFFrame, + InterruptionTaskFrame, StartFrame, TranscriptionFrame, ) -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.utils.time import time_now_iso8601 @@ -105,7 +105,7 @@ class DTMFAggregator(FrameProcessor): # For first digit, schedule interruption. if is_first_digit: - await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM) + await self.push_frame(InterruptionTaskFrame(), FrameDirection.UPSTREAM) # Check for immediate flush conditions if frame.button == self._termination_digit: diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 1f9bb9b71..d38233275 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -22,7 +22,6 @@ from pipecat.audio.interruptions.base_interruption_strategy import BaseInterrupt from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams from pipecat.audio.vad.vad_analyzer import VADParams from pipecat.frames.frames import ( - BotInterruptionFrame, BotStartedSpeakingFrame, BotStoppedSpeakingFrame, CancelFrame, @@ -36,7 +35,7 @@ from pipecat.frames.frames import ( FunctionCallsStartedFrame, InputAudioRawFrame, InterimTranscriptionFrame, - InterruptionFrame, + InterruptionTaskFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesAppendFrame, @@ -532,9 +531,9 @@ class LLMUserContextAggregator(LLMContextResponseAggregator): if should_interrupt: logger.debug( - "Interruption conditions met - pushing BotInterruptionFrame and aggregation" + "Interruption conditions met - pushing InterruptionTaskFrame and aggregation" ) - await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM) + await self.push_frame(InterruptionTaskFrame(), FrameDirection.UPSTREAM) await self._process_aggregation() else: logger.debug("Interruption conditions not met - not pushing aggregation") diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 0e14e7f10..70e4b46d0 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -13,7 +13,6 @@ LLM processing, and text-to-speech components in conversational AI pipelines. import asyncio import json -from dataclasses import dataclass from typing import Any, Dict, List, Literal, Optional, Set from loguru import logger @@ -23,7 +22,6 @@ from pipecat.audio.interruptions.base_interruption_strategy import BaseInterrupt from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams from pipecat.audio.vad.vad_analyzer import VADParams from pipecat.frames.frames import ( - BotInterruptionFrame, BotStartedSpeakingFrame, BotStoppedSpeakingFrame, CancelFrame, @@ -37,7 +35,7 @@ from pipecat.frames.frames import ( FunctionCallsStartedFrame, InputAudioRawFrame, InterimTranscriptionFrame, - InterruptionFrame, + InterruptionTaskFrame, LLMContextAssistantTimestampFrame, LLMContextFrame, LLMFullResponseEndFrame, @@ -311,9 +309,9 @@ class LLMUserAggregator(LLMContextAggregator): if should_interrupt: logger.debug( - "Interruption conditions met - pushing BotInterruptionFrame and aggregation" + "Interruption conditions met - pushing InterruptionTaskFrame and aggregation" ) - await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM) + await self.push_frame(InterruptionTaskFrame(), FrameDirection.UPSTREAM) await self._process_aggregation() else: logger.debug("Interruption conditions not met - not pushing aggregation") diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 8d6e67551..bd63cd0c2 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -30,7 +30,6 @@ from loguru import logger from pydantic import BaseModel, Field, PrivateAttr, ValidationError from pipecat.frames.frames import ( - BotInterruptionFrame, BotStartedSpeakingFrame, BotStoppedSpeakingFrame, CancelFrame, @@ -42,6 +41,7 @@ from pipecat.frames.frames import ( FunctionCallResultFrame, InputAudioRawFrame, InterimTranscriptionFrame, + InterruptionTaskFrame, LLMContextFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, @@ -1206,7 +1206,7 @@ class RTVIProcessor(FrameProcessor): async def interrupt_bot(self): """Send a bot interruption frame upstream.""" - await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM) + await self.push_frame(InterruptionTaskFrame(), FrameDirection.UPSTREAM) async def send_server_message(self, data: Any): """Send a server message to the client.""" diff --git a/src/pipecat/processors/transcript_processor.py b/src/pipecat/processors/transcript_processor.py index 9a57aad35..5900ffc7f 100644 --- a/src/pipecat/processors/transcript_processor.py +++ b/src/pipecat/processors/transcript_processor.py @@ -86,7 +86,7 @@ class AssistantTranscriptProcessor(BaseTranscriptProcessor): transcript messages. Utterances are completed when: - The bot stops speaking (BotStoppedSpeakingFrame) - - The bot is interrupted (StartInterruptionFrame) + - The bot is interrupted (InterruptionFrame) - The pipeline ends (EndFrame) """ @@ -185,7 +185,7 @@ class AssistantTranscriptProcessor(BaseTranscriptProcessor): - TTSTextFrame: Aggregates text for current utterance - BotStoppedSpeakingFrame: Completes current utterance - - StartInterruptionFrame: Completes current utterance due to interruption + - InterruptionFrame: Completes current utterance due to interruption - EndFrame: Completes current utterance at pipeline end - CancelFrame: Completes current utterance due to cancellation diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py index ab37a9add..3d6d5bd4c 100644 --- a/src/pipecat/services/elevenlabs/tts.py +++ b/src/pipecat/services/elevenlabs/tts.py @@ -558,7 +558,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService): logger.trace(f"Closing context {self._context_id} due to interruption") try: # ElevenLabs requires that Pipecat manages the contexts and closes them - # when they're not longer in use. Since a StartInterruptionFrame is pushed + # when they're not longer in use. Since an InterruptionFrame is pushed # every time the user speaks, we'll use this as a trigger to close the context # and reset the state. # Note: We do not need to call remove_audio_context here, as the context is diff --git a/src/pipecat/services/speechmatics/stt.py b/src/pipecat/services/speechmatics/stt.py index 7136bbb0c..3cd912093 100644 --- a/src/pipecat/services/speechmatics/stt.py +++ b/src/pipecat/services/speechmatics/stt.py @@ -19,12 +19,12 @@ from loguru import logger from pydantic import BaseModel from pipecat.frames.frames import ( - BotInterruptionFrame, CancelFrame, EndFrame, ErrorFrame, Frame, InterimTranscriptionFrame, + InterruptionTaskFrame, StartFrame, TranscriptionFrame, UserStartedSpeakingFrame, @@ -756,7 +756,7 @@ class SpeechmaticsSTTService(STTService): if self._params.enable_vad and not self._is_speaking: logger.debug("User started speaking") self._is_speaking = True - upstream_frames += [BotInterruptionFrame()] + upstream_frames += [InterruptionTaskFrame()] downstream_frames += [UserStartedSpeakingFrame()] # If final, then re-parse into TranscriptionFrame diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 228b8f6e7..04f74e4f8 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -22,7 +22,6 @@ 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, @@ -34,6 +33,7 @@ from pipecat.frames.frames import ( InputAudioRawFrame, InputImageRawFrame, InterruptionFrame, + InterruptionTaskFrame, MetricsFrame, SpeechControlParamsFrame, StartFrame, @@ -289,8 +289,6 @@ class BaseInputTransport(FrameProcessor): elif isinstance(frame, CancelFrame): await self.cancel(frame) 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, direction) @@ -335,13 +333,6 @@ class BaseInputTransport(FrameProcessor): # Handle interruptions # - async def _handle_bot_interruption(self, frame: BotInterruptionFrame): - """Handle bot interruption frames.""" - logger.debug("Bot interruption") - if self.interruptions_allowed: - await self._start_interruption() - await self.push_frame(InterruptionFrame()) - async def _handle_user_interruption(self, vad_state: VADState, emulated: bool = False): """Handle user interruption events based on speaking state.""" if vad_state == VADState.SPEAKING: @@ -353,7 +344,7 @@ class BaseInputTransport(FrameProcessor): await self.push_frame(downstream_frame) await self.push_frame(upstream_frame, FrameDirection.UPSTREAM) - # Only push StartInterruptionFrame if: + # Only push InterruptionFrame if: # 1. No interruption config is set, OR # 2. Interruption config is set but bot is not speaking should_push_immediate_interruption = ( diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 34c66571c..95db2b71b 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -287,9 +287,8 @@ class BaseOutputTransport(FrameProcessor): await super().process_frame(frame, direction) # - # System frames (like StartInterruptionFrame) are pushed - # immediately. Other frames require order so they are put in the sink - # queue. + # System frames (like InterruptionFrame) are pushed immediately. Other + # frames require order so they are put in the sink queue. # if isinstance(frame, StartFrame): # Push StartFrame before start(), because we want StartFrame to be