> ## Documentation Index
> Fetch the complete documentation index at: https://docs.ai-coustics.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Performance

> Measure processing time, capacity and queue behavior on your deployment hardware.

Measure the model and integration on the CPU, operating system and resource limits you will deploy. Throughput measured on a quiet development machine is not a production capacity guarantee.

## CPU requirements

The SDK runs inference on the CPU. Available operating-system and architecture builds vary by binding; check your [language guide](/reference/sdk/language-bindings) and the [compatibility matrix](/reference/sdk/compatibility-matrix).

Model complexity, host-rate conversion, block adaptation, wrapper copies, concurrent streams and other workloads all contribute to CPU cost. Test smaller models if the larger variant exceeds your capacity budget. Choose by task quality as well as runtime cost using the [evaluation guide](/guides/evaluate-audio).

## Real-time guarantees

A serial processing path must keep pace with incoming audio. Define the frame budget and processing ratio as:

$$
\text{frame\_duration} = \frac{\text{block\_size}}{\text{sample\_rate}},\qquad
\text{ratio} = \frac{\text{processing\_call\_duration}}{\text{frame\_duration}}
$$

A sustained ratio above 1 means that path cannot keep up. A ratio below 1 on average does not rule out deadline misses: inspect high percentiles, maximum duration and queue growth. Reserve capacity for capture, resampling, transport and downstream work.

For example, 160 samples at 16 kHz represent 10 ms. This is an illustrative budget calculation, not a measured inference time.

## Performance (CPU usage)

Changing the host sample rate can change conversion cost even when the model stays the same. Changing block size can change buffering and scheduling overhead. Keep both fixed when comparing model variants, then measure the actual production configuration separately.

Distinguish [audio delay](/reference/concepts/latency) from time spent computing a block. Reducing one does not prove that the other improved.

## Parallel processing

<Tooltip headline="Python-specific behavior" tip="This behavior applies to the Python SDK bindings. Other bindings can use a different threading model.">
  <Badge icon="python" color="yellow" className="underline decoration-dotted">Python only</Badge>
</Tooltip>

In Python SDK 3.3.0, `Processor.process()` blocks its calling thread while native processing releases the Python interpreter lock. `ProcessorAsync.process_async()` schedules work on the native blocking pool so the event loop can continue.

Use one instance per stream and await frames sequentially within each stream. More concurrent instances improve capacity only while the machine has resources to run them. Reuse loaded model weights where supported; each stream still owns processing state.

## Async execution footguns

<Tooltip headline="Python-specific behavior" tip="This behavior applies to the Python SDK bindings. Other bindings can use a different threading model.">
  <Badge icon="python" color="yellow" className="underline decoration-dotted">Python only</Badge>
</Tooltip>

* **Submitting one stream concurrently:** A mutex prevents simultaneous mutation but does not make submission order an audio-order guarantee. Await one frame before sending the next.
* **Unbounded work queues:** Pending calls retain audio buffers and add waiting time. Bound admission and measure queue depth.
* **Reinitializing during processing:** Stop input and settle in-flight work before changing configuration or resetting state.
* **Blocking the event loop with control calls:** Create contexts during setup and keep initialization, synchronous teardown and other blocking work off the loop.
* **Treating cancellation as rollback:** A cancelled await does not prove native work stopped. Follow the [lifecycle guidance](/reference/concepts/streams-and-state#ordering-cancellation-and-teardown).

## Run a bounded comparison

Use Python 3.12 or newer, an SDK key and a finite, non-silent mono WAV recording. The script below targets Python SDK 3.3.0/core 0.25.0 and compares one stream with two independent streams by running it twice. It measures call duration, scheduling lateness and process memory/CPU; it does not measure enhancement quality.

```bash theme={null}
python -m venv .venv
source .venv/bin/activate
python -m pip install 'aic-sdk==3.3.0' 'numpy==2.5.3' 'soundfile==0.13.1' 'psutil==7.2.2'
```

Set `AIC_SDK_LICENSE` through your normal secret mechanism. Save this as `measure_runtime.py`:

```python theme={null}
import argparse
import asyncio
import hashlib
import importlib.metadata
import json
import os
import platform
import time
from pathlib import Path

import aic_sdk as aic
import numpy as np
import psutil
import soundfile as sf

MODEL = "quail-vf-2.2-l-16khz"


async def measure(model, audio, config, seconds):
    processor = aic.ProcessorAsync(model, os.environ["AIC_SDK_LICENSE"], config)
    size = config.block_size
    period = size / config.sample_rate
    frames = len(audio) // size
    durations, late = [], []
    try:
        for i in range(10):
            await processor.process_async(audio[(i % frames) * size:((i % frames) + 1) * size])
        start = time.monotonic()
        index = 0
        while time.monotonic() - start < seconds:
            due = start + index * period
            await asyncio.sleep(max(0, due - time.monotonic()))
            before = time.monotonic()
            frame = index % frames
            output = await processor.process_async(audio[frame * size:(frame + 1) * size])
            after = time.monotonic()
            if len(output) != size or not np.isfinite(output).all():
                raise RuntimeError("Invalid processor output")
            durations.append(after - before)
            late.append(max(0, after - (due + period)))
            index += 1
        return {
            "frames": index,
            "call_ms_p50_p95_p99_max": np.percentile(durations, [50, 95, 99, 100]).tolist(),
            "max_scheduling_lateness_ms": max(late) * 1000,
            "deadline_misses": sum(value > 0 for value in late),
            "max_call_frame_ratio": max(durations) / period,
        }
    finally:
        await processor.terminate_session_async()


async def main(args):
    if not os.environ.get("AIC_SDK_LICENSE"):
        raise ValueError("Set AIC_SDK_LICENSE before running")
    audio, rate = sf.read(args.input, dtype="float32")
    if audio.ndim != 1 or not np.isfinite(audio).all() or not np.any(audio):
        raise ValueError("Use finite, non-silent mono audio")
    model = aic.Model.from_file(aic.Model.download(MODEL, "./models"))
    config = aic.ProcessorConfig(rate, model.get_optimal_block_size(rate))
    if len(audio) < config.block_size:
        raise ValueError("Input must contain at least one complete frame")
    process = psutil.Process()
    process.cpu_percent(None)
    rss_before = process.memory_info().rss
    tasks = [measure(model, audio, config, args.seconds) for _ in range(args.streams)]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    failures = [type(result).__name__ for result in results if isinstance(result, BaseException)]
    if failures:
        raise RuntimeError(f"Run failed: {failures}")
    for result in results:
        result["call_ms_p50_p95_p99_max"] = [value * 1000 for value in result["call_ms_p50_p95_p99_max"]]
    print(json.dumps({
        "binding": importlib.metadata.version("aic-sdk"), "core": aic.get_sdk_version(),
        "model": MODEL, "platform": platform.platform(), "machine": platform.machine(),
        "input_sha256": hashlib.sha256(Path(args.input).read_bytes()).hexdigest(),
        "sample_rate": rate, "block_size": config.block_size,
        "seconds_per_stream": args.seconds, "streams": args.streams,
        "process_cpu_percent": process.cpu_percent(None),
        "rss_before_bytes": rss_before, "rss_after_bytes": process.memory_info().rss,
        "results": results,
    }, indent=2))


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("input")
    parser.add_argument("--streams", type=int, choices=[1, 2], default=1)
    parser.add_argument("--seconds", type=int, choices=range(1, 121), default=10)
    asyncio.run(main(parser.parse_args()))
```

Run with a controlled recording, then compare the JSON results:

```bash theme={null}
python measure_runtime.py input.wav --streams 1 --seconds 10 > one-stream.json
python measure_runtime.py input.wav --streams 2 --seconds 10 > two-streams.json
```

The file is repeated to fill the requested duration; the final incomplete input block is excluded. Warmup and model download are outside per-call timing. CPU percentage covers the process and can exceed 100% across cores; before/after resident memory is not peak memory. Record the CPU model and deployment CPU/memory limits alongside the output.

For sustained-load validation, test your pipeline under deployment resource limits at the intended concurrency. This script holds at most one call per stream; measure queueing, drops, bypass and processing errors in your application. Stop runs that fall behind; their nominal stream count is not supported capacity.

## Record and compare results

Retain the fixture hash, exact package/core/model versions, settings, hardware, resource limits and repetitions with each result. Report p50/p95/p99 and maximum call duration, deadline misses, queue depth, peak memory and output/error checks. Compare the same fixtures before and after an upgrade.

Use the [SDK benchmark example](/reference/sdk/examples) for further experiments and [deployment verification](/production/deployment#verify-the-deployment) for the release checklist. Every published number should identify what was measured and what the measurement excludes.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.