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

transports: remove deprecated VAD and turn analyzers

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2026-04-02 11:51:08 -07:00
parent f93d13efff
commit cc3af59db4
4 changed files with 3 additions and 348 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

9
src/pipecat/transports/base_transport.py
View File

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

11
src/pipecat/transports/daily/transport.py
View File

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

10
src/pipecat/transports/livekit/transport.py
View File

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