5b67dcd9e7
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.
581 lines
23 KiB
Python
581 lines
23 KiB
Python
#
|
|
# Copyright (c) 2024-2026, Daily
|
|
#
|
|
# SPDX-License-Identifier: BSD 2-Clause License
|
|
#
|
|
|
|
"""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.
|
|
"""
|
|
|
|
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
|
|
|
|
AUDIO_INPUT_TIMEOUT_SECS = 0.5
|
|
|
|
|
|
class BaseInputTransport(FrameProcessor):
|
|
"""Base class for input transport implementations.
|
|
|
|
Handles audio and video input processing including Voice Activity Detection,
|
|
turn analysis, audio filtering, and user interaction management. Supports
|
|
interruption handling and provides hooks for transport-specific implementations.
|
|
"""
|
|
|
|
def __init__(self, params: TransportParams, **kwargs):
|
|
"""Initialize the base input transport.
|
|
|
|
Args:
|
|
params: Transport configuration parameters.
|
|
**kwargs: Additional arguments passed to parent class.
|
|
"""
|
|
super().__init__(**kwargs)
|
|
|
|
self._params = params
|
|
|
|
# Input sample rate. It will be initialized on StartFrame.
|
|
self._sample_rate = 0
|
|
|
|
# Track bot speaking state for interruption logic
|
|
self._bot_speaking = False
|
|
|
|
# Track user speaking state for interruption logic
|
|
self._user_speaking = False
|
|
# Last time a UserSpeakingFrame was pushed.
|
|
self._user_speaking_frame_time = 0
|
|
# How often a UserSpeakingFrame should be pushed (value should be
|
|
# greater than the audio chunks to have any effect).
|
|
self._user_speaking_frame_period = 0.2
|
|
|
|
# Task to process incoming audio (VAD) and push audio frames downstream
|
|
# if passthrough is enabled.
|
|
self._audio_task = None
|
|
|
|
# If the transport is stopped with `StopFrame` we might still be
|
|
# receiving frames from the transport but we really don't want to push
|
|
# 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.
|
|
|
|
Args:
|
|
enabled: Whether to start audio streaming immediately on transport start.
|
|
"""
|
|
logger.debug(f"Enabling audio on start. {enabled}")
|
|
self._params.audio_in_stream_on_start = enabled
|
|
|
|
async def start_audio_in_streaming(self):
|
|
"""Start audio input streaming.
|
|
|
|
Override in subclasses to implement transport-specific audio streaming.
|
|
"""
|
|
pass
|
|
|
|
@property
|
|
def sample_rate(self) -> int:
|
|
"""Get the current audio sample rate.
|
|
|
|
Returns:
|
|
The sample rate in Hz.
|
|
"""
|
|
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.
|
|
|
|
Args:
|
|
frame: The start frame containing initialization parameters.
|
|
"""
|
|
self._paused = False
|
|
self._user_speaking = False
|
|
|
|
self._sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate
|
|
|
|
# Start audio filter.
|
|
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.
|
|
|
|
Args:
|
|
frame: The end frame signaling transport shutdown.
|
|
"""
|
|
# Cancel and wait for the audio input task to finish.
|
|
await self._cancel_audio_task()
|
|
# Stop audio filter.
|
|
if self._params.audio_in_filter:
|
|
await self._params.audio_in_filter.stop()
|
|
|
|
async def pause(self, frame: StopFrame):
|
|
"""Pause the input transport temporarily.
|
|
|
|
Args:
|
|
frame: The stop frame signaling transport pause.
|
|
"""
|
|
self._paused = True
|
|
# Cancel task so we clear the queue
|
|
await self._cancel_audio_task()
|
|
# Retart the task
|
|
self._create_audio_task()
|
|
|
|
async def cancel(self, frame: CancelFrame):
|
|
"""Cancel the input transport and stop all processing.
|
|
|
|
Args:
|
|
frame: The cancel frame signaling immediate cancellation.
|
|
"""
|
|
# Cancel and wait for the audio input task to finish.
|
|
await self._cancel_audio_task()
|
|
# Stop audio filter.
|
|
if self._params.audio_in_filter:
|
|
await self._params.audio_in_filter.stop()
|
|
|
|
async def set_transport_ready(self, frame: StartFrame):
|
|
"""Called when the transport is ready to stream.
|
|
|
|
Args:
|
|
frame: The start frame containing initialization parameters.
|
|
"""
|
|
# Create audio input queue and task if needed.
|
|
self._create_audio_task()
|
|
|
|
async def push_video_frame(self, frame: InputImageRawFrame):
|
|
"""Push a video frame downstream if video input is enabled.
|
|
|
|
Args:
|
|
frame: The input video frame to process.
|
|
"""
|
|
if self._params.video_in_enabled and not self._paused:
|
|
await self.push_frame(frame)
|
|
|
|
async def push_audio_frame(self, frame: InputAudioRawFrame):
|
|
"""Push an audio frame to the processing queue if audio input is enabled.
|
|
|
|
Args:
|
|
frame: The input audio frame to process.
|
|
"""
|
|
if self._params.audio_in_enabled and not self._paused:
|
|
await self._audio_in_queue.put(frame)
|
|
|
|
#
|
|
# Frame processor
|
|
#
|
|
|
|
async def process_frame(self, frame: Frame, direction: FrameDirection):
|
|
"""Process incoming frames and handle transport-specific logic.
|
|
|
|
Args:
|
|
frame: The frame to process.
|
|
direction: The direction of frame flow in the pipeline.
|
|
"""
|
|
await super().process_frame(frame, direction)
|
|
|
|
# Specific system frames
|
|
if isinstance(frame, StartFrame):
|
|
# Push StartFrame before start(), because we want StartFrame to be
|
|
# processed by every processor before any other frame is processed.
|
|
await self.push_frame(frame, direction)
|
|
await self.start(frame)
|
|
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)
|
|
# Control frames
|
|
elif isinstance(frame, EndFrame):
|
|
# Push EndFrame before stop(), because stop() waits on the task to
|
|
# finish and the task finishes when EndFrame is processed.
|
|
await self.push_frame(frame, direction)
|
|
await self.stop(frame)
|
|
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
|
|
else:
|
|
await self.push_frame(frame, direction)
|
|
|
|
#
|
|
# Audio input
|
|
#
|
|
|
|
def _create_audio_task(self):
|
|
"""Create the audio processing task if audio input is enabled."""
|
|
if not self._audio_task and self._params.audio_in_enabled:
|
|
self._audio_in_queue = asyncio.Queue()
|
|
self._audio_task = self.create_task(self._audio_task_handler())
|
|
|
|
async def _cancel_audio_task(self):
|
|
"""Cancel and cleanup the audio processing task."""
|
|
if self._audio_task:
|
|
await self.cancel_task(self._audio_task)
|
|
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
|
|
# Skip timeout handling until the first audio frame arrives (e.g. client
|
|
# not yet connected).
|
|
audio_received = False
|
|
while True:
|
|
try:
|
|
frame: InputAudioRawFrame = await asyncio.wait_for(
|
|
self._audio_in_queue.get(), timeout=AUDIO_INPUT_TIMEOUT_SECS
|
|
)
|
|
|
|
# From now on, timeout should warn if there's no audio.
|
|
audio_received = True
|
|
|
|
# If an audio filter is available, run it before VAD.
|
|
if self._params.audio_in_filter:
|
|
frame.audio = await self._params.audio_in_filter.filter(frame.audio)
|
|
|
|
# Skip frames with no audio data (e.g. filter is buffering).
|
|
if not frame.audio:
|
|
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)
|
|
|
|
self._audio_in_queue.task_done()
|
|
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()
|