wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove deprecated interruption_strategies plumbing

Browse Source 
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.
This commit is contained in:
Mark Backman
2026-04-02 11:19:17 -04:00
parent fa30268b84
commit d503383c23
8 changed files with 1 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
src/pipecat/audio/interruptions/__init__.py
View File

58
src/pipecat/audio/interruptions/base_interruption_strategy.py
View File

@@ -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

75
src/pipecat/audio/interruptions/min_words_interruption_strategy.py
View File

@@ -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 = ""

7
src/pipecat/frames/frames.py
View File

@@ -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

8
src/pipecat/pipeline/task.py
View File

@@ -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()

17
src/pipecat/processors/frame_processor.py
View File

@@ -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()

15
src/pipecat/transports/base_input.py
View File

@@ -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

28
tests/test_interruption_strategies.py
View File

@@ -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()