arctan-vi 0.7.2

Arctan Voice Isolation for voice agents

Arctan Voice Isolation

Arctan Voice Isolation is a Python SDK for real-time voice isolation in voice agents and audio pipelines.

The SDK supports:

• standard Numpy audio processing
• LiveKit audio input processing
• Pipecat audio input filtering

An Arctan SDK key is required before audio can be processed.

Installation

Install the base SDK:

pip install arctan-vi

Install with framework adapters:

pip install "arctan-vi[livekit]"
pip install "arctan-vi[pipecat]"

The base SDK and LiveKit adapter support Python 3.10+. The Pipecat adapter requires Python 3.11+.

Set your SDK key in the runtime environment:

export ARCTAN_SDK_KEY="arc_sk_live_..."

You can also pass license_key="arc_sk_live_..." directly to the processor or adapter constructors. An explicit license_key takes precedence over ARCTAN_SDK_KEY.

Standard Python

Process one stream

The existing single-stream API remains supported. It creates and owns its model internally:

import numpy as np
import arctan

config = arctan.ProcessorConfig(sample_rate=24000, num_channels=1)
processor = arctan.Processor(config=config)

try:
    audio = np.zeros((config.num_channels, config.num_frames), dtype=np.float32)
    enhanced = processor.process(audio)
finally:
    processor.close()

Process concurrent streams

For multiple streams in one worker process, create one shared Model and one ProcessorAsync per call or audio stream. ProcessorAsync runs blocking native inference off the event loop while keeping frames within each stream ordered.

import asyncio
import numpy as np
import arctan


async def main() -> None:
    model = arctan.Model()
    configs = [arctan.ProcessorConfig.optimal(model) for _ in range(2)]
    processors = [
        arctan.ProcessorAsync(model=model, config=config)
        for config in configs
    ]

    try:
        frames = [
            np.zeros((config.num_channels, config.num_frames), dtype=np.float32)
            for config in configs
        ]
        enhanced = await asyncio.gather(
            *(
                processor.process_async(frame)
                for processor, frame in zip(processors, frames)
            )
        )
        print(len(enhanced))
    finally:
        for processor in processors:
            processor.close()
        model.close()


asyncio.run(main())

Use one shared Model per worker process and exactly one processor per call or audio stream. Do not share one processor between independent streams or submit one stream's frames out of order. Different processors may execute concurrently, but calls on the same ProcessorAsync are serialized. Close every processor before closing its shared model.

The number of processors is the number of active streams, not a universal capacity guarantee. Benchmark the deployment machine using representative audio and retain CPU and RSS headroom for transport, codecs, logging, and application work.

Processor.process() accepts float32 audio shaped as channels x frames. Samples must be finite and in [-1.0, 1.0]. The output is a new float32 array with the same shape.

Mono and stereo input are supported. Stereo input is downmixed for voice isolation and returned as dual-mono stereo. The sample rate must remain stable for the lifetime of a processor.

If num_frames is omitted, initialization fills it with the native frame size for the selected sample rate. If you set num_frames, it must contain a whole number of native processing blocks.

LiveKit

Install the LiveKit extra:

pip install "arctan-vi[livekit]"

Create an Arctan noise-cancellation processor and pass it to LiveKit audio input options:

from arctan.livekit import arctan_enhancer
from livekit.agents import room_io

await session.start(
    agent=Assistant(),
    room=ctx.room,
    room_options=room_io.RoomOptions(
        audio_input=room_io.AudioInputOptions(
            noise_cancellation=arctan_enhancer.noise_canceller(),
        ),
    ),
)

You can pass SDK options directly to Arctan:

noise_cancellation = arctan_enhancer.noise_canceller(
    license_key="arc_sk_live_...",
)

LiveKit mono and stereo frames are supported. Stereo frames are downmixed for voice isolation and returned as dual-mono stereo. The sample rate must remain stable, and each input frame must contain a whole number of native processing blocks.

Create a separate noise-canceller instance for every participant, call, or independent audio input. Do not share one instance across concurrent streams.

Pipecat

Install the Pipecat extra:

pip install "arctan-vi[pipecat]"

Create an ArctanAudioFilter and pass it as the input audio filter for your transport:

from arctan.pipecat import arctan_enhancer
from pipecat.transports.base_transport import TransportParams

audio_filter = arctan_enhancer.ArctanAudioFilter()

transport_params = TransportParams(
    audio_in_enabled=True,
    audio_in_filter=audio_filter,
)

ArctanAudioFilter implements Pipecat's BaseAudioFilter. Pipecat calls start(sample_rate), filter(audio), and stop() through the transport lifecycle.

The Pipecat adapter accepts mono PCM16 input. Arbitrary byte chunks are buffered until a complete native processing block is available, so filter() may return b"". FilterEnableFrame can bypass or re-enable processing; toggling clears any incomplete buffered block.

Create a separate ArctanAudioFilter for every call or pipeline. Do not share one filter across independent concurrent streams.