> ## 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.

# Real-time analysis with Tyto

> Collect a live audio stream and run Tyto analysis outside the audio callback.

[Tyto](/models/audio-insight/tyto) scores incoming audio so you can inspect conditions associated with downstream voice AI failures. This guide uses Python SDK 3.3.0/core 0.25.0 and `tyto-1.1-l-16khz`. For complete files, see [batch call analysis](/models/audio-insight/batch-call-analysis).

For LiveKit Agents, the plugin schedules analysis and exposes its collector as a frame processor. See [LiveKit audio-quality analysis](/models/get-started/livekit-quickstart#audio-quality-analysis).

## Two objects, two threads

| Object | Role | Scheduling |
| :- | :- | :- |
| **Collector** | Copies mono audio into a rolling analysis window without modifying the input | Native collection is suitable for the audio path. Check wrapper allocation and your callback's constraints. |
| **Analyzer** | Runs the analysis model over the current window | Use a separate worker; do not run inference in an audio callback. |

The collector and analyzer may run concurrently. Each independent stream needs its own pair. Tyto's window is 5 s; newer samples replace the oldest ones. Nothing is analyzed until your application calls `analyze_buffered()`.

CPU cost grows with analysis frequency and concurrent streams. Start at a 5 s interval and measure before reducing it; 5 s is not a minimum. A 1 s interval analyzes overlapping audio. If inference falls behind, reduce frequency or concurrency instead of accumulating jobs.

## Set up the pair

Create the collector and analyzer with [`analyzer_pair()`](/reference/sdk/api/python/analysis#analyzer_pair).

Use Python 3.12 or newer, a valid SDK key in `AIC_SDK_LICENSE` and a non-silent mono WAV file containing at least 5 s of user speech. Create an isolated environment:

```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'
```

Save the script below as `analyze_stream.py`. It feeds a recording at its original rate to demonstrate the same collector/worker division you use with live input. It does not capture a microphone or make a call.

## Buffer and analyze

```python theme={null}
import argparse
import json
import math
import os
import queue
import threading
import time

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

MODEL = "tyto-1.1-l-16khz"
WINDOW_SECONDS = 5


def main(path, interval):
    if not math.isfinite(interval) or interval <= 0:
        raise ValueError("Analysis interval must be finite and positive")
    key = os.environ["AIC_SDK_LICENSE"]
    audio, rate = sf.read(path, 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")
    if len(audio) < WINDOW_SECONDS * rate:
        raise ValueError("Use at least 5 s to fill this example's analysis window")
    model = aic.Model.from_file(aic.Model.download(MODEL, "./models"))
    collector, analyzer = aic.analyzer_pair(model, key)
    config = aic.ProcessorConfig(rate, model.get_optimal_block_size(rate))
    collector.initialize(config)
    ready, stop = threading.Event(), threading.Event()
    first_pass_done = threading.Event()
    failures = queue.SimpleQueue()
    scores = []

    def analyze():
        while not ready.wait(0.1):
            if stop.is_set():
                return
        while not stop.is_set():
            started = time.monotonic()
            try:
                result = analyzer.analyze_buffered()
                record = {"risk_score": result.risk_score, "noise": result.noise}
                scores.append(record)
                print(json.dumps(record))
            except Exception as error:
                failures.put(type(error).__name__)
                stop.set()
                return
            finally:
                first_pass_done.set()
            elapsed = time.monotonic() - started
            if elapsed > interval:
                print("Analysis exceeded its interval; reduce frequency or concurrency")
            stop.wait(max(0, interval - elapsed))

    worker = threading.Thread(target=analyze)
    worker.start()
    try:
        started = time.monotonic()
        buffered = 0
        for offset in range(0, len(audio), config.block_size):
            if stop.is_set():
                break
            chunk = audio[offset:offset + config.block_size]
            block = np.zeros(config.block_size, dtype=np.float32)
            block[:len(chunk)] = chunk
            collector.buffer(block)
            buffered += len(chunk)
            if buffered >= WINDOW_SECONDS * rate:
                ready.set()
            stop.wait(max(0, started + buffered / rate - time.monotonic()))
        # A full window must get one analysis attempt before shutdown.
        if ready.is_set() and not stop.is_set():
            first_pass_done.wait()
    finally:
        stop.set()
        worker.join()
        analyzer.terminate_session()
    if not failures.empty():
        raise RuntimeError(f"Analysis failed: {failures.get()}")
    if not scores:
        raise RuntimeError("No score was produced; use a longer recording")
    print(f"Completed {len(scores)} analysis pass(es)")


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("input")
    parser.add_argument("--interval", type=float, default=5.0)
    args = parser.parse_args()
    main(args.input, args.interval)
```

Run:

```bash theme={null}
python analyze_stream.py input.wav --interval 5
```

Expect JSON scores after the first full window and a completed-pass count. Scores depend on your input. Authorization or analysis errors fail the run.

The script allocates blocks and uses Python synchronization to keep the example clear. It is a paced file demonstration, not allocation-free callback code. In a native real-time callback, preallocate buffers and keep logging, waits, model loading and analysis off the audio thread.

### The first few seconds

Before the collector has a full window, the analyzer fills missing context with silence. The example waits until it has buffered 5 s of real input before accepting a score. This avoids interpreting a partially empty startup window as a representative call segment.

Keep sample count, audio timestamps and analysis timestamps separate. A new score describes the latest buffered window, not necessarily a new non-overlapping span. Use the [smoothing guidance](/models/audio-insight/tyto#smoothing-for-real-time-use-cases) when designing interventions.

## Reset between streams

Call `analyzer.reset()` on discontinuity or before reusing a pair for another call. This requests clearing of analyzer and collector history while preserving collector configuration. Clear your application warmup count and smoothing state too.

Coordinate reset with the owner of the input stream and analysis worker so old frames cannot arrive after reset. A native reset operation is real-time safe, but the surrounding Python/task coordination is not an allocation-free callback guarantee. A terminated session needs a new pair.

## Running alongside enhancement and VAD

Give the collector and dedicated voice activity detector (VAD) the original input. The Python processor returns enhanced output; native in-place APIs should run enhancement after the read-only consumers:

```python theme={null}
collector.buffer(audio)
vad.process(audio)
enhanced = processor.process(audio)
```

Initialize all consumers for the actual host rate and block size. The model's native rate is not necessarily the input rate; the core collector handles supported non-native host rates. See [audio format](/reference/concepts/audio-format).

## When analysis stops being allowed

`ProcessingNotAllowedError` can occur after startup as well as on initial analysis. Inspect authorization, session termination and required reporting/connectivity conditions using [troubleshooting](/production/troubleshooting#processing-becomes-disallowed).

Report degraded analysis, preserve your application's audio policy and bound recovery attempts. Do not treat stale scores as current results. Enhancement and VAD have separate processing-error paths.

## Entry points per language

| Language | Create | Buffer | Analyze |
| :- | :- | :- | :- |
| Python | [`aic.analyzer_pair(model, key)`](/reference/sdk/api/python/analysis#analyzer_pair) | [`collector.buffer(audio)`](/reference/sdk/api/python/analysis#collector-buffer) | [`analyzer.analyze_buffered()`](/reference/sdk/api/python/analysis#analyzer-analyze_buffered) |
| Rust | [`aic_sdk::analyzer_pair(&model, key)`](/reference/sdk/api/rust/analysis#rust-aic_sdk-analyzer_pair) | [`collector.buffer(&audio)`](/reference/sdk/api/rust/analysis#rust-aic_sdk-Collector-buffer) | [`analyzer.analyze_buffered()`](/reference/sdk/api/rust/analysis#rust-aic_sdk-Analyzer-analyze_buffered) |
| Node.js | [`new Analyzer(model, key)`](/reference/sdk/api/node/analysis#node-analyzer-constructor) | [`analyzer.buffer(samples)`](/reference/sdk/api/node/analysis#node-analyzer-buffer) | [`analyzer.analyze()`](/reference/sdk/api/node/analysis#node-analyzer-analyze) / [`await analyzer.analyzeAsync()`](/reference/sdk/api/node/analysis#node-analyzer-analyzeasync) |
| C++ | [`aic::AnalyzerPair::create(model, key)`](/reference/sdk/api/cpp/analysis#aic-AnalyzerPair-create) | [`collector.buffer(audio.data(), audio.size())`](/reference/sdk/api/cpp/analysis#aic-Collector-buffer) | [`analyzer.analyze_buffered()`](/reference/sdk/api/cpp/analysis#aic-Analyzer-analyze_buffered) |
| C | [`aic_analyzer_pair_create(&collector, &analyzer, model, key)`](/reference/sdk/api/c/analysis#aic_analyzer_pair_create) | [`aic_collector_buffer(...)`](/reference/sdk/api/c/analysis#aic_collector_buffer) | [`aic_analyzer_analyze_buffered(...)`](/reference/sdk/api/c/analysis#aic_analyzer_analyze_buffered) |
| WASM | [`new Analyzer(model, key)`](/reference/sdk/api/wasm/analysis#wasm-analyzer-constructor) | [`analyzer.buffer(audio)`](/reference/sdk/api/wasm/analysis#wasm-analyzer-buffer) | [`analyzer.analyze()`](/reference/sdk/api/wasm/analysis#wasm-analyzer-analyze) |

Check initialization, errors and supported versions in your [binding guide](/reference/sdk/language-bindings). Node.js `analyze()` blocks its thread; `analyzeAsync()` uses a worker. Browser analysis also needs scheduling appropriate to its execution context.

## Find out more

* [Tyto scores and limitations](/models/audio-insight/tyto)
* [Batch call analysis](/models/audio-insight/batch-call-analysis)
* [Audio evaluation](/guides/evaluate-audio)
* [SDK examples](/reference/sdk/examples)


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