diff --git a/CHANGELOG.md b/CHANGELOG.md index c882fa7c5..7524b0d53 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `FrameProcessor.push_interruption_task_frame_and_wait()`. Use this + method to programatically interrupt the bot from any part of the + pipeline. This guarantees that all the processors in the pipeline are + interrupted in order (from upstream to downstream). Internally, this works by + first pushing an `InterruptionTaskFrame` upstream until it reaches the + pipeline task. The pipeline task then generates an `InterruptionFrame`, which + flows downstream through all processors. Once the `InterruptionFrame` has + reaches the processor waiting for the interruption, the function returns and + execution continues after the call. Think of it as sending an upstream request + for interruption and waiting until the acknowledgment flows back downstream. + +- Added new base `TaskFrame` (which is a system frame). This is the base class + for all task frames (`EndTaskFrame`, `CancelTaskFrame`, etc.) that are meant + to be pushed upstream to reach the pipeline task. + - Expanded support for universal `LLMContext` to the AWS Bedrock LLM service. Using the universal `LLMContext` and associated `LLMContextAggregatorPair` is a pre-requisite for using `LLMSwitcher` to switch between LLMs at runtime. @@ -25,6 +40,10 @@ 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 removed. See the `12*` examples for the new recommended replacement pattern. 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/07c-interruptible-deepgram-vad.py b/examples/foundational/07c-interruptible-deepgram-vad.py index 3569fc440..3f9330bbd 100644 --- a/examples/foundational/07c-interruptible-deepgram-vad.py +++ b/examples/foundational/07c-interruptible-deepgram-vad.py @@ -12,8 +12,8 @@ from dotenv import load_dotenv from loguru import logger from pipecat.frames.frames import ( + InterruptionFrame, LLMRunFrame, - StartInterruptionFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, ) @@ -97,7 +97,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): @stt.event_handler("on_speech_started") async def on_speech_started(stt, *args, **kwargs): - await task.queue_frames([StartInterruptionFrame(), UserStartedSpeakingFrame()]) + await task.queue_frames([InterruptionFrame(), UserStartedSpeakingFrame()]) @stt.event_handler("on_utterance_end") async def on_utterance_end(stt, *args, **kwargs): diff --git a/examples/foundational/07s-interruptible-google-audio-in.py b/examples/foundational/07s-interruptible-google-audio-in.py index d7aaf07bd..0265c969f 100644 --- a/examples/foundational/07s-interruptible-google-audio-in.py +++ b/examples/foundational/07s-interruptible-google-audio-in.py @@ -16,10 +16,10 @@ from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import ( Frame, InputAudioRawFrame, + InterruptionFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMRunFrame, - StartInterruptionFrame, TextFrame, TranscriptionFrame, UserStartedSpeakingFrame, @@ -181,9 +181,7 @@ class TranscriptionContextFixup(FrameProcessor): if isinstance(frame, MagicDemoTranscriptionFrame): self._transcript = frame.text - elif isinstance(frame, LLMFullResponseEndFrame) or isinstance( - frame, StartInterruptionFrame - ): + elif isinstance(frame, LLMFullResponseEndFrame) or isinstance(frame, InterruptionFrame): self.swap_user_audio() self.add_transcript_back_to_inference_output() self._transcript = "" diff --git a/examples/foundational/22b-natural-conversation-proposal.py b/examples/foundational/22b-natural-conversation-proposal.py index dc70d0379..417aeca76 100644 --- a/examples/foundational/22b-natural-conversation-proposal.py +++ b/examples/foundational/22b-natural-conversation-proposal.py @@ -18,9 +18,9 @@ from pipecat.frames.frames import ( Frame, FunctionCallInProgressFrame, FunctionCallResultFrame, + InterruptionFrame, LLMRunFrame, StartFrame, - StartInterruptionFrame, SystemFrame, TextFrame, TranscriptionFrame, @@ -144,7 +144,7 @@ class OutputGate(FrameProcessor): await self._start() if isinstance(frame, (EndFrame, CancelFrame)): await self._stop() - if isinstance(frame, StartInterruptionFrame): + if isinstance(frame, InterruptionFrame): self._frames_buffer = [] self.close_gate() await self.push_frame(frame, direction) @@ -232,7 +232,7 @@ class TurnDetectionLLM(Pipeline): async def pass_only_llm_trigger_frames(frame): return ( isinstance(frame, OpenAILLMContextFrame) - or isinstance(frame, StartInterruptionFrame) + or isinstance(frame, InterruptionFrame) or isinstance(frame, FunctionCallInProgressFrame) or isinstance(frame, FunctionCallResultFrame) ) diff --git a/examples/foundational/22c-natural-conversation-mixed-llms.py b/examples/foundational/22c-natural-conversation-mixed-llms.py index 44f3f1349..e4c554b26 100644 --- a/examples/foundational/22c-natural-conversation-mixed-llms.py +++ b/examples/foundational/22c-natural-conversation-mixed-llms.py @@ -18,9 +18,9 @@ from pipecat.frames.frames import ( Frame, FunctionCallInProgressFrame, FunctionCallResultFrame, + InterruptionFrame, LLMRunFrame, StartFrame, - StartInterruptionFrame, SystemFrame, TextFrame, TranscriptionFrame, @@ -347,7 +347,7 @@ class OutputGate(FrameProcessor): await self._start() if isinstance(frame, (EndFrame, CancelFrame)): await self._stop() - if isinstance(frame, StartInterruptionFrame): + if isinstance(frame, InterruptionFrame): self._frames_buffer = [] self.close_gate() await self.push_frame(frame, direction) @@ -426,7 +426,7 @@ class TurnDetectionLLM(Pipeline): async def pass_only_llm_trigger_frames(frame): return ( isinstance(frame, OpenAILLMContextFrame) - or isinstance(frame, StartInterruptionFrame) + or isinstance(frame, InterruptionFrame) or isinstance(frame, FunctionCallInProgressFrame) or isinstance(frame, FunctionCallResultFrame) ) diff --git a/examples/foundational/22d-natural-conversation-gemini-audio.py b/examples/foundational/22d-natural-conversation-gemini-audio.py index 5eed49092..d7ecf2ba7 100644 --- a/examples/foundational/22d-natural-conversation-gemini-audio.py +++ b/examples/foundational/22d-natural-conversation-gemini-audio.py @@ -20,10 +20,10 @@ from pipecat.frames.frames import ( FunctionCallInProgressFrame, FunctionCallResultFrame, InputAudioRawFrame, + InterruptionFrame, LLMFullResponseStartFrame, LLMRunFrame, StartFrame, - StartInterruptionFrame, SystemFrame, TextFrame, TranscriptionFrame, @@ -570,7 +570,7 @@ class OutputGate(FrameProcessor): await self._start() if isinstance(frame, (EndFrame, CancelFrame)): await self._stop() - if isinstance(frame, StartInterruptionFrame): + if isinstance(frame, InterruptionFrame): self._frames_buffer = [] self.close_gate() await self.push_frame(frame, direction) diff --git a/examples/foundational/30-observer.py b/examples/foundational/30-observer.py index 34557ca56..2f0ee11ca 100644 --- a/examples/foundational/30-observer.py +++ b/examples/foundational/30-observer.py @@ -15,8 +15,8 @@ from pipecat.frames.frames import ( BotStartedSpeakingFrame, BotStoppedSpeakingFrame, EndFrame, + InterruptionFrame, LLMRunFrame, - StartInterruptionFrame, TTSTextFrame, UserStartedSpeakingFrame, ) @@ -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 @@ -69,7 +69,7 @@ class CustomObserver(BaseObserver): # Create direction arrow arrow = "→" if direction == FrameDirection.DOWNSTREAM else "←" - if isinstance(frame, StartInterruptionFrame) and isinstance(src, BaseOutputTransport): + if isinstance(frame, InterruptionFrame) and isinstance(src, BaseOutputTransport): logger.info(f"⚡ INTERRUPTION START: {src} {arrow} {dst} at {time_sec:.2f}s") elif isinstance(frame, BotStartedSpeakingFrame): logger.info(f"🤖 BOT START SPEAKING: {src} {arrow} {dst} at {time_sec:.2f}s") diff --git a/src/pipecat/extensions/voicemail/voicemail_detector.py b/src/pipecat/extensions/voicemail/voicemail_detector.py index 12c429a3f..1b460404b 100644 --- a/src/pipecat/extensions/voicemail/voicemail_detector.py +++ b/src/pipecat/extensions/voicemail/voicemail_detector.py @@ -21,7 +21,6 @@ from typing import List, Optional from loguru import logger from pipecat.frames.frames import ( - BotInterruptionFrame, EndFrame, Frame, LLMFullResponseEndFrame, @@ -360,7 +359,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_interruption_task_frame_and_wait() # 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 83488f2a4..012ab9af7 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -788,43 +788,6 @@ class FatalErrorFrame(ErrorFrame): fatal: bool = field(default=True, init=False) -@dataclass -class EndTaskFrame(SystemFrame): - """Frame to request graceful pipeline task closure. - - This is used to notify the pipeline task that the pipeline should be - closed nicely (flushing all the queued frames) by pushing an EndFrame - downstream. This frame should be pushed upstream. - """ - - pass - - -@dataclass -class CancelTaskFrame(SystemFrame): - """Frame to request immediate pipeline task cancellation. - - This is used to notify the pipeline task that the pipeline should be - stopped immediately by pushing a CancelFrame downstream. This frame - should be pushed upstream. - """ - - pass - - -@dataclass -class StopTaskFrame(SystemFrame): - """Frame to request pipeline task stop while keeping processors running. - - This is used to notify the pipeline task that it should be stopped as - soon as possible (flushing all the queued frames) but that the pipeline - processors should be kept in a running state. This frame should be pushed - upstream. - """ - - pass - - @dataclass class FrameProcessorPauseUrgentFrame(SystemFrame): """Frame to pause frame processing immediately. @@ -857,7 +820,7 @@ class FrameProcessorResumeUrgentFrame(SystemFrame): @dataclass -class StartInterruptionFrame(SystemFrame): +class InterruptionFrame(SystemFrame): """Frame indicating user started speaking (interruption detected). Emitted by the BaseInputTransport to indicate that a user has started @@ -869,6 +832,34 @@ class StartInterruptionFrame(SystemFrame): pass +@dataclass +class StartInterruptionFrame(InterruptionFrame): + """Frame indicating user started speaking (interruption detected). + + .. deprecated:: 0.0.85 + This frame is deprecated and will be removed in a future version. + Instead, use `InterruptionFrame`. + + Emitted by the BaseInputTransport to indicate that a user has started + speaking (i.e. is interrupting). This is similar to + UserStartedSpeakingFrame except that it should be pushed concurrently + with other frames (so the order is not guaranteed). + """ + + def __post_init__(self): + super().__post_init__() + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "StartInterruptionFrame is deprecated and will be removed in a future version. " + "Instead, use InterruptionFrame.", + DeprecationWarning, + stacklevel=2, + ) + + @dataclass class UserStartedSpeakingFrame(SystemFrame): """Frame indicating user has started speaking. @@ -944,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. @@ -1289,6 +1266,103 @@ class SpeechControlParamsFrame(SystemFrame): turn_params: Optional[SmartTurnParams] = None +# +# Task frames +# + + +@dataclass +class TaskFrame(SystemFrame): + """Base frame for task frames. + + This is a base class for frames that are meant to be sent and handled + upstream by the pipeline task. This might result in a corresponding frame + sent downstream (e.g. `InterruptionTaskFrame` / `InterruptionFrame` or + `EndTaskFrame` / `EndFrame`). + + """ + + pass + + +@dataclass +class EndTaskFrame(TaskFrame): + """Frame to request graceful pipeline task closure. + + This is used to notify the pipeline task that the pipeline should be + closed nicely (flushing all the queued frames) by pushing an EndFrame + downstream. This frame should be pushed upstream. + """ + + pass + + +@dataclass +class CancelTaskFrame(TaskFrame): + """Frame to request immediate pipeline task cancellation. + + This is used to notify the pipeline task that the pipeline should be + stopped immediately by pushing a CancelFrame downstream. This frame + should be pushed upstream. + """ + + pass + + +@dataclass +class StopTaskFrame(TaskFrame): + """Frame to request pipeline task stop while keeping processors running. + + This is used to notify the pipeline task that it should be stopped as + soon as possible (flushing all the queued frames) but that the pipeline + processors should be kept in a running state. This frame should be pushed + upstream. + """ + + 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..8eec80ce4 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,23 @@ 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. Note that we are + # bypassing the push queue and directly queue into the + # pipeline. This is in case the push task is blocked waiting for a + # pipeline-ending frame to finish traversing the pipeline. + logger.debug(f"{self}: received interruption task frame {frame}") + await self._pipeline.queue_frame(InterruptionFrame()) elif isinstance(frame, ErrorFrame): if frame.fatal: logger.error(f"A fatal error occurred: {frame}") @@ -642,7 +654,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..1aa0760b4 100644 --- a/src/pipecat/processors/aggregators/dtmf_aggregator.py +++ b/src/pipecat/processors/aggregators/dtmf_aggregator.py @@ -16,7 +16,6 @@ from typing import Optional from pipecat.audio.dtmf.types import KeypadEntry from pipecat.frames.frames import ( - BotInterruptionFrame, CancelFrame, EndFrame, Frame, @@ -24,7 +23,7 @@ from pipecat.frames.frames import ( 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 +104,7 @@ class DTMFAggregator(FrameProcessor): # For first digit, schedule interruption. if is_first_digit: - await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM) + await self.push_interruption_task_frame_and_wait() # 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 d058a4334..ace7b94fd 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,6 +35,7 @@ from pipecat.frames.frames import ( FunctionCallsStartedFrame, InputAudioRawFrame, InterimTranscriptionFrame, + InterruptionFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesAppendFrame, @@ -48,7 +48,6 @@ from pipecat.frames.frames import ( OpenAILLMContextAssistantTimestampFrame, SpeechControlParamsFrame, StartFrame, - StartInterruptionFrame, TextFrame, TranscriptionFrame, UserImageRawFrame, @@ -138,7 +137,7 @@ class LLMFullResponseAggregator(FrameProcessor): """ await super().process_frame(frame, direction) - if isinstance(frame, StartInterruptionFrame): + if isinstance(frame, InterruptionFrame): await self._call_event_handler("on_completion", self._aggregation, False) self._aggregation = "" self._started = False @@ -532,9 +531,9 @@ class LLMUserContextAggregator(LLMContextResponseAggregator): if should_interrupt: logger.debug( - "Interruption conditions met - pushing BotInterruptionFrame and aggregation" + "Interruption conditions met - pushing interruption and aggregation" ) - await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM) + await self.push_interruption_task_frame_and_wait() await self._process_aggregation() else: logger.debug("Interruption conditions not met - not pushing aggregation") @@ -838,7 +837,7 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): """ await super().process_frame(frame, direction) - if isinstance(frame, StartInterruptionFrame): + if isinstance(frame, InterruptionFrame): await self._handle_interruptions(frame) await self.push_frame(frame, direction) elif isinstance(frame, LLMFullResponseStartFrame): @@ -904,7 +903,7 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): if frame.run_llm: await self.push_context_frame(FrameDirection.UPSTREAM) - async def _handle_interruptions(self, frame: StartInterruptionFrame): + async def _handle_interruptions(self, frame: InterruptionFrame): await self.push_aggregation() self._started = 0 await self.reset() diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index bda761ebd..7cb101fa1 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, @@ -48,7 +46,6 @@ from pipecat.frames.frames import ( LLMSetToolsFrame, SpeechControlParamsFrame, StartFrame, - StartInterruptionFrame, TextFrame, TranscriptionFrame, UserImageRawFrame, @@ -311,9 +308,9 @@ class LLMUserAggregator(LLMContextAggregator): if should_interrupt: logger.debug( - "Interruption conditions met - pushing BotInterruptionFrame and aggregation" + "Interruption conditions met - pushing interruption and aggregation" ) - await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM) + await self.push_interruption_task_frame_and_wait() await self._process_aggregation() else: logger.debug("Interruption conditions not met - not pushing aggregation") @@ -579,7 +576,7 @@ class LLMAssistantAggregator(LLMContextAggregator): """ await super().process_frame(frame, direction) - if isinstance(frame, StartInterruptionFrame): + if isinstance(frame, InterruptionFrame): await self._handle_interruptions(frame) await self.push_frame(frame, direction) elif isinstance(frame, LLMFullResponseStartFrame): @@ -645,7 +642,7 @@ class LLMAssistantAggregator(LLMContextAggregator): if frame.run_llm: await self.push_context_frame(FrameDirection.UPSTREAM) - async def _handle_interruptions(self, frame: StartInterruptionFrame): + async def _handle_interruptions(self, frame: InterruptionFrame): await self._push_aggregation() self._started = 0 await self.reset() diff --git a/src/pipecat/processors/filters/stt_mute_filter.py b/src/pipecat/processors/filters/stt_mute_filter.py index d6baac1f7..613d1ef51 100644 --- a/src/pipecat/processors/filters/stt_mute_filter.py +++ b/src/pipecat/processors/filters/stt_mute_filter.py @@ -25,8 +25,8 @@ from pipecat.frames.frames import ( FunctionCallResultFrame, InputAudioRawFrame, InterimTranscriptionFrame, + InterruptionFrame, StartFrame, - StartInterruptionFrame, STTMuteFrame, TranscriptionFrame, UserStartedSpeakingFrame, @@ -204,7 +204,7 @@ class STTMuteFilter(FrameProcessor): if isinstance( frame, ( - StartInterruptionFrame, + InterruptionFrame, VADUserStartedSpeakingFrame, VADUserStoppedSpeakingFrame, UserStartedSpeakingFrame, diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index cac1a0806..76400fc22 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -28,8 +28,9 @@ from pipecat.frames.frames import ( FrameProcessorPauseUrgentFrame, FrameProcessorResumeFrame, FrameProcessorResumeUrgentFrame, + InterruptionFrame, + InterruptionTaskFrame, StartFrame, - StartInterruptionFrame, SystemFrame, ) from pipecat.metrics.metrics import LLMTokenUsage, MetricsData @@ -219,6 +220,9 @@ class FrameProcessor(BaseObject): self.__process_event: Optional[asyncio.Event] = None self.__process_frame_task: Optional[asyncio.Task] = None + self._wait_for_interruption = False + self._wait_interruption_event = asyncio.Event() + @property def id(self) -> int: """Get the unique identifier for this processor. @@ -542,6 +546,14 @@ class FrameProcessor(BaseObject): if self._cancelling: return + # If we are waiting for an interruption we will bypass all queued system + # frames and we will process the frame right away. This is because a + # previous system frame might be waiting for the interruption frame and + # it's blocking the input task. + if self._wait_for_interruption and isinstance(frame, InterruptionFrame): + await self.__process_frame(frame, direction, callback) + return + if self._enable_direct_mode: await self.__process_frame(frame, direction, callback) else: @@ -588,7 +600,7 @@ class FrameProcessor(BaseObject): if isinstance(frame, StartFrame): await self.__start(frame) - elif isinstance(frame, StartInterruptionFrame): + elif isinstance(frame, InterruptionFrame): await self._start_interruption() await self.stop_all_metrics() elif isinstance(frame, CancelFrame): @@ -620,6 +632,32 @@ class FrameProcessor(BaseObject): await self.__internal_push_frame(frame, direction) + if isinstance(frame, InterruptionFrame): + self._wait_interruption_event.set() + + async def push_interruption_task_frame_and_wait(self): + """Push an interruption task frame upstream and wait for the interruption. + + This function sends an `InterruptionTaskFrame` upstream to the pipeline + task and waits to receive the corresponding `InterruptionFrame`. When + the function finishes it is guaranteed that the `InterruptionFrame` has + been pushed downstream. + """ + self._wait_for_interruption = True + + await self.push_frame(InterruptionTaskFrame(), FrameDirection.UPSTREAM) + + # Wait for an `InterruptionFrame` to come to this processor and be + # pushed. Take a look at `push_frame()` to see how we first push the + # `InterruptionFrame` and then we set the event in order to maintain + # frame ordering. + await self._wait_interruption_event.wait() + + # Clean the event. + self._wait_interruption_event.clear() + + self._wait_for_interruption = False + async def __start(self, frame: StartFrame): """Handle the start frame to initialize processor state. @@ -669,20 +707,22 @@ class FrameProcessor(BaseObject): async def _start_interruption(self): """Start handling an interruption by cancelling current tasks.""" try: - # Cancel the process task. This will stop processing queued frames. - await self.__cancel_process_task() + if self._wait_for_interruption: + # If we get here we know the process task was just waiting for + # an interruption (push_interruption_task_frame_and_wait()), so + # we can't cancel the task because it might still need to do + # more things (e.g. pushing a frame after the + # interruption). Instead we just drain the queue because this is + # an interruption. + self.__reset_process_task() + else: + # Cancel and re-create the process task including the queue. + await self.__cancel_process_task() + self.__create_process_task() except Exception as e: logger.exception(f"Uncaught exception in {self} when handling _start_interruption: {e}") await self.push_error(ErrorFrame(str(e))) - # Create a new process queue and task. - self.__create_process_task() - - async def _stop_interruption(self): - """Stop handling an interruption.""" - # Nothing to do right now. - pass - async def __internal_push_frame(self, frame: Frame, direction: FrameDirection): """Internal method to push frames to adjacent processors. @@ -764,6 +804,17 @@ class FrameProcessor(BaseObject): self.__process_queue = asyncio.Queue() self.__process_frame_task = self.create_task(self.__process_frame_task_handler()) + def __reset_process_task(self): + """Reset non-system frame processing task.""" + if self._enable_direct_mode: + return + + self.__should_block_frames = False + self.__process_event = asyncio.Event() + while not self.__process_queue.empty(): + self.__process_queue.get_nowait() + self.__process_queue.task_done() + async def __cancel_process_task(self): """Cancel the non-system frame processing task.""" if self.__process_frame_task: diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 8d6e67551..f469c3431 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, @@ -1206,7 +1205,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_interruption_task_frame_and_wait() 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 424374faf..5900ffc7f 100644 --- a/src/pipecat/processors/transcript_processor.py +++ b/src/pipecat/processors/transcript_processor.py @@ -19,7 +19,7 @@ from pipecat.frames.frames import ( CancelFrame, EndFrame, Frame, - StartInterruptionFrame, + InterruptionFrame, TranscriptionFrame, TranscriptionMessage, TranscriptionUpdateFrame, @@ -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 @@ -195,7 +195,7 @@ class AssistantTranscriptProcessor(BaseTranscriptProcessor): """ await super().process_frame(frame, direction) - if isinstance(frame, (StartInterruptionFrame, CancelFrame)): + if isinstance(frame, (InterruptionFrame, CancelFrame)): # Push frame first otherwise our emitted transcription update frame # might get cleaned up. await self.push_frame(frame, direction) diff --git a/src/pipecat/serializers/exotel.py b/src/pipecat/serializers/exotel.py index 9ed342631..1a5859211 100644 --- a/src/pipecat/serializers/exotel.py +++ b/src/pipecat/serializers/exotel.py @@ -20,8 +20,8 @@ from pipecat.frames.frames import ( Frame, InputAudioRawFrame, InputDTMFFrame, + InterruptionFrame, StartFrame, - StartInterruptionFrame, TransportMessageFrame, TransportMessageUrgentFrame, ) @@ -98,7 +98,7 @@ class ExotelFrameSerializer(FrameSerializer): Returns: Serialized data as string or bytes, or None if the frame isn't handled. """ - if isinstance(frame, StartInterruptionFrame): + if isinstance(frame, InterruptionFrame): answer = {"event": "clear", "streamSid": self._stream_sid} return json.dumps(answer) elif isinstance(frame, AudioRawFrame): diff --git a/src/pipecat/serializers/plivo.py b/src/pipecat/serializers/plivo.py index aa8b4b27e..519e0893a 100644 --- a/src/pipecat/serializers/plivo.py +++ b/src/pipecat/serializers/plivo.py @@ -22,8 +22,8 @@ from pipecat.frames.frames import ( Frame, InputAudioRawFrame, InputDTMFFrame, + InterruptionFrame, StartFrame, - StartInterruptionFrame, TransportMessageFrame, TransportMessageUrgentFrame, ) @@ -122,7 +122,7 @@ class PlivoFrameSerializer(FrameSerializer): self._hangup_attempted = True await self._hang_up_call() return None - elif isinstance(frame, StartInterruptionFrame): + elif isinstance(frame, InterruptionFrame): answer = {"event": "clearAudio", "streamId": self._stream_id} return json.dumps(answer) elif isinstance(frame, AudioRawFrame): diff --git a/src/pipecat/serializers/telnyx.py b/src/pipecat/serializers/telnyx.py index 467c01ba2..7603c9bf7 100644 --- a/src/pipecat/serializers/telnyx.py +++ b/src/pipecat/serializers/telnyx.py @@ -29,8 +29,8 @@ from pipecat.frames.frames import ( Frame, InputAudioRawFrame, InputDTMFFrame, + InterruptionFrame, StartFrame, - StartInterruptionFrame, ) from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType @@ -137,7 +137,7 @@ class TelnyxFrameSerializer(FrameSerializer): self._hangup_attempted = True await self._hang_up_call() return None - elif isinstance(frame, StartInterruptionFrame): + elif isinstance(frame, InterruptionFrame): answer = {"event": "clear"} return json.dumps(answer) elif isinstance(frame, AudioRawFrame): diff --git a/src/pipecat/serializers/twilio.py b/src/pipecat/serializers/twilio.py index 57e7c8dba..f83787468 100644 --- a/src/pipecat/serializers/twilio.py +++ b/src/pipecat/serializers/twilio.py @@ -22,8 +22,8 @@ from pipecat.frames.frames import ( Frame, InputAudioRawFrame, InputDTMFFrame, + InterruptionFrame, StartFrame, - StartInterruptionFrame, TransportMessageFrame, TransportMessageUrgentFrame, ) @@ -122,7 +122,7 @@ class TwilioFrameSerializer(FrameSerializer): self._hangup_attempted = True await self._hang_up_call() return None - elif isinstance(frame, StartInterruptionFrame): + elif isinstance(frame, InterruptionFrame): answer = {"event": "clear", "streamSid": self._stream_sid} return json.dumps(answer) elif isinstance(frame, AudioRawFrame): diff --git a/src/pipecat/services/asyncai/tts.py b/src/pipecat/services/asyncai/tts.py index d536263bd..a453d6820 100644 --- a/src/pipecat/services/asyncai/tts.py +++ b/src/pipecat/services/asyncai/tts.py @@ -20,8 +20,8 @@ from pipecat.frames.frames import ( EndFrame, ErrorFrame, Frame, + InterruptionFrame, StartFrame, - StartInterruptionFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, @@ -275,7 +275,7 @@ class AsyncAITTSService(InterruptibleTTSService): direction: The direction to push the frame. """ await super().push_frame(frame, direction) - if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)): + if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)): self._started = False async def _receive_messages(self): diff --git a/src/pipecat/services/aws_nova_sonic/context.py b/src/pipecat/services/aws_nova_sonic/context.py index e23a18362..0ce5ce033 100644 --- a/src/pipecat/services/aws_nova_sonic/context.py +++ b/src/pipecat/services/aws_nova_sonic/context.py @@ -21,13 +21,13 @@ from pipecat.frames.frames import ( DataFrame, Frame, FunctionCallResultFrame, + InterruptionFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesAppendFrame, LLMMessagesUpdateFrame, LLMSetToolChoiceFrame, LLMSetToolsFrame, - StartInterruptionFrame, TextFrame, UserImageRawFrame, ) @@ -306,7 +306,7 @@ class AWSNovaSonicAssistantContextAggregator(OpenAIAssistantContextAggregator): if isinstance( frame, ( - StartInterruptionFrame, + InterruptionFrame, LLMFullResponseStartFrame, LLMFullResponseEndFrame, TextFrame, diff --git a/src/pipecat/services/cartesia/tts.py b/src/pipecat/services/cartesia/tts.py index 5efda600c..3b81da5d4 100644 --- a/src/pipecat/services/cartesia/tts.py +++ b/src/pipecat/services/cartesia/tts.py @@ -20,8 +20,8 @@ from pipecat.frames.frames import ( EndFrame, ErrorFrame, Frame, + InterruptionFrame, StartFrame, - StartInterruptionFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, @@ -371,7 +371,7 @@ class CartesiaTTSService(AudioContextWordTTSService): return self._websocket raise Exception("Websocket not connected") - async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection): + async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection): await super()._handle_interruption(frame, direction) await self.stop_all_metrics() if self._context_id: diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py index 8551fc5de..3d6d5bd4c 100644 --- a/src/pipecat/services/elevenlabs/tts.py +++ b/src/pipecat/services/elevenlabs/tts.py @@ -25,9 +25,9 @@ from pipecat.frames.frames import ( EndFrame, ErrorFrame, Frame, + InterruptionFrame, LLMFullResponseEndFrame, StartFrame, - StartInterruptionFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, @@ -460,7 +460,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService): direction: The direction to push the frame. """ await super().push_frame(frame, direction) - if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)): + if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)): self._started = False if isinstance(frame, TTSStoppedFrame): await self.add_word_timestamps([("Reset", 0)]) @@ -549,7 +549,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService): return self._websocket raise Exception("Websocket not connected") - async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection): + async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection): """Handle interruption by closing the current context.""" await super()._handle_interruption(frame, direction) @@ -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 @@ -856,7 +856,7 @@ class ElevenLabsHttpTTSService(WordTTSService): direction: The direction to push the frame. """ await super().push_frame(frame, direction) - if isinstance(frame, (StartInterruptionFrame, TTSStoppedFrame)): + if isinstance(frame, (InterruptionFrame, TTSStoppedFrame)): # Reset timing on interruption or stop self._reset_state() diff --git a/src/pipecat/services/fish/tts.py b/src/pipecat/services/fish/tts.py index 305c14884..b39b775e5 100644 --- a/src/pipecat/services/fish/tts.py +++ b/src/pipecat/services/fish/tts.py @@ -21,8 +21,8 @@ from pipecat.frames.frames import ( EndFrame, ErrorFrame, Frame, + InterruptionFrame, StartFrame, - StartInterruptionFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, @@ -259,7 +259,7 @@ class FishAudioTTSService(InterruptibleTTSService): return self._websocket raise Exception("Websocket not connected") - async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection): + async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection): await super()._handle_interruption(frame, direction) await self.stop_all_metrics() self._request_id = None diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 106f668e9..df560358f 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -33,6 +33,7 @@ from pipecat.frames.frames import ( InputAudioRawFrame, InputImageRawFrame, InputTextRawFrame, + InterruptionFrame, LLMContextFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, @@ -41,7 +42,6 @@ from pipecat.frames.frames import ( LLMTextFrame, LLMUpdateSettingsFrame, StartFrame, - StartInterruptionFrame, TranscriptionFrame, TTSAudioRawFrame, TTSStartedFrame, @@ -752,7 +752,7 @@ class GeminiMultimodalLiveLLMService(LLMService): elif isinstance(frame, InputImageRawFrame): await self._send_user_video(frame) await self.push_frame(frame, direction) - elif isinstance(frame, StartInterruptionFrame): + elif isinstance(frame, InterruptionFrame): await self._handle_interruption() await self.push_frame(frame, direction) elif isinstance(frame, UserStartedSpeakingFrame): diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 502570a83..e309609fe 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -36,12 +36,12 @@ from pipecat.frames.frames import ( FunctionCallResultFrame, FunctionCallResultProperties, FunctionCallsStartedFrame, + InterruptionFrame, LLMConfigureOutputFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMTextFrame, StartFrame, - StartInterruptionFrame, UserImageRequestFrame, ) from pipecat.processors.aggregators.llm_context import LLMContext @@ -269,7 +269,7 @@ class LLMService(AIService): """ await super().process_frame(frame, direction) - if isinstance(frame, StartInterruptionFrame): + if isinstance(frame, InterruptionFrame): await self._handle_interruptions(frame) elif isinstance(frame, LLMConfigureOutputFrame): self._skip_tts = frame.skip_tts @@ -286,7 +286,7 @@ class LLMService(AIService): await super().push_frame(frame, direction) - async def _handle_interruptions(self, _: StartInterruptionFrame): + async def _handle_interruptions(self, _: InterruptionFrame): for function_name, entry in self._functions.items(): if entry.cancel_on_interruption: await self._cancel_function_call(function_name) diff --git a/src/pipecat/services/lmnt/tts.py b/src/pipecat/services/lmnt/tts.py index 187ef6b84..a602789fd 100644 --- a/src/pipecat/services/lmnt/tts.py +++ b/src/pipecat/services/lmnt/tts.py @@ -16,8 +16,8 @@ from pipecat.frames.frames import ( EndFrame, ErrorFrame, Frame, + InterruptionFrame, StartFrame, - StartInterruptionFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, @@ -180,7 +180,7 @@ class LmntTTSService(InterruptibleTTSService): direction: The direction to push the frame. """ await super().push_frame(frame, direction) - if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)): + if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)): self._started = False async def _connect(self): diff --git a/src/pipecat/services/neuphonic/tts.py b/src/pipecat/services/neuphonic/tts.py index f4777b8d9..46d805086 100644 --- a/src/pipecat/services/neuphonic/tts.py +++ b/src/pipecat/services/neuphonic/tts.py @@ -25,9 +25,9 @@ from pipecat.frames.frames import ( EndFrame, ErrorFrame, Frame, + InterruptionFrame, LLMFullResponseEndFrame, StartFrame, - StartInterruptionFrame, TTSAudioRawFrame, TTSSpeakFrame, TTSStartedFrame, @@ -224,7 +224,7 @@ class NeuphonicTTSService(InterruptibleTTSService): direction: The direction to push the frame. """ await super().push_frame(frame, direction) - if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)): + if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)): self._started = False async def process_frame(self, frame: Frame, direction: FrameDirection): diff --git a/src/pipecat/services/openai_realtime/openai.py b/src/pipecat/services/openai_realtime/openai.py index 67e3ac9e6..8b3d500eb 100644 --- a/src/pipecat/services/openai_realtime/openai.py +++ b/src/pipecat/services/openai_realtime/openai.py @@ -23,6 +23,7 @@ from pipecat.frames.frames import ( Frame, InputAudioRawFrame, InterimTranscriptionFrame, + InterruptionFrame, LLMContextFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, @@ -31,7 +32,6 @@ from pipecat.frames.frames import ( LLMTextFrame, LLMUpdateSettingsFrame, StartFrame, - StartInterruptionFrame, TranscriptionFrame, TTSAudioRawFrame, TTSStartedFrame, @@ -366,7 +366,7 @@ class OpenAIRealtimeLLMService(LLMService): elif isinstance(frame, InputAudioRawFrame): if not self._audio_input_paused: await self._send_user_audio(frame) - elif isinstance(frame, StartInterruptionFrame): + elif isinstance(frame, InterruptionFrame): await self._handle_interruption() elif isinstance(frame, UserStartedSpeakingFrame): await self._handle_user_started_speaking(frame) @@ -716,14 +716,12 @@ class OpenAIRealtimeLLMService(LLMService): async def _handle_evt_speech_started(self, evt): await self._truncate_current_audio_response() - await self._start_interruption() # cancels this processor task - await self.push_frame(StartInterruptionFrame()) # cancels downstream tasks + await self.push_interruption_task_frame_and_wait() await self.push_frame(UserStartedSpeakingFrame()) async def _handle_evt_speech_stopped(self, evt): await self.start_ttfb_metrics() await self.start_processing_metrics() - await self._stop_interruption() await self.push_frame(UserStoppedSpeakingFrame()) async def _maybe_handle_evt_retrieve_conversation_item_error(self, evt: events.ErrorEvent): diff --git a/src/pipecat/services/openai_realtime_beta/openai.py b/src/pipecat/services/openai_realtime_beta/openai.py index ef0ea92d6..922f9a572 100644 --- a/src/pipecat/services/openai_realtime_beta/openai.py +++ b/src/pipecat/services/openai_realtime_beta/openai.py @@ -24,6 +24,7 @@ from pipecat.frames.frames import ( Frame, InputAudioRawFrame, InterimTranscriptionFrame, + InterruptionFrame, LLMContextFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, @@ -32,7 +33,6 @@ from pipecat.frames.frames import ( LLMTextFrame, LLMUpdateSettingsFrame, StartFrame, - StartInterruptionFrame, TranscriptionFrame, TTSAudioRawFrame, TTSStartedFrame, @@ -364,7 +364,7 @@ class OpenAIRealtimeBetaLLMService(LLMService): elif isinstance(frame, InputAudioRawFrame): if not self._audio_input_paused: await self._send_user_audio(frame) - elif isinstance(frame, StartInterruptionFrame): + elif isinstance(frame, InterruptionFrame): await self._handle_interruption() elif isinstance(frame, UserStartedSpeakingFrame): await self._handle_user_started_speaking(frame) @@ -658,14 +658,12 @@ class OpenAIRealtimeBetaLLMService(LLMService): async def _handle_evt_speech_started(self, evt): await self._truncate_current_audio_response() - await self._start_interruption() # cancels this processor task - await self.push_frame(StartInterruptionFrame()) # cancels downstream tasks + await self.push_interruption_task_frame_and_wait() await self.push_frame(UserStartedSpeakingFrame()) async def _handle_evt_speech_stopped(self, evt): await self.start_ttfb_metrics() await self.start_processing_metrics() - await self._stop_interruption() await self.push_frame(UserStoppedSpeakingFrame()) async def _maybe_handle_evt_retrieve_conversation_item_error(self, evt: events.ErrorEvent): diff --git a/src/pipecat/services/playht/tts.py b/src/pipecat/services/playht/tts.py index aa92df055..0f23b7b5e 100644 --- a/src/pipecat/services/playht/tts.py +++ b/src/pipecat/services/playht/tts.py @@ -25,8 +25,8 @@ from pipecat.frames.frames import ( EndFrame, ErrorFrame, Frame, + InterruptionFrame, StartFrame, - StartInterruptionFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, @@ -312,7 +312,7 @@ class PlayHTTTSService(InterruptibleTTSService): return self._websocket raise Exception("Websocket not connected") - async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection): + async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection): """Handle interruption by stopping metrics and clearing request ID.""" await super()._handle_interruption(frame, direction) await self.stop_all_metrics() diff --git a/src/pipecat/services/rime/tts.py b/src/pipecat/services/rime/tts.py index be979b7f9..917716545 100644 --- a/src/pipecat/services/rime/tts.py +++ b/src/pipecat/services/rime/tts.py @@ -24,15 +24,14 @@ from pipecat.frames.frames import ( EndFrame, ErrorFrame, Frame, + InterruptionFrame, StartFrame, - StartInterruptionFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.tts_service import AudioContextWordTTSService, TTSService -from pipecat.transcriptions import language from pipecat.transcriptions.language import Language from pipecat.utils.text.base_text_aggregator import BaseTextAggregator from pipecat.utils.text.skip_tags_aggregator import SkipTagsAggregator @@ -280,7 +279,7 @@ class RimeTTSService(AudioContextWordTTSService): return self._websocket raise Exception("Websocket not connected") - async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection): + async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection): """Handle interruption by clearing current context.""" await super()._handle_interruption(frame, direction) await self.stop_all_metrics() @@ -375,7 +374,7 @@ class RimeTTSService(AudioContextWordTTSService): direction: The direction to push the frame. """ await super().push_frame(frame, direction) - if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)): + if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)): if isinstance(frame, TTSStoppedFrame): await self.add_word_timestamps([("Reset", 0)]) diff --git a/src/pipecat/services/sarvam/tts.py b/src/pipecat/services/sarvam/tts.py index 01702fd04..a9fedcc58 100644 --- a/src/pipecat/services/sarvam/tts.py +++ b/src/pipecat/services/sarvam/tts.py @@ -20,9 +20,9 @@ from pipecat.frames.frames import ( EndFrame, ErrorFrame, Frame, + InterruptionFrame, LLMFullResponseEndFrame, StartFrame, - StartInterruptionFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, @@ -455,7 +455,7 @@ class SarvamTTSService(InterruptibleTTSService): direction: The direction to push the frame. """ await super().push_frame(frame, direction) - if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)): + if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)): self._started = False async def process_frame(self, frame: Frame, direction: FrameDirection): diff --git a/src/pipecat/services/simli/video.py b/src/pipecat/services/simli/video.py index e35dad4c6..d48a744e0 100644 --- a/src/pipecat/services/simli/video.py +++ b/src/pipecat/services/simli/video.py @@ -15,8 +15,8 @@ from pipecat.frames.frames import ( CancelFrame, EndFrame, Frame, + InterruptionFrame, OutputImageRawFrame, - StartInterruptionFrame, TTSAudioRawFrame, TTSStoppedFrame, UserStartedSpeakingFrame, @@ -179,7 +179,7 @@ class SimliVideoService(FrameProcessor): return elif isinstance(frame, (EndFrame, CancelFrame)): await self._stop() - elif isinstance(frame, (StartInterruptionFrame, UserStartedSpeakingFrame)): + elif isinstance(frame, (InterruptionFrame, UserStartedSpeakingFrame)): if not self._previously_interrupted: await self._simli_client.clearBuffer() self._previously_interrupted = self._is_trinity_avatar diff --git a/src/pipecat/services/speechmatics/stt.py b/src/pipecat/services/speechmatics/stt.py index 7136bbb0c..4028dd248 100644 --- a/src/pipecat/services/speechmatics/stt.py +++ b/src/pipecat/services/speechmatics/stt.py @@ -19,7 +19,6 @@ from loguru import logger from pydantic import BaseModel from pipecat.frames.frames import ( - BotInterruptionFrame, CancelFrame, EndFrame, ErrorFrame, @@ -749,14 +748,13 @@ class SpeechmaticsSTTService(STTService): return # Frames to send - upstream_frames: list[Frame] = [] downstream_frames: list[Frame] = [] # If VAD is enabled, then send a speaking frame if self._params.enable_vad and not self._is_speaking: logger.debug("User started speaking") self._is_speaking = True - upstream_frames += [BotInterruptionFrame()] + await self.push_interruption_task_frame_and_wait() downstream_frames += [UserStartedSpeakingFrame()] # If final, then re-parse into TranscriptionFrame @@ -794,10 +792,6 @@ class SpeechmaticsSTTService(STTService): self._is_speaking = False downstream_frames += [UserStoppedSpeakingFrame()] - # Send UPSTREAM frames - for frame in upstream_frames: - await self.push_frame(frame, FrameDirection.UPSTREAM) - # Send the DOWNSTREAM frames for frame in downstream_frames: await self.push_frame(frame, FrameDirection.DOWNSTREAM) diff --git a/src/pipecat/services/tavus/video.py b/src/pipecat/services/tavus/video.py index aff5778d9..68309a5c9 100644 --- a/src/pipecat/services/tavus/video.py +++ b/src/pipecat/services/tavus/video.py @@ -23,12 +23,12 @@ from pipecat.frames.frames import ( CancelFrame, EndFrame, Frame, + InterruptionFrame, OutputAudioRawFrame, OutputImageRawFrame, OutputTransportReadyFrame, SpeechOutputAudioRawFrame, StartFrame, - StartInterruptionFrame, TTSAudioRawFrame, TTSStartedFrame, ) @@ -222,7 +222,7 @@ class TavusVideoService(AIService): """ await super().process_frame(frame, direction) - if isinstance(frame, StartInterruptionFrame): + if isinstance(frame, InterruptionFrame): await self._handle_interruptions() await self.push_frame(frame, direction) elif isinstance(frame, TTSAudioRawFrame): diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index 93800338b..02b80b609 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -20,10 +20,10 @@ from pipecat.frames.frames import ( ErrorFrame, Frame, InterimTranscriptionFrame, + InterruptionFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, StartFrame, - StartInterruptionFrame, TextFrame, TranscriptionFrame, TTSAudioRawFrame, @@ -309,7 +309,7 @@ class TTSService(AIService): and not isinstance(frame, TranscriptionFrame) ): await self._process_text_frame(frame) - elif isinstance(frame, StartInterruptionFrame): + elif isinstance(frame, InterruptionFrame): await self._handle_interruption(frame, direction) await self.push_frame(frame, direction) elif isinstance(frame, (LLMFullResponseEndFrame, EndFrame)): @@ -367,14 +367,14 @@ class TTSService(AIService): await super().push_frame(frame, direction) if self._push_stop_frames and ( - isinstance(frame, StartInterruptionFrame) + isinstance(frame, InterruptionFrame) or isinstance(frame, TTSStartedFrame) or isinstance(frame, TTSAudioRawFrame) or isinstance(frame, TTSStoppedFrame) ): await self._stop_frame_queue.put(frame) - async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection): + async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection): self._processing_text = False await self._text_aggregator.handle_interruption() for filter in self._text_filters: @@ -438,7 +438,7 @@ class TTSService(AIService): ) if isinstance(frame, TTSStartedFrame): has_started = True - elif isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)): + elif isinstance(frame, (TTSStoppedFrame, InterruptionFrame)): has_started = False except asyncio.TimeoutError: if has_started: @@ -523,7 +523,7 @@ class WordTTSService(TTSService): elif isinstance(frame, (LLMFullResponseEndFrame, EndFrame)): await self.flush_audio() - async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection): + async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection): await super()._handle_interruption(frame, direction) self._llm_response_started = False self.reset_word_timestamps() @@ -613,7 +613,7 @@ class InterruptibleTTSService(WebsocketTTSService): # user interrupts we need to reconnect. self._bot_speaking = False - async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection): + async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection): await super()._handle_interruption(frame, direction) if self._bot_speaking: await self._disconnect() @@ -685,7 +685,7 @@ class InterruptibleWordTTSService(WebsocketWordTTSService): # user interrupts we need to reconnect. self._bot_speaking = False - async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection): + async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection): await super()._handle_interruption(frame, direction) if self._bot_speaking: await self._disconnect() @@ -813,7 +813,7 @@ class AudioContextWordTTSService(WebsocketWordTTSService): await super().cancel(frame) await self._stop_audio_context_task() - async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection): + async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection): await super()._handle_interruption(frame, direction) await self._stop_audio_context_task() self._create_audio_context_task() diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index cb3808bf9..f2ccd1d96 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, @@ -36,7 +35,6 @@ from pipecat.frames.frames import ( MetricsFrame, SpeechControlParamsFrame, StartFrame, - StartInterruptionFrame, StopFrame, SystemFrame, UserSpeakingFrame, @@ -289,8 +287,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 +331,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(StartInterruptionFrame()) - 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 +342,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 = ( @@ -362,11 +351,7 @@ class BaseInputTransport(FrameProcessor): # Make sure we notify about interruptions quickly out-of-band. 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()) + await self.push_interruption_task_frame_and_wait() elif self.interruption_strategies and self._bot_speaking: logger.debug( "User started speaking while bot is speaking with interruption config - " @@ -381,9 +366,6 @@ class BaseInputTransport(FrameProcessor): await self.push_frame(downstream_frame) await self.push_frame(upstream_frame, FrameDirection.UPSTREAM) - if self.interruptions_allowed: - await self._stop_interruption() - # # Handle bot speaking state # diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 3830aa9c9..95db2b71b 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -30,6 +30,7 @@ from pipecat.frames.frames import ( EndFrame, Frame, InputTransportMessageUrgentFrame, + InterruptionFrame, MixerControlFrame, OutputAudioRawFrame, OutputDTMFFrame, @@ -39,7 +40,6 @@ from pipecat.frames.frames import ( SpeechOutputAudioRawFrame, SpriteFrame, StartFrame, - StartInterruptionFrame, SystemFrame, TransportMessageFrame, TransportMessageUrgentFrame, @@ -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 @@ -299,7 +298,7 @@ class BaseOutputTransport(FrameProcessor): elif isinstance(frame, CancelFrame): await self.cancel(frame) await self.push_frame(frame, direction) - elif isinstance(frame, StartInterruptionFrame): + elif isinstance(frame, InterruptionFrame): await self.push_frame(frame, direction) await self._handle_frame(frame) elif isinstance(frame, TransportMessageUrgentFrame) and not isinstance( @@ -340,7 +339,7 @@ class BaseOutputTransport(FrameProcessor): sender = self._media_senders[frame.transport_destination] - if isinstance(frame, StartInterruptionFrame): + if isinstance(frame, InterruptionFrame): await sender.handle_interruptions(frame) elif isinstance(frame, OutputAudioRawFrame): await sender.handle_audio_frame(frame) @@ -491,7 +490,7 @@ class BaseOutputTransport(FrameProcessor): await self._cancel_clock_task() await self._cancel_video_task() - async def handle_interruptions(self, _: StartInterruptionFrame): + async def handle_interruptions(self, _: InterruptionFrame): """Handle interruption events by restarting tasks and clearing buffers. Args: diff --git a/src/pipecat/transports/tavus/transport.py b/src/pipecat/transports/tavus/transport.py index aea30f72f..23e14f97b 100644 --- a/src/pipecat/transports/tavus/transport.py +++ b/src/pipecat/transports/tavus/transport.py @@ -25,9 +25,9 @@ from pipecat.frames.frames import ( EndFrame, Frame, InputAudioRawFrame, + InterruptionFrame, OutputAudioRawFrame, StartFrame, - StartInterruptionFrame, TransportMessageFrame, TransportMessageUrgentFrame, ) @@ -618,7 +618,7 @@ class TavusOutputTransport(BaseOutputTransport): direction: The direction of frame flow in the pipeline. """ await super().process_frame(frame, direction) - if isinstance(frame, StartInterruptionFrame): + if isinstance(frame, InterruptionFrame): await self._handle_interruptions() async def _handle_interruptions(self): diff --git a/src/pipecat/transports/websocket/fastapi.py b/src/pipecat/transports/websocket/fastapi.py index 8287783c2..474f70de8 100644 --- a/src/pipecat/transports/websocket/fastapi.py +++ b/src/pipecat/transports/websocket/fastapi.py @@ -26,9 +26,9 @@ from pipecat.frames.frames import ( EndFrame, Frame, InputAudioRawFrame, + InterruptionFrame, OutputAudioRawFrame, StartFrame, - StartInterruptionFrame, TransportMessageFrame, TransportMessageUrgentFrame, ) @@ -398,7 +398,7 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): """ await super().process_frame(frame, direction) - if isinstance(frame, StartInterruptionFrame): + if isinstance(frame, InterruptionFrame): await self._write_frame(frame) self._next_send_time = 0 diff --git a/src/pipecat/transports/websocket/server.py b/src/pipecat/transports/websocket/server.py index 8e73fb47e..67631ab04 100644 --- a/src/pipecat/transports/websocket/server.py +++ b/src/pipecat/transports/websocket/server.py @@ -25,9 +25,9 @@ from pipecat.frames.frames import ( EndFrame, Frame, InputAudioRawFrame, + InterruptionFrame, OutputAudioRawFrame, StartFrame, - StartInterruptionFrame, TransportMessageFrame, TransportMessageUrgentFrame, ) @@ -334,7 +334,7 @@ class WebsocketServerOutputTransport(BaseOutputTransport): """ await super().process_frame(frame, direction) - if isinstance(frame, StartInterruptionFrame): + if isinstance(frame, InterruptionFrame): await self._write_frame(frame) self._next_send_time = 0 diff --git a/tests/test_context_aggregators.py b/tests/test_context_aggregators.py index e04487f69..9c29c78e8 100644 --- a/tests/test_context_aggregators.py +++ b/tests/test_context_aggregators.py @@ -17,11 +17,11 @@ from pipecat.frames.frames import ( FunctionCallResultFrame, FunctionCallResultProperties, InterimTranscriptionFrame, + InterruptionFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, OpenAILLMContextAssistantTimestampFrame, SpeechControlParamsFrame, - StartInterruptionFrame, TextFrame, TranscriptionFrame, UserStartedSpeakingFrame, @@ -618,7 +618,7 @@ class BaseTestAssistantContextAggreagator: TextFrame(text="Pipecat."), LLMFullResponseEndFrame(), SleepFrame(AGGREGATION_SLEEP), - StartInterruptionFrame(), + InterruptionFrame(), LLMFullResponseStartFrame(), TextFrame(text="How are "), TextFrame(text="you?"), @@ -626,7 +626,7 @@ class BaseTestAssistantContextAggreagator: ] expected_down_frames = [ *self.EXPECTED_CONTEXT_FRAMES, - StartInterruptionFrame, + InterruptionFrame, *self.EXPECTED_CONTEXT_FRAMES, ] await run_test( diff --git a/tests/test_dtmf_aggregator.py b/tests/test_dtmf_aggregator.py index 5d9d1346a..c7590ae47 100644 --- a/tests/test_dtmf_aggregator.py +++ b/tests/test_dtmf_aggregator.py @@ -10,6 +10,7 @@ from pipecat.audio.dtmf.types import KeypadEntry from pipecat.frames.frames import ( EndFrame, InputDTMFFrame, + InterruptionFrame, TranscriptionFrame, ) from pipecat.processors.aggregators.dtmf_aggregator import DTMFAggregator @@ -28,6 +29,7 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase): ] expected_down_frames = [ InputDTMFFrame, + InterruptionFrame, InputDTMFFrame, InputDTMFFrame, InputDTMFFrame, @@ -59,9 +61,11 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase): ] expected_down_frames = [ InputDTMFFrame, + InterruptionFrame, InputDTMFFrame, TranscriptionFrame, # First aggregation "12" InputDTMFFrame, + InterruptionFrame, TranscriptionFrame, # Second aggregation "3" ] @@ -93,10 +97,12 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase): ] expected_down_frames = [ InputDTMFFrame, + InterruptionFrame, InputDTMFFrame, InputDTMFFrame, TranscriptionFrame, # "12#" InputDTMFFrame, + InterruptionFrame, InputDTMFFrame, TranscriptionFrame, # "45" ] @@ -125,6 +131,7 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase): ] expected_down_frames = [ InputDTMFFrame, + InterruptionFrame, InputDTMFFrame, TranscriptionFrame, # Should flush before EndFrame EndFrame, @@ -152,6 +159,7 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase): ] expected_down_frames = [ InputDTMFFrame, + InterruptionFrame, InputDTMFFrame, TranscriptionFrame, ] @@ -178,6 +186,7 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase): ] expected_down_frames = [ InputDTMFFrame, + InterruptionFrame, InputDTMFFrame, InputDTMFFrame, TranscriptionFrame, @@ -214,7 +223,11 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase): ] # All the InputDTMFFrames plus one TranscriptionFrame - expected_down_frames = [InputDTMFFrame] * len(frames_to_send) + [TranscriptionFrame] + expected_down_frames = ( + [InputDTMFFrame, InterruptionFrame] + + [InputDTMFFrame] * (len(frames_to_send) - 1) + + [TranscriptionFrame] + ) received_down_frames, _ = await run_test( aggregator, diff --git a/tests/test_llm_response.py b/tests/test_llm_response.py index 93838a658..663bd6671 100644 --- a/tests/test_llm_response.py +++ b/tests/test_llm_response.py @@ -7,10 +7,10 @@ import unittest from pipecat.frames.frames import ( + InterruptionFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMTextFrame, - StartInterruptionFrame, ) from pipecat.processors.aggregators.llm_response import LLMFullResponseAggregator from pipecat.tests.utils import SleepFrame, run_test @@ -113,7 +113,7 @@ class TestLLMFullResponseAggregator(unittest.IsolatedAsyncioTestCase): LLMFullResponseStartFrame(), LLMTextFrame("Hello "), SleepFrame(), - StartInterruptionFrame(), + InterruptionFrame(), LLMFullResponseStartFrame(), LLMTextFrame("Hello "), LLMTextFrame("there!"), @@ -122,7 +122,7 @@ class TestLLMFullResponseAggregator(unittest.IsolatedAsyncioTestCase): expected_down_frames = [ LLMFullResponseStartFrame, LLMTextFrame, - StartInterruptionFrame, + InterruptionFrame, LLMFullResponseStartFrame, LLMTextFrame, LLMTextFrame, diff --git a/tests/test_transcript_processor.py b/tests/test_transcript_processor.py index a2a057a13..b433951ce 100644 --- a/tests/test_transcript_processor.py +++ b/tests/test_transcript_processor.py @@ -14,7 +14,7 @@ from pipecat.frames.frames import ( BotStartedSpeakingFrame, BotStoppedSpeakingFrame, CancelFrame, - StartInterruptionFrame, + InterruptionFrame, TranscriptionFrame, TranscriptionMessage, TranscriptionUpdateFrame, @@ -238,7 +238,7 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase): TTSTextFrame(text="Hello"), TTSTextFrame(text="world!"), SleepFrame(), - StartInterruptionFrame(), # User interrupts here + InterruptionFrame(), # User interrupts here SleepFrame(), BotStartedSpeakingFrame(), TTSTextFrame(text="New"), @@ -252,7 +252,7 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase): BotStartedSpeakingFrame, TTSTextFrame, # "Hello" TTSTextFrame, # "world!" - StartInterruptionFrame, + InterruptionFrame, TranscriptionUpdateFrame, # First message (emitted due to interruption) BotStartedSpeakingFrame, TTSTextFrame, # "New"