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

Merge main into feature/genesys_serializer

Browse Source 
Incorporates latest changes from main branch including:
- AIC filter and VAD updates
- STT service improvements
- Base serializer changes
- Various bug fixes
This commit is contained in:
ssillerom
2026-01-28 10:48:56 +01:00
parent 55e0d4ecc4 bf9b166464
commit 105824a372
34 changed files with 1362 additions and 352 deletions
Download Patch File Download Diff File Expand all files Collapse all files

310
src/pipecat/audio/filters/aic_filter.py
View File

@@ -9,129 +9,145 @@
This module provides an audio filter implementation using ai-coustics' AIC SDK to
enhance audio streams in real time. It mirrors the structure of other filters like
the Koala filter and integrates with Pipecat's input transport pipeline.
Classes:
AICFilter: For aic-sdk (uses 'aic_sdk' module)
"""
from pathlib import Path
from typing import List, Optional
import numpy as np
from aic_sdk import (
Model,
ParameterFixedError,
ProcessorAsync,
ProcessorConfig,
ProcessorParameter,
set_sdk_id,
)
from loguru import logger
from pipecat.audio.filters.base_audio_filter import BaseAudioFilter
from pipecat.audio.vad.aic_vad import AICVADAnalyzer
from pipecat.frames.frames import FilterControlFrame, FilterEnableFrame
try:
# AIC SDK (https://ai-coustics.github.io/aic-sdk-py/api/)
from aic import AICModelType, AICParameter, Model
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error("In order to use the AIC filter, you need to `pip install pipecat-ai[aic]`.")
raise Exception(f"Missing module: {e}")
class AICFilter(BaseAudioFilter):
"""Audio filter using ai-coustics' AIC SDK for real-time enhancement.
Buffers incoming audio to the model's preferred block size and processes
planar frames in-place using float32 samples in the linear -1..+1 range.
frames using float32 samples normalized to the range -1 to +1.
"""
def __init__(
self,
*,
license_key: str = "",
model_type: AICModelType = AICModelType.QUAIL_STT,
enhancement_level: Optional[float] = 1.0,
voice_gain: Optional[float] = 1.0,
noise_gate_enable: Optional[bool] = True,
license_key: str,
model_id: Optional[str] = None,
model_path: Optional[Path] = None,
model_download_dir: Optional[Path] = None,
) -> None:
"""Initialize the AIC filter.
Args:
license_key: ai-coustics license key for authentication.
model_type: Model variant to load.
enhancement_level: Optional overall enhancement strength (0.0..1.0).
voice_gain: Optional linear gain applied to detected speech (0.0..4.0).
noise_gate_enable: Optional enable/disable noise gate (default: True).
model_id: Model identifier to download from CDN. Required if model_path
is not provided. See https://artifacts.ai-coustics.io/ for available models.
model_path: Optional path to a local .aicmodel file. If provided,
model_id is ignored and no download occurs.
model_download_dir: Directory for downloading models as a Path object.
Defaults to a cache directory in user's home folder.
.. deprecated:: 1.3.0
The `noise_gate_enable` parameter is deprecated and no longer has any effect.
It will be removed in a future version.
Raises:
ValueError: If neither model_id nor model_path is provided.
"""
# Set SDK ID for telemetry identification (6 = pipecat)
set_sdk_id(6)
if model_id is None and model_path is None:
raise ValueError(
"Either 'model_id' or 'model_path' must be provided. "
"See https://artifacts.ai-coustics.io/ for available models."
)
self._license_key = license_key
self._model_type = model_type
self._model_id = model_id
self._model_path = model_path
self._model_download_dir = model_download_dir or (
Path.home() / ".cache" / "pipecat" / "aic-models"
)
self._enhancement_level = enhancement_level
self._voice_gain = voice_gain
if noise_gate_enable is not None:
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"Parameter `noise_gate_enable` is deprecated and no longer has any effect. "
"It will be removed in a future version. Use AIC VAD instead (create_vad_analyzer()).",
DeprecationWarning,
)
self._noise_gate_enable = noise_gate_enable
self._enabled = True
self._bypass = False
self._sample_rate = 0
self._aic_ready = False
self._frames_per_block = 0
self._audio_buffer = bytearray()
# Model will be created in start() since the API now requires sample_rate
self._aic = None
def get_vad_factory(self):
"""Return a zero-arg factory that will create the VAD once the model exists.
# Audio format constants
self._bytes_per_sample = 2 # int16 = 2 bytes
self._dtype = np.int16
self._scale = (
32768.0 # 2^15, for normalizing int16 (-32768 to 32767) to float32 (-1.0 to 1.0)
)
# AIC SDK objects
self._model = None
self._processor = None
self._processor_ctx = None
self._vad_ctx = None
# Pre-allocated buffers (resized in start() once frames_per_block is known)
self._in_f32 = None
self._out_i16 = None
def get_vad_context(self):
"""Return the VAD context once the processor exists.
Returns:
A zero-argument callable that, when invoked, returns an initialized
VoiceActivityDetector bound to the underlying AIC model. Raises a
RuntimeError if the model has not been initialized (i.e. start()
has not been called successfully).
The VadContext instance bound to the underlying processor.
Raises RuntimeError if the processor has not been initialized.
"""
def _factory():
if self._aic is None:
raise RuntimeError("AIC model not initialized yet. Call start(sample_rate) first.")
return self._aic.create_vad()
return _factory
if self._vad_ctx is None:
raise RuntimeError("AIC processor not initialized yet. Call start(sample_rate) first.")
return self._vad_ctx
def create_vad_analyzer(
self,
*,
lookback_buffer_size: Optional[float] = None,
speech_hold_duration: Optional[float] = None,
minimum_speech_duration: Optional[float] = None,
sensitivity: Optional[float] = None,
):
"""Return an analyzer that will lazily instantiate the AIC VAD when ready.
AIC VAD parameters:
- lookback_buffer_size:
Number of window-length audio buffers used as a lookback buffer.
Higher values increase prediction stability but add latency.
Range: 1.0 .. 20.0, Default (SDK): 6.0
- speech_hold_duration:
How long VAD continues detecting after speech ends (in seconds).
Range: 0.0 to 100x model window length, Default (SDK): 0.05s
- minimum_speech_duration:
Minimum duration of speech required before VAD reports speech detected
(in seconds). Range: 0.0 to 1.0, Default (SDK): 0.0s
- sensitivity:
Energy threshold sensitivity. Energy threshold = 10 ** (-sensitivity).
Range: 1.0 .. 15.0, Default (SDK): 6.0
Range: 1.0 to 15.0, Default (SDK): 6.0
Args:
lookback_buffer_size: Optional lookback buffer size to configure on the VAD.
Range: 1.0 .. 20.0. If None, SDK default is used.
speech_hold_duration: Optional speech hold duration to configure on the VAD.
If None, SDK default (0.05s) is used.
minimum_speech_duration: Optional minimum speech duration before VAD reports
speech detected. If None, SDK default (0.0s) is used.
sensitivity: Optional sensitivity (energy threshold) to configure on the VAD.
Range: 1.0 .. 15.0. If None, SDK default is used.
Range: 1.0 to 15.0. If None, SDK default (6.0) is used.
Returns:
A lazily-initialized AICVADAnalyzer that will bind to the VAD backend
once the filter's model has been created (after start(sample_rate)).
A lazily-initialized AICVADAnalyzer that will bind to the VAD context
once the filter's processor has been created (after start(sample_rate)).
"""
from pipecat.audio.vad.aic_vad import AICVADAnalyzer
return AICVADAnalyzer(
vad_factory=self.get_vad_factory(),
lookback_buffer_size=lookback_buffer_size,
vad_context_factory=lambda: self.get_vad_context(),
speech_hold_duration=speech_hold_duration,
minimum_speech_duration=minimum_speech_duration,
sensitivity=sensitivity,
)
@@ -146,52 +162,83 @@ class AICFilter(BaseAudioFilter):
"""
self._sample_rate = sample_rate
# Load or download model
if self._model_path:
logger.debug(f"Loading AIC model from: {self._model_path}")
self._model = Model.from_file(str(self._model_path))
else:
logger.debug(f"Downloading AIC model: {self._model_id}")
self._model_download_dir.mkdir(parents=True, exist_ok=True)
model_path = await Model.download_async(self._model_id, str(self._model_download_dir))
logger.debug(f"Model downloaded to: {model_path}")
self._model = Model.from_file(model_path)
# Get optimal frames for this sample rate
self._frames_per_block = self._model.get_optimal_num_frames(self._sample_rate)
# Allocate processing buffers now that we know the block size
self._in_f32 = np.zeros((1, self._frames_per_block), dtype=np.float32)
self._out_i16 = np.zeros(self._frames_per_block, dtype=np.int16)
# Create configuration
config = ProcessorConfig.optimal(
self._model,
sample_rate=self._sample_rate,
)
# Create async processor
try:
# Create model with required runtime parameters
self._aic = Model(
model_type=self._model_type,
license_key=self._license_key or None,
sample_rate=self._sample_rate,
channels=1,
)
self._frames_per_block = self._aic.optimal_num_frames()
# Optional parameter configuration
if self._enhancement_level is not None:
self._aic.set_parameter(
AICParameter.ENHANCEMENT_LEVEL,
float(self._enhancement_level if self._enabled else 0.0),
)
if self._voice_gain is not None:
self._aic.set_parameter(AICParameter.VOICE_GAIN, float(self._voice_gain))
self._aic_ready = True
# Log processor information
logger.debug(f"ai-coustics filter started:")
logger.debug(f" Sample rate: {self._sample_rate} Hz")
logger.debug(f" Frames per chunk: {self._frames_per_block}")
logger.debug(f" Enhancement strength: {int(self._enhancement_level * 100)}%")
logger.debug(f" Optimal input buffer size: {self._aic.optimal_num_frames()} samples")
logger.debug(f" Optimal sample rate: {self._aic.optimal_sample_rate()} Hz")
logger.debug(
f" Current algorithmic latency: {self._aic.processing_latency() / self._sample_rate * 1000:.2f}ms"
)
self._processor = ProcessorAsync(self._model, self._license_key, config)
except Exception as e: # noqa: BLE001 - surfacing SDK initialization errors
logger.error(f"AIC model initialization failed: {e}")
self._aic_ready = False
self._processor = None
self._aic_ready = self._processor is not None
if not self._aic_ready:
logger.debug(f"ai-coustics filter is not ready.")
return
# Get contexts for parameter control and VAD
self._processor_ctx = self._processor.get_processor_context()
self._vad_ctx = self._processor.get_vad_context()
# Apply initial parameters
try:
self._processor_ctx.set_parameter(
ProcessorParameter.Bypass, 1.0 if self._bypass else 0.0
)
except ParameterFixedError as e:
logger.error(f"AIC parameter update failed: {e}")
# Log processor information
logger.debug(f"ai-coustics filter started:")
logger.debug(f" Model ID: {self._model.get_id()}")
logger.debug(f" Sample rate: {self._sample_rate} Hz")
logger.debug(f" Frames per chunk: {self._frames_per_block}")
logger.debug(f" Optimal sample rate: {self._model.get_optimal_sample_rate()} Hz")
logger.debug(
f" Optimal number of frames for {self._sample_rate} Hz: {self._model.get_optimal_num_frames(self._sample_rate)}"
)
logger.debug(
f" Output delay: {self._processor_ctx.get_output_delay()} samples "
f"({self._processor_ctx.get_output_delay() / self._sample_rate * 1000:.2f}ms)"
)
async def stop(self):
"""Clean up the AIC model when stopping.
"""Clean up the AIC processor when stopping.
Returns:
None
"""
try:
if self._aic is not None:
self._aic.close()
if self._processor_ctx is not None:
self._processor_ctx.reset()
finally:
self._aic = None
self._processor = None
self._processor_ctx = None
self._vad_ctx = None
self._model = None
self._aic_ready = False
self._audio_buffer.clear()
@@ -205,11 +252,12 @@ class AICFilter(BaseAudioFilter):
None
"""
if isinstance(frame, FilterEnableFrame):
self._enabled = frame.enable
if self._aic is not None:
self._bypass = not frame.enable
if self._processor_ctx is not None:
try:
level = float(self._enhancement_level if self._enabled else 0.0)
self._aic.set_parameter(AICParameter.ENHANCEMENT_LEVEL, level)
self._processor_ctx.set_parameter(
ProcessorParameter.Bypass, 1.0 if self._bypass else 0.0
)
except Exception as e: # noqa: BLE001
logger.error(f"AIC set_parameter failed: {e}")
@@ -220,43 +268,41 @@ class AICFilter(BaseAudioFilter):
model's required block length. Returns enhanced audio data.
Args:
audio: Raw audio data as bytes to be filtered (int16 PCM, planar).
audio: Raw audio data as bytes (int16 PCM).
Returns:
Enhanced audio data as bytes (int16 PCM, planar).
Enhanced audio data as bytes (int16 PCM).
"""
if not self._aic_ready or self._aic is None:
if not self._aic_ready or self._processor is None:
return audio
self._audio_buffer.extend(audio)
available_frames = len(self._audio_buffer) // self._bytes_per_sample
num_blocks = available_frames // self._frames_per_block
if num_blocks == 0:
return b""
filtered_chunks: List[bytes] = []
mv = memoryview(self._audio_buffer)
block_size = self._frames_per_block * self._bytes_per_sample
# Number of int16 samples currently buffered
available_frames = len(self._audio_buffer) // 2
for i in range(num_blocks):
start = i * block_size
block_i16 = np.frombuffer(mv[start : start + block_size], dtype=self._dtype)
while available_frames >= self._frames_per_block:
# Consume exactly one block worth of frames
samples_to_consume = self._frames_per_block * 1
bytes_to_consume = samples_to_consume * 2
block_bytes = bytes(self._audio_buffer[:bytes_to_consume])
# Reuse input buffer, in-place divide
np.copyto(self._in_f32[0], block_i16)
self._in_f32 /= self._scale
# Convert to float32 in -1..+1 range and reshape to planar (channels, frames)
block_i16 = np.frombuffer(block_bytes, dtype=np.int16)
block_f32 = (block_i16.astype(np.float32) / 32768.0).reshape(
(1, self._frames_per_block)
)
out_f32 = await self._processor.process_async(self._in_f32)
# Process planar in-place; returns ndarray (same shape)
out_f32 = await self._aic.process_async(block_f32)
# Convert float32 output back to int16
np.multiply(out_f32, self._scale, out=self._in_f32) # reuse in_f32 as temp
np.clip(self._in_f32, -self._scale, self._scale - 1, out=self._in_f32)
np.copyto(self._out_i16, self._in_f32[0].astype(self._dtype))
# Convert back to int16 bytes, planar layout
out_i16 = np.clip(out_f32 * 32768.0, -32768, 32767).astype(np.int16)
filtered_chunks.append(out_i16.reshape(-1).tobytes())
filtered_chunks.append(self._out_i16.tobytes())
# Slide buffer
self._audio_buffer = self._audio_buffer[bytes_to_consume:]
available_frames = len(self._audio_buffer) // 2
# Do not flush incomplete frames; keep them buffered for the next call
self._audio_buffer = self._audio_buffer[num_blocks * block_size :]
return b"".join(filtered_chunks)

153
src/pipecat/audio/vad/aic_vad.py
View File

@@ -1,44 +1,44 @@
"""AIC-integrated VAD analyzer that lazily binds to the AIC SDK backend.
This analyzer queries the backend's is_speech_detected() and maps it to a float
confidence (1.0/0.0). It uses 10 ms windows based on the sample rate and applies
optional AIC VAD parameters (lookback_buffer_size, sensitivity) when available.
This module provides VAD analyzer implementations that query the AIC SDK's
is_speech_detected() and map it to a float confidence (1.0/0.0).
Classes:
AICVADAnalyzer: For aic-sdk (uses 'aic_sdk' module)
"""
from typing import Any, Callable, Optional
from aic_sdk import VadParameter
from loguru import logger
from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADParams
try:
from aic import AICVadParameter
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error("In order to use the AIC filter, you need to `pip install pipecat-ai[aic]`.")
raise Exception(f"Missing module: {e}")
class AICVADAnalyzer(VADAnalyzer):
"""VAD analyzer that lazily instantiates the AIC VoiceActivityDetector via a factory.
"""VAD analyzer that lazily binds to the AIC VadContext via a factory.
The analyzer can be constructed before the AIC Model exists. Once the filter has
started and the Model is available, the provided factory will succeed and the
backend VAD will be created. We then switch to single-sample updates where
num_frames_required() returns 1 and confidence is derived from the backend's
boolean is_speech_detected() state.
The analyzer can be constructed before the AIC Processor exists. Once the filter has
started and the Processor is available, the provided factory will succeed and the
VadContext will be obtained. The context's is_speech_detected() boolean state is
then mapped to 1.0 (speech) or 0.0 (no speech) to satisfy the VADAnalyzer interface.
AIC VAD runtime parameters:
- lookback_buffer_size:
Controls the lookback buffer size used by the VAD, i.e. the number of
window-length audio buffers used as a lookback buffer. Larger values improve
stability but increase latency.
Range: 1.0 .. 20.0
Default (SDK): 6.0
- speech_hold_duration:
Controls for how long the VAD continues to detect speech after the audio signal
no longer contains speech (in seconds).
Range: 0.0 to 100x model window length
Default (SDK): 0.05s (50ms)
- minimum_speech_duration:
Controls for how long speech needs to be present in the audio signal before the
VAD considers it speech (in seconds).
Range: 0.0 to 1.0
Default (SDK): 0.0s
- sensitivity:
Controls the energy threshold sensitivity. Higher values make the detector
less sensitive (require more energy to count as speech).
Range: 1.0 .. 15.0
Controls the sensitivity (energy threshold) of the VAD. This value is used by
the VAD as the threshold a speech audio signal's energy has to exceed in order
to be considered speech.
Range: 1.0 to 15.0
Formula: Energy threshold = 10 ** (-sensitivity)
Default (SDK): 6.0
"""
@@ -46,69 +46,80 @@ class AICVADAnalyzer(VADAnalyzer):
def __init__(
self,
*,
vad_factory: Optional[Callable[[], Any]] = None,
lookback_buffer_size: Optional[float] = None,
vad_context_factory: Optional[Callable[[], Any]] = None,
speech_hold_duration: Optional[float] = None,
minimum_speech_duration: Optional[float] = None,
sensitivity: Optional[float] = None,
):
"""Create an AIC VAD analyzer.
Args:
vad_factory:
Zero-arg callable that returns an initialized AIC VoiceActivityDetector.
This may raise until the filter's Model has been created; the analyzer
vad_context_factory:
Zero-arg callable that returns the AIC VadContext.
This may raise until the filter's Processor has been created; the analyzer
will retry on set_sample_rate/first use.
lookback_buffer_size:
Optional override for AIC VAD lookback buffer size.
Range: 1.0 .. 20.0. Larger values increase stability at the cost of latency.
If None, the SDK default (6.0) is used.
speech_hold_duration:
Optional override for AIC VAD speech hold duration (in seconds).
Range: 0.0 to 100x model window length.
If None, the SDK default (0.05s) is used.
minimum_speech_duration:
Optional override for minimum speech duration before VAD reports
speech detected (in seconds).
Range: 0.0 to 1.0.
If None, the SDK default (0.0s) is used.
sensitivity:
Optional override for AIC VAD sensitivity (energy threshold).
Range: 1.0 .. 15.0. Energy threshold = 10 ** (-sensitivity).
Range: 1.0 to 15.0. Energy threshold = 10 ** (-sensitivity).
If None, the SDK default (6.0) is used.
"""
# Use fixed VAD parameters for AIC: no user override
fixed_params = VADParams(confidence=0.5, start_secs=0.0, stop_secs=0.0, min_volume=0.0)
super().__init__(sample_rate=None, params=fixed_params)
self._vad_factory = vad_factory
self._backend_vad: Optional[Any] = None
self._pending_lookback: Optional[float] = lookback_buffer_size
self._vad_context_factory = vad_context_factory
self._vad_ctx: Optional[Any] = None
self._pending_speech_hold_duration: Optional[float] = speech_hold_duration
self._pending_minimum_speech_duration: Optional[float] = minimum_speech_duration
self._pending_sensitivity: Optional[float] = sensitivity
def bind_vad_factory(self, vad_factory: Callable[[], Any]):
def bind_vad_context_factory(self, vad_context_factory: Callable[[], Any]):
"""Attach or replace the factory post-construction."""
self._vad_factory = vad_factory
self._ensure_backend_initialized()
self._vad_context_factory = vad_context_factory
self._ensure_vad_context_initialized()
def _apply_backend_params(self):
def _apply_vad_params(self):
"""Apply optional AIC VAD parameters if available."""
if self._backend_vad is None or AICVadParameter is None:
if self._vad_ctx is None or VadParameter is None:
return
try:
if self._pending_lookback is not None:
self._backend_vad.set_parameter(
AICVadParameter.LOOKBACK_BUFFER_SIZE, float(self._pending_lookback)
if self._pending_speech_hold_duration is not None:
self._vad_ctx.set_parameter(
VadParameter.SpeechHoldDuration, self._pending_speech_hold_duration
)
if self._pending_minimum_speech_duration is not None:
self._vad_ctx.set_parameter(
VadParameter.MinimumSpeechDuration, self._pending_minimum_speech_duration
)
if self._pending_sensitivity is not None:
self._backend_vad.set_parameter(
AICVadParameter.SENSITIVITY, float(self._pending_sensitivity)
)
self._vad_ctx.set_parameter(VadParameter.Sensitivity, self._pending_sensitivity)
except Exception as e: # noqa: BLE001
logger.debug(f"AIC VAD parameter application deferred/failed: {e}")
def _ensure_backend_initialized(self):
if self._backend_vad is not None:
def _ensure_vad_context_initialized(self):
if self._vad_ctx is not None:
return
if not self._vad_factory:
if not self._vad_context_factory:
return
try:
self._backend_vad = self._vad_factory()
self._apply_backend_params()
# With backend ready, recompute internal frame sizing
self._vad_ctx = self._vad_context_factory()
self._apply_vad_params()
# With VAD context ready, recompute internal frame sizing
super().set_params(self._params)
logger.debug("AIC VAD backend initialized in analyzer.")
logger.debug("AIC VAD context initialized in analyzer.")
except Exception as e: # noqa: BLE001
# Filter may not be started yet; try again later
logger.debug(f"Deferring AIC VAD backend initialization: {e}")
logger.debug(f"Deferring AIC VAD context initialization: {e}")
def set_sample_rate(self, sample_rate: int):
"""Set the sample rate for audio processing.
@@ -116,10 +127,10 @@ class AICVADAnalyzer(VADAnalyzer):
Args:
sample_rate: Audio sample rate in Hz.
"""
# Set rate and attempt backend initialization once we know SR
# Set rate and attempt VAD context initialization once we know SR
self._sample_rate = self._init_sample_rate or sample_rate
self._ensure_backend_initialized()
# Ensure params are initialized even if backend not ready yet
self._ensure_vad_context_initialized()
# Ensure params are initialized even if VAD context not ready yet
try:
super().set_params(self._params)
except Exception:
@@ -135,23 +146,29 @@ class AICVADAnalyzer(VADAnalyzer):
return int(self.sample_rate * 0.01) if self.sample_rate > 0 else 160
def voice_confidence(self, buffer: bytes) -> float:
"""Calculate voice activity confidence for the given audio buffer.
"""Return voice activity detection result for the given audio buffer.
Note:
The AIC SDK provides binary speech detection (not a probability score).
This method returns 1.0 when speech is detected and 0.0 otherwise,
rather than a true confidence value.
Args:
buffer: Audio buffer to analyze.
buffer: Audio buffer (unused - AIC VAD state is updated internally
by the enhancement pipeline).
Returns:
Voice confidence score is 0.0 or 1.0.
1.0 if speech is detected, 0.0 otherwise.
"""
# Ensure backend exists (filter might have started since last call)
self._ensure_backend_initialized()
if self._backend_vad is None:
# Ensure VAD context exists (filter might have started since last call)
self._ensure_vad_context_initialized()
if self._vad_ctx is None:
return 0.0
# We do not need to analyze 'buffer' here since the model's VAD is updated
# We do not need to analyze 'buffer' here since the processor's VAD is updated
# as part of the enhancement pipeline. Simply query the boolean and map it.
try:
is_speech = self._backend_vad.is_speech_detected()
is_speech = self._vad_ctx.is_speech_detected()
return 1.0 if is_speech else 0.0
except Exception as e: # noqa: BLE001
logger.error(f"AIC VAD inference error: {e}")

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

@@ -464,9 +464,11 @@ class LLMUserAggregator(LLMContextAggregator):
await s.setup(self.task_manager)
async def _stop(self, frame: EndFrame):
await self._maybe_emit_user_turn_stopped(on_session_end=True)
await self._cleanup()
async def _cancel(self, frame: CancelFrame):
await self._maybe_emit_user_turn_stopped(on_session_end=True)
await self._cleanup()
async def _cleanup(self):
@@ -602,14 +604,7 @@ class LLMUserAggregator(LLMContextAggregator):
if params.enable_user_speaking_frames:
await self.broadcast_frame(UserStoppedSpeakingFrame)
# Always push context frame.
aggregation = await self.push_aggregation()
message = UserTurnStoppedMessage(
content=aggregation, timestamp=self._user_turn_start_timestamp
)
await self._call_event_handler("on_user_turn_stopped", strategy, message)
self._user_turn_start_timestamp = ""
await self._maybe_emit_user_turn_stopped(strategy)
async def _on_user_turn_stop_timeout(self, controller):
await self._call_event_handler("on_user_turn_stop_timeout")
@@ -617,6 +612,26 @@ class LLMUserAggregator(LLMContextAggregator):
async def _on_user_turn_idle(self, controller):
await self._call_event_handler("on_user_turn_idle")
async def _maybe_emit_user_turn_stopped(
self,
strategy: Optional[BaseUserTurnStopStrategy] = None,
on_session_end: bool = False,
):
"""Maybe emit user turn stopped event.
Args:
strategy: The strategy that triggered the turn stop.
on_session_end: If True, only emit if there's unemitted content
(avoids duplicate events when session ends).
"""
aggregation = await self.push_aggregation()
if not on_session_end or aggregation:
message = UserTurnStoppedMessage(
content=aggregation, timestamp=self._user_turn_start_timestamp
)
await self._call_event_handler("on_user_turn_stopped", strategy, message)
self._user_turn_start_timestamp = ""
class LLMAssistantAggregator(LLMContextAggregator):
"""Assistant LLM aggregator that processes bot responses and function calls.
@@ -739,6 +754,9 @@ class LLMAssistantAggregator(LLMContextAggregator):
if isinstance(frame, InterruptionFrame):
await self._handle_interruptions(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, (EndFrame, CancelFrame)):
await self._handle_end_or_cancel(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, LLMFullResponseStartFrame):
await self._handle_llm_start(frame)
elif isinstance(frame, LLMFullResponseEndFrame):
@@ -813,6 +831,10 @@ class LLMAssistantAggregator(LLMContextAggregator):
self._started = 0
await self.reset()
async def _handle_end_or_cancel(self, frame: Frame):
await self._trigger_assistant_turn_stopped()
self._started = 0
async def _handle_function_calls_started(self, frame: FunctionCallsStartedFrame):
function_names = [f"{f.function_name}:{f.tool_call_id}" for f in frame.function_calls]
logger.debug(f"{self} FunctionCallsStartedFrame: {function_names}")
@@ -833,7 +855,7 @@ class LLMAssistantAggregator(LLMContextAggregator):
"id": frame.tool_call_id,
"function": {
"name": frame.function_name,
"arguments": json.dumps(frame.arguments),
"arguments": json.dumps(frame.arguments, ensure_ascii=False),
},
"type": "function",
}
@@ -866,7 +888,7 @@ class LLMAssistantAggregator(LLMContextAggregator):
# Update context with the function call result
if frame.result:
result = json.dumps(frame.result)
result = json.dumps(frame.result, ensure_ascii=False)
self._update_function_call_result(frame.function_name, frame.tool_call_id, result)
else:
self._update_function_call_result(frame.function_name, frame.tool_call_id, "COMPLETED")

80
src/pipecat/processors/audio/audio_buffer_processor.py
View File

@@ -11,7 +11,6 @@ of audio from both user input and bot output sources, with support for various a
configurations and event-driven processing.
"""
import time
from typing import Optional
from pipecat.audio.utils import create_stream_resampler, interleave_stereo_audio, mix_audio
@@ -104,10 +103,6 @@ class AudioBufferProcessor(FrameProcessor):
self._user_turn_audio_buffer = bytearray()
self._bot_turn_audio_buffer = bytearray()
# Intermittent (non continous user stream variables)
self._last_user_frame_at = 0
self._last_bot_frame_at = 0
self._recording = False
self._input_resampler = create_stream_resampler()
@@ -211,23 +206,31 @@ class AudioBufferProcessor(FrameProcessor):
"""Process audio frames for recording."""
resampled = None
if isinstance(frame, InputAudioRawFrame):
# Add silence if we need to.
silence = self._compute_silence(self._last_user_frame_at)
self._user_audio_buffer.extend(silence)
# Add user audio.
resampled = await self._resample_input_audio(frame)
self._user_audio_buffer.extend(resampled)
# Save time of frame so we can compute silence.
self._last_user_frame_at = time.time()
# Ignoring in case we don't have audio
if len(resampled) > 0:
# Sync bot buffer to current user position before adding user audio.
# We sync BEFORE extending to align both buffers at the same starting timestamp.
# For example, user buffer is at 100 bytes, and you receive 20 bytes of new audio
# - Bot buffer sees User is at 100. Bot pads itself to 100.
# - User buffer adds 20. User is now at 120.
# - Outcome: At index 100-120, we have User Audio and (potentially) Bot Audio or silence. They are aligned
# This gives the opportunity to the bot to send audio.
#
# If we synced AFTER, we'd pad the bot buffer with silence for the same
# window we just gave to the user, effectively "overwriting" that time slot
# with silence and causing the bot's audio to flicker or cut out.
self._sync_buffer_to_position(self._bot_audio_buffer, len(self._user_audio_buffer))
# Add user audio.
self._user_audio_buffer.extend(resampled)
elif self._recording and isinstance(frame, OutputAudioRawFrame):
# Add silence if we need to.
silence = self._compute_silence(self._last_bot_frame_at)
self._bot_audio_buffer.extend(silence)
# Add bot audio.
resampled = await self._resample_output_audio(frame)
self._bot_audio_buffer.extend(resampled)
# Save time of frame so we can compute silence.
self._last_bot_frame_at = time.time()
# Ignoring in case we don't have audio
if len(resampled) > 0:
# Sync user buffer to current bot position before adding bot audio
self._sync_buffer_to_position(self._user_audio_buffer, len(self._bot_audio_buffer))
# Add bot audio.
self._bot_audio_buffer.extend(resampled)
if self._buffer_size > 0 and (
len(self._user_audio_buffer) >= self._buffer_size
@@ -240,6 +243,21 @@ class AudioBufferProcessor(FrameProcessor):
if self._enable_turn_audio:
await self._process_turn_recording(frame, resampled)
def _sync_buffer_to_position(self, buffer: bytearray, target_position: int):
"""Pad buffer with silence if it's behind the target position.
This ensures both buffers stay synchronized by padding the lagging
buffer before new audio is added to the other buffer.
Args:
buffer: The buffer to potentially pad.
target_position: The position (in bytes) the buffer should reach.
"""
current_len = len(buffer)
if current_len < target_position:
silence_needed = target_position - current_len
buffer.extend(b"\x00" * silence_needed)
async def _process_turn_recording(self, frame: Frame, resampled_audio: Optional[bytes] = None):
"""Process frames for turn-based audio recording."""
if isinstance(frame, UserStartedSpeakingFrame):
@@ -281,8 +299,8 @@ class AudioBufferProcessor(FrameProcessor):
if len(self._user_audio_buffer) == 0 and len(self._bot_audio_buffer) == 0:
return
# Final alignment before we send the audio
self._align_track_buffers()
flush_time = time.time()
# Call original handler with merged audio
merged_audio = self.merge_audio_buffers()
@@ -299,9 +317,6 @@ class AudioBufferProcessor(FrameProcessor):
self._num_channels,
)
self._last_user_frame_at = flush_time
self._last_bot_frame_at = flush_time
def _buffer_has_audio(self, buffer: bytearray) -> bool:
"""Check if a buffer contains audio data."""
return buffer is not None and len(buffer) > 0
@@ -309,8 +324,6 @@ class AudioBufferProcessor(FrameProcessor):
def _reset_recording(self):
"""Reset recording state and buffers."""
self._reset_all_audio_buffers()
self._last_user_frame_at = time.time()
self._last_bot_frame_at = time.time()
def _reset_all_audio_buffers(self):
"""Reset all audio buffers to empty state."""
@@ -336,11 +349,9 @@ class AudioBufferProcessor(FrameProcessor):
target_len = max(user_len, bot_len)
if user_len < target_len:
self._user_audio_buffer.extend(b"\x00" * (target_len - user_len))
self._last_user_frame_at = max(self._last_user_frame_at, self._last_bot_frame_at)
self._sync_buffer_to_position(self._user_audio_buffer, target_len)
if bot_len < target_len:
self._bot_audio_buffer.extend(b"\x00" * (target_len - bot_len))
self._last_bot_frame_at = max(self._last_bot_frame_at, self._last_user_frame_at)
self._sync_buffer_to_position(self._bot_audio_buffer, target_len)
async def _resample_input_audio(self, frame: InputAudioRawFrame) -> bytes:
"""Resample audio frame to the target sample rate."""
@@ -353,14 +364,3 @@ class AudioBufferProcessor(FrameProcessor):
return await self._output_resampler.resample(
frame.audio, frame.sample_rate, self._sample_rate
)
def _compute_silence(self, from_time: float) -> bytes:
"""Compute silence to insert based on time gap."""
quiet_time = time.time() - from_time
# We should get audio frames very frequently. We introduce silence only
# if there's a big enough gap of 1s.
if from_time == 0 or quiet_time < 1.0:
return b""
num_bytes = int(quiet_time * self._sample_rate) * 2
silence = b"\x00" * num_bytes
return silence

3
src/pipecat/serializers/base_serializer.py
View File

@@ -9,9 +9,10 @@
from abc import ABC, abstractmethod
from pipecat.frames.frames import Frame, StartFrame
from pipecat.utils.base_object import BaseObject
class FrameSerializer(ABC):
class FrameSerializer(BaseObject):
"""Abstract base class for frame serialization implementations.
Defines the interface for converting frames to/from serialized formats

49
src/pipecat/services/azure/tts.py
View File

@@ -90,7 +90,7 @@ class AzureBaseTTSService:
emphasis: Emphasis level for speech ("strong", "moderate", "reduced").
language: Language for synthesis. Defaults to English (US).
pitch: Voice pitch adjustment (e.g., "+10%", "-5Hz", "high").
rate: Speech rate multiplier. Defaults to "1.05".
rate: Speech rate adjustment (e.g., "1.0", "1.25", "slow", "fast").
role: Voice role for expression (e.g., "YoungAdultFemale").
style: Speaking style (e.g., "cheerful", "sad", "excited").
style_degree: Intensity of the speaking style (0.01 to 2.0).
@@ -100,7 +100,7 @@ class AzureBaseTTSService:
emphasis: Optional[str] = None
language: Optional[Language] = Language.EN_US
pitch: Optional[str] = None
rate: Optional[str] = "1.05"
rate: Optional[str] = None
role: Optional[str] = None
style: Optional[str] = None
style_degree: Optional[str] = None
@@ -185,7 +185,9 @@ class AzureBaseTTSService:
if self._settings["volume"]:
prosody_attrs.append(f"volume='{self._settings['volume']}'")
ssml += f"<prosody {' '.join(prosody_attrs)}>"
# Only wrap in prosody tag if there are prosody attributes
if prosody_attrs:
ssml += f"<prosody {' '.join(prosody_attrs)}>"
if self._settings["emphasis"]:
ssml += f"<emphasis level='{self._settings['emphasis']}'>"
@@ -195,7 +197,8 @@ class AzureBaseTTSService:
if self._settings["emphasis"]:
ssml += "</emphasis>"
ssml += "</prosody>"
if prosody_attrs:
ssml += "</prosody>"
if self._settings["style"]:
ssml += "</mstts:express-as>"
@@ -277,6 +280,11 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService):
self._started = False
self._first_chunk = True
self._cumulative_audio_offset: float = 0.0 # Cumulative audio duration in seconds
self._current_sentence_base_offset: float = 0.0 # Base offset for current sentence
self._current_sentence_duration: float = 0.0 # Duration from Azure callback
self._current_sentence_max_word_offset: float = (
0.0 # Max word boundary offset seen in current sentence (for 8kHz workaround)
)
self._last_word: Optional[str] = None # Track last word for punctuation merging
self._last_timestamp: Optional[float] = None # Track last timestamp
@@ -386,8 +394,14 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService):
word = evt.text
sentence_relative_seconds = evt.audio_offset / 10_000_000.0
# Add cumulative offset to get absolute timestamp across sentences
absolute_seconds = self._cumulative_audio_offset + sentence_relative_seconds
# Use base offset captured at start of run_tts to avoid race conditions
# with callbacks from overlapping TTS requests
absolute_seconds = self._current_sentence_base_offset + sentence_relative_seconds
# Track max word offset for accurate cumulative timing
# (audio_duration from Azure doesn't always match word boundary offsets at 8kHz)
if sentence_relative_seconds > self._current_sentence_max_word_offset:
self._current_sentence_max_word_offset = sentence_relative_seconds
if not word:
return
@@ -492,9 +506,9 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService):
self._last_word = None
self._last_timestamp = None
# Update cumulative audio offset for next sentence
# Store duration for cumulative offset calculation
if evt.result and evt.result.audio_duration:
self._cumulative_audio_offset += evt.result.audio_duration.total_seconds()
self._current_sentence_duration = evt.result.audio_duration.total_seconds()
self._audio_queue.put_nowait(None) # Signal completion
@@ -530,6 +544,9 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService):
self._started = False
self._first_chunk = True
self._cumulative_audio_offset = 0.0
self._current_sentence_base_offset = 0.0
self._current_sentence_duration = 0.0
self._current_sentence_max_word_offset = 0.0
self._last_word = None
self._last_timestamp = None
@@ -604,6 +621,12 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService):
self._started = True
self._first_chunk = True
# Capture base offset BEFORE starting synthesis to avoid race conditions
# Word boundary callbacks will use this value
self._current_sentence_base_offset = self._cumulative_audio_offset
self._current_sentence_duration = 0.0
self._current_sentence_max_word_offset = 0.0
ssml = self._construct_ssml(text)
self._speech_synthesizer.speak_ssml_async(ssml)
await self.start_tts_usage_metrics(text)
@@ -627,6 +650,16 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService):
)
yield frame
# Update cumulative offset for next sentence
# At 8kHz, Azure's audio_duration doesn't match word boundary offsets,
# so we use max_word_offset as a workaround. At other sample rates,
# audio_duration is accurate.
# TODO: Remove after Azure fixes word boundary timing at 8kHz
if self.sample_rate == 8000:
self._cumulative_audio_offset += self._current_sentence_max_word_offset
else:
self._cumulative_audio_offset += self._current_sentence_duration
except Exception as e:
yield ErrorFrame(error=f"Unknown error occurred: {e}")
yield TTSStoppedFrame()

3
src/pipecat/services/cartesia/stt.py
View File

@@ -221,13 +221,10 @@ class CartesiaSTTService(WebsocketSTTService):
await super().process_frame(frame, direction)
if isinstance(frame, VADUserStartedSpeakingFrame):
# Reset finalize state for new utterance
self.set_finalize_pending(False)
await self._start_metrics()
elif isinstance(frame, VADUserStoppedSpeakingFrame):
# Send finalize command to flush the transcription session
if self._websocket and self._websocket.state is State.OPEN:
self.set_finalize_pending(True)
await self._websocket.send("finalize")
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:

4
src/pipecat/services/elevenlabs/stt.py
View File

@@ -551,8 +551,6 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService):
await super().process_frame(frame, direction)
if isinstance(frame, VADUserStartedSpeakingFrame):
# Reset finalize state for new utterance
self.set_finalize_pending(False)
# Start metrics when user starts speaking
await self._start_metrics()
elif isinstance(frame, VADUserStoppedSpeakingFrame):
@@ -560,8 +558,6 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService):
if self._params.commit_strategy == CommitStrategy.MANUAL:
if self._websocket and self._websocket.state is State.OPEN:
try:
# Mark that the next committed transcript should be finalized
self.set_finalize_pending(True)
commit_message = {
"message_type": "input_audio_chunk",
"audio_base_64": "",

8
src/pipecat/services/openai_realtime_beta/openai.py
View File

@@ -525,6 +525,14 @@ class OpenAIRealtimeBetaLLMService(LLMService):
# note: ttfb is faster by 1/2 RTT than ttfb as measured for other services, since we're getting
# this event from the server
await self.stop_ttfb_metrics()
if self._current_audio_response and self._current_audio_response.item_id != evt.item_id:
logger.warning(
f"Received a new audio delta for an already completed audio response before receiving the BotStoppedSpeakingFrame."
)
logger.debug("Forcing previous audio response to None")
self._current_audio_response = None
if not self._current_audio_response:
self._current_audio_response = CurrentAudioResponse(
item_id=evt.item_id,

2
src/pipecat/services/sarvam/stt.py
View File

@@ -193,11 +193,9 @@ class SarvamSTTService(STTService):
# Only handle VAD frames when not using Sarvam's VAD signals
if not self._vad_signals:
if isinstance(frame, VADUserStartedSpeakingFrame):
self.set_finalize_pending(False)
await self._start_metrics()
elif isinstance(frame, VADUserStoppedSpeakingFrame):
if self._socket_client:
self.set_finalize_pending(True)
await self._socket_client.flush()
async def set_language(self, language: Language):

55
src/pipecat/services/speechmatics/stt.py
View File

@@ -8,7 +8,6 @@
import asyncio
import os
import time
from enum import Enum
from typing import Any, AsyncGenerator
@@ -67,7 +66,7 @@ class TurnDetectionMode(str, Enum):
"""Endpoint and turn detection handling mode.
How the STT engine handles the endpointing of speech. If using Pipecat's built-in endpointing,
then use `TurnDetectionMode.FIXED` (default).
then use `TurnDetectionMode.EXTERNAL` (default).
To use the STT engine's built-in endpointing, then use `TurnDetectionMode.ADAPTIVE` for simple
voice activity detection or `TurnDetectionMode.SMART_TURN` for more advanced ML-based
@@ -107,7 +106,7 @@ class SpeechmaticsSTTService(STTService):
turn_detection_mode: Endpoint handling, one of `TurnDetectionMode.FIXED`,
`TurnDetectionMode.EXTERNAL`, `TurnDetectionMode.ADAPTIVE` and
`TurnDetectionMode.SMART_TURN`. Defaults to `TurnDetectionMode.FIXED`.
`TurnDetectionMode.SMART_TURN`. Defaults to `TurnDetectionMode.EXTERNAL`.
speaker_active_format: Formatter for active speaker ID. This formatter is used to format
the text output for individual speakers and ensures that the context is clear for
@@ -201,6 +200,7 @@ class SpeechmaticsSTTService(STTService):
extra_params: Extra parameters to pass to the STT engine. This is a dictionary of
additional parameters that can be used to configure the STT engine.
Default to None.
"""
# Service configuration
@@ -208,7 +208,7 @@ class SpeechmaticsSTTService(STTService):
language: Language | str = Language.EN
# Endpointing mode
turn_detection_mode: TurnDetectionMode = TurnDetectionMode.FIXED
turn_detection_mode: TurnDetectionMode = TurnDetectionMode.EXTERNAL
# Output formatting
speaker_active_format: str | None = None
@@ -346,7 +346,7 @@ class SpeechmaticsSTTService(STTService):
params.speaker_passive_format or params.speaker_active_format
)
# Metrics
# Model + metrics
self.set_model_name(self._config.operating_point.value)
# Message queue
@@ -598,9 +598,6 @@ class SpeechmaticsSTTService(STTService):
if segments:
await self._send_frames(segments)
# Update metrics
await self._emit_metrics(message.get("metadata", {}).get("processing_time", 0.0))
async def _handle_segment(self, message: dict[str, Any]) -> None:
"""Handle AddSegment events.
@@ -695,6 +692,7 @@ class SpeechmaticsSTTService(STTService):
f"{self} VADUserStoppedSpeakingFrame received but internal VAD is being used"
)
elif not self._enable_vad and self._client is not None:
self.request_finalize()
self._client.finalize()
async def _send_frames(self, segments: list[dict[str, Any]], finalized: bool = False) -> None:
@@ -738,16 +736,33 @@ class SpeechmaticsSTTService(STTService):
# If final, then re-parse into TranscriptionFrame
if finalized:
# Do any segments have `is_eou` set to True?
if (
any(segment.get("is_eou", False) for segment in segments)
and self._finalize_requested
):
self.confirm_finalize()
# Add the finalized frames
frames += [TranscriptionFrame(**attr_from_segment(segment)) for segment in segments]
# Handle the text (for metrics reporting)
finalized_text = "|".join([s["text"] for s in segments])
await self._handle_transcription(finalized_text, True, segments[0]["language"])
await self._handle_transcription(
finalized_text, is_final=True, language=segments[0]["language"]
)
# Log the frames
logger.debug(f"{self} finalized transcript: {[f.text for f in frames]}")
# Return as interim results (unformatted)
else:
# Add the interim frames
frames += [
InterimTranscriptionFrame(**attr_from_segment(segment)) for segment in segments
]
# Log the frames
logger.debug(f"{self} interim transcript: {[f.text for f in frames]}")
# Send the frames
@@ -804,28 +819,6 @@ class SpeechmaticsSTTService(STTService):
yield ErrorFrame(f"Speechmatics error: {e}")
await self._disconnect()
async def _emit_metrics(self, processing_time: float) -> None:
"""Create TTFB metrics.
The TTFB is the seconds between the person speaking and the STT
engine emitting the first partial. This is only calculated at the
start of an utterance.
"""
# Skip if metrics not available
if not self._metrics or processing_time == 0.0:
return
# Calculate time as time.time() - ttfb (which is seconds)
start_time = time.time() - processing_time
# Update internal metrics
self._metrics._start_ttfb_time = start_time
self._metrics._start_processing_time = start_time
# Stop TTFB metrics
await self.stop_ttfb_metrics()
await self.stop_processing_metrics()
# ============================================================================
# HELPERS
# ============================================================================

27
src/pipecat/services/stt_service.py
View File

@@ -119,28 +119,15 @@ class STTService(AIService):
"""
return self._muted
def set_finalize_pending(self, value: bool):
"""Set whether the next TranscriptionFrame should be marked as finalized.
When True, the next TranscriptionFrame pushed will have its `finalized`
field set to True, and this flag will automatically reset to False.
This is used to signal that a transcript is the final result for an
utterance, enabling immediate TTFB reporting.
Args:
value: True to mark the next transcription as finalized.
"""
self._finalize_pending = value
def request_finalize(self):
"""Mark that a finalize request has been sent, awaiting server confirmation.
For providers that require server confirmation before marking transcripts
as finalized (e.g., Deepgram's from_finalize field), call this when sending
the finalize request. Then call confirm_finalize() when the server confirms.
For providers that have explicit server confirmation of finalization
(e.g., Deepgram's from_finalize field), call this when sending the finalize
request. Then call confirm_finalize() when the server confirms.
This is an alternative to set_finalize_pending() for providers that need
two-step finalization.
For providers without server confirmation, don't call this method - just
send the finalize/flush/commit command and rely on the TTFB timeout.
"""
self._finalize_requested = True
@@ -298,7 +285,7 @@ class STTService(AIService):
"""Push a frame downstream, tracking TranscriptionFrame timestamps for TTFB.
Stores the timestamp of each TranscriptionFrame for TTFB calculation.
If the frame is marked as finalized (either directly or via set_finalize_pending),
If the frame is marked as finalized (via request_finalize/confirm_finalize),
reports TTFB immediately and cancels any pending timeout. Otherwise, TTFB is
reported after a timeout.
@@ -361,6 +348,7 @@ class STTService(AIService):
"""Handle VAD user started speaking frame to start tracking transcriptions.
Cancels any pending TTFB timeout, resets TTFB tracking state, and marks user as speaking.
Also resets finalization state to prevent stale finalization from a previous utterance.
Args:
frame: The VAD user started speaking frame.
@@ -368,6 +356,7 @@ class STTService(AIService):
await self._reset_stt_ttfb_state()
self._user_speaking = True
self._finalize_requested = False
self._finalize_pending = False
async def _handle_vad_user_stopped_speaking(self, frame: VADUserStoppedSpeakingFrame):
"""Handle VAD user stopped speaking frame.

23
src/pipecat/services/websocket_service.py
View File

@@ -123,26 +123,29 @@ class WebsocketService(ABC):
async def _maybe_try_reconnect(
self,
error: Exception,
error_message: str,
report_error: Callable[[ErrorFrame], Awaitable[None]],
error: Optional[Exception] = None,
) -> bool:
"""Check if reconnection should be attempted and try if appropriate.
Args:
error: The exception that occurred.
error_message: Human-readable error message for logging.
report_error: Callback function to report connection errors.
error: The exception that occurred (optional, may be None for graceful closes).
Returns:
True if should continue the receive loop, False if should break.
"""
# Don't reconnect if we're intentionally disconnecting
if self._disconnecting:
logger.warning(f"{self} error during disconnect: {error}")
if error:
logger.warning(f"{self} error during disconnect: {error}")
else:
logger.debug(f"{self} receive loop ended during disconnect")
return False
# Log the error
# Log the message
logger.warning(error_message)
# Try to reconnect if enabled
@@ -167,6 +170,14 @@ class WebsocketService(ABC):
while True:
try:
await self._receive_messages()
# _receive_messages() returned normally. This happens when the websocket
# closes gracefully (server sent close frame). The async for loop over
# the websocket exits without raising an exception in this case.
# We must handle this to avoid an infinite loop.
message = f"{self} connection closed by server"
should_continue = await self._maybe_try_reconnect(message, report_error)
if not should_continue:
break
except ConnectionClosedOK as e:
# Normal closure, don't retry
logger.debug(f"{self} connection closed normally: {e}")
@@ -175,13 +186,13 @@ class WebsocketService(ABC):
# Connection closed with error (e.g., no close frame received/sent)
# This often indicates network issues, server problems, or abrupt disconnection
message = f"{self} connection closed, but with an error: {e}"
should_continue = await self._maybe_try_reconnect(e, message, report_error)
should_continue = await self._maybe_try_reconnect(message, report_error, e)
if not should_continue:
break
except Exception as e:
# General error during message receiving
message = f"{self} error receiving messages: {e}"
should_continue = await self._maybe_try_reconnect(e, message, report_error)
should_continue = await self._maybe_try_reconnect(message, report_error, e)
if not should_continue:
break

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

@@ -1733,7 +1733,7 @@ class DailyInputTransport(BaseInputTransport):
message: The message data to send.
sender: ID of the message sender.
"""
await self.broadcast_frame_class(
await self.broadcast_frame(
DailyInputTransportMessageFrame, message=message, participant_id=sender
)

2
src/pipecat/transports/smallwebrtc/transport.py
View File

@@ -698,7 +698,7 @@ class SmallWebRTCInputTransport(BaseInputTransport):
message: The application message to process.
"""
logger.debug(f"Received app message inside SmallWebRTCInputTransport {message}")
await self.broadcast_frame_class(InputTransportMessageFrame, message=message)
await self.broadcast_frame(InputTransportMessageFrame, message=message)
# Add this method similar to DailyInputTransport.request_participant_image
async def request_participant_image(self, frame: UserImageRequestFrame):