From cc3af59db4f9b7a66f233c96ed689e300113f8c2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 2 Apr 2026 11:51:08 -0700 Subject: [PATCH] transports: remove deprecated VAD and turn analyzers --- src/pipecat/transports/base_input.py | 321 +------------------- src/pipecat/transports/base_transport.py | 9 - src/pipecat/transports/daily/transport.py | 11 - src/pipecat/transports/livekit/transport.py | 10 - 4 files changed, 3 insertions(+), 348 deletions(-) diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index d4adb74e1..4fb4a1736 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -7,42 +7,24 @@ """Base input transport implementation for Pipecat. This module provides the BaseInputTransport class which handles audio and video -input processing, including VAD, turn analysis, and interruption management. +input processing. """ import asyncio -import time -from typing import Optional from loguru import logger -from pipecat.audio.turn.base_turn_analyzer import ( - BaseTurnAnalyzer, - EndOfTurnState, -) -from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADState from pipecat.frames.frames import ( - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, CancelFrame, EndFrame, FilterUpdateSettingsFrame, Frame, InputAudioRawFrame, InputImageRawFrame, - MetricsFrame, - SpeechControlParamsFrame, StartFrame, StopFrame, SystemFrame, - UserSpeakingFrame, - UserStartedSpeakingFrame, - UserStoppedSpeakingFrame, - VADParamsUpdateFrame, - VADUserStartedSpeakingFrame, - VADUserStoppedSpeakingFrame, ) -from pipecat.metrics.metrics import MetricsData from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.transports.base_transport import TransportParams @@ -91,29 +73,6 @@ class BaseInputTransport(FrameProcessor): # them downstream until we get another `StartFrame`. self._paused = False - if self._params.turn_analyzer: - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "Parameter 'turn_analyzer' is deprecated, use `LLMUserAggregator`'s new " - "`user_turn_strategies` parameter instead.", - DeprecationWarning, - ) - - if self._params.vad_analyzer: - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "Parameter 'vad_analyzer' is deprecated. Use `LLMUserAggregator`'s " - "`vad_analyzer` parameter, or `VADProcessor` if no `LLMUserAggregator` " - "is needed.", - DeprecationWarning, - ) - def enable_audio_in_stream_on_start(self, enabled: bool) -> None: """Enable or disable audio streaming on transport start. @@ -139,52 +98,6 @@ class BaseInputTransport(FrameProcessor): """ return self._sample_rate - @property - def vad_analyzer(self) -> Optional[VADAnalyzer]: - """Get the Voice Activity Detection analyzer. - - .. deprecated:: 0.0.101 - This method is deprecated and will be removed in a future version. - Use `LLMUserAggregator`'s new `vad_analyzer` parameter instead. - - Returns: - The VAD analyzer instance if configured, None otherwise. - """ - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "Method 'vad_analyzer' is deprecated. Use `LLMUserAggregator`'s new " - "`vad_analyzer` parameter instead.", - DeprecationWarning, - ) - - return self._params.vad_analyzer - - @property - def turn_analyzer(self) -> Optional[BaseTurnAnalyzer]: - """Get the turn-taking analyzer. - - .. deprecated:: 0.0.99 - This method is deprecated and will be removed in a future version. - Use `LLMUserAggregator`'s new `user_turn_strategies` parameter instead. - - Returns: - The turn analyzer instance if configured, None otherwise. - """ - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "Method 'turn_analyzer' is deprecated. Use `LLMUserAggregator`'s new " - "`user_turn_strategies` parameter instead.", - DeprecationWarning, - ) - - return self._params.turn_analyzer - async def start(self, frame: StartFrame): """Start the input transport and initialize components. @@ -200,26 +113,6 @@ class BaseInputTransport(FrameProcessor): if self._params.audio_in_filter: await self._params.audio_in_filter.start(self._sample_rate) - ################################################################### - # DEPRECATED. - - # Configure VAD analyzer. - if self._params.vad_analyzer: - self._params.vad_analyzer.set_sample_rate(self._sample_rate) - - # Configure End of turn analyzer. - if self._params.turn_analyzer: - self._params.turn_analyzer.set_sample_rate(self._sample_rate) - - if self._params.vad_analyzer or self._params.turn_analyzer: - vad_params = self._params.vad_analyzer.params if self._params.vad_analyzer else None - turn_params = self._params.turn_analyzer.params if self._params.turn_analyzer else None - - await self.broadcast_frame( - SpeechControlParamsFrame, vad_params=vad_params, turn_params=turn_params - ) - ################################################################### - async def stop(self, frame: EndFrame): """Stop the input transport and cleanup resources. @@ -305,12 +198,6 @@ class BaseInputTransport(FrameProcessor): elif isinstance(frame, CancelFrame): await self.cancel(frame) await self.push_frame(frame, direction) - elif isinstance(frame, BotStartedSpeakingFrame): - await self._deprecated_handle_bot_started_speaking(frame) - await self.push_frame(frame, direction) - elif isinstance(frame, BotStoppedSpeakingFrame): - await self._deprecated_handle_bot_stopped_speaking(frame) - await self.push_frame(frame, direction) # All other system frames elif isinstance(frame, SystemFrame): await self.push_frame(frame, direction) @@ -323,19 +210,6 @@ class BaseInputTransport(FrameProcessor): elif isinstance(frame, StopFrame): await self.push_frame(frame, direction) await self.pause(frame) - ################################################################### - # DEPRECATED. - elif isinstance(frame, VADParamsUpdateFrame): - if self._params.vad_analyzer: - self._params.vad_analyzer.set_params(frame.params) - await self.broadcast_frame( - SpeechControlParamsFrame, - vad_params=frame.params, - turn_params=self._params.turn_analyzer.params - if self._params.turn_analyzer - else None, - ) - ################################################################### elif isinstance(frame, FilterUpdateSettingsFrame) and self._params.audio_in_filter: await self._params.audio_in_filter.process_frame(frame) # Other frames @@ -359,8 +233,7 @@ class BaseInputTransport(FrameProcessor): self._audio_task = None async def _audio_task_handler(self): - """Main audio processing task handler for VAD and turn analysis.""" - vad_state: VADState = VADState.QUIET + """Main audio processing task handler.""" # Skip timeout handling until the first audio frame arrives (e.g. client # not yet connected). audio_received = False @@ -373,7 +246,7 @@ class BaseInputTransport(FrameProcessor): # From now on, timeout should warn if there's no audio. audio_received = True - # If an audio filter is available, run it before VAD. + # Filter audio, if an audio filter is available. if self._params.audio_in_filter: frame.audio = await self._params.audio_in_filter.filter(frame.audio) @@ -382,22 +255,6 @@ class BaseInputTransport(FrameProcessor): self._audio_in_queue.task_done() continue - ################################################################### - # DEPRECATED. - # - # Check VAD and push event if necessary. We just care about - # changes from QUIET to SPEAKING and vice versa. - previous_vad_state = vad_state - if self._params.vad_analyzer: - vad_state = await self._deprecated_handle_vad(frame, vad_state) - - if self._params.turn_analyzer: - await self._deprecated_run_turn_analyzer(frame, vad_state, previous_vad_state) - - if self._params.vad_analyzer and vad_state == VADState.SPEAKING: - await self._deprecated_user_currently_speaking() - ################################################################### - # Push audio downstream if passthrough is set. if self._params.audio_in_passthrough: await self.push_frame(frame) @@ -406,175 +263,3 @@ class BaseInputTransport(FrameProcessor): except asyncio.TimeoutError: if not audio_received: continue - - ################################################################### - # DEPRECATED. - if self._user_speaking: - logger.warning( - "Forcing VAD user stopped speaking due to timeout receiving audio frame!" - ) - vad_state = VADState.QUIET - if self._params.turn_analyzer: - self._params.turn_analyzer.clear() - - if self._params.turn_analyzer: - await self._deprecated_handle_user_interruption(VADState.QUIET) - else: - stop_secs = ( - self._params.vad_analyzer.params.stop_secs - if self._params.vad_analyzer - else 0.0 - ) - await self.push_frame(VADUserStoppedSpeakingFrame(stop_secs=stop_secs)) - ################################################################### - - # - # DEPRECATED. - # - # The functions below are deprecated and should be removed once the old - # interruption strategies and turn analyzer are removed. - # - - async def _deprecated_vad_analyze(self, audio_frame: InputAudioRawFrame) -> VADState: - """Analyze audio frame for voice activity.""" - state = VADState.QUIET - if self._params.vad_analyzer: - state = await self._params.vad_analyzer.analyze_audio(audio_frame.audio) - return state - - async def _deprecated_new_handle_vad( - self, audio_frame: InputAudioRawFrame, vad_state: VADState - ) -> VADState: - """Handle Voice Activity Detection results and generate appropriate frames.""" - new_vad_state = await self._deprecated_vad_analyze(audio_frame) - if ( - new_vad_state != vad_state - and new_vad_state != VADState.STARTING - and new_vad_state != VADState.STOPPING - ): - if new_vad_state == VADState.SPEAKING: - start_secs = ( - self._params.vad_analyzer.params.start_secs - if self._params.vad_analyzer - else 0.0 - ) - await self.push_frame(VADUserStartedSpeakingFrame(start_secs=start_secs)) - elif new_vad_state == VADState.QUIET: - stop_secs = ( - self._params.vad_analyzer.params.stop_secs if self._params.vad_analyzer else 0.0 - ) - await self.push_frame(VADUserStoppedSpeakingFrame(stop_secs=stop_secs)) - - vad_state = new_vad_state - return vad_state - - async def _deprecated_handle_vad( - self, audio_frame: InputAudioRawFrame, vad_state: VADState - ) -> VADState: - """Handle Voice Activity Detection results and generate appropriate frames.""" - if self._params.turn_analyzer: - return await self._deprecated_old_handle_vad(audio_frame, vad_state) - else: - return await self._deprecated_new_handle_vad(audio_frame, vad_state) - - async def _deprecated_user_currently_speaking(self): - """Handle user speaking frame.""" - diff_time = time.time() - self._user_speaking_frame_time - if diff_time >= self._user_speaking_frame_period: - await self.broadcast_frame(UserSpeakingFrame) - self._user_speaking_frame_time = time.time() - - async def _deprecated_handle_bot_started_speaking(self, frame: BotStartedSpeakingFrame): - """Update bot speaking state when bot starts speaking.""" - self._bot_speaking = True - - async def _deprecated_handle_bot_stopped_speaking(self, frame: BotStoppedSpeakingFrame): - """Update bot speaking state when bot stops speaking.""" - self._bot_speaking = 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) - - # Make sure we notify about interruptions quickly out-of-band. - await self.broadcast_interruption() - elif vad_state == VADState.QUIET: - logger.debug("User stopped speaking") - self._user_speaking = False - - await self.broadcast_frame(UserStoppedSpeakingFrame) - - async def _deprecated_old_handle_vad( - self, audio_frame: InputAudioRawFrame, vad_state: VADState - ) -> VADState: - """Handle Voice Activity Detection results and generate appropriate frames.""" - new_vad_state = await self._deprecated_vad_analyze(audio_frame) - if ( - new_vad_state != vad_state - and new_vad_state != VADState.STARTING - and new_vad_state != VADState.STOPPING - ): - interruption_state = None - - # If the turn analyser is enabled, this will prevent: - # - Creating the UserStoppedSpeakingFrame - # - Creating the UserStartedSpeakingFrame multiple times - can_create_user_frames = ( - self._params.turn_analyzer is None - or not self._params.turn_analyzer.speech_triggered - ) - if new_vad_state == VADState.SPEAKING: - start_secs = ( - self._params.vad_analyzer.params.start_secs - if self._params.vad_analyzer - else 0.0 - ) - await self.push_frame(VADUserStartedSpeakingFrame(start_secs=start_secs)) - if can_create_user_frames: - interruption_state = VADState.SPEAKING - elif new_vad_state == VADState.QUIET: - stop_secs = ( - self._params.vad_analyzer.params.stop_secs if self._params.vad_analyzer else 0.0 - ) - await self.push_frame(VADUserStoppedSpeakingFrame(stop_secs=stop_secs)) - if can_create_user_frames: - interruption_state = VADState.QUIET - - if interruption_state: - await self._deprecated_handle_user_interruption(interruption_state) - - vad_state = new_vad_state - return vad_state - - async def _deprecated_handle_end_of_turn(self): - """Handle end-of-turn analysis and generate prediction results.""" - if self._params.turn_analyzer: - state, prediction = await self._params.turn_analyzer.analyze_end_of_turn() - await self._deprecated_handle_prediction_result(prediction) - await self._deprecated_handle_end_of_turn_complete(state) - - async def _deprecated_handle_end_of_turn_complete(self, state: EndOfTurnState): - """Handle completion of end-of-turn analysis.""" - if state == EndOfTurnState.COMPLETE: - await self._deprecated_handle_user_interruption(VADState.QUIET) - - async def _deprecated_handle_prediction_result(self, result: MetricsData): - """Handle a prediction result event from the turn analyzer.""" - await self.push_frame(MetricsFrame(data=[result])) - - async def _deprecated_run_turn_analyzer( - self, frame: InputAudioRawFrame, vad_state: VADState, previous_vad_state: VADState - ): - """Run turn analysis on audio frame and handle results.""" - is_speech = vad_state == VADState.SPEAKING or vad_state == VADState.STARTING - # If silence exceeds threshold, we are going to receive EndOfTurnState.COMPLETE - end_of_turn_state = self._params.turn_analyzer.append_audio(frame.audio, is_speech) - if end_of_turn_state == EndOfTurnState.COMPLETE: - await self._deprecated_handle_end_of_turn_complete(end_of_turn_state) - # Otherwise we are going to trigger to check if the turn is completed based on the VAD - elif vad_state == VADState.QUIET and vad_state != previous_vad_state: - await self._deprecated_handle_end_of_turn() diff --git a/src/pipecat/transports/base_transport.py b/src/pipecat/transports/base_transport.py index 3e760cd32..4d7d3caa0 100644 --- a/src/pipecat/transports/base_transport.py +++ b/src/pipecat/transports/base_transport.py @@ -54,13 +54,6 @@ class TransportParams(BaseModel): video_out_color_format: Video output color format string. video_out_codec: Preferred video codec for output (e.g., 'VP8', 'H264', 'H265'). video_out_destinations: List of video output destination identifiers. - vad_analyzer: Voice Activity Detection analyzer instance. - - .. deprecated:: 0.0.101 - The `vad_analyzer` parameter is deprecated. Use `LLMUserAggregator`'s - `vad_analyzer` parameter, or `VADProcessor` if no `LLMUserAggregator` - is needed. - turn_analyzer: Turn-taking analyzer instance for conversation management. .. deprecated:: 0.0.99 @@ -95,8 +88,6 @@ class TransportParams(BaseModel): video_out_color_format: str = "RGB" video_out_codec: Optional[str] = None video_out_destinations: List[str] = Field(default_factory=list) - vad_analyzer: Optional[VADAnalyzer] = None - turn_analyzer: Optional[BaseTurnAnalyzer] = None class BaseTransport(BaseObject): diff --git a/src/pipecat/transports/daily/transport.py b/src/pipecat/transports/daily/transport.py index 6b5f3e909..ae91db7b5 100644 --- a/src/pipecat/transports/daily/transport.py +++ b/src/pipecat/transports/daily/transport.py @@ -1711,17 +1711,6 @@ class DailyInputTransport(BaseInputTransport): # Audio task when using a virtual speaker (i.e. no user tracks). self._audio_in_task: Optional[asyncio.Task] = None - self._vad_analyzer: Optional[VADAnalyzer] = params.vad_analyzer - - @property - def vad_analyzer(self) -> Optional[VADAnalyzer]: - """Get the Voice Activity Detection analyzer. - - Returns: - The VAD analyzer instance if configured. - """ - return self._vad_analyzer - async def start_audio_in_streaming(self): """Start receiving audio from participants.""" if not self._params.audio_in_enabled: diff --git a/src/pipecat/transports/livekit/transport.py b/src/pipecat/transports/livekit/transport.py index 07a7e3f39..34a23f16d 100644 --- a/src/pipecat/transports/livekit/transport.py +++ b/src/pipecat/transports/livekit/transport.py @@ -652,21 +652,11 @@ class LiveKitInputTransport(BaseInputTransport): self._audio_in_task = None self._video_in_task = None - self._vad_analyzer: Optional[VADAnalyzer] = params.vad_analyzer self._resampler = create_stream_resampler() # Whether we have seen a StartFrame already. self._initialized = False - @property - def vad_analyzer(self) -> Optional[VADAnalyzer]: - """Get the Voice Activity Detection analyzer. - - Returns: - The VAD analyzer instance if configured. - """ - return self._vad_analyzer - async def start(self, frame: StartFrame): """Start the input transport and connect to LiveKit room.