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

Merge pull request #4229 from pipecat-ai/aleix/deprecate-transport-vad-turn-analyzers

Browse Source 
Remove deprecated transport VAD/turn analyzers and ExternalUserTurnStrategies
This commit is contained in:
Aleix Conchillo Flaqué
2026-04-02 14:30:12 -07:00
committed by GitHub
parent f93d13efff a29be38f48
commit 57068f1b38
8 changed files with 17 additions and 386 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog/4229.removed.2.md Normal file
View File

@@ -0,0 +1 @@
- ⚠️ Removed `ExternalUserTurnStrategies` and the automatic fallback to it in `LLMUserAggregator` when a `SpeechControlParamsFrame` was received from the transport.

1
changelog/4229.removed.md Normal file
View File

@@ -0,0 +1 @@
- ⚠️ Removed `vad_analyzer` and `turn_analyzer` parameters from `TransportParams` and all transport input classes, along with all deprecated VAD/turn analysis logic in `BaseInputTransport`. VAD and turn detection are now handled entirely by `LLMUserAggregator`.

37
src/pipecat/processors/aggregators/llm_response_universal.py
View File

@@ -50,7 +50,6 @@ from pipecat.frames.frames import (
LLMThoughtStartFrame,
LLMThoughtTextFrame,
LLMUpdateSettingsFrame,
SpeechControlParamsFrame,
StartFrame,
TextFrame,
TranscriptionFrame,
@@ -74,7 +73,7 @@ from pipecat.processors.aggregators.llm_context_summarizer import (
LLMContextSummarizer,
SummaryAppliedEvent,
)
from pipecat.processors.frame_processor import FrameCallback, FrameDirection, FrameProcessor
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.settings import LLMSettings
from pipecat.turns.user_idle_controller import UserIdleController
from pipecat.turns.user_mute import BaseUserMuteStrategy
@@ -82,7 +81,7 @@ from pipecat.turns.user_start import BaseUserTurnStartStrategy, UserTurnStartedP
from pipecat.turns.user_stop import BaseUserTurnStopStrategy, UserTurnStoppedParams
from pipecat.turns.user_turn_completion_mixin import UserTurnCompletionConfig
from pipecat.turns.user_turn_controller import UserTurnController
from pipecat.turns.user_turn_strategies import ExternalUserTurnStrategies, UserTurnStrategies
from pipecat.turns.user_turn_strategies import UserTurnStrategies
from pipecat.utils.context.llm_context_summarization import (
LLMAutoContextSummarizationConfig,
LLMContextSummarizationConfig,
@@ -468,11 +467,6 @@ class LLMUserAggregator(LLMContextAggregator):
self._vad_controller.add_event_handler("on_push_frame", self._on_push_frame)
self._vad_controller.add_event_handler("on_broadcast_frame", self._on_broadcast_frame)
# NOTE(aleix): Probably just needed temporarily. This was added to
# prevent processing self-queued frames (SpeechControlParamsFrame)
# pushed by strategies.
self._self_queued_frames = set()
async def cleanup(self):
"""Clean up processor resources."""
await super().cleanup()
@@ -528,8 +522,6 @@ class LLMUserAggregator(LLMContextAggregator):
await self.push_frame(frame, direction)
elif isinstance(frame, LLMSetToolChoiceFrame):
self.set_tool_choice(frame.tool_choice)
elif isinstance(frame, SpeechControlParamsFrame):
await self._handle_speech_control_params(frame)
else:
await self.push_frame(frame, direction)
@@ -643,17 +635,6 @@ class LLMUserAggregator(LLMContextAggregator):
if frame.run_llm:
await self.push_context_frame()
async def _handle_speech_control_params(self, frame: SpeechControlParamsFrame):
if frame.id in self._self_queued_frames:
return
if not frame.turn_params:
return
logger.warning(f"{self}: `turn_analyzer` in base input transport is deprecated.")
await self._user_turn_controller.update_strategies(ExternalUserTurnStrategies())
async def _handle_transcription(self, frame: TranscriptionFrame):
text = frame.text
@@ -668,16 +649,6 @@ class LLMUserAggregator(LLMContextAggregator):
)
)
async def _internal_queue_frame(
self,
frame: Frame,
direction: FrameDirection = FrameDirection.DOWNSTREAM,
callback: Optional[FrameCallback] = None,
):
"""Queues the given frame to ourselves."""
self._self_queued_frames.add(frame.id)
await self.queue_frame(frame, direction, callback)
async def _queued_broadcast_frame(self, frame_cls: Type[Frame], **kwargs):
"""Broadcasts a frame upstream and queues it for internal processing.
@@ -690,13 +661,13 @@ class LLMUserAggregator(LLMContextAggregator):
**kwargs: Keyword arguments to be passed to the frame's constructor.
"""
await self._internal_queue_frame(frame_cls(**kwargs))
await self.queue_frame(frame_cls(**kwargs))
await self.push_frame(frame_cls(**kwargs), FrameDirection.UPSTREAM)
async def _on_push_frame(
self, controller, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM
):
await self._internal_queue_frame(frame, direction)
await self.queue_frame(frame, direction)
async def _on_broadcast_frame(self, controller, frame_cls: Type[Frame], **kwargs):
await self._queued_broadcast_frame(frame_cls, **kwargs)

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.

13
tests/test_context_aggregators_universal.py
View File

@@ -28,6 +28,7 @@ from pipecat.frames.frames import (
LLMThoughtEndFrame,
LLMThoughtStartFrame,
LLMThoughtTextFrame,
SpeechControlParamsFrame,
StartFrame,
TextFrame,
TranscriptionFrame,
@@ -67,7 +68,7 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
pipeline = Pipeline([LLMUserAggregator(context)])
frames_to_send = [LLMRunFrame()]
expected_down_frames = [LLMContextFrame]
expected_down_frames = [SpeechControlParamsFrame, LLMContextFrame]
await run_test(
pipeline,
frames_to_send=frames_to_send,
@@ -110,7 +111,7 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
run_llm=True,
)
]
expected_down_frames = [LLMContextFrame]
expected_down_frames = [SpeechControlParamsFrame, LLMContextFrame]
await run_test(
pipeline,
frames_to_send=frames_to_send,
@@ -450,7 +451,7 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
(down_frames, _) = await run_test(
pipeline,
frames_to_send=[],
expected_down_frames=[StartFrame, UserMuteStartedFrame],
expected_down_frames=[StartFrame, UserMuteStartedFrame, SpeechControlParamsFrame],
ignore_start=False,
)
@@ -467,6 +468,7 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
# TranscriptionUserTurnStartStrategy, so we expect turn-related frames
# but NOT the InterimTranscriptionFrame itself.
expected_down_frames = [
SpeechControlParamsFrame,
UserStartedSpeakingFrame,
InterruptionFrame,
]
@@ -485,11 +487,12 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
frames_to_send = [
TranslationFrame(text="Hola!", user_id="", timestamp="now", language="es"),
]
# No downstream frames expected — translations are consumed.
# Only the SpeechControlParamsFrame from the default turn strategy on
# start is expected — the translation itself is consumed.
await run_test(
pipeline,
frames_to_send=frames_to_send,
expected_down_frames=[],
expected_down_frames=[SpeechControlParamsFrame],
)