Audio runs on the mock backend

Goal: explain a message that holds speech, with a backend that reacts to the audio it hears, without any model.

Why it matters: audio pipelines need deterministic, offline tests; a backend that ignores audio would hide masking bugs behind constant outputs.

Prerequisites: the core install only; the clip is synthesised as 16-bit WAV.

Output: the heard tokens of a clip, how silencing one word changes them, and an explanation run through mock_audio.

import io
import wave

import numpy as np

from mllm_shap import (
    AudioPart,
    ConversationSnapshot,
    ExplainConfig,
    ExplainRunner,
    Message,
    MockAudioBackend,
    Role,
    TextPart,
    Turn,
)
from mllm_shap.domain.config import GenerationConfig

A clip with two spoken words

Two tone bursts stand in for two words, separated by 50 ms of silence. speech writes 16-bit WAV with the standard library, which mock_audio reads without the audio extra.

RATE = 8000


def speech(amplitudes: list[float]) -> bytes:
    t = np.arange(RATE // 20) / RATE
    waveform = np.concatenate([a * np.sin(2 * np.pi * 300 * t) for a in amplitudes])
    buffer = io.BytesIO()
    with wave.open(buffer, "wb") as writer:
        writer.setnchannels(1)
        writer.setsampwidth(2)
        writer.setframerate(RATE)
        writer.writeframes((waveform * 32767).astype("<i2").tobytes())
    return buffer.getvalue()


def snapshot(audio: bytes) -> ConversationSnapshot:
    spoken = Message(
        role=Role.USER, content=(TextPart("hello there"), AudioPart(audio, RATE))
    )
    return ConversationSnapshot.from_turns((Turn(messages=(spoken,)),))

What the backend hears

The answer echoes the message words, then adds one token per voiced run. Silencing the second word drops only its token; max_new_tokens truncates the answer.

backend = MockAudioBackend(model_id="demo")


async def heard(audio: bytes, max_new_tokens: int = 16) -> str:
    record = await backend.generate(
        snapshot(audio), GenerationConfig(max_new_tokens=max_new_tokens)
    )
    return record.text


print("both words  :", await heard(speech([0.6, 0, 0.4])))
print("second muted:", await heard(speech([0.6, 0, 0])))
print("truncated   :", await heard(speech([0.6, 0, 0.4]), max_new_tokens=3))

both words  : hello there a32e 14a3
second muted: hello there a32e
truncated   : hello there a32e

An explanation through mock_audio

backend.kind: mock_audio selects the backend from the config. Here the words of the transcript are players; with players: sgpa_segments the spoken words are players and the absence policy edits the audio the backend hears (see the alignment guide).

config = ExplainConfig.model_validate({
    "backend": {"kind": "mock_audio", "model_id": "demo"},
    "explanation": {"archive": {"path": ""}},
})
output = await ExplainRunner(config).explain(snapshot(speech([0.6, 0, 0.4])))
result = output.result
print([round(v, 4) for v in result.values])
print("efficiency gap:", round(sum(result.values) - (result.grand_value - result.base_value), 12))

[0.2101, 0.2101]
efficiency gap: 0.0

Audio it cannot read

Bytes that are not 16-bit WAV go to the audio extra’s decoder: undecodable bytes raise ValueError, and without the extra an ImportError names it.

try:
    await heard(b"not audio")
except (ValueError, ImportError) as error:
    print(type(error).__name__, "-", error)

ValueError - could not decode the audio bytes