Transcribe a WAV file and read recognition events separately

This guide first produces a reproducible local transcript from the supplied WAV file. It then shows where microphone capture and a streaming model fit. Recognition, turn-taking and model interaction remain separate actions, so an existing recorder or chat interface can adopt one layer at a time.

Hear the input, then run recognition#

Download the sample recording and play it with your usual audio player. It says: “Build AI applications like normal software. Your functions. Your architecture.” The recording was synthesized for this example. Whisper produces the transcript during the run.

Use the commands in Run this guide above from a fresh directory with an active Python 3.11+ environment. They download both the client and the exact WAV used below.

The first run resolves and downloads the tiny.en Whisper model if it is not already cached. Recognition runs on your machine. The program reads the supplied recording and leaves the microphone inactive.

You will see audio upload messages, recognizer events and a transcript:

[audio] 4000 samples
[audio] 4000 samples
…
[recognizer] inference_started
[transcript] Build AI applications like normal software, your functions, your architecture.
[recognizer] inference_stopped
[completed]

That transcript came from running this file against the supplied recording. Punctuation may vary. For this short clip, recognition can arrive after all audio has been supplied; streaming inputs do not guarantee a transcript for every audio packet.

Feed a saved recording into the recognizer#

You can test recognition without a live microphone by sending the same recording on every run. That makes it possible to compare a model or preprocessing change against a known input. The sender supplies audio buffers through the action’s audio port.

The complete file registers the native audio actions and creates transcribe_audio:

registry = a11.ActionRegistry()
actions.register(registry)
action = registry.make_action("transcribe_audio").run()
await action["asr_options"].finalize(
    SpeechRecognizerOptions(model=args.model, language="en")
)

The recognizer is a native A11 action, backed by whisper.cpp. Python supplies its configuration and audio input; the runtime executes the recognizer. You do not need an agent, graph or conversation framework to call it.

Inside send_audio(), wave.open supplies recording; the file first checks that it is mono, 16-bit PCM at 16 kHz. A WAV file stores signed 16-bit samples. The sender converts those values to floating-point samples between approximately −1 and 1 and supplies AudioBuffer objects:

while raw := recording.readframes(4000):
    pcm = array.array("h", raw)
    if sys.byteorder != "little":
        pcm.byteswap()
    samples = array.array("f", (value / 32768.0 for value in pcm))
    await action["audio"].put(
        AudioBuffer(samples, sample_rate=16000.0)
    )
    print(f"[audio] {len(samples)} samples", flush=True)
    await asyncio.sleep(0.05)
await action["audio"].finalize()

Each batch contains up to 4,000 samples: a quarter of a second at 16 kHz. The demo sends batches with a short delay so the stream is visible. After the last batch, finalize() closes the audio input and lets the recognizer finish.

Show recognized words and recognizer activity separately#

While someone waits for a transcript, your UI can show that recognition has started even if no words are ready yet. Activity events belong in that indicator; recognized words belong in the editable transcript. Mixing them would force the transcript view to sort out which messages are actually text to keep.

Three coroutines run together: one sends audio, one reads transcript pieces and one drains recognizer events. The two output consumers are:

async def read_text():
    async for text in action["transcription_pieces"]:
        print(f"[transcript] {text}", flush=True)

async def read_events():
    async for event in action["events"]:
        print(f"[recognizer] {event.kind}", flush=True)

A transcript view needs strings. An activity indicator needs events such as inference_started. Keeping those outputs separate means the transcript view never decodes model events or audio bytes. The same readers work when audio comes from a microphone instead of a file.

The sender and both readers share one lifetime. Here is the caller’s task group and cleanup:

try:
    async with asyncio.timeout(180):
        async with asyncio.TaskGroup() as tasks:
            tasks.create_task(send_audio())
            tasks.create_task(read_text())
            tasks.create_task(read_events())
        await action.wait()
    print("[completed]")
finally:
    action.cancel()

Closing audio is normal end-of-input; action.cancel() stops the invocation if sending or reading fails, the deadline expires, or the user interrupts. [completed] appears only after the final status succeeds.

Try your own mono, 16-bit PCM WAV at 16,000 Hz:

python transcribe_file.py my-recording.wav

The script rejects another format explicitly. If you already use FFmpeg, conversion is:

ffmpeg -i my-recording.m4a -ar 16000 -ac 1 -c:a pcm_s16le my-recording.wav

The supplied fixture reproduces the recognition path with no microphone recording. Press Ctrl+C to stop a run.

Speak a question, review the transcript and submit it#

Now use recognition as the input method for a conversation. Reviewing the transcript before submission lets you correct a misheard name or number before the model answers the wrong question.

The installed a11 chat CLI already connects speech input to model interaction. With the shared hosted provider, launch:

a11 chat ollama glm-5.3-flash:cloud --gateway wss://a11.to/ws/demoserver --header x-a11-llm-base-url https://ollama.com --header x-a11-llm-api-key use-a11-demo-resources --no-shell-tools --voice-model tiny.en

The speech model is cached from the first exercise. The CLI may also download its voice-activity detector. Allow microphone access for your terminal if the operating system asks. When the CLI reports that microphone input is active during your turns, speak a short question, review the recognized text in the prompt and press Enter. The model's answer streams back as text. Type /exit to finish.

Your audio is recognized locally; the submitted transcript goes to the hosted model. The public demo resource token selects the credentials configured on the demo server. No provider SDK or private API key is needed for this hosted route.

If the microphone is unavailable, add --no-voice and type the prompt. This uses the same hosted model path and skips the audio device and recognition-model download. Microphone capture depends on the device and operating-system permission.

Stop microphone capture when the question is submitted#

If the microphone kept appending words after you pressed Enter, you could not tell whether those words belonged to the submitted question or the next one. The CLI uses the submit event to end capture for that turn, then reads the model’s reply.

The CLI's chat_ui.py owns the turn loop. It starts local recognition while you compose a prompt, inserts transcript pieces into the editable input, stops capture when you submit, and consumes the model response. voice.py resolves cached recognition models. The native audio action bindings expose capture and transcription independently.

The program you ran demonstrates recognition and streaming model text. It does not yet generate spoken replies or implement a full duplex voice assistant. To add speech output, connect a speech-generation function to the reply and give audio its own output. If the assistant is reading an answer aloud and the user says “Stop, I meant tomorrow,” stop the old playback and cancel generation for that answer. A tool outcome that already happened—such as a calendar entry being created—must remain in conversation state; stopping speech cannot undo it.

The CLI command above disables shell tools because this first interaction only needs speech and a reply. An application's own tools can be ordinary registered actions, made available to its model interaction. The action-tools guide shows how to add them.

Complete source#

Download transcribe_file.py and hello-a11.wav. The source below is the exact file used by the launch commands.

"""Stream a WAV recording into A11's local speech recognizer."""

import argparse
import array
import asyncio
import sys
import wave

import a11
from a11.sdk.audio import AudioBuffer, SpeechRecognizerOptions, actions
from a11.status import Status, StatusCode


async def main(args):
    registry = a11.ActionRegistry()
    actions.register(registry)
    action = registry.make_action("transcribe_audio").run()
    await action["asr_options"].finalize(
        SpeechRecognizerOptions(model=args.model, language="en")
    )

    async def send_audio():
        with wave.open(args.file, "rb") as recording:
            if (
                recording.getnchannels(),
                recording.getsampwidth(),
                recording.getframerate(),
            ) != (1, 2, 16000):
                raise Status(
                    code=StatusCode.INVALID_ARGUMENT,
                    message="Use a mono, 16-bit PCM WAV at 16000 Hz.",
                ).to_exception()
            while raw := recording.readframes(4000):
                pcm = array.array("h", raw)
                if sys.byteorder != "little":
                    pcm.byteswap()
                samples = array.array("f", (value / 32768.0 for value in pcm))
                await action["audio"].put(
                    AudioBuffer(samples, sample_rate=16000.0)
                )
                print(f"[audio] {len(samples)} samples", flush=True)
                await asyncio.sleep(0.05)
        await action["audio"].finalize()

    async def read_text():
        async for text in action["transcription_pieces"]:
            print(f"[transcript] {text}", flush=True)

    async def read_events():
        async for event in action["events"]:
            print(f"[recognizer] {event.kind}", flush=True)

    try:
        async with asyncio.timeout(180):
            async with asyncio.TaskGroup() as tasks:
                tasks.create_task(send_audio())
                tasks.create_task(read_text())
                tasks.create_task(read_events())
            await action.wait()
        print("[completed]")
    finally:
        action.cancel()


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("file", nargs="?", default="hello-a11.wav")
    parser.add_argument("--model", default="tiny.en")
    args = parser.parse_args()
    try:
        asyncio.run(main(args))
    except KeyboardInterrupt:
        print("\nStopped.")

If a larger recognition model will not fit on the client device, run it on a machine with enough memory and keep sending audio buffers and reading transcript pieces through the same contract. The remote walkthrough teaches that move with a function that needs no microphone or model download.