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

aic-sdk-py v2.

Browse Source 
# Conflicts:
#	uv.lock

# Conflicts:
#	examples/foundational/07zd-interruptible-aicoustics.py
#	pyproject.toml
#	src/pipecat/audio/filters/aic_filter.py
#	src/pipecat/audio/vad/aic_vad.py
This commit is contained in:
Gökmen Görgen
2026-01-11 20:14:59 +01:00
parent 4e85e81d9b
commit ec17dc6626
3 changed files with 516 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

296
src/pipecat/audio/filters/aic_filter_v2.py Normal file
View File

@@ -0,0 +1,296 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""ai-coustics AIC SDK audio filter for Pipecat (aic-sdk >= 2.0.0).
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.
.. note::
This module is compatible with aic-sdk versions >= 2.0.0.
For aic-sdk < 2.0.0, use :mod:`pipecat.audio.filters.aic_filter` instead.
"""
import os
from typing import List, Optional
import numpy as np
from loguru import logger
from pipecat.audio.filters.base_audio_filter import BaseAudioFilter
from pipecat.audio.utils import check_aic_sdk_version
from pipecat.frames.frames import FilterControlFrame, FilterEnableFrame
# Check aic-sdk is installed and version is compatible (>= 2.0.0)
check_aic_sdk_version("v2")
# AIC SDK v2 (https://ai-coustics.github.io/aic-sdk-py/api/)
import aic
from aic import Model, ProcessorAsync, ProcessorConfig, ProcessorParameter, VadParameter
class AICFilter(BaseAudioFilter):
"""Audio filter using ai-coustics' AIC SDK v2 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.
.. note::
This class requires aic-sdk >= 2.0.0.
"""
def __init__(
self,
*,
license_key: str = "",
model_id: Optional[str] = None,
model_path: Optional[str] = None,
model_download_dir: Optional[str] = None,
enhancement_level: Optional[float] = 1.0,
voice_gain: Optional[float] = 1.0,
) -> None:
"""Initialize the AIC filter.
Args:
license_key: ai-coustics license key for authentication.
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. Defaults to
a cache directory in user's home folder.
enhancement_level: Optional overall enhancement strength (0.0..1.0).
voice_gain: Optional linear gain applied to detected speech (0.1..4.0).
Raises:
ValueError: If neither model_id nor model_path is provided.
"""
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_id = model_id
self._model_path = model_path
self._model_download_dir = model_download_dir or os.path.expanduser(
"~/.cache/pipecat/aic-models"
)
self._enhancement_level = enhancement_level
self._voice_gain = voice_gain
self._enabled = True
self._sample_rate = 0
self._aic_ready = False
self._frames_per_block = 0
self._audio_buffer = bytearray()
# v2 API objects
self._model: Optional[Model] = None
self._processor: Optional[ProcessorAsync] = None
self._processor_ctx = None
self._vad_ctx = None
def get_vad_context(self):
"""Return the VAD context once the processor exists.
Returns:
The VadContext instance bound to the underlying processor.
Raises RuntimeError if the processor has not been initialized.
"""
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,
*,
speech_hold_duration: Optional[float] = None,
sensitivity: Optional[float] = None,
):
"""Return an analyzer that will lazily instantiate the AIC VAD when ready.
AIC VAD parameters (v2):
- speech_hold_duration:
How long VAD continues detecting after speech ends (in seconds).
Range: 0.0 .. 20x model window length, Default (SDK): 0.05s
- sensitivity:
Energy threshold sensitivity. Energy threshold = 10 ** (-sensitivity).
Range: 1.0 .. 15.0, Default (SDK): 6.0
Args:
speech_hold_duration: Optional speech hold duration to configure on the VAD.
If None, SDK default (0.05s) is used.
sensitivity: Optional sensitivity (energy threshold) to configure on the VAD.
Range: 1.0 .. 15.0. If None, SDK default (6.0) is used.
Returns:
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_v2 import AICVADAnalyzer
return AICVADAnalyzer(
vad_context_factory=lambda: self.get_vad_context(),
speech_hold_duration=speech_hold_duration,
sensitivity=sensitivity,
)
async def start(self, sample_rate: int):
"""Initialize the filter with the transport's sample rate.
Args:
sample_rate: The sample rate of the input transport in Hz.
Returns:
None
"""
self._sample_rate = sample_rate
try:
# Load or download model
if self._model_path:
logger.debug(f"Loading AIC model from: {self._model_path}")
self._model = Model.from_file(self._model_path)
else:
logger.debug(f"Downloading AIC model: {self._model_id}")
os.makedirs(self._model_download_dir, exist_ok=True)
model_path = await Model.download_async(self._model_id, self._model_download_dir)
logger.debug(f"Model downloaded to: {model_path}")
self._model = Model.from_file(model_path)
# Create async processor
self._processor = ProcessorAsync(self._model, self._license_key or "")
# Get optimal frames for this sample rate
self._frames_per_block = self._model.get_optimal_num_frames(self._sample_rate)
# Create configuration
config = ProcessorConfig(
sample_rate=self._sample_rate,
num_channels=1,
num_frames=self._frames_per_block,
allow_variable_frames=False,
)
# Initialize processor
await self._processor.initialize_async(config)
# 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
if self._enhancement_level is not None:
level = float(self._enhancement_level if self._enabled else 0.0)
self._processor_ctx.set_parameter(ProcessorParameter.EnhancementLevel, level)
if self._voice_gain is not None:
self._processor_ctx.set_parameter(
ProcessorParameter.VoiceGain, float(self._voice_gain)
)
self._aic_ready = True
# Log processor information
logger.debug(f"ai-coustics filter (v2) 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" Enhancement strength: {int((self._enhancement_level or 1.0) * 100)}%")
logger.debug(f" Optimal sample rate: {self._model.get_optimal_sample_rate()} Hz")
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)"
)
except Exception as e: # noqa: BLE001 - surfacing SDK initialization errors
logger.error(f"AIC model initialization failed: {e}")
self._aic_ready = False
async def stop(self):
"""Clean up the AIC processor when stopping.
Returns:
None
"""
try:
if self._processor_ctx is not None:
self._processor_ctx.reset()
finally:
self._processor = None
self._processor_ctx = None
self._vad_ctx = None
self._model = None
self._aic_ready = False
self._audio_buffer.clear()
async def process_frame(self, frame: FilterControlFrame):
"""Process control frames to enable/disable filtering.
Args:
frame: The control frame containing filter commands.
Returns:
None
"""
if isinstance(frame, FilterEnableFrame):
self._enabled = frame.enable
if self._processor_ctx is not None:
try:
level = float(self._enhancement_level if self._enabled else 0.0)
self._processor_ctx.set_parameter(ProcessorParameter.EnhancementLevel, level)
except Exception as e: # noqa: BLE001
logger.error(f"AIC set_parameter failed: {e}")
async def filter(self, audio: bytes) -> bytes:
"""Apply AIC enhancement to audio data.
Buffers incoming audio and processes it in chunks that match the AIC
model's required block length. Returns enhanced audio data.
Args:
audio: Raw audio data as bytes to be filtered (int16 PCM, planar).
Returns:
Enhanced audio data as bytes (int16 PCM, planar).
"""
if not self._aic_ready or self._processor is None:
return audio
self._audio_buffer.extend(audio)
filtered_chunks: List[bytes] = []
# Number of int16 samples currently buffered
available_frames = len(self._audio_buffer) // 2
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])
# Convert to float32 in -1..+1 range and reshape to (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)
)
# Process via async processor; returns ndarray (same shape)
out_f32 = await self._processor.process_async(block_f32)
# Convert back to int16 bytes
out_i16 = np.clip(out_f32 * 32768.0, -32768, 32767).astype(np.int16)
filtered_chunks.append(out_i16.reshape(-1).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
return b"".join(filtered_chunks)

57
src/pipecat/audio/utils.py
View File

@@ -12,6 +12,9 @@ various audio formats used in Pipecat pipelines.
"""
import audioop
from typing import Literal
from loguru import logger
import numpy as np
import pyloudnorm as pyln
@@ -311,3 +314,57 @@ def is_silence(pcm_bytes: bytes) -> bool:
# If max value is lower than SPEAKING_THRESHOLD, consider it as silence
return max_value <= SPEAKING_THRESHOLD
def check_aic_sdk_version(required_version: Literal["v1", "v2"]) -> None:
"""Check if the aic-sdk is installed and compatible with the module.
This function checks both that the aic-sdk is installed and that its version
is compatible with the module requirements.
Args:
required_version: Either "v1" (for aic-sdk < 2.0.0) or "v2" (for aic-sdk >= 2.0.0).
Raises:
ImportError: If aic-sdk is not installed or version is incompatible.
"""
try:
import aic # noqa: F401 - check if module is installed
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 ImportError(f"Missing module: {e}") from e
try:
from importlib.metadata import version as get_version
aic_version = get_version("aic-sdk")
major_version = int(aic_version.split(".")[0])
if required_version == "v1" and major_version >= 2:
error_msg = (
f"aic-sdk version {aic_version} detected, but aic-sdk < 2.0.0 is required. "
"Please use the v2 modules instead: "
"'from pipecat.audio.filters.aic_filter_v2 import AICFilter' or "
"'from pipecat.audio.vad.aic_vad_v2 import AICVADAnalyzer'."
)
logger.error(error_msg)
raise ImportError(error_msg)
if required_version == "v2" and major_version < 2:
error_msg = (
f"aic-sdk version {aic_version} detected, but aic-sdk >= 2.0.0 is required. "
"Please update with 'pip install --upgrade aic-sdk>=2.0.0' "
"or use the v1 modules: "
"'from pipecat.audio.filters.aic_filter import AICFilter' or "
"'from pipecat.audio.vad.aic_vad import AICVADAnalyzer'."
)
logger.error(error_msg)
raise ImportError(error_msg)
except ImportError:
# Re-raise if it's our version mismatch error
raise
except Exception:
# If we can't determine version for other reasons, log warning and allow to proceed
logger.warning("Could not determine aic-sdk version. Proceeding anyway.")

163
src/pipecat/audio/vad/aic_vad_v2.py Normal file
View File

@@ -0,0 +1,163 @@
"""AIC-integrated VAD analyzer that lazily binds to the AIC SDK v2 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 (speech_hold_duration, sensitivity) when available.
.. note::
This module is compatible with aic-sdk versions >= 2.0.0.
For aic-sdk < 2.0.0, use :mod:`pipecat.audio.vad.aic_vad` instead.
"""
from typing import Any, Callable, Optional
from loguru import logger
from pipecat.audio.utils import check_aic_sdk_version
from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADParams
# Check aic-sdk is installed and version is compatible (>= 2.0.0)
check_aic_sdk_version("v2")
from aic import VadParameter
class AICVADAnalyzer(VADAnalyzer):
"""VAD analyzer that lazily binds to the AIC VadContext via a factory.
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. We then use the context's is_speech_detected() state
to derive confidence values.
AIC VAD runtime parameters (v2):
- 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 .. 20x model window length
Default (SDK): 0.05s (50ms)
- 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
Formula: Energy threshold = 10 ** (-sensitivity)
Default (SDK): 6.0
.. note::
This class requires aic-sdk >= 2.0.0.
"""
def __init__(
self,
*,
vad_context_factory: Optional[Callable[[], Any]] = None,
speech_hold_duration: Optional[float] = None,
sensitivity: Optional[float] = None,
):
"""Create an AIC VAD analyzer.
Args:
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.
speech_hold_duration:
Optional override for AIC VAD speech hold duration (in seconds).
Range: 0.0 .. 20x model window length.
If None, the SDK default (0.05s) is used.
sensitivity:
Optional override for AIC VAD sensitivity (energy threshold).
Range: 1.0 .. 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_context_factory = vad_context_factory
self._vad_ctx: Optional[Any] = None
self._pending_speech_hold_duration: Optional[float] = speech_hold_duration
self._pending_sensitivity: Optional[float] = sensitivity
def bind_vad_context_factory(self, vad_context_factory: Callable[[], Any]):
"""Attach or replace the factory post-construction."""
self._vad_context_factory = vad_context_factory
self._ensure_vad_context_initialized()
def _apply_vad_params(self):
"""Apply optional AIC VAD parameters if available."""
if self._vad_ctx is None or VadParameter is None:
return
try:
if self._pending_speech_hold_duration is not None:
self._vad_ctx.set_parameter(
VadParameter.SpeechHoldDuration, float(self._pending_speech_hold_duration)
)
if self._pending_sensitivity is not None:
self._vad_ctx.set_parameter(
VadParameter.Sensitivity, float(self._pending_sensitivity)
)
except Exception as e: # noqa: BLE001
logger.debug(f"AIC VAD parameter application deferred/failed: {e}")
def _ensure_vad_context_initialized(self):
if self._vad_ctx is not None:
return
if not self._vad_context_factory:
return
try:
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 context (v2) initialized in analyzer.")
except Exception as e: # noqa: BLE001
# Filter may not be started yet; try again later
logger.debug(f"Deferring AIC VAD context initialization: {e}")
def set_sample_rate(self, sample_rate: int):
"""Set the sample rate for audio processing.
Args:
sample_rate: Audio sample rate in Hz.
"""
# Set rate and attempt VAD context initialization once we know SR
self._sample_rate = self._init_sample_rate or sample_rate
self._ensure_vad_context_initialized()
# Ensure params are initialized even if VAD context not ready yet
try:
super().set_params(self._params)
except Exception:
pass
def num_frames_required(self) -> int:
"""Get the number of audio frames required for analysis.
Returns:
Number of frames needed for VAD processing.
"""
# Use 10 ms windows based on sample rate
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.
Args:
buffer: Audio buffer to analyze.
Returns:
Voice confidence score is 0.0 or 1.0.
"""
# 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 processor's VAD is updated
# as part of the enhancement pipeline. Simply query the boolean and map it.
try:
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}")
return 0.0