Add AIC SDK audio filter

env.example

@@ -1,3 +1,6 @@
+# AI-COUSTICS
+AICOUSTICS_LICENSE_KEY=...
+
 # Anthropic
 ANTHROPIC_API_KEY=...
 

examples/foundational/07ad-interruptible-aicoustics.py

@@ -0,0 +1,159 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+
+import datetime
+import os
+import wave
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.filters.aic_filter import AICFilter
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.deepgram.tts import DeepgramTTSService
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams
+from pipecat.transports.services.daily import DailyParams
+
+load_dotenv(override=True)
+
+
+# Create audio buffer processor with default settings
+audiobuffer = AudioBufferProcessor(
+    num_channels=2,  # 1 for mono, 2 for stereo (user left, bot right)
+    enable_turn_audio=False,  # Enable per-turn audio recording
+    user_continuous_stream=True,  # User has continuous audio stream
+)
+
+
+def _create_aic_filter() -> AICFilter:
+    license_key = os.getenv("AICOUSTICS_LICENSE_KEY", "")
+
+    return AICFilter(
+        license_key=license_key,
+        enhancement_level=1.0,
+    )
+
+
+# We store functions so objects (e.g. SileroVADAnalyzer) don't get
+# instantiated. The function will be called when the desired transport gets
+# selected.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+        audio_in_filter=_create_aic_filter(),
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+        audio_in_filter=_create_aic_filter(),
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+        audio_in_filter=_create_aic_filter(),
+    ),
+}
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+    logger.info(f"Starting bot")
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = DeepgramTTSService(api_key=os.getenv("DEEPGRAM_API_KEY"), voice="aura-helios-en")
+
+    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+
+    messages = [
+        {
+            "role": "system",
+            "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+        },
+    ]
+
+    context = OpenAILLMContext(messages)
+    context_aggregator = llm.create_context_aggregator(context)
+
+    pipeline = Pipeline(
+        [
+            transport.input(),  # Transport user input
+            stt,  # STT
+            context_aggregator.user(),  # User responses
+            llm,  # LLM
+            tts,  # TTS
+            transport.output(),  # Transport bot output
+            audiobuffer,  # write audio data to a file
+            context_aggregator.assistant(),  # Assistant spoken responses
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            enable_metrics=True,
+            enable_usage_metrics=True,
+        ),
+        idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
+    )
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info(f"Client connected")
+        await audiobuffer.start_recording()
+        # Kick off the conversation.
+        messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+        await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+    @audiobuffer.event_handler("on_audio_data")
+    async def on_audio_data(buffer, audio, sample_rate, num_channels):
+        # Save or process the composite audio
+        timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
+        filename = f"./conversation_{timestamp}.wav"
+
+        # Create the WAV file
+        with wave.open(filename, "wb") as wf:
+            wf.setnchannels(num_channels)
+            wf.setsampwidth(2)  # 16-bit audio
+            wf.setframerate(sample_rate)
+            wf.writeframes(audio)
+
+        logger.info(f"Saved recording to {filename}")
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info(f"Client disconnected")
+        await task.cancel()
+
+    runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
+
+    await runner.run(task)
+
+
+async def bot(runner_args: RunnerArguments):
+    """Main bot entry point compatible with Pipecat Cloud."""
+    transport = await create_transport(runner_args, transport_params)
+    await run_bot(transport, runner_args)
+
+
+if __name__ == "__main__":
+    from pipecat.runner.run import main
+
+    main()

pyproject.toml

@@ -45,6 +45,7 @@ Source = "https://github.com/pipecat-ai/pipecat"
 Website = "https://pipecat.ai"
 
 [project.optional-dependencies]
+aic = [ "aic-sdk~=0.6.1" ]
 anthropic = [ "anthropic~=0.49.0" ]
 assemblyai = [ "websockets>=13.1,<15.0" ]
 asyncai = [ "websockets>=13.1,<15.0" ]

src/pipecat/audio/filters/aic_filter.py

@@ -0,0 +1,202 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""ai-coustics AIC SDK audio filter for Pipecat.
+
+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.
+"""
+
+from typing import List, Optional
+
+import numpy as np
+from loguru import logger
+
+from pipecat.audio.filters.base_audio_filter import BaseAudioFilter
+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.
+    """
+
+    def __init__(
+        self,
+        *,
+        license_key: str = "",
+        model_type: AICModelType = AICModelType.QUAIL_L,
+        enhancement_level: Optional[float] = 1.0,
+        voice_gain: Optional[float] = 1.0,
+        noise_gate_enable: Optional[bool] = True,
+    ) -> 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).
+        """
+        self._license_key = license_key
+        self._model_type = model_type
+
+        self._enhancement_level = enhancement_level
+        self._voice_gain = voice_gain
+        self._noise_gate_enable = noise_gate_enable
+
+        self._enabled = True
+        self._sample_rate = 0
+        self._aic_ready = False
+        self._frames_per_block = 0
+        self._audio_buffer = bytearray()
+        # Create model and configure it
+        try:
+            self._aic = Model(model_type=self._model_type, license_key=self._license_key)
+        except Exception as e:  # noqa: BLE001 - surfacing SDK initialization errors
+            logger.error(f"AIC model creation failed: {e}")
+            self._aic = None
+            self._aic_ready = False
+            return
+
+    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:
+            self._aic.initialize(
+                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))
+            if self._noise_gate_enable is not None:
+                self._aic.set_parameter(
+                    AICParameter.NOISE_GATE_ENABLE, 1.0 if bool(self._noise_gate_enable) else 0.0
+                )
+
+            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"
+            )
+        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 model when stopping.
+
+        Returns:
+            None
+        """
+        try:
+            if self._aic is not None:
+                self._aic.close()
+        finally:
+            self._aic = 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._aic is not None:
+                try:
+                    level = float(self._enhancement_level if self._enabled else 0.0)
+                    self._aic.set_parameter(AICParameter.ENHANCEMENT_LEVEL, 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._aic 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 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)
+            )
+
+            # Process planar in-place; returns ndarray (same shape)
+            out_f32 = self._aic.process(block_f32)
+
+            # 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())
+
+            # 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)

uv.lock

@@ -42,6 +42,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/e3/52/6ad8f63ec8da1bf40f96996d25d5b650fdd38f5975f8c813732c47388f18/aenum-3.1.16-py3-none-any.whl", hash = "sha256:9035092855a98e41b66e3d0998bd7b96280e85ceb3a04cc035636138a1943eaf", size = 165627, upload-time = "2025-04-25T03:17:58.89Z" },
 ]
 
+[[package]]
+name = "aic-sdk"
+version = "0.6.1"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "numpy" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/8a/40/a307063543a59be1ebec640027666d1180ccf3434f69d890e33f55f78066/aic_sdk-0.6.1.tar.gz", hash = "sha256:9b4a48e0dcdb3ad0ef702c64b5930c5ce1c34e11235861b3ba4a8aaa337bb777", size = 29368, upload-time = "2025-08-18T16:24:05.348Z" }
+
 [[package]]
 name = "aioboto3"
 version = "15.0.0"
@@ -4209,6 +4218,9 @@ dependencies = [
 ]
 
 [package.optional-dependencies]
+aic = [
+    { name = "aic-sdk" },
+]
 anthropic = [
     { name = "anthropic" },
 ]
@@ -4409,6 +4421,7 @@ docs = [
 [package.metadata]
 requires-dist = [
     { name = "accelerate", marker = "extra == 'moondream'", specifier = "~=1.10.0" },
+    { name = "aic-sdk", marker = "extra == 'aic'", specifier = "~=0.6.1" },
     { name = "aioboto3", marker = "extra == 'aws'", specifier = "~=15.0.0" },
     { name = "aiofiles", specifier = ">=24.1.0,<25" },
     { name = "aiohttp", specifier = ">=3.11.12,<4" },
@@ -4500,7 +4513,7 @@ requires-dist = [
     { name = "websockets", marker = "extra == 'soniox'", specifier = ">=13.1,<15.0" },
     { name = "websockets", marker = "extra == 'websocket'", specifier = ">=13.1,<15.0" },
 ]
-provides-extras = ["anthropic", "assemblyai", "asyncai", "aws", "aws-nova-sonic", "azure", "cartesia", "cerebras", "deepseek", "daily", "deepgram", "elevenlabs", "fal", "fireworks", "fish", "gladia", "google", "grok", "groq", "gstreamer", "heygen", "inworld", "krisp", "koala", "langchain", "livekit", "lmnt", "local", "mcp", "mem0", "mistral", "mlx-whisper", "moondream", "nim", "neuphonic", "noisereduce", "openai", "openpipe", "openrouter", "perplexity", "playht", "qwen", "rime", "riva", "runner", "sambanova", "sarvam", "sentry", "local-smart-turn", "remote-smart-turn", "silero", "simli", "soniox", "soundfile", "speechmatics", "tavus", "together", "tracing", "ultravox", "webrtc", "websocket", "whisper"]
+provides-extras = ["aic", "anthropic", "assemblyai", "asyncai", "aws", "aws-nova-sonic", "azure", "cartesia", "cerebras", "deepseek", "daily", "deepgram", "elevenlabs", "fal", "fireworks", "fish", "gladia", "google", "grok", "groq", "gstreamer", "heygen", "inworld", "krisp", "koala", "langchain", "livekit", "lmnt", "local", "mcp", "mem0", "mistral", "mlx-whisper", "moondream", "nim", "neuphonic", "noisereduce", "openai", "openpipe", "openrouter", "perplexity", "playht", "qwen", "rime", "riva", "runner", "sambanova", "sarvam", "sentry", "local-smart-turn", "remote-smart-turn", "silero", "simli", "soniox", "soundfile", "speechmatics", "tavus", "together", "tracing", "ultravox", "webrtc", "websocket", "whisper"]
 
 [package.metadata.requires-dev]
 dev = [