Capture the Microphone with PyAudio

Course updated November 3, 2026 for Voxtral 3, based on Mistral’s Migrate from Voxtral 2 guide: Voxtral 2 models (voxtral-mini-2602, voxtral-mini-transcribe-realtime-2602, voxtral-mini-tts-2603), deprecated October 31, 2026 and retired April 30, 2027, are replaced in examples by voxtral-mini-transcribe-3, voxtral-mini-transcribe-realtime-3, and voxtral-mini-tts-3-0. Key changes: timestamps are segment-only (word returns a 400 error), real-time diarization is supported, real-time session settings are locked after the first audio chunk, and context_bias inputs are flexible (phrases with spaces). Sources: Migrate from Voxtral 2 · Speech to Text.

From File to Microphone

The previous lesson established an architecture where the audio source is a simple, interchangeable asynchronous generator. You’ll leverage that immediately: by swapping the file-reading generator for one that reads from your sound card, your program starts transcribing your voice live without changing a single line of the rest of the code.

The tool for this lesson is PyAudio, the standard Python library for accessing audio devices. It relies on PortAudio, a native layer you may need to install separately depending on your system.

Installing PyAudio

# Linux (Debian/Ubuntu)
sudo apt-get install portaudio19-dev
pip install pyaudio

# macOS
brew install portaudio
pip install pyaudio

# Windows
pip install pyaudio

The Microphone Generator

The core of this lesson fits in about thirty lines. The function opens an input stream on the default device, loops indefinitely reading fixed-size blocks, and yields each one to the transcription engine. The finally block isn’t decorative: it guarantees the stream stops, closes, and PyAudio releases the hardware even if the program is interrupted abruptly. Without it, the microphone can remain locked, blocking other applications.

import asyncio
import pyaudio

async def iter_microphone(
    sample_rate: int = 16000,
    chunk_duration_ms: int = 480
):
    """Capture the microphone and yield PCM chunks.

    Args:
        sample_rate: Sampling rate (16000 Hz for Voxtral)
        chunk_duration_ms: Duration of each chunk in milliseconds

    Yields:
        bytes: 16-bit mono PCM audio chunks
    """
    p = pyaudio.PyAudio()

    # Number of samples per chunk
    chunk_samples = int(sample_rate * chunk_duration_ms / 1000)

    # Open the microphone stream
    stream = p.open(
        format=pyaudio.paInt16,    # 16-bit PCM
        channels=1,                 # Mono
        rate=sample_rate,           # 16 kHz
        input=True,                 # Capture mode
        frames_per_buffer=chunk_samples
    )

    loop = asyncio.get_running_loop()

    try:
        print("Microphone active — speak...")
        while True:
            # Read the chunk non-blocking
            data = await loop.run_in_executor(
                None,
                stream.read,
                chunk_samples,
                False  # exception_on_overflow
            )
            yield data
    finally:
        stream.stop_stream()
        stream.close()
        p.terminate()
        print("Microphone closed")

Every parameter has a specific reason. sample_rate=16000 reflects a model requirement, not a preference: Voxtral expects 16 kHz, so don’t change this value. pyaudio.paInt16 matches the signed 16-bit PCM format, exactly equivalent to the pcm_s16le declared on the API side, and channels=1 enforces mono, which is also required. That leaves chunk_duration_ms=480, the only truly debatable setting: each chunk represents 480 milliseconds of audio, a good compromise between responsiveness and network efficiency. A shorter chunk, say 100 ms, improves responsiveness but multiplies network requests; a longer chunk saves bandwidth at the cost of added latency before the model sees the audio.

The loop.run_in_executor call also deserves explanation. stream.read is a blocking function: it holds the thread until the requested number of samples is available. Running it directly in a coroutine would freeze the entire asyncio loop, including reception of transcription events. By offloading it to an executor, you make the reading compatible with the rest of the asynchronous program.

The block size calculation follows mechanically from the two settings, for sample_rate=16000 and chunk_duration_ms=480:

chunk_samples = 16000 × 480 / 1000 = 7680 samples
byte_size = 7680 × 2 bytes = 15360 bytes per chunk

Complete Program: From Microphone to Transcription

The following script combines capture and transcription into a ready-to-use dictation tool. The new element compared to the previous generator is clean shutdown handling: a signal handler intercepts Ctrl+C, flips the running flag to False, which exits the capture loop, closes the stream, and allows the program to save the accumulated text. Without this mechanism, a keyboard interrupt would kill the process before saving, and you’d lose the dictation.

import asyncio
import os
import signal
from mistralai.client import Mistral
from mistralai.client.models import (
    AudioFormat,
    TranscriptionStreamTextDelta,
    TranscriptionStreamDone
)
import pyaudio

# Flag for clean shutdown
running = True

def signal_handler(sig, frame):
    global running
    print("\n\nShutting down...")
    running = False

signal.signal(signal.SIGINT, signal_handler)

async def iter_microphone(sample_rate=16000, chunk_duration_ms=480):
    """Capture the microphone until interrupted (Ctrl+C)."""
    p = pyaudio.PyAudio()
    chunk_samples = int(sample_rate * chunk_duration_ms / 1000)

    stream = p.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=sample_rate,
        input=True,
        frames_per_buffer=chunk_samples
    )

    loop = asyncio.get_running_loop()

    try:
        while running:
            data = await loop.run_in_executor(
                None, stream.read, chunk_samples, False
            )
            yield data
    finally:
        stream.stop_stream()
        stream.close()
        p.terminate()

async def main():
    client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
    audio_format = AudioFormat(encoding="pcm_s16le", sample_rate=16000)

    print("=" * 50)
    print("LIVE TRANSCRIPTION")
    print("Speak into the microphone (Ctrl+C to stop)")
    print("=" * 50)
    print()

    full_text = []

    async for event in client.audio.realtime.transcribe_stream(
        audio_stream=iter_microphone(),
        model="voxtral-mini-transcribe-realtime-3",
        audio_format=audio_format,
    ):
        if isinstance(event, TranscriptionStreamTextDelta):
            print(event.text, end="", flush=True)
            full_text.append(event.text)

        elif isinstance(event, TranscriptionStreamDone):
            break

    # Save
    final_text = "".join(full_text)
    if final_text.strip():
        with open("dictee.txt", "w", encoding="utf-8") as f:
            f.write(final_text)
        print(f"\n\nTranscription saved to dictee.txt")
        print(f"Length: {len(final_text)} characters")
    else:
        print("\n\nNo text transcribed.")

asyncio.run(main())

Run it, speak, stop with Ctrl+C: the text appears as you speak, then ends up in dictee.txt.

python dictee_vocale.py
# Speak into your microphone
# Text appears in real time
# Ctrl+C to stop

Try it with about thirty seconds of continuous French, varying your pace, and compare what you said with what the file contains. It’s the best way to calibrate your expectations before tuning latency in the next lesson.

Choosing the Right Device

On a machine with a built-in mic, a USB headset, and a webcam, the default device isn’t always the one you think, and an empty transcription usually comes down to this detail. First, list available inputs with their index:

def list_microphones():
    """List all available audio input devices."""
    p = pyaudio.PyAudio()
    print("Available input devices:")
    for i in range(p.get_device_count()):
        info = p.get_device_info_by_index(i)
        if info["maxInputChannels"] > 0:
            print(f"  [{i}] {info['name']} "
                  f"(channels: {info['maxInputChannels']}, "
                  f"rate: {int(info['defaultSampleRate'])} Hz)")
    p.terminate()

list_microphones()

Then simply enforce the chosen index when opening the stream, via the input_device_index parameter:

stream = p.open(
    format=pyaudio.paInt16,
    channels=1,
    rate=sample_rate,
    input=True,
    input_device_index=2,  # Device index
    frames_per_buffer=chunk_samples
)

Key Takeaways

• PyAudio captures the microphone in 16-bit mono PCM at 16 kHz, exactly what Voxtral expects
• The iter_microphone generator yields 480 ms chunks by default
• loop.run_in_executor makes microphone reading non-blocking in asyncio
• Ctrl+C cleanly stops capture via a signal handler
• List devices with get_device_count() to choose the right microphone