From 41e3afbc2f7fe5ddd29bcedca3866f7ce8b4e63c Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 2 Apr 2026 10:28:01 -0400 Subject: [PATCH 01/11] Remove deprecated add_pattern_pair method from PatternPairAggregator --- .../utils/text/pattern_pair_aggregator.py | 40 ------------------- tests/test_pattern_pair_aggregator.py | 4 +- 2 files changed, 2 insertions(+), 42 deletions(-) diff --git a/src/pipecat/utils/text/pattern_pair_aggregator.py b/src/pipecat/utils/text/pattern_pair_aggregator.py index 835bb8591..c69622e8b 100644 --- a/src/pipecat/utils/text/pattern_pair_aggregator.py +++ b/src/pipecat/utils/text/pattern_pair_aggregator.py @@ -161,46 +161,6 @@ class PatternPairAggregator(SimpleTextAggregator): } return self - def add_pattern_pair( - self, pattern_id: str, start_pattern: str, end_pattern: str, remove_match: bool = True - ): - """Add a pattern pair to detect in the text. - - .. deprecated:: 0.0.95 - This function is deprecated and will be removed in a future version. - Use `add_pattern` with a type and MatchAction instead. - - This method calls `add_pattern` setting type with the provided pattern_id and action - to either MatchAction.REMOVE or MatchAction.KEEP based on `remove_match`. - - Args: - pattern_id: Identifier for this pattern pair. Should be unique and ideally descriptive. - (e.g., 'code', 'speaker', 'custom'). pattern_id can not be 'sentence' or 'word' - as those arereserved for the default behavior. - start_pattern: Pattern that marks the beginning of content. - end_pattern: Pattern that marks the end of content. - remove_match: If True, the matched pattern will be removed from the text. (Same as MatchAction.REMOVE) - If False, it will be kept and treated as normal text. (Same as MatchAction.KEEP) - """ - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("once") - warnings.warn( - "add_pattern_pair with a pattern_id or remove_match is deprecated and will be" - " removed in a future version. Use add_pattern with a type and MatchAction instead", - DeprecationWarning, - stacklevel=2, - ) - - action = MatchAction.REMOVE if remove_match else MatchAction.KEEP - return self.add_pattern( - type=pattern_id, - start_pattern=start_pattern, - end_pattern=end_pattern, - action=action, - ) - def on_pattern_match( self, type: str, handler: Callable[[PatternMatch], Awaitable[None]] ) -> "PatternPairAggregator": diff --git a/tests/test_pattern_pair_aggregator.py b/tests/test_pattern_pair_aggregator.py index 6c9e23552..6dd055c20 100644 --- a/tests/test_pattern_pair_aggregator.py +++ b/tests/test_pattern_pair_aggregator.py @@ -21,8 +21,8 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.code_handler = AsyncMock() # Add a test pattern - self.aggregator.add_pattern_pair( - pattern_id="test_pattern", + self.aggregator.add_pattern( + type="test_pattern", start_pattern="", end_pattern="", ) From 74acb0b7d0fe34a26d6c3019ae711a6a9111117c Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 2 Apr 2026 10:31:15 -0400 Subject: [PATCH 02/11] Remove deprecated class_decorators tracing module --- src/pipecat/utils/tracing/class_decorators.py | 257 ------------------ .../utils/tracing/service_decorators.py | 5 - 2 files changed, 262 deletions(-) delete mode 100644 src/pipecat/utils/tracing/class_decorators.py diff --git a/src/pipecat/utils/tracing/class_decorators.py b/src/pipecat/utils/tracing/class_decorators.py deleted file mode 100644 index 73dacbe51..000000000 --- a/src/pipecat/utils/tracing/class_decorators.py +++ /dev/null @@ -1,257 +0,0 @@ -# -# Copyright (c) 2024-2026, Daily -# Portions Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -"""Base OpenTelemetry tracing decorators and utilities for Pipecat. - -.. deprecated:: 0.0.103 - This module is unused and will be removed in a future release. - Service tracing is handled by the decorators in - :mod:`pipecat.utils.tracing.service_decorators`. - -This module provides class and method level tracing capabilities -similar to the original NVIDIA implementation. -""" - -import asyncio -import contextlib -import enum -import functools -import inspect -import warnings -from typing import Callable, Optional, TypeVar - -warnings.warn( - "pipecat.utils.tracing.class_decorators is deprecated and will be removed in a future " - "release. Use pipecat.utils.tracing.service_decorators instead.", - DeprecationWarning, - stacklevel=2, -) - -from pipecat.utils.tracing.setup import is_tracing_available - -# Import OpenTelemetry if available -if is_tracing_available(): - import opentelemetry.trace - from opentelemetry import metrics, trace - -# Type variables for better typing support -T = TypeVar("T") -C = TypeVar("C", bound=type) - - -class AttachmentStrategy(enum.Enum): - """Controls how spans are attached to the trace hierarchy. - - Parameters: - CHILD: Attached to class span if no parent, otherwise to parent. - LINK: Attached to class span with link to parent. - NONE: Always attached to class span regardless of context. - """ - - CHILD = enum.auto() - LINK = enum.auto() - NONE = enum.auto() - - -class Traceable: - """Base class for objects that can be traced with OpenTelemetry. - - Provides the foundational tracing capabilities used by @traced methods. - """ - - def __init__(self, name: str, **kwargs): - """Initialize a traceable object. - - Args: - name: Name of the traceable object for the span. - **kwargs: Additional arguments passed to parent class. - """ - super().__init__(**kwargs) - - if not is_tracing_available(): - self._tracer = self._meter = self._parent_span_id = self._span = None - return - - self._tracer = trace.get_tracer("pipecat") - self._meter = metrics.get_meter("pipecat") - self._parent_span_id = trace.get_current_span().get_span_context().span_id - self._span = self._tracer.start_span(name) - self._span.end() - - @property - def meter(self): - """Get the OpenTelemetry meter instance. - - Returns: - The OpenTelemetry meter instance for this object. - """ - return self._meter - - -@contextlib.contextmanager -def __traced_context_manager( - self: Traceable, func: Callable, name: str | None, attachment_strategy: AttachmentStrategy -): - """Internal context manager for the traced decorator. - - Args: - self: The Traceable instance. - func: The function being traced. - name: Custom span name or None to use function name. - attachment_strategy: How to attach this span to the trace hierarchy. - - Raises: - RuntimeError: If used in a class not inheriting from Traceable. - """ - if not isinstance(self, Traceable): - raise RuntimeError( - "@traced annotation can only be used in classes inheriting from Traceable" - ) - - stack = contextlib.ExitStack() - try: - current_span = trace.get_current_span() - is_span_class_parent_span = current_span.get_span_context().span_id == self._parent_span_id - match attachment_strategy: - case AttachmentStrategy.CHILD if not is_span_class_parent_span: - stack.enter_context( - self._tracer.start_as_current_span(func.__name__ if name is None else name) # type: ignore - ) - case AttachmentStrategy.LINK: - if is_span_class_parent_span: - link = trace.Link(self._span.get_span_context()) # type: ignore - else: - link = trace.Link(current_span.get_span_context()) - stack.enter_context( - opentelemetry.trace.use_span(span=self._span, end_on_exit=False) # type: ignore - ) - stack.enter_context( - self._tracer.start_as_current_span( # type: ignore - func.__name__ if name is None else name, links=[link] - ) - ) - case AttachmentStrategy.NONE | AttachmentStrategy.CHILD: - stack.enter_context( - opentelemetry.trace.use_span(span=self._span, end_on_exit=False) # type: ignore - ) - stack.enter_context( - self._tracer.start_as_current_span(func.__name__ if name is None else name) # type: ignore - ) - yield - finally: - stack.close() - - -def __traced_decorator(func, name, attachment_strategy: AttachmentStrategy): - """Implementation of the traced decorator. - - Args: - func: The function to trace. - name: Custom span name. - attachment_strategy: How to attach this span. - - Returns: - The wrapped function with tracing capabilities. - """ - - @functools.wraps(func) - async def coroutine_wrapper(self: Traceable, *args, **kwargs): - exception = None - with __traced_context_manager(self, func, name, attachment_strategy): - try: - return await func(self, *args, **kwargs) - except asyncio.CancelledError as e: - exception = e - if exception: - raise exception - - @functools.wraps(func) - async def generator_wrapper(self: Traceable, *args, **kwargs): - exception = None - with __traced_context_manager(self, func, name, attachment_strategy): - try: - async for v in func(self, *args, **kwargs): - yield v - except asyncio.CancelledError as e: - exception = e - if exception: - raise exception - - if inspect.iscoroutinefunction(func): - return coroutine_wrapper - if inspect.isasyncgenfunction(func): - return generator_wrapper - - raise ValueError("@traced annotation can only be used on async or async generator functions") - - -def traced( - func: Optional[Callable] = None, - *, - name: Optional[str] = None, - attachment_strategy: AttachmentStrategy = AttachmentStrategy.CHILD, -) -> Callable: - """Add tracing to an async function in a Traceable class. - - Args: - func: The async function to trace. - name: Custom span name. Defaults to function name. - attachment_strategy: How to attach this span (CHILD, LINK, NONE). - - Returns: - Wrapped async function with tracing. - - Raises: - RuntimeError: If used in a class not inheriting from Traceable. - ValueError: If used on a non-async function. - """ - if not is_tracing_available(): - # Just return the original function or a simple decorator - def decorator(f): - return f - - return decorator if func is None else func - - if func is not None: - return __traced_decorator(func, name=name, attachment_strategy=attachment_strategy) - else: - return functools.partial( - __traced_decorator, name=name, attachment_strategy=attachment_strategy - ) - - -def traceable(cls: C) -> C: - """Make a class traceable for OpenTelemetry. - - Creates a new class that inherits from both the original class - and Traceable, enabling tracing for class methods. - - Args: - cls: The class to make traceable. - - Returns: - A new class with tracing capabilities. - """ - if not is_tracing_available(): - return cls - - @functools.wraps(cls, updated=()) - class TracedClass(cls, Traceable): - def __init__(self, *args, **kwargs): - """Initialize the traced class instance. - - Args: - *args: Positional arguments passed to parent classes. - **kwargs: Keyword arguments passed to parent classes. - """ - cls.__init__(self, *args, **kwargs) - if hasattr(self, "name"): - Traceable.__init__(self, self.name) - else: - Traceable.__init__(self, cls.__name__) - - return TracedClass diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py index d87955755..256626e34 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -100,11 +100,6 @@ def _get_parent_service_context(self): if not is_tracing_available(): return None - # TODO: Remove this block and delete class_decorators.py once Traceable is removed. - # Legacy: support for classes inheriting from Traceable (currently unused, deprecated). - if hasattr(self, "_span") and self._span: - return trace.set_span_in_context(self._span) - # Use the conversation context set by TurnTraceObserver via TracingContext. tracing_ctx = getattr(self, "_tracing_context", None) conversation_context = tracing_ctx.get_conversation_context() if tracing_ctx else None From f8267f1ea64523677d7c465b28df640c20594fb4 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 2 Apr 2026 10:47:44 -0400 Subject: [PATCH 03/11] Remove deprecated allow_interruptions parameter This field was deprecated in v0.0.99 in favor of LLMUserAggregator's user_turn_strategies / user_mute_strategies parameters. Since the default was True (interruptions allowed), removing the guards keeps the current default behavior. --- src/pipecat/frames/frames.py | 6 ----- src/pipecat/pipeline/task.py | 7 ----- .../aggregators/llm_response_universal.py | 2 +- src/pipecat/processors/frame_processor.py | 26 ------------------- src/pipecat/transports/base_input.py | 2 +- src/pipecat/transports/base_output.py | 3 --- src/pipecat/turns/user_turn_processor.py | 2 +- 7 files changed, 3 insertions(+), 45 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index bf5319405..f3e8a3ad9 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -878,11 +878,6 @@ class StartFrame(SystemFrame): Parameters: audio_in_sample_rate: Input audio sample rate in Hz. audio_out_sample_rate: Output audio sample rate in Hz. - allow_interruptions: Whether to allow user interruptions. - - .. deprecated:: 0.0.99 - Use `LLMUserAggregator`'s new `user_mute_strategies` parameter instead. - enable_metrics: Whether to enable performance metrics collection. enable_tracing: Whether to enable OpenTelemetry tracing. enable_usage_metrics: Whether to enable usage metrics collection. @@ -897,7 +892,6 @@ class StartFrame(SystemFrame): audio_in_sample_rate: int = 16000 audio_out_sample_rate: int = 24000 - allow_interruptions: bool = False enable_metrics: bool = False enable_tracing: bool = False enable_usage_metrics: bool = False diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index e5a6ad7ff..f3e7804f6 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -111,11 +111,6 @@ class PipelineParams(BaseModel): constructor arguments instead. Parameters: - allow_interruptions: Whether to allow pipeline interruptions. - - .. deprecated:: 0.0.99 - Use `LLMUserAggregator`'s new `user_turn_strategies` parameter instead. - audio_in_sample_rate: Input audio sample rate in Hz. audio_out_sample_rate: Output audio sample rate in Hz. enable_heartbeats: Whether to enable heartbeat monitoring. @@ -136,7 +131,6 @@ class PipelineParams(BaseModel): model_config = ConfigDict(arbitrary_types_allowed=True) - allow_interruptions: bool = True audio_in_sample_rate: int = 16000 audio_out_sample_rate: int = 24000 enable_heartbeats: bool = False @@ -778,7 +772,6 @@ class PipelineTask(BasePipelineTask): self._maybe_start_idle_task() start_frame = StartFrame( - allow_interruptions=self._params.allow_interruptions, audio_in_sample_rate=self._params.audio_in_sample_rate, audio_out_sample_rate=self._params.audio_out_sample_rate, enable_metrics=self._params.enable_metrics, diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 605db31f6..f8c2ffd43 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -731,7 +731,7 @@ class LLMUserAggregator(LLMContextAggregator): await self._user_idle_controller.process_frame(UserStartedSpeakingFrame()) - if params.enable_interruptions and self._allow_interruptions: + if params.enable_interruptions: await self.broadcast_interruption() await self._call_event_handler("on_user_turn_started", strategy) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 29b613527..3fe31ec01 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -193,8 +193,6 @@ class FrameProcessor(BaseObject): self._enable_metrics = False self._enable_usage_metrics = False self._report_only_initial_ttfb = False - # Other properties (deprecated) - self._allow_interruptions = False self._interruption_strategies: List[BaseInterruptionStrategy] = [] # Indicates whether we have received the StartFrame. @@ -307,29 +305,6 @@ class FrameProcessor(BaseObject): """ return self._prev - @property - def interruptions_allowed(self): - """Check if interruptions are allowed for this processor. - - .. deprecated:: 0.0.99 - Use `LLMUserAggregator`'s new `user_mute_strategies` parameter instead. - - Returns: - True if interruptions are allowed. - """ - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "`FrameProcessor.interruptions_allowed` is deprecated. " - "Use `LLMUserAggregator`'s new `user_mute_strategies` parameter instead.", - DeprecationWarning, - stacklevel=2, - ) - - return self._allow_interruptions - @property def metrics_enabled(self): """Check if metrics collection is enabled. @@ -819,7 +794,6 @@ class FrameProcessor(BaseObject): frame: The start frame containing initialization parameters. """ self.__started = True - self._allow_interruptions = frame.allow_interruptions self._enable_metrics = frame.enable_metrics self._enable_usage_metrics = frame.enable_usage_metrics self._interruption_strategies = frame.interruption_strategies diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index d4a1241f3..ff7026673 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -518,7 +518,7 @@ class BaseInputTransport(FrameProcessor): ) # Make sure we notify about interruptions quickly out-of-band. - if should_push_immediate_interruption and self._allow_interruptions: + if should_push_immediate_interruption: await self.broadcast_interruption() elif self.interruption_strategies and self._bot_speaking: logger.debug( diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 01af97be8..1d26d5e53 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -517,9 +517,6 @@ class BaseOutputTransport(FrameProcessor): Args: _: The start interruption frame (unused). """ - if not self._transport._allow_interruptions: - return - # Cancel tasks. await self._cancel_audio_task() await self._cancel_clock_task() diff --git a/src/pipecat/turns/user_turn_processor.py b/src/pipecat/turns/user_turn_processor.py index 85bc658dd..a3501d2c8 100644 --- a/src/pipecat/turns/user_turn_processor.py +++ b/src/pipecat/turns/user_turn_processor.py @@ -181,7 +181,7 @@ class UserTurnProcessor(FrameProcessor): await self._user_idle_controller.process_frame(UserStartedSpeakingFrame()) - if params.enable_interruptions and self._allow_interruptions: + if params.enable_interruptions: await self.broadcast_interruption() await self._call_event_handler("on_user_turn_started", strategy) From a5e1bbf4a3b09205c696b0332f7d3bb898fe3ef4 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 2 Apr 2026 10:50:05 -0400 Subject: [PATCH 04/11] Remove deprecated UserResponseAggregator class --- .../processors/aggregators/user_response.py | 64 ------------------- 1 file changed, 64 deletions(-) delete mode 100644 src/pipecat/processors/aggregators/user_response.py diff --git a/src/pipecat/processors/aggregators/user_response.py b/src/pipecat/processors/aggregators/user_response.py deleted file mode 100644 index 705e48853..000000000 --- a/src/pipecat/processors/aggregators/user_response.py +++ /dev/null @@ -1,64 +0,0 @@ -# -# Copyright (c) 2024-2026, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -"""User response aggregation for text frames. - -This module provides an aggregator that collects user responses and outputs -them as TextFrame objects, useful for capturing and processing user input -in conversational pipelines. -""" - -from pipecat.frames.frames import TextFrame -from pipecat.processors.aggregators.llm_context import LLMContext -from pipecat.processors.aggregators.llm_response_universal import LLMUserAggregator - - -class UserResponseAggregator(LLMUserAggregator): - """Aggregates user responses into TextFrame objects. - - This aggregator extends LLMUserAggregator to specifically handle - user input by collecting text responses and outputting them as TextFrame - objects when the aggregation is complete. - """ - - def __init__(self, **kwargs): - """Initialize the user response aggregator. - - .. deprecated:: 0.0.92 - `UserResponseAggregator` is deprecated and will be removed in a future version. - - Args: - **kwargs: Additional arguments passed to parent LLMUserAggregator. - """ - super().__init__(context=LLMContext(), **kwargs) - - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "`UserResponseAggregator` is deprecated and will be removed in a future version.", - DeprecationWarning, - stacklevel=2, - ) - - async def push_aggregation(self): - """Push the aggregated user response as a TextFrame. - - Creates a TextFrame from the current aggregation if it contains content, - resets the aggregation state, and pushes the frame downstream. - """ - if len(self._aggregation) > 0: - frame = TextFrame(self._aggregation.strip()) - - # 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._aggregation = "" - - await self.push_frame(frame) - - # Reset our accumulator state. - await self.reset() From 87e8ed109ad7ef8c55fe9ba21f12f57aef2928c9 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 2 Apr 2026 10:52:41 -0400 Subject: [PATCH 05/11] Remove deprecated STTMuteFilter, STTMuteConfig, and STTMuteStrategy --- .../processors/filters/stt_mute_filter.py | 243 ------------ src/pipecat/services/google/stt.py | 2 +- tests/test_stt_mute_filter.py | 354 ------------------ 3 files changed, 1 insertion(+), 598 deletions(-) delete mode 100644 src/pipecat/processors/filters/stt_mute_filter.py delete mode 100644 tests/test_stt_mute_filter.py diff --git a/src/pipecat/processors/filters/stt_mute_filter.py b/src/pipecat/processors/filters/stt_mute_filter.py deleted file mode 100644 index 9f522a20d..000000000 --- a/src/pipecat/processors/filters/stt_mute_filter.py +++ /dev/null @@ -1,243 +0,0 @@ -# -# Copyright (c) 2024-2026, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -"""Speech-to-text (STT) muting control module. - -This module provides functionality to control STT muting based on different strategies, -such as during function calls, bot speech, or custom conditions. It helps manage when -the STT service should be active or inactive during a conversation. -""" - -from dataclasses import dataclass -from enum import Enum -from typing import Awaitable, Callable, Optional - -from loguru import logger - -from pipecat.frames.frames import ( - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, - Frame, - FunctionCallCancelFrame, - FunctionCallResultFrame, - FunctionCallsStartedFrame, - InputAudioRawFrame, - InterimTranscriptionFrame, - InterruptionFrame, - StartFrame, - TranscriptionFrame, - UserStartedSpeakingFrame, - UserStoppedSpeakingFrame, - VADUserStartedSpeakingFrame, - VADUserStoppedSpeakingFrame, -) -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor - - -class STTMuteStrategy(Enum): - """Strategies determining when STT should be muted. - - Each strategy defines different conditions under which speech-to-text - processing should be temporarily disabled to prevent unwanted audio - processing during specific conversation states. - - Parameters: - FIRST_SPEECH: Mute STT until the first bot speech is detected. - MUTE_UNTIL_FIRST_BOT_COMPLETE: Mute STT until the first bot completes speaking, - regardless of whether it is the first speech. - FUNCTION_CALL: Mute STT during function calls to prevent interruptions. - ALWAYS: Always mute STT when the bot is speaking. - CUSTOM: Use a custom callback to determine muting logic dynamically. - - .. deprecated:: 0.0.99 - `STTMuteStrategy` is deprecated and will be removed in a future version. - Use `LLMUserAggregator`'s new `user_mute_strategies` instead. - """ - - FIRST_SPEECH = "first_speech" - MUTE_UNTIL_FIRST_BOT_COMPLETE = "mute_until_first_bot_complete" - FUNCTION_CALL = "function_call" - ALWAYS = "always" - CUSTOM = "custom" - - -@dataclass -class STTMuteConfig: - """Configuration for STT muting behavior. - - Defines which muting strategies to apply and provides optional custom - callback for advanced muting logic. Multiple strategies can be combined - to create sophisticated muting behavior. - - Parameters: - strategies: Set of muting strategies to apply simultaneously. - should_mute_callback: Optional callback for custom muting logic. - Only required when using STTMuteStrategy.CUSTOM. Called with - the STTMuteFilter instance to determine muting state. - - Note: - MUTE_UNTIL_FIRST_BOT_COMPLETE and FIRST_SPEECH strategies should not be used together - as they handle the first bot speech differently. - - .. deprecated:: 0.0.99 - `STTMuteConfig` is deprecated and will be removed in a future version. - Use `LLMUserAggregator`'s new `user_mute_strategies` instead. - """ - - strategies: set[STTMuteStrategy] - should_mute_callback: Optional[Callable[["STTMuteFilter"], Awaitable[bool]]] = None - - def __post_init__(self): - """Validate configuration after initialization. - - Raises: - ValueError: If incompatible strategies are used together. - """ - if ( - STTMuteStrategy.MUTE_UNTIL_FIRST_BOT_COMPLETE in self.strategies - and STTMuteStrategy.FIRST_SPEECH in self.strategies - ): - raise ValueError( - "MUTE_UNTIL_FIRST_BOT_COMPLETE and FIRST_SPEECH strategies should not be used together" - ) - - -class STTMuteFilter(FrameProcessor): - """A processor that handles STT muting and interruption control. - - This processor combines STT muting and interruption control as a coordinated - feature. When STT is muted, interruptions are automatically disabled by - suppressing VAD-related frames. This prevents unwanted speech detection - during bot speech, function calls, or other specified conditions. - - .. deprecated:: 0.0.99 - `STTMuteFilter` is deprecated and will be removed in a future version. - Use `LLMUserAggregator`'s new `user_mute_strategies` instead. - """ - - def __init__(self, *, config: STTMuteConfig, **kwargs): - """Initialize the STT mute filter. - - Args: - config: Configuration specifying muting strategies and behavior. - **kwargs: Additional arguments passed to parent class. - """ - super().__init__(**kwargs) - self._config = config - self._first_speech_handled = False - self._bot_is_speaking = False - self._function_call_in_progress = set() - self._is_muted = False - - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "`STTMuteFilter` is deprecated and will be removed in a future version. " - "Use `LLMUserAggregator`'s new `user_mute_strategies` instead.", - DeprecationWarning, - ) - - async def _handle_mute_state(self, should_mute: bool): - """Handle STT muting and interruption control state changes.""" - if should_mute != self._is_muted: - logger.debug(f"STTMuteFilter {'muting' if should_mute else 'unmuting'}") - self._is_muted = should_mute - # Note: We don't send STTMuteFrame to the STT service itself. - # The filter blocks frames locally, but the STT service continues - # processing audio to keep streaming connections alive (e.g., Google STT). - - async def _should_mute(self) -> bool: - """Determine if STT should be muted based on current state and strategies.""" - for strategy in self._config.strategies: - match strategy: - case STTMuteStrategy.FUNCTION_CALL: - if self._function_call_in_progress: - return True - - case STTMuteStrategy.ALWAYS: - if self._bot_is_speaking: - return True - - case STTMuteStrategy.FIRST_SPEECH: - if self._bot_is_speaking and not self._first_speech_handled: - self._first_speech_handled = True - return True - - case STTMuteStrategy.MUTE_UNTIL_FIRST_BOT_COMPLETE: - if not self._first_speech_handled: - return True - - case STTMuteStrategy.CUSTOM: - if self._bot_is_speaking and self._config.should_mute_callback: - should_mute = await self._config.should_mute_callback(self) - if should_mute: - return True - - return False - - async def process_frame(self, frame: Frame, direction: FrameDirection): - """Process incoming frames and manage muting state. - - Monitors conversation state through frame types and applies muting - strategies accordingly. Suppresses VAD-related frames when muted - while allowing other frames to pass through. - - Args: - frame: The incoming frame to process. - direction: The direction of frame flow in the pipeline. - """ - await super().process_frame(frame, direction) - - # Determine if we need to change mute state based on frame type - should_mute = None - - # Process frames to determine mute state - if isinstance(frame, StartFrame): - should_mute = await self._should_mute() - elif isinstance(frame, FunctionCallsStartedFrame): - for f in frame.function_calls: - self._function_call_in_progress.add(f.tool_call_id) - should_mute = await self._should_mute() - elif isinstance(frame, (FunctionCallCancelFrame, FunctionCallResultFrame)): - self._function_call_in_progress.remove(frame.tool_call_id) - should_mute = await self._should_mute() - elif isinstance(frame, BotStartedSpeakingFrame): - self._bot_is_speaking = True - should_mute = await self._should_mute() - elif isinstance(frame, BotStoppedSpeakingFrame): - self._bot_is_speaking = False - if not self._first_speech_handled: - self._first_speech_handled = True - should_mute = await self._should_mute() - - # Then push the original frame - if isinstance( - frame, - ( - InterruptionFrame, - VADUserStartedSpeakingFrame, - VADUserStoppedSpeakingFrame, - UserStartedSpeakingFrame, - UserStoppedSpeakingFrame, - InputAudioRawFrame, - InterimTranscriptionFrame, - TranscriptionFrame, - ), - ): - # Only pass VAD-related frames when not muted - if not self._is_muted: - await self.push_frame(frame, direction) - else: - logger.trace(f"{frame.__class__.__name__} suppressed - STT currently muted") - else: - # Pass all other frames through - await self.push_frame(frame, direction) - - # Finally handle mute state change if needed - if should_mute is not None and should_mute != self._is_muted: - await self._handle_mute_state(should_mute) diff --git a/src/pipecat/services/google/stt.py b/src/pipecat/services/google/stt.py index 9bf2685b7..282389a3a 100644 --- a/src/pipecat/services/google/stt.py +++ b/src/pipecat/services/google/stt.py @@ -1004,7 +1004,7 @@ class GoogleSTTService(STTService): except Aborted as e: # Handle stream abort due to inactivity (409 error). # This occurs when no audio is sent to the stream for 10+ seconds, - # which can happen when InputAudioRawFrames are blocked (e.g., by STTMuteFilter). + # which can happen when InputAudioRawFrames are blocked. # Google's STT service automatically closes the stream in this case. # We log at DEBUG level (not ERROR) since this is recoverable, then re-raise # to trigger automatic reconnection in _stream_audio. diff --git a/tests/test_stt_mute_filter.py b/tests/test_stt_mute_filter.py deleted file mode 100644 index 8f55bdecb..000000000 --- a/tests/test_stt_mute_filter.py +++ /dev/null @@ -1,354 +0,0 @@ -# -# Copyright (c) 2024-2026, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -import unittest - -from pipecat.frames.frames import ( - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, - FunctionCallFromLLM, - FunctionCallResultFrame, - FunctionCallsStartedFrame, - InputAudioRawFrame, - InterimTranscriptionFrame, - InterruptionFrame, - TranscriptionFrame, - UserStartedSpeakingFrame, - UserStoppedSpeakingFrame, - VADUserStartedSpeakingFrame, - VADUserStoppedSpeakingFrame, -) -from pipecat.processors.filters.stt_mute_filter import STTMuteConfig, STTMuteFilter, STTMuteStrategy -from pipecat.tests.utils import SleepFrame, run_test - - -class TestSTTMuteFilter(unittest.IsolatedAsyncioTestCase): - async def test_first_speech_strategy(self): - filter = STTMuteFilter(config=STTMuteConfig(strategies={STTMuteStrategy.FIRST_SPEECH})) - - frames_to_send = [ - BotStartedSpeakingFrame(), # First bot speech starts - VADUserStartedSpeakingFrame(), # Should be suppressed - UserStartedSpeakingFrame(), # Should be suppressed - InputAudioRawFrame( - audio=b"", sample_rate=16000, num_channels=1 - ), # Should be suppressed - VADUserStoppedSpeakingFrame(), # Should be suppressed - UserStoppedSpeakingFrame(), # Should be suppressed - BotStoppedSpeakingFrame(), # First bot speech ends - BotStartedSpeakingFrame(), # Second bot speech - VADUserStartedSpeakingFrame(), # Should pass through - UserStartedSpeakingFrame(), # Should pass through - InputAudioRawFrame(audio=b"", sample_rate=16000, num_channels=1), # Should pass through - VADUserStoppedSpeakingFrame(), # Should pass through - UserStoppedSpeakingFrame(), # Should pass through - BotStoppedSpeakingFrame(), - ] - - expected_returned_frames = [ - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, - BotStartedSpeakingFrame, - VADUserStartedSpeakingFrame, # Now passes through - UserStartedSpeakingFrame, # Now passes through - InputAudioRawFrame, # Now passes through - VADUserStoppedSpeakingFrame, # Now passes through - UserStoppedSpeakingFrame, # Now passes through - BotStoppedSpeakingFrame, - ] - - await run_test( - filter, - frames_to_send=frames_to_send, - expected_down_frames=expected_returned_frames, - ) - - async def test_always_strategy(self): - filter = STTMuteFilter(config=STTMuteConfig(strategies={STTMuteStrategy.ALWAYS})) - - frames_to_send = [ - BotStartedSpeakingFrame(), # First speech starts - VADUserStartedSpeakingFrame(), # Should be suppressed - UserStartedSpeakingFrame(), # Should be suppressed - InputAudioRawFrame( - audio=b"", sample_rate=16000, num_channels=1 - ), # Should be suppressed - VADUserStoppedSpeakingFrame(), # Should be suppressed - UserStoppedSpeakingFrame(), # Should be suppressed - BotStoppedSpeakingFrame(), # First speech ends - VADUserStartedSpeakingFrame(), # Should pass through - UserStartedSpeakingFrame(), # Should pass through - InputAudioRawFrame(audio=b"", sample_rate=16000, num_channels=1), # Should pass through - VADUserStoppedSpeakingFrame(), # Should pass through - UserStoppedSpeakingFrame(), # Should pass through - BotStartedSpeakingFrame(), # Second speech starts - VADUserStartedSpeakingFrame(), # Should be suppressed again - UserStartedSpeakingFrame(), # Should be suppressed again - InputAudioRawFrame( - audio=b"", sample_rate=16000, num_channels=1 - ), # Should be suppressed again - VADUserStoppedSpeakingFrame(), # Should be suppressed again - UserStoppedSpeakingFrame(), # Should be suppressed again - BotStoppedSpeakingFrame(), # Second speech ends - ] - - expected_returned_frames = [ - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, - VADUserStartedSpeakingFrame, - UserStartedSpeakingFrame, - InputAudioRawFrame, - VADUserStoppedSpeakingFrame, - UserStoppedSpeakingFrame, - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, - ] - - await run_test( - filter, - frames_to_send=frames_to_send, - expected_down_frames=expected_returned_frames, - ) - - async def test_transcription_frames_with_always_strategy(self): - filter = STTMuteFilter(config=STTMuteConfig(strategies={STTMuteStrategy.ALWAYS})) - - frames_to_send = [ - # Bot speaking - should mute - BotStartedSpeakingFrame(), - SleepFrame(), # Wait for StartedSpeaking to process - InterimTranscriptionFrame( - user_id="user1", text="This should be suppressed", timestamp="1234567890" - ), - TranscriptionFrame( - user_id="user1", text="This should be suppressed", timestamp="1234567890" - ), - SleepFrame(), # Wait for transcription frames to queue - BotStoppedSpeakingFrame(), - # Bot not speaking - should pass through - InterimTranscriptionFrame( - user_id="user1", text="This should pass", timestamp="1234567891" - ), - TranscriptionFrame( - user_id="user1", text="This should pass through", timestamp="1234567891" - ), - ] - - expected_returned_frames = [ - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, - InterimTranscriptionFrame, # Only passes through after bot stops speaking - TranscriptionFrame, # Only passes through after bot stops speaking - ] - - await run_test( - filter, - frames_to_send=frames_to_send, - expected_down_frames=expected_returned_frames, - ) - - async def test_function_call_strategy(self): - filter = STTMuteFilter(config=STTMuteConfig(strategies={STTMuteStrategy.FUNCTION_CALL})) - - frames_to_send = [ - VADUserStartedSpeakingFrame(), # Should pass through initially - UserStartedSpeakingFrame(), # Should pass through initially - VADUserStoppedSpeakingFrame(), # Should pass through initially - UserStoppedSpeakingFrame(), # Should pass through initially - FunctionCallsStartedFrame( - function_calls=[ - FunctionCallFromLLM( - function_name="get_weather", - tool_call_id="call_123", - arguments='{"location": "San Francisco"}', - context=None, - ) - ] - ), # Start function call - VADUserStartedSpeakingFrame(), # Should be suppressed - UserStartedSpeakingFrame(), # Should be suppressed - VADUserStoppedSpeakingFrame(), # Should be suppressed - UserStoppedSpeakingFrame(), # Should be suppressed - FunctionCallResultFrame( - function_name="get_weather", - tool_call_id="call_123", - arguments='{"location": "San Francisco"}', - result={"temperature": 22}, - ), # End function call - SleepFrame(), - VADUserStartedSpeakingFrame(), # Should pass through again - UserStartedSpeakingFrame(), # Should pass through again - VADUserStoppedSpeakingFrame(), - UserStoppedSpeakingFrame(), - ] - - expected_returned_frames = [ - VADUserStartedSpeakingFrame, - UserStartedSpeakingFrame, - VADUserStoppedSpeakingFrame, - UserStoppedSpeakingFrame, - FunctionCallsStartedFrame, - FunctionCallResultFrame, - VADUserStartedSpeakingFrame, - UserStartedSpeakingFrame, - VADUserStoppedSpeakingFrame, - UserStoppedSpeakingFrame, - ] - - await run_test( - filter, - frames_to_send=frames_to_send, - expected_down_frames=expected_returned_frames, - ) - - async def test_mute_until_first_bot_complete_strategy(self): - filter = STTMuteFilter( - config=STTMuteConfig(strategies={STTMuteStrategy.MUTE_UNTIL_FIRST_BOT_COMPLETE}) - ) - - frames_to_send = [ - VADUserStartedSpeakingFrame(), # Should be suppressed (starts muted) - UserStartedSpeakingFrame(), # Should be suppressed (starts muted) - InputAudioRawFrame( - audio=b"", sample_rate=16000, num_channels=1 - ), # Should be suppressed - VADUserStoppedSpeakingFrame(), # Should be suppressed - UserStoppedSpeakingFrame(), # Should be suppressed - BotStartedSpeakingFrame(), # First bot speech - VADUserStartedSpeakingFrame(), # Should be suppressed - UserStartedSpeakingFrame(), # Should be suppressed - InputAudioRawFrame( - audio=b"", sample_rate=16000, num_channels=1 - ), # Should be suppressed - VADUserStoppedSpeakingFrame(), # Should be suppressed - UserStoppedSpeakingFrame(), # Should be suppressed - BotStoppedSpeakingFrame(), # First speech ends, unmutes - VADUserStartedSpeakingFrame(), # Should pass through - UserStartedSpeakingFrame(), # Should pass through - InputAudioRawFrame(audio=b"", sample_rate=16000, num_channels=1), # Should pass through - VADUserStoppedSpeakingFrame(), # Should pass through - UserStoppedSpeakingFrame(), # Should pass through - BotStartedSpeakingFrame(), # Second speech - VADUserStartedSpeakingFrame(), # Should pass through - UserStartedSpeakingFrame(), # Should pass through - InputAudioRawFrame(audio=b"", sample_rate=16000, num_channels=1), # Should pass through - VADUserStoppedSpeakingFrame(), # Should pass through - UserStoppedSpeakingFrame(), # Should pass through - BotStoppedSpeakingFrame(), - ] - - expected_returned_frames = [ - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, - VADUserStartedSpeakingFrame, - UserStartedSpeakingFrame, - InputAudioRawFrame, - VADUserStoppedSpeakingFrame, - UserStoppedSpeakingFrame, - BotStartedSpeakingFrame, - VADUserStartedSpeakingFrame, - UserStartedSpeakingFrame, - InputAudioRawFrame, - VADUserStoppedSpeakingFrame, - UserStoppedSpeakingFrame, - BotStoppedSpeakingFrame, - ] - - await run_test( - filter, - frames_to_send=frames_to_send, - expected_down_frames=expected_returned_frames, - ) - - async def test_incompatible_strategies(self): - with self.assertRaises(ValueError): - STTMuteFilter( - config=STTMuteConfig( - strategies={ - STTMuteStrategy.FIRST_SPEECH, - STTMuteStrategy.MUTE_UNTIL_FIRST_BOT_COMPLETE, - } - ) - ) - - async def test_custom_strategy(self): - async def custom_mute_logic(processor: STTMuteFilter) -> bool: - return processor._bot_is_speaking - - filter = STTMuteFilter( - config=STTMuteConfig( - strategies={STTMuteStrategy.CUSTOM}, - should_mute_callback=custom_mute_logic, - ) - ) - - frames_to_send = [ - VADUserStartedSpeakingFrame(), # Should pass through - UserStartedSpeakingFrame(), # Should pass through - InputAudioRawFrame(audio=b"", sample_rate=16000, num_channels=1), # Should pass through - VADUserStoppedSpeakingFrame(), # Should pass through - UserStoppedSpeakingFrame(), # Should pass through - BotStartedSpeakingFrame(), # Bot starts speaking - VADUserStartedSpeakingFrame(), # Should be suppressed - UserStartedSpeakingFrame(), # Should be suppressed - InputAudioRawFrame( - audio=b"", sample_rate=16000, num_channels=1 - ), # Should be suppressed - VADUserStoppedSpeakingFrame(), # Should be suppressed - UserStoppedSpeakingFrame(), # Should be suppressed - BotStoppedSpeakingFrame(), # Bot stops speaking - VADUserStartedSpeakingFrame(), # Should pass through - UserStartedSpeakingFrame(), # Should pass through - InputAudioRawFrame(audio=b"", sample_rate=16000, num_channels=1), # Should pass through - VADUserStoppedSpeakingFrame(), # Should pass through - UserStoppedSpeakingFrame(), # Should pass through - ] - - expected_returned_frames = [ - VADUserStartedSpeakingFrame, - UserStartedSpeakingFrame, - InputAudioRawFrame, - VADUserStoppedSpeakingFrame, - UserStoppedSpeakingFrame, - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, - VADUserStartedSpeakingFrame, - UserStartedSpeakingFrame, - InputAudioRawFrame, - VADUserStoppedSpeakingFrame, - UserStoppedSpeakingFrame, - ] - - await run_test( - filter, - frames_to_send=frames_to_send, - expected_down_frames=expected_returned_frames, - ) - - async def test_interruption_frame_suppressed_when_muted(self): - """Test that InterruptionFrame is suppressed when the filter is muted.""" - filter = STTMuteFilter(config=STTMuteConfig(strategies={STTMuteStrategy.ALWAYS})) - - frames_to_send = [ - BotStartedSpeakingFrame(), - InterruptionFrame(), - BotStoppedSpeakingFrame(), - ] - - expected_returned_frames = [ - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, - ] - - await run_test( - filter, - frames_to_send=frames_to_send, - expected_down_frames=expected_returned_frames, - ) - - -if __name__ == "__main__": - unittest.main() From 2a118084bda1b0bcc8c63a8bb47167545ba677b7 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 2 Apr 2026 10:54:13 -0400 Subject: [PATCH 06/11] Remove deprecated transcript_processor module --- .../processors/transcript_processor.py | 370 -------- tests/test_transcript_processor.py | 798 ------------------ 2 files changed, 1168 deletions(-) delete mode 100644 src/pipecat/processors/transcript_processor.py delete mode 100644 tests/test_transcript_processor.py diff --git a/src/pipecat/processors/transcript_processor.py b/src/pipecat/processors/transcript_processor.py deleted file mode 100644 index 2c9fe2917..000000000 --- a/src/pipecat/processors/transcript_processor.py +++ /dev/null @@ -1,370 +0,0 @@ -# -# Copyright (c) 2024-2026, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -"""Transcript processing utilities for conversation recording and analysis. - -This module provides processors that convert speech and text frames into structured -transcript messages with timestamps, enabling conversation history tracking and analysis. -""" - -from typing import List, Optional - -from loguru import logger - -from pipecat.frames.frames import ( - BotStoppedSpeakingFrame, - CancelFrame, - EndFrame, - Frame, - InterruptionFrame, - LLMThoughtEndFrame, - LLMThoughtStartFrame, - LLMThoughtTextFrame, - ThoughtTranscriptionMessage, - TranscriptionFrame, - TranscriptionMessage, - TranscriptionUpdateFrame, - TTSTextFrame, -) -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.utils.string import TextPartForConcatenation, concatenate_aggregated_text -from pipecat.utils.time import time_now_iso8601 - - -class BaseTranscriptProcessor(FrameProcessor): - """Base class for processing conversation transcripts. - - Provides common functionality for handling transcript messages and updates. - """ - - def __init__(self, **kwargs): - """Initialize processor with empty message store. - - Args: - **kwargs: Additional arguments passed to parent class. - """ - super().__init__(**kwargs) - self._processed_messages: List[TranscriptionMessage] = [] - self._register_event_handler("on_transcript_update") - - async def _emit_update(self, messages: List[TranscriptionMessage]): - """Emit transcript updates for new messages. - - Args: - messages: New messages to emit in update. - """ - if messages: - self._processed_messages.extend(messages) - update_frame = TranscriptionUpdateFrame(messages=messages) - await self._call_event_handler("on_transcript_update", update_frame) - await self.push_frame(update_frame) - - -class UserTranscriptProcessor(BaseTranscriptProcessor): - """Processes user transcription frames into timestamped conversation messages.""" - - async def process_frame(self, frame: Frame, direction: FrameDirection): - """Process TranscriptionFrames into user conversation messages. - - Args: - frame: Input frame to process. - direction: Frame processing direction. - """ - await super().process_frame(frame, direction) - - if isinstance(frame, TranscriptionFrame): - message = TranscriptionMessage( - role="user", user_id=frame.user_id, content=frame.text, timestamp=frame.timestamp - ) - await self._emit_update([message]) - - await self.push_frame(frame, direction) - - -class AssistantTranscriptProcessor(BaseTranscriptProcessor): - """Processes assistant TTS text frames and LLM thought frames into timestamped messages. - - This processor aggregates both TTS text frames and LLM thought frames into - complete utterances and thoughts, emitting them as transcript messages. - - An assistant utterance is completed when: - - The bot stops speaking (BotStoppedSpeakingFrame) - - The bot is interrupted (InterruptionFrame) - - The pipeline ends (EndFrame, CancelFrame) - - A thought is completed when: - - The thought ends (LLMThoughtEndFrame) - - The bot is interrupted (InterruptionFrame) - - The pipeline ends (EndFrame, CancelFrame) - """ - - def __init__(self, *, process_thoughts: bool = False, **kwargs): - """Initialize processor with aggregation state. - - Args: - process_thoughts: Whether to process LLM thought frames. Defaults to False. - **kwargs: Additional arguments passed to parent class. - """ - super().__init__(**kwargs) - - self._process_thoughts = process_thoughts - self._current_assistant_text_parts: List[TextPartForConcatenation] = [] - self._assistant_text_start_time: Optional[str] = None - - self._current_thought_parts: List[TextPartForConcatenation] = [] - self._thought_start_time: Optional[str] = None - self._thought_active = False - - async def _emit_aggregated_assistant_text(self): - """Aggregates and emits text fragments as a transcript message. - - This method aggregates text fragments that may arrive in multiple - TTSTextFrame instances and emits them as a single TranscriptionMessage. - """ - if self._current_assistant_text_parts and self._assistant_text_start_time: - content = concatenate_aggregated_text(self._current_assistant_text_parts) - if content: - logger.trace(f"Emitting aggregated assistant message: {content}") - message = TranscriptionMessage( - role="assistant", - content=content, - timestamp=self._assistant_text_start_time, - ) - await self._emit_update([message]) - else: - logger.trace("No content to emit after stripping whitespace") - - # Reset aggregation state - self._current_assistant_text_parts = [] - self._assistant_text_start_time = None - - async def _emit_aggregated_thought(self): - """Aggregates and emits thought text fragments as a thought transcript message. - - This method aggregates thought fragments that may arrive in multiple - LLMThoughtTextFrame instances and emits them as a single ThoughtTranscriptionMessage. - """ - if self._current_thought_parts and self._thought_start_time: - content = concatenate_aggregated_text(self._current_thought_parts) - if content: - logger.trace(f"Emitting aggregated thought message: {content}") - message = ThoughtTranscriptionMessage( - content=content, - timestamp=self._thought_start_time, - ) - await self._emit_update([message]) - else: - logger.trace("No thought content to emit after stripping whitespace") - - # Reset aggregation state - self._current_thought_parts = [] - self._thought_start_time = None - self._thought_active = False - - async def process_frame(self, frame: Frame, direction: FrameDirection): - """Process frames into assistant conversation messages and thought messages. - - Handles different frame types: - - - TTSTextFrame: Aggregates text for current utterance - - LLMThoughtStartFrame: Begins aggregating a new thought - - LLMThoughtTextFrame: Aggregates text for current thought - - LLMThoughtEndFrame: Completes current thought - - BotStoppedSpeakingFrame: Completes current utterance - - InterruptionFrame: Completes current utterance and thought due to interruption - - EndFrame: Completes current utterance and thought at pipeline end - - CancelFrame: Completes current utterance and thought due to cancellation - - Args: - frame: Input frame to process. - direction: Frame processing direction. - """ - await super().process_frame(frame, direction) - - if isinstance(frame, (InterruptionFrame, CancelFrame)): - # Push frame first otherwise our emitted transcription update frame - # might get cleaned up. - await self.push_frame(frame, direction) - # Emit accumulated text and thought with interruptions - await self._emit_aggregated_assistant_text() - if self._process_thoughts and self._thought_active: - await self._emit_aggregated_thought() - elif isinstance(frame, LLMThoughtStartFrame): - # Start a new thought - if self._process_thoughts: - self._thought_active = True - self._thought_start_time = time_now_iso8601() - self._current_thought_parts = [] - # Push frame. - await self.push_frame(frame, direction) - elif isinstance(frame, LLMThoughtTextFrame): - # Aggregate thought text if we have an active thought - if self._process_thoughts and self._thought_active: - self._current_thought_parts.append( - TextPartForConcatenation( - frame.text, includes_inter_part_spaces=frame.includes_inter_frame_spaces - ) - ) - # Push frame. - await self.push_frame(frame, direction) - elif isinstance(frame, LLMThoughtEndFrame): - # Emit accumulated thought when thought ends - if self._process_thoughts and self._thought_active: - await self._emit_aggregated_thought() - # Push frame. - await self.push_frame(frame, direction) - elif isinstance(frame, TTSTextFrame): - # Start timestamp on first text part - if not self._assistant_text_start_time: - self._assistant_text_start_time = time_now_iso8601() - - self._current_assistant_text_parts.append( - TextPartForConcatenation( - frame.text, includes_inter_part_spaces=frame.includes_inter_frame_spaces - ) - ) - - # Push frame. - await self.push_frame(frame, direction) - elif isinstance(frame, (BotStoppedSpeakingFrame, EndFrame)): - # Emit accumulated text when bot finishes speaking or pipeline ends. - await self._emit_aggregated_assistant_text() - # Emit accumulated thought at pipeline end if still active - if isinstance(frame, EndFrame) and self._process_thoughts and self._thought_active: - await self._emit_aggregated_thought() - # Push frame. - await self.push_frame(frame, direction) - else: - await self.push_frame(frame, direction) - - -class TranscriptProcessor: - """Factory for creating and managing transcript processors. - - Provides unified access to user and assistant transcript processors - with shared event handling. The assistant processor handles both TTS text - and LLM thought frames. - - Example:: - - transcript = TranscriptProcessor() - - pipeline = Pipeline( - [ - transport.input(), - stt, - transcript.user(), # User transcripts - context_aggregator.user(), - llm, - tts, - transport.output(), - transcript.assistant(), # Assistant transcripts (including thoughts) - context_aggregator.assistant(), - ] - ) - - @transcript.event_handler("on_transcript_update") - async def handle_update(processor, frame): - print(f"New messages: {frame.messages}") - - .. deprecated:: 0.0.99 - `TranscriptProcessor` is deprecated and will be removed in a future version. - Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events instead. - """ - - def __init__(self, *, process_thoughts: bool = False): - """Initialize factory. - - Args: - process_thoughts: Whether the assistant processor should handle LLM thought - frames. Defaults to False. - """ - self._process_thoughts = process_thoughts - self._user_processor = None - self._assistant_processor = None - self._event_handlers = {} - - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "`TranscriptProcessor` is deprecated and will be removed in a future version. " - "Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events instead.", - DeprecationWarning, - ) - - def user(self, **kwargs) -> UserTranscriptProcessor: - """Get the user transcript processor. - - Args: - **kwargs: Arguments specific to UserTranscriptProcessor. - - Returns: - The user transcript processor instance. - """ - if self._user_processor is None: - self._user_processor = UserTranscriptProcessor(**kwargs) - # Apply any registered event handlers - for event_name, handler in self._event_handlers.items(): - - @self._user_processor.event_handler(event_name) - async def user_handler(processor, frame): - return await handler(processor, frame) - - return self._user_processor - - def assistant(self, **kwargs) -> AssistantTranscriptProcessor: - """Get the assistant transcript processor. - - Args: - **kwargs: Arguments specific to AssistantTranscriptProcessor. - - Returns: - The assistant transcript processor instance. - """ - if self._assistant_processor is None: - self._assistant_processor = AssistantTranscriptProcessor( - process_thoughts=self._process_thoughts, **kwargs - ) - # Apply any registered event handlers - for event_name, handler in self._event_handlers.items(): - - @self._assistant_processor.event_handler(event_name) - async def assistant_handler(processor, frame): - return await handler(processor, frame) - - return self._assistant_processor - - def event_handler(self, event_name: str): - """Register event handler for both processors. - - Args: - event_name: Name of event to handle. - - Returns: - Decorator function that registers handler with both processors. - """ - - def decorator(handler): - self._event_handlers[event_name] = handler - - # Apply handler to existing processors if they exist - if self._user_processor: - - @self._user_processor.event_handler(event_name) - async def user_handler(processor, frame): - return await handler(processor, frame) - - if self._assistant_processor: - - @self._assistant_processor.event_handler(event_name) - async def assistant_handler(processor, frame): - return await handler(processor, frame) - - return handler - - return decorator diff --git a/tests/test_transcript_processor.py b/tests/test_transcript_processor.py deleted file mode 100644 index 1dfdd58a3..000000000 --- a/tests/test_transcript_processor.py +++ /dev/null @@ -1,798 +0,0 @@ -# -# Copyright (c) 2024-2026, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - - -import asyncio -import unittest -from datetime import datetime, timezone -from typing import List, Tuple, cast - -from pipecat.frames.frames import ( - AggregationType, - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, - CancelFrame, - InterruptionFrame, - LLMThoughtEndFrame, - LLMThoughtStartFrame, - LLMThoughtTextFrame, - ThoughtTranscriptionMessage, - TranscriptionFrame, - TranscriptionMessage, - TranscriptionUpdateFrame, - TTSTextFrame, -) -from pipecat.processors.transcript_processor import ( - AssistantTranscriptProcessor, - UserTranscriptProcessor, -) -from pipecat.tests.utils import SleepFrame, run_test - - -class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase): - """Tests for UserTranscriptProcessor""" - - async def test_basic_transcription(self): - """Test basic transcription frame processing""" - # Create processor - processor = UserTranscriptProcessor() - - # Create test timestamp - timestamp = datetime.now(timezone.utc).isoformat() - - # Create frames to send - frames_to_send = [ - TranscriptionFrame(text="Hello, world!", user_id="test_user", timestamp=timestamp) - ] - - # Expected frames downstream - note the order: - # 1. TranscriptionUpdateFrame (processor emits the update first) - # 2. TranscriptionFrame (original frame is passed through) - expected_down_frames = [TranscriptionUpdateFrame, TranscriptionFrame] - - # Run test - received_frames, _ = await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Verify the content of the TranscriptionUpdateFrame - update_frame = cast( - TranscriptionUpdateFrame, received_frames[0] - ) # Note: now checking first frame - self.assertIsInstance(update_frame, TranscriptionUpdateFrame) - self.assertEqual(len(update_frame.messages), 1) - message = update_frame.messages[0] - self.assertEqual(message.role, "user") - self.assertEqual(message.content, "Hello, world!") - self.assertEqual(message.user_id, "test_user") - self.assertEqual(message.timestamp, timestamp) - - async def test_event_handler(self): - """Test that event handlers are called with transcript updates""" - # Create processor - processor = UserTranscriptProcessor() - - # Track received updates - received_updates: List[TranscriptionMessage] = [] - - # Register event handler - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.extend(frame.messages) - - # Create test data - timestamp = datetime.now(timezone.utc).isoformat() - frames_to_send = [ - TranscriptionFrame(text="First message", user_id="test_user", timestamp=timestamp), - TranscriptionFrame(text="Second message", user_id="test_user", timestamp=timestamp), - ] - - expected_down_frames = [ - TranscriptionUpdateFrame, - TranscriptionFrame, # First message - TranscriptionUpdateFrame, - TranscriptionFrame, # Second message - ] - - # Run test - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Verify event handler received updates - self.assertEqual(len(received_updates), 2) - - # Check first message - self.assertEqual(received_updates[0].role, "user") - self.assertEqual(received_updates[0].content, "First message") - self.assertEqual(received_updates[0].timestamp, timestamp) - - # Check second message - self.assertEqual(received_updates[1].role, "user") - self.assertEqual(received_updates[1].content, "Second message") - self.assertEqual(received_updates[1].timestamp, timestamp) - - async def test_text_aggregation(self): - """Test that TTSTextFrames are properly aggregated into a single message""" - # Create processor - processor = AssistantTranscriptProcessor() - - # Track received updates - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - # Create test frames simulating bot speaking multiple text chunks - frames_to_send = [ - BotStartedSpeakingFrame(), - SleepFrame(), # Wait for StartedSpeaking to process - TTSTextFrame(text="Hello", aggregated_by=AggregationType.WORD), - TTSTextFrame(text="world!", aggregated_by=AggregationType.WORD), - TTSTextFrame(text="How", aggregated_by=AggregationType.WORD), - TTSTextFrame(text="are", aggregated_by=AggregationType.WORD), - TTSTextFrame(text="you?", aggregated_by=AggregationType.WORD), - SleepFrame(), # Wait for text frames to queue - BotStoppedSpeakingFrame(), - ] - - # Expected order: - # 1. BotStartedSpeakingFrame (system frame, immediate) - # 2. All queued TTSTextFrames - # 3. BotStoppedSpeakingFrame (system frame, immediate) - # 4. TranscriptionUpdateFrame (after aggregation) - expected_down_frames = [ - BotStartedSpeakingFrame, - TTSTextFrame, - TTSTextFrame, - TTSTextFrame, - TTSTextFrame, - TTSTextFrame, - TranscriptionUpdateFrame, - BotStoppedSpeakingFrame, - ] - - # Run test - received_frames, _ = await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Verify update was received - self.assertEqual(len(received_updates), 1) - - # Get the update frame - update_frame = received_updates[0] - - # Should have one aggregated message - self.assertEqual(len(update_frame.messages), 1) - - message = update_frame.messages[0] - self.assertEqual(message.role, "assistant") - self.assertEqual(message.content, "Hello world! How are you?") - - # Verify timestamp exists - self.assertIsNotNone(message.timestamp) - - # All frames should be passed through in order, with update at end - downstream_update = cast(TranscriptionUpdateFrame, received_frames[-2]) - self.assertEqual(downstream_update.messages[0].content, "Hello world! How are you?") - - async def test_empty_text_handling(self): - """Test that empty messages are not emitted""" - processor = AssistantTranscriptProcessor() - - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - frames_to_send = [ - BotStartedSpeakingFrame(), - SleepFrame(), - TTSTextFrame(text="", aggregated_by=AggregationType.WORD), # Empty text - TTSTextFrame(text=" ", aggregated_by=AggregationType.WORD), # Just whitespace - TTSTextFrame(text="\n", aggregated_by=AggregationType.WORD), # Just newline - BotStoppedSpeakingFrame(), - # Pipeline ends here; run_test will automatically send EndFrame - ] - - # From our earlier tests, we know BotStoppedSpeakingFrame comes before TTSTextFrames - expected_down_frames = [ - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, - TTSTextFrame, # empty - TTSTextFrame, # whitespace - TTSTextFrame, # newline - # No TranscriptionUpdateFrame since content is empty after stripping - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - self.assertEqual(len(received_updates), 0, "No updates should be emitted for empty content") - - async def test_interruption_handling(self): - """Test that messages are properly captured when bot is interrupted""" - processor = AssistantTranscriptProcessor() - - # Track received updates - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - # Simulate bot being interrupted mid-sentence - frames_to_send = [ - BotStartedSpeakingFrame(), - SleepFrame(), - TTSTextFrame(text="Hello", aggregated_by=AggregationType.WORD), - TTSTextFrame(text="world!", aggregated_by=AggregationType.WORD), - SleepFrame(), - InterruptionFrame(), # User interrupts here - SleepFrame(), - BotStartedSpeakingFrame(), - TTSTextFrame(text="New", aggregated_by=AggregationType.WORD), - TTSTextFrame(text="response", aggregated_by=AggregationType.WORD), - SleepFrame(), - BotStoppedSpeakingFrame(), - ] - - # Actual order of frames: - expected_down_frames = [ - BotStartedSpeakingFrame, - TTSTextFrame, # "Hello" - TTSTextFrame, # "world!" - InterruptionFrame, - TranscriptionUpdateFrame, # First message (emitted due to interruption) - BotStartedSpeakingFrame, - TTSTextFrame, # "New" - TTSTextFrame, # "response" - TranscriptionUpdateFrame, # Second message - BotStoppedSpeakingFrame, - ] - - # Run test - received_frames, _ = await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Should have received two updates - self.assertEqual(len(received_updates), 2) - - # First update should be interrupted message - first_message = received_updates[0].messages[0] - self.assertEqual(first_message.role, "assistant") - self.assertEqual(first_message.content, "Hello world!") - self.assertIsNotNone(first_message.timestamp) - - # Second update should be new response - second_message = received_updates[1].messages[0] - self.assertEqual(second_message.role, "assistant") - self.assertEqual(second_message.content, "New response") - self.assertIsNotNone(second_message.timestamp) - - # Verify timestamps are different - self.assertNotEqual(first_message.timestamp, second_message.timestamp) - - async def test_end_frame_handling(self): - """Test that final messages are captured when pipeline ends normally""" - processor = AssistantTranscriptProcessor() - - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - frames_to_send = [ - BotStartedSpeakingFrame(), - SleepFrame(), - TTSTextFrame(text="Hello", aggregated_by=AggregationType.WORD), - TTSTextFrame(text="world", aggregated_by=AggregationType.WORD), - # Pipeline ends here; run_test will automatically send EndFrame - ] - - expected_down_frames = [ - BotStartedSpeakingFrame, - TTSTextFrame, - TTSTextFrame, - TranscriptionUpdateFrame, # Final message emitted due to EndFrame - ] - - # Run test - EndFrame will be sent automatically - received_frames, _ = await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - self.assertEqual(len(received_updates), 1) - message = received_updates[0].messages[0] - self.assertEqual(message.role, "assistant") - self.assertEqual(message.content, "Hello world") - - async def test_cancel_frame_handling(self): - """Test that messages are properly captured when pipeline is cancelled""" - processor = AssistantTranscriptProcessor() - - # Track updates with timestamps to verify order - received_updates: List[Tuple[str, float]] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - # Record message content and time received - received_updates.append((frame.messages[0].content, asyncio.get_event_loop().time())) - - frames_to_send = [ - BotStartedSpeakingFrame(), - SleepFrame(), - TTSTextFrame(text="Hello", aggregated_by=AggregationType.WORD), - TTSTextFrame(text="world", aggregated_by=AggregationType.WORD), - SleepFrame(), # Ensure messages are processed - CancelFrame(), - ] - - # We don't need to verify frame order, just that CancelFrame triggers message emission - expected_down_frames = [ - BotStartedSpeakingFrame, - TTSTextFrame, - TTSTextFrame, - CancelFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - send_end_frame=False, - ) - - # Verify that we received an update - self.assertEqual(len(received_updates), 1, "Should receive one update before cancellation") - content, _ = received_updates[0] - self.assertEqual(content, "Hello world") - - async def test_transcript_processor_factory(self): - """Test that factory properly manages processors and event handlers""" - from pipecat.processors.transcript_processor import TranscriptProcessor - - factory = TranscriptProcessor() - received_updates: List[TranscriptionMessage] = [] - - # Register handler with factory - @factory.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.extend(frame.messages) - - # Get processors and verify they're reused - user_proc1 = factory.user() - user_proc2 = factory.user() - self.assertIs(user_proc1, user_proc2, "User processor should be reused") - - asst_proc1 = factory.assistant() - asst_proc2 = factory.assistant() - self.assertIs(asst_proc1, asst_proc2, "Assistant processor should be reused") - - # Test user processor - timestamp = datetime.now(timezone.utc).isoformat() - frames_to_send = [ - TranscriptionFrame(text="User message", user_id="user1", timestamp=timestamp) - ] - - await run_test( - user_proc1, - frames_to_send=frames_to_send, - expected_down_frames=[TranscriptionUpdateFrame, TranscriptionFrame], - ) - - # Test assistant processor - frames_to_send = [ - BotStartedSpeakingFrame(), - SleepFrame(), - TTSTextFrame(text="Assistant", aggregated_by=AggregationType.WORD), - TTSTextFrame(text="message", aggregated_by=AggregationType.WORD), - BotStoppedSpeakingFrame(), - ] - - # The actual order we see in the output: - await run_test( - asst_proc1, - frames_to_send=frames_to_send, - expected_down_frames=[ - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, - TTSTextFrame, - TTSTextFrame, - TranscriptionUpdateFrame, - ], - ) - - # Verify both processors triggered the same handler - self.assertEqual(len(received_updates), 2) - self.assertEqual(received_updates[0].role, "user") - self.assertEqual(received_updates[0].content, "User message") - self.assertEqual(received_updates[1].role, "assistant") - self.assertEqual(received_updates[1].content, "Assistant message") - - async def test_text_fragments_with_spaces(self): - """Test aggregating text fragments with various spacing patterns""" - processor = AssistantTranscriptProcessor() - - # Track received updates - received_updates = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - # Test the specific pattern shared - def make_tts_text_frame(text: str) -> TTSTextFrame: - frame = TTSTextFrame(text=text, aggregated_by=AggregationType.WORD) - frame.includes_inter_frame_spaces = True - return frame - - frames_to_send = [ - BotStartedSpeakingFrame(), - SleepFrame(), - make_tts_text_frame("Hello"), - make_tts_text_frame(" there"), - make_tts_text_frame("!"), - make_tts_text_frame(" How"), - make_tts_text_frame("'s"), - make_tts_text_frame(" it"), - make_tts_text_frame(" going"), - make_tts_text_frame("?"), - BotStoppedSpeakingFrame(), - ] - - expected_down_frames = [ - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, - TTSTextFrame, - TTSTextFrame, - TTSTextFrame, - TTSTextFrame, - TTSTextFrame, - TTSTextFrame, - TTSTextFrame, - TTSTextFrame, - TranscriptionUpdateFrame, - ] - - # Run test - received_frames, _ = await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Verify result - self.assertEqual(len(received_updates), 1) - message = received_updates[0].messages[0] - self.assertEqual(message.role, "assistant") - # Should be properly joined without extra spaces - self.assertEqual(message.content, "Hello there! How's it going?") - - -class TestThoughtTranscription(unittest.IsolatedAsyncioTestCase): - """Tests for thought transcription in AssistantTranscriptProcessor""" - - async def test_basic_thought_transcription(self): - """Test basic thought frame processing""" - processor = AssistantTranscriptProcessor(process_thoughts=True) - - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - # Create frames for a simple thought - frames_to_send = [ - LLMThoughtStartFrame(), - LLMThoughtTextFrame(text="Let me think about this..."), - LLMThoughtEndFrame(), - ] - - expected_down_frames = [ - LLMThoughtStartFrame, - LLMThoughtTextFrame, - TranscriptionUpdateFrame, - LLMThoughtEndFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Verify update was received - self.assertEqual(len(received_updates), 1) - message = received_updates[0].messages[0] - self.assertIsInstance(message, ThoughtTranscriptionMessage) - self.assertEqual(message.content, "Let me think about this...") - self.assertIsNotNone(message.timestamp) - - async def test_thought_aggregation(self): - """Test that thought text frames are properly aggregated""" - processor = AssistantTranscriptProcessor(process_thoughts=True) - - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - # Create frames simulating chunked thought text - frames_to_send = [ - LLMThoughtStartFrame(), - LLMThoughtTextFrame(text="The user "), - LLMThoughtTextFrame(text="is asking "), - LLMThoughtTextFrame(text="about electric "), - LLMThoughtTextFrame(text="cars."), - LLMThoughtEndFrame(), - ] - - expected_down_frames = [ - LLMThoughtStartFrame, - LLMThoughtTextFrame, - LLMThoughtTextFrame, - LLMThoughtTextFrame, - LLMThoughtTextFrame, - TranscriptionUpdateFrame, - LLMThoughtEndFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Verify aggregation - self.assertEqual(len(received_updates), 1) - message = received_updates[0].messages[0] - self.assertIsInstance(message, ThoughtTranscriptionMessage) - self.assertEqual(message.content, "The user is asking about electric cars.") - - async def test_thought_with_interruption(self): - """Test that thoughts are properly captured when interrupted""" - processor = AssistantTranscriptProcessor(process_thoughts=True) - - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - frames_to_send = [ - LLMThoughtStartFrame(), - LLMThoughtTextFrame(text="I need to consider "), - LLMThoughtTextFrame(text="multiple factors"), - SleepFrame(), - InterruptionFrame(), # User interrupts - ] - - expected_down_frames = [ - LLMThoughtStartFrame, - LLMThoughtTextFrame, - LLMThoughtTextFrame, - InterruptionFrame, - TranscriptionUpdateFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Verify thought was captured on interruption - self.assertEqual(len(received_updates), 1) - message = received_updates[0].messages[0] - self.assertIsInstance(message, ThoughtTranscriptionMessage) - self.assertEqual(message.content, "I need to consider multiple factors") - - async def test_thought_with_cancel(self): - """Test that thoughts are properly captured when cancelled""" - processor = AssistantTranscriptProcessor(process_thoughts=True) - - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - frames_to_send = [ - LLMThoughtStartFrame(), - LLMThoughtTextFrame(text="Starting analysis"), - SleepFrame(), - CancelFrame(), - ] - - expected_down_frames = [ - LLMThoughtStartFrame, - LLMThoughtTextFrame, - CancelFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - send_end_frame=False, - ) - - # Verify thought was captured on cancellation - self.assertEqual(len(received_updates), 1) - message = received_updates[0].messages[0] - self.assertIsInstance(message, ThoughtTranscriptionMessage) - self.assertEqual(message.content, "Starting analysis") - - async def test_thought_with_end_frame(self): - """Test that thoughts are captured when pipeline ends normally""" - processor = AssistantTranscriptProcessor(process_thoughts=True) - - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - frames_to_send = [ - LLMThoughtStartFrame(), - LLMThoughtTextFrame(text="Final thought"), - # Pipeline ends here; run_test will automatically send EndFrame - ] - - expected_down_frames = [ - LLMThoughtStartFrame, - LLMThoughtTextFrame, - TranscriptionUpdateFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Verify thought was captured on EndFrame - self.assertEqual(len(received_updates), 1) - message = received_updates[0].messages[0] - self.assertIsInstance(message, ThoughtTranscriptionMessage) - self.assertEqual(message.content, "Final thought") - - async def test_multiple_thoughts(self): - """Test multiple separate thoughts in sequence""" - processor = AssistantTranscriptProcessor(process_thoughts=True) - - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - frames_to_send = [ - # First thought - LLMThoughtStartFrame(), - LLMThoughtTextFrame(text="First consideration"), - LLMThoughtEndFrame(), - # Second thought - LLMThoughtStartFrame(), - LLMThoughtTextFrame(text="Second consideration"), - LLMThoughtEndFrame(), - ] - - expected_down_frames = [ - LLMThoughtStartFrame, - LLMThoughtTextFrame, - TranscriptionUpdateFrame, - LLMThoughtEndFrame, - LLMThoughtStartFrame, - LLMThoughtTextFrame, - TranscriptionUpdateFrame, - LLMThoughtEndFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Verify both thoughts were captured - self.assertEqual(len(received_updates), 2) - - first_message = received_updates[0].messages[0] - self.assertIsInstance(first_message, ThoughtTranscriptionMessage) - self.assertEqual(first_message.content, "First consideration") - - second_message = received_updates[1].messages[0] - self.assertIsInstance(second_message, ThoughtTranscriptionMessage) - self.assertEqual(second_message.content, "Second consideration") - - async def test_empty_thought_handling(self): - """Test that empty thoughts are not emitted""" - processor = AssistantTranscriptProcessor(process_thoughts=True) - - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - frames_to_send = [ - LLMThoughtStartFrame(), - LLMThoughtTextFrame(text=""), # Empty - LLMThoughtTextFrame(text=" "), # Just whitespace - LLMThoughtEndFrame(), - ] - - expected_down_frames = [ - LLMThoughtStartFrame, - LLMThoughtTextFrame, - LLMThoughtTextFrame, - LLMThoughtEndFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Verify no updates emitted for empty content - self.assertEqual(len(received_updates), 0) - - async def test_thought_without_start_frame(self): - """Test that thought text without start frame is ignored""" - processor = AssistantTranscriptProcessor(process_thoughts=True) - - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - # Send thought text without start frame - frames_to_send = [ - LLMThoughtTextFrame(text="This should be ignored"), - LLMThoughtEndFrame(), - ] - - expected_down_frames = [ - LLMThoughtTextFrame, - LLMThoughtEndFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Verify no updates since thought wasn't properly started - self.assertEqual(len(received_updates), 0) - - -if __name__ == "__main__": - unittest.main() From fa30268b84147a9a0a33c924b14463e64228c975 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 2 Apr 2026 11:03:23 -0400 Subject: [PATCH 07/11] Remove deprecated TranscriptionMessage, ThoughtTranscriptionMessage, and TranscriptionUpdateFrame --- src/pipecat/frames/frames.py | 131 ----------------------------------- 1 file changed, 131 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index f3e8a3ad9..41b930ee5 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -462,137 +462,6 @@ class LLMContextAssistantTimestampFrame(DataFrame): timestamp: str -@dataclass -class TranscriptionMessage: - """A message in a conversation transcript. - - A message in a conversation transcript containing the role and content. - Messages are in standard format with roles normalized to user/assistant. - - Parameters: - role: The role of the message sender (user or assistant). - content: The message content/text. - user_id: Optional identifier for the user. - timestamp: Optional timestamp when the message was created. - - .. deprecated:: 0.0.99 - `TranscriptionMessage` is deprecated and will be removed in a future version. - Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events instead. - """ - - role: Literal["user", "assistant"] - content: str - user_id: Optional[str] = None - timestamp: Optional[str] = None - - def __post_init__(self): - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "TranscriptionMessage is deprecated and will be removed in a future version. " - "Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events instead.", - DeprecationWarning, - stacklevel=2, - ) - - -@dataclass -class ThoughtTranscriptionMessage: - """An LLM thought message in a conversation transcript. - - .. deprecated:: 0.0.99 - `ThoughtTranscriptionMessage` is deprecated and will be removed in a future version. - Use `LLMAssistantAggregator`'s new events instead. - """ - - role: Literal["assistant"] = field(default="assistant", init=False) - content: str - timestamp: Optional[str] = None - - def __post_init__(self): - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "ThoughtTranscriptionMessage is deprecated and will be removed in a future version. " - "Use `LLMAssistantAggregator`'s new events instead.", - DeprecationWarning, - stacklevel=2, - ) - - -@dataclass -class TranscriptionUpdateFrame(DataFrame): - """Frame containing new messages added to conversation transcript. - - A frame containing new messages added to the conversation transcript. - This frame is emitted when new messages are added to the conversation history, - containing only the newly added messages rather than the full transcript. - Messages have normalized roles (user/assistant) regardless of the LLM service used. - Messages are always in the OpenAI standard message format, which supports both: - - Examples: - Simple format:: - - [ - { - "role": "user", - "content": "Hi, how are you?" - }, - { - "role": "assistant", - "content": "Great! And you?" - } - ] - - Content list format:: - - [ - { - "role": "user", - "content": [{"type": "text", "text": "Hi, how are you?"}] - }, - { - "role": "assistant", - "content": [{"type": "text", "text": "Great! And you?"}] - } - ] - - OpenAI supports both formats. Anthropic and Google messages are converted to the - content list format. - - Parameters: - messages: List of new transcript messages that were added. - - .. deprecated:: 0.0.99 - `TranscriptionUpdateFrame` is deprecated and will be removed in a future version. - Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events instead. - """ - - messages: List[TranscriptionMessage | ThoughtTranscriptionMessage] - - def __post_init__(self): - super().__post_init__() - - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "TranscriptionUpdateFrame is deprecated and will be removed in a future version. " - "Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events instead.", - DeprecationWarning, - stacklevel=2, - ) - - def __str__(self): - pts = format_pts(self.pts) - return f"{self.name}(pts: {pts}, messages: {len(self.messages)})" - - @dataclass class LLMContextFrame(Frame): """Frame containing a universal LLM context. From d503383c23d8823bbc1110f7c5092b1a88101593 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 2 Apr 2026 11:19:17 -0400 Subject: [PATCH 08/11] Remove deprecated interruption_strategies plumbing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The interruption_strategies mechanism was deprecated in v0.0.99 in favor of LLMUserAggregator's user_turn_strategies. All evaluation logic was already removed — this removes the remaining field definitions, property, StartFrame propagation, conditional check in base_input.py, strategy files, and test. --- src/pipecat/audio/interruptions/__init__.py | 0 .../base_interruption_strategy.py | 58 -------------- .../min_words_interruption_strategy.py | 75 ------------------- src/pipecat/frames/frames.py | 7 -- src/pipecat/pipeline/task.py | 8 -- src/pipecat/processors/frame_processor.py | 17 ----- src/pipecat/transports/base_input.py | 15 +--- tests/test_interruption_strategies.py | 28 ------- 8 files changed, 1 insertion(+), 207 deletions(-) delete mode 100644 src/pipecat/audio/interruptions/__init__.py delete mode 100644 src/pipecat/audio/interruptions/base_interruption_strategy.py delete mode 100644 src/pipecat/audio/interruptions/min_words_interruption_strategy.py delete mode 100644 tests/test_interruption_strategies.py diff --git a/src/pipecat/audio/interruptions/__init__.py b/src/pipecat/audio/interruptions/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/src/pipecat/audio/interruptions/base_interruption_strategy.py b/src/pipecat/audio/interruptions/base_interruption_strategy.py deleted file mode 100644 index fba6cde33..000000000 --- a/src/pipecat/audio/interruptions/base_interruption_strategy.py +++ /dev/null @@ -1,58 +0,0 @@ -# -# Copyright (c) 2024-2026, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -"""Base interruption strategy for determining when users can interrupt bot speech.""" - -from abc import ABC, abstractmethod - - -class BaseInterruptionStrategy(ABC): - """Base class for interruption strategies. - - This is a base class for interruption strategies. Interruption strategies - decide when the user can interrupt the bot while the bot is speaking. For - example, there could be strategies based on audio volume or strategies based - on the number of words the user spoke. - """ - - async def append_audio(self, audio: bytes, sample_rate: int): - """Append audio data to the strategy for analysis. - - Not all strategies handle audio. Default implementation does nothing. - - Args: - audio: Raw audio bytes to append. - sample_rate: Sample rate of the audio data in Hz. - """ - pass - - async def append_text(self, text: str): - """Append text data to the strategy for analysis. - - Not all strategies handle text. Default implementation does nothing. - - Args: - text: Text string to append for analysis. - """ - pass - - @abstractmethod - async def should_interrupt(self) -> bool: - """Determine if the user should interrupt the bot. - - This is called when the user stops speaking and it's time to decide - whether the user should interrupt the bot. The decision will be based on - the aggregated audio and/or text. - - Returns: - True if the user should interrupt the bot, False otherwise. - """ - pass - - @abstractmethod - async def reset(self): - """Reset the current accumulated text and/or audio.""" - pass diff --git a/src/pipecat/audio/interruptions/min_words_interruption_strategy.py b/src/pipecat/audio/interruptions/min_words_interruption_strategy.py deleted file mode 100644 index 36f8e8903..000000000 --- a/src/pipecat/audio/interruptions/min_words_interruption_strategy.py +++ /dev/null @@ -1,75 +0,0 @@ -# -# Copyright (c) 2024-2026, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -"""Minimum words interruption strategy for word count-based interruptions.""" - -from loguru import logger - -from pipecat.audio.interruptions.base_interruption_strategy import BaseInterruptionStrategy - - -class MinWordsInterruptionStrategy(BaseInterruptionStrategy): - """Interruption strategy based on minimum number of words spoken. - - This is an interruption strategy based on a minimum number of words said - by the user. That is, the strategy will be true if the user has said at - least that amount of words. - - .. deprecated:: 0.0.99 - - This class is deprecated, use - `pipecat.turns.user_start.MinWordsUserTurnStartStrategy` with `PipelineTask`'s - new `user_turn_strategies` parameter instead. - - """ - - def __init__(self, *, min_words: int): - """Initialize the minimum words interruption strategy. - - Args: - min_words: Minimum number of words required to trigger an interruption. - """ - super().__init__() - self._min_words = min_words - self._text = "" - - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "'pipecat.audio.interruptions' is deprecated. " - "Use `pipecat.turns.user_start.MinWordsUserTurnStartStrategy` with `PipelineTask`'s " - "new `user_turn_strategies` parameter instead.", - DeprecationWarning, - ) - - async def append_text(self, text: str): - """Append text for word count analysis. - - Args: - text: Text string to append to the accumulated text. - - Note: Not all strategies need to handle text. - """ - self._text += text - - async def should_interrupt(self) -> bool: - """Check if the minimum word count has been reached. - - Returns: - True if the user has spoken at least the minimum number of words. - """ - word_count = len(self._text.split()) - interrupt = word_count >= self._min_words - logger.debug( - f"should_interrupt={interrupt} num_spoken_words={word_count} min_words={self._min_words}" - ) - return interrupt - - async def reset(self): - """Reset the accumulated text for the next analysis cycle.""" - self._text = "" diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 41b930ee5..76ed58682 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -29,7 +29,6 @@ from typing import ( from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.audio.dtmf.types import KeypadEntry -from pipecat.audio.interruptions.base_interruption_strategy import BaseInterruptionStrategy from pipecat.audio.turn.base_turn_analyzer import BaseTurnParams from pipecat.audio.vad.vad_analyzer import VADParams from pipecat.metrics.metrics import MetricsData @@ -750,11 +749,6 @@ class StartFrame(SystemFrame): enable_metrics: Whether to enable performance metrics collection. enable_tracing: Whether to enable OpenTelemetry tracing. enable_usage_metrics: Whether to enable usage metrics collection. - interruption_strategies: List of interruption handling strategies. - - .. deprecated:: 0.0.99 - Use `LLMUserAggregator`'s new `user_turn_strategies` parameter instead. - report_only_initial_ttfb: Whether to report only initial time-to-first-byte. tracing_context: Pipeline-scoped tracing context for span hierarchy. """ @@ -764,7 +758,6 @@ class StartFrame(SystemFrame): enable_metrics: bool = False enable_tracing: bool = False enable_usage_metrics: bool = False - interruption_strategies: List[BaseInterruptionStrategy] = field(default_factory=list) report_only_initial_ttfb: bool = False tracing_context: Optional["TracingContext"] = None diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index f3e7804f6..b3a034c91 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -20,7 +20,6 @@ from typing import Any, AsyncIterable, Dict, Iterable, List, Optional, Set, Tupl from loguru import logger from pydantic import BaseModel, ConfigDict, Field -from pipecat.audio.interruptions.base_interruption_strategy import BaseInterruptionStrategy from pipecat.clocks.base_clock import BaseClock from pipecat.clocks.system_clock import SystemClock from pipecat.frames.frames import ( @@ -119,11 +118,6 @@ class PipelineParams(BaseModel): heartbeats_period_secs: Period between heartbeats in seconds. heartbeats_monitor_secs: Timeout (in seconds) before warning about missed heartbeats. Defaults to 10 seconds. - interruption_strategies: [deprecated] Strategies for bot interruption behavior. - - .. deprecated:: 0.0.99 - Use `LLMUserAggregator`'s new `user_turn_strategies` parameter instead. - 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. @@ -138,7 +132,6 @@ class PipelineParams(BaseModel): enable_usage_metrics: bool = False heartbeats_period_secs: float = HEARTBEAT_SECS heartbeats_monitor_secs: float = HEARTBEAT_MONITOR_SECS - interruption_strategies: List[BaseInterruptionStrategy] = Field(default_factory=list) report_only_initial_ttfb: bool = False send_initial_empty_metrics: bool = True start_metadata: Dict[str, Any] = Field(default_factory=dict) @@ -778,7 +771,6 @@ class PipelineTask(BasePipelineTask): enable_tracing=self._enable_tracing, enable_usage_metrics=self._params.enable_usage_metrics, report_only_initial_ttfb=self._params.report_only_initial_ttfb, - interruption_strategies=self._params.interruption_strategies, tracing_context=self._tracing_context, ) start_frame.metadata = self._create_start_metadata() diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 3fe31ec01..02cf6ce7b 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -23,14 +23,12 @@ from typing import ( Coroutine, List, Optional, - Sequence, Tuple, Type, ) from loguru import logger -from pipecat.audio.interruptions.base_interruption_strategy import BaseInterruptionStrategy from pipecat.clocks.base_clock import BaseClock from pipecat.frames.frames import ( CancelFrame, @@ -193,7 +191,6 @@ class FrameProcessor(BaseObject): self._enable_metrics = False self._enable_usage_metrics = False self._report_only_initial_ttfb = False - self._interruption_strategies: List[BaseInterruptionStrategy] = [] # Indicates whether we have received the StartFrame. self.__started = False @@ -332,19 +329,6 @@ class FrameProcessor(BaseObject): """ return self._report_only_initial_ttfb - @property - def interruption_strategies(self) -> Sequence[BaseInterruptionStrategy]: - """Get the interruption strategies for this processor. - - .. deprecated:: 0.0.99 - This function is deprecated, use the new user and bot turn start - strategies insted. - - Returns: - Sequence of interruption strategies. - """ - return self._interruption_strategies - @property def task_manager(self) -> BaseTaskManager: """Get the task manager for this processor. @@ -796,7 +780,6 @@ class FrameProcessor(BaseObject): self.__started = True self._enable_metrics = frame.enable_metrics self._enable_usage_metrics = frame.enable_usage_metrics - self._interruption_strategies = frame.interruption_strategies self._report_only_initial_ttfb = frame.report_only_initial_ttfb self.__create_process_task() diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index ff7026673..3d26925cb 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -510,21 +510,8 @@ class BaseInputTransport(FrameProcessor): await self.broadcast_frame(UserStartedSpeakingFrame, emulated=emulated) - # 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 = ( - not self.interruption_strategies or not self._bot_speaking - ) - # Make sure we notify about interruptions quickly out-of-band. - if should_push_immediate_interruption: - await self.broadcast_interruption() - elif self.interruption_strategies and self._bot_speaking: - logger.debug( - "User started speaking while bot is speaking with interruption config - " - "deferring interruption to aggregator" - ) + await self.broadcast_interruption() elif vad_state == VADState.QUIET: logger.debug("User stopped speaking") self._user_speaking = False diff --git a/tests/test_interruption_strategies.py b/tests/test_interruption_strategies.py deleted file mode 100644 index d4aff2a7a..000000000 --- a/tests/test_interruption_strategies.py +++ /dev/null @@ -1,28 +0,0 @@ -# -# Copyright (c) 2024-2026, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -import unittest - -from pipecat.audio.interruptions.min_words_interruption_strategy import MinWordsInterruptionStrategy - - -class TestMinWordsInterruptionStrategy(unittest.IsolatedAsyncioTestCase): - async def test_min_words(self): - strategy = MinWordsInterruptionStrategy(min_words=2) - await strategy.append_text("Hello") - self.assertEqual(await strategy.should_interrupt(), False) - await strategy.append_text(" there!") - self.assertEqual(await strategy.should_interrupt(), True) - # Reset and check again - await strategy.reset() - await strategy.append_text("Hello!") - self.assertEqual(await strategy.should_interrupt(), False) - await strategy.append_text(" How are you?") - self.assertEqual(await strategy.should_interrupt(), True) - - -if __name__ == "__main__": - unittest.main() From 5b67dcd9e78d14d77797cbe172bfe9b83a70b483 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 2 Apr 2026 11:31:29 -0400 Subject: [PATCH 09/11] Remove deprecated EmulateUser{Started,Stopped}SpeakingFrame and emulated field Remove EmulateUserStartedSpeakingFrame, EmulateUserStoppedSpeakingFrame (deprecated since v0.0.99), and the emulated field from UserStartedSpeakingFrame and UserStoppedSpeakingFrame. Clean up the handling code in base_input.py and a stale comment in nova_sonic/llm.py. --- src/pipecat/frames/frames.py | 68 +--------------------- src/pipecat/services/aws/nova_sonic/llm.py | 14 +---- src/pipecat/transports/base_input.py | 16 +---- 3 files changed, 7 insertions(+), 91 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 76ed58682..86a93825b 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -866,16 +866,9 @@ class UserStartedSpeakingFrame(SystemFrame): Emitted when the user turn starts, which usually means that some transcriptions are already available. - - Parameters: - emulated: Whether this event was emulated rather than detected by VAD. - - .. deprecated:: 0.0.99 - This field is deprecated and will be removed in a future version. - """ - emulated: bool = False + pass @dataclass @@ -884,16 +877,9 @@ class UserStoppedSpeakingFrame(SystemFrame): Emitted when the user turn ends. This usually coincides with the start of the bot turn. - - Parameters: - emulated: Whether this event was emulated rather than detected by VAD. - - .. deprecated:: 0.0.99 - This field is deprecated and will be removed in a future version. - """ - emulated: bool = False + pass @dataclass @@ -928,56 +914,6 @@ class UserSpeakingFrame(SystemFrame): pass -@dataclass -class EmulateUserStartedSpeakingFrame(SystemFrame): - """Frame to emulate user started speaking behavior. - - Emitted by internal processors upstream to emulate VAD behavior when a - user starts speaking. - - .. deprecated:: 0.0.99 - This frame is deprecated and will be removed in a future version. - """ - - def __post_init__(self): - super().__post_init__() - - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "EmulateUserStartedSpeakingFrame is deprecated and will be removed in a future version.", - DeprecationWarning, - stacklevel=2, - ) - - -@dataclass -class EmulateUserStoppedSpeakingFrame(SystemFrame): - """Frame to emulate user stopped speaking behavior. - - Emitted by internal processors upstream to emulate VAD behavior when a - user stops speaking. - - .. deprecated:: 0.0.99 - This frame is deprecated and will be removed in a future version. - """ - - def __post_init__(self): - super().__post_init__() - - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "EmulateUserStoppedSpeakingFrame is deprecated and will be removed in a future version.", - DeprecationWarning, - stacklevel=2, - ) - - @dataclass class VADUserStartedSpeakingFrame(SystemFrame): """Frame emitted when VAD definitively detects user started speaking. diff --git a/src/pipecat/services/aws/nova_sonic/llm.py b/src/pipecat/services/aws/nova_sonic/llm.py index 6f330babb..2465624f5 100644 --- a/src/pipecat/services/aws/nova_sonic/llm.py +++ b/src/pipecat/services/aws/nova_sonic/llm.py @@ -1281,18 +1281,8 @@ class AWSNovaSonicLLMService(LLMService): # HACK: Check if this transcription was triggered by our own # assistant response trigger. If so, we need to wrap it with # UserStarted/StoppedSpeakingFrames; otherwise the user aggregator - # would fire an EmulatedUserStartedSpeakingFrame, which would - # trigger an interruption, which would prevent us from writing the - # assistant response to context. - # - # Sending an EmulateUserStartedSpeakingFrame ourselves doesn't - # work: it just causes the interruption we're trying to avoid. - # - # Setting enable_emulated_vad_interruptions also doesn't work: at - # the time the user aggregator receives the TranscriptionFrame, it - # doesn't yet know the assistant has started responding, so it - # doesn't know that emulating the user starting to speak would - # cause an interruption. + # would trigger an interruption, which would prevent us from + # writing the assistant response to context. should_wrap_in_user_started_stopped_speaking_frames = ( self._waiting_for_trigger_transcription and self._user_text_buffer.strip().lower() == "ready" diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 3d26925cb..d4adb74e1 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -25,8 +25,6 @@ from pipecat.frames.frames import ( BotStartedSpeakingFrame, BotStoppedSpeakingFrame, CancelFrame, - EmulateUserStartedSpeakingFrame, - EmulateUserStoppedSpeakingFrame, EndFrame, FilterUpdateSettingsFrame, Frame, @@ -313,12 +311,6 @@ class BaseInputTransport(FrameProcessor): elif isinstance(frame, BotStoppedSpeakingFrame): await self._deprecated_handle_bot_stopped_speaking(frame) await self.push_frame(frame, direction) - elif isinstance(frame, EmulateUserStartedSpeakingFrame): - logger.debug("Emulating user started speaking") - await self._deprecated_handle_user_interruption(VADState.SPEAKING, emulated=True) - elif isinstance(frame, EmulateUserStoppedSpeakingFrame): - logger.debug("Emulating user stopped speaking") - await self._deprecated_handle_user_interruption(VADState.QUIET, emulated=True) # All other system frames elif isinstance(frame, SystemFrame): await self.push_frame(frame, direction) @@ -500,15 +492,13 @@ class BaseInputTransport(FrameProcessor): """Update bot speaking state when bot stops speaking.""" self._bot_speaking = False - async def _deprecated_handle_user_interruption( - self, vad_state: VADState, emulated: bool = False - ): + async def _deprecated_handle_user_interruption(self, vad_state: VADState): """Handle user interruption events based on speaking state.""" if vad_state == VADState.SPEAKING: logger.debug("User started speaking") self._user_speaking = True - await self.broadcast_frame(UserStartedSpeakingFrame, emulated=emulated) + await self.broadcast_frame(UserStartedSpeakingFrame) # Make sure we notify about interruptions quickly out-of-band. await self.broadcast_interruption() @@ -516,7 +506,7 @@ class BaseInputTransport(FrameProcessor): logger.debug("User stopped speaking") self._user_speaking = False - await self.broadcast_frame(UserStoppedSpeakingFrame, emulated=emulated) + await self.broadcast_frame(UserStoppedSpeakingFrame) async def _deprecated_old_handle_vad( self, audio_frame: InputAudioRawFrame, vad_state: VADState From 0c598196820300ff243bc4f0e3116e8c20d43d29 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 2 Apr 2026 11:32:44 -0400 Subject: [PATCH 10/11] Remove allow_interruptions from voice-sarvam example This was missed from the allow_interruptions removal commit. --- examples/voice/voice-sarvam.py | 1 - 1 file changed, 1 deletion(-) diff --git a/examples/voice/voice-sarvam.py b/examples/voice/voice-sarvam.py index bd007d89e..feda4fe38 100644 --- a/examples/voice/voice-sarvam.py +++ b/examples/voice/voice-sarvam.py @@ -96,7 +96,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): params=PipelineParams( enable_metrics=True, enable_usage_metrics=True, - allow_interruptions=True, ), ) From 3be0ea05ef101e6eea7cb1f3ca50bb5aae678c21 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 2 Apr 2026 11:34:22 -0400 Subject: [PATCH 11/11] Add changelog entries for #4228 --- changelog/4228.removed.10.md | 1 + changelog/4228.removed.2.md | 1 + changelog/4228.removed.3.md | 1 + changelog/4228.removed.4.md | 1 + changelog/4228.removed.5.md | 1 + changelog/4228.removed.6.md | 1 + changelog/4228.removed.7.md | 1 + changelog/4228.removed.8.md | 1 + changelog/4228.removed.9.md | 1 + changelog/4228.removed.md | 1 + 10 files changed, 10 insertions(+) create mode 100644 changelog/4228.removed.10.md create mode 100644 changelog/4228.removed.2.md create mode 100644 changelog/4228.removed.3.md create mode 100644 changelog/4228.removed.4.md create mode 100644 changelog/4228.removed.5.md create mode 100644 changelog/4228.removed.6.md create mode 100644 changelog/4228.removed.7.md create mode 100644 changelog/4228.removed.8.md create mode 100644 changelog/4228.removed.9.md create mode 100644 changelog/4228.removed.md diff --git a/changelog/4228.removed.10.md b/changelog/4228.removed.10.md new file mode 100644 index 000000000..259275b49 --- /dev/null +++ b/changelog/4228.removed.10.md @@ -0,0 +1 @@ +- ⚠️ Removed deprecated `add_pattern_pair` method from `PatternPairAggregator`. Use `add_pattern` instead. diff --git a/changelog/4228.removed.2.md b/changelog/4228.removed.2.md new file mode 100644 index 000000000..c662fc08d --- /dev/null +++ b/changelog/4228.removed.2.md @@ -0,0 +1 @@ +- ⚠️ Removed deprecated `interruption_strategies` parameter from `PipelineParams`, `StartFrame`, and `FrameProcessor`. Use `LLMUserAggregator`'s `user_turn_strategies` parameter instead. diff --git a/changelog/4228.removed.3.md b/changelog/4228.removed.3.md new file mode 100644 index 000000000..7dfd35a1d --- /dev/null +++ b/changelog/4228.removed.3.md @@ -0,0 +1 @@ +- ⚠️ Removed deprecated `EmulateUserStartedSpeakingFrame` and `EmulateUserStoppedSpeakingFrame` frames, and the `emulated` field from `UserStartedSpeakingFrame` / `UserStoppedSpeakingFrame`. diff --git a/changelog/4228.removed.4.md b/changelog/4228.removed.4.md new file mode 100644 index 000000000..7a376b817 --- /dev/null +++ b/changelog/4228.removed.4.md @@ -0,0 +1 @@ +- ⚠️ Removed deprecated `pipecat.audio.interruptions` module (`BaseInterruptionStrategy`, `MinWordsInterruptionStrategy`). Use `pipecat.turns.user_start.MinWordsUserTurnStartStrategy` with `LLMUserAggregator`'s `user_turn_strategies` parameter instead. diff --git a/changelog/4228.removed.5.md b/changelog/4228.removed.5.md new file mode 100644 index 000000000..cb94d7151 --- /dev/null +++ b/changelog/4228.removed.5.md @@ -0,0 +1 @@ +- ⚠️ Removed deprecated `pipecat.processors.transcript_processor` module (`TranscriptProcessor`, `TranscriptProcessorConfig`). Use pipeline observers instead. diff --git a/changelog/4228.removed.6.md b/changelog/4228.removed.6.md new file mode 100644 index 000000000..2879461d9 --- /dev/null +++ b/changelog/4228.removed.6.md @@ -0,0 +1 @@ +- ⚠️ Removed deprecated `TranscriptionMessage`, `ThoughtTranscriptionMessage`, and `TranscriptionUpdateFrame` from `pipecat.frames.frames`. diff --git a/changelog/4228.removed.7.md b/changelog/4228.removed.7.md new file mode 100644 index 000000000..1e51a9316 --- /dev/null +++ b/changelog/4228.removed.7.md @@ -0,0 +1 @@ +- ⚠️ Removed deprecated `STTMuteFilter`, `STTMuteConfig`, and `STTMuteStrategy` from `pipecat.processors.filters.stt_mute_filter`. Use `pipecat.turns.user_mute` strategies with `LLMUserAggregator`'s `user_mute_strategies` parameter instead. diff --git a/changelog/4228.removed.8.md b/changelog/4228.removed.8.md new file mode 100644 index 000000000..926a827a9 --- /dev/null +++ b/changelog/4228.removed.8.md @@ -0,0 +1 @@ +- ⚠️ Removed deprecated `UserResponseAggregator` class from `pipecat.processors.aggregators.user_response`. Use `LLMUserAggregator` instead. diff --git a/changelog/4228.removed.9.md b/changelog/4228.removed.9.md new file mode 100644 index 000000000..5a6aa32ad --- /dev/null +++ b/changelog/4228.removed.9.md @@ -0,0 +1 @@ +- ⚠️ Removed deprecated `pipecat.utils.tracing.class_decorators` module. Use `pipecat.utils.tracing.service_decorators` instead. diff --git a/changelog/4228.removed.md b/changelog/4228.removed.md new file mode 100644 index 000000000..176ed6b63 --- /dev/null +++ b/changelog/4228.removed.md @@ -0,0 +1 @@ +- ⚠️ Removed deprecated `allow_interruptions` parameter from `PipelineParams`, `StartFrame`, and `FrameProcessor`. Interruptions are now always allowed by default. Use `LLMUserAggregator`'s `user_turn_strategies` / `user_mute_strategies` parameters to control interruption behavior.