Implement Real-Time Transcription in Python

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), diarization is possible in real time, real-time session settings are locked after the first audio chunk, and context_bias inputs are free-form (phrases with spaces). Sources: Migrate from Voxtral 2 · Speech to Text.

The Complete Python Async Stream

You are now going to write a complete real-time transcription pipeline, from the first audio byte to the final text file. Everything relies on asyncio: the program must read audio and receive model events simultaneously, without either operation blocking the other. Standard synchronous code would simply wait here, and the latency you carefully configured would vanish into the dead time of your own program.

Streaming Program Architecture

Regardless of the application, the structure remains the same and reads from left to right. Upstream, an audio source: an asynchronous generator that produces PCM audio chunks. Today it might be a file, tomorrow a microphone or a network stream. In the center, the Voxtral client, meaning the SDK, which pushes these chunks to the model and yields events to you. Downstream, event processing: your business logic, which displays the text, accumulates it, saves it, or forwards it to another service.

Audio Source → [PCM chunks] → Mistral SDK → [events] → Your Code

This separation has a practical benefit: the audio source is interchangeable. The transcription code you write below will work identically with the microphone generator from the next lesson, without changing a single line.

Basic Implementation

Two details in this first program deserve your attention. The source_audio_fichier generator slices the file into 32,000-byte blocks, which corresponds exactly to one second of 16-bit mono PCM audio at 16 kHz: 16,000 samples multiplied by 2 bytes. The await asyncio.sleep following each yield simulates real time; comment out this line and run the program on a few-minute excerpt, and you will see the file consumed in a fraction of a second, teaching you nothing about streaming behavior. On the reception side, texte_complet accumulates fragments so you have the full text available at the end, in addition to the live display.

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

async def source_audio_fichier(chemin: str, taille_chunk: int = 32000):
    """Asynchronous generator that reads a PCM file in chunks.

    taille_chunk = 32000 bytes = 1 second of 16-bit mono PCM audio at 16kHz
    (16000 samples × 2 bytes = 32000 bytes)
    """
    with open(chemin, "rb") as f:
        while True:
            chunk = f.read(taille_chunk)
            if not chunk:
                break
            yield chunk
            # Simulate real time (1 chunk = 1 second)
            await asyncio.sleep(len(chunk) / 32000)

async def transcrire_realtime(source_audio):
    """Transcribes an audio stream in real time with Voxtral."""
    client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

    audio_format = AudioFormat(
        encoding="pcm_s16le",
        sample_rate=16000
    )

    texte_complet = []

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

        elif isinstance(event, TranscriptionStreamDone):
            print("\n\n--- Transcription finished ---")

    return "".join(texte_complet)

async def main():
    source = source_audio_fichier("audio.pcm")
    texte = await transcrire_realtime(source)

    # Save the result
    with open("transcription_realtime.txt", "w", encoding="utf-8") as f:
        f.write(texte)
    print(f"\nText saved ({len(texte)} characters)")

asyncio.run(main())

The live display relies on print(event.text, end="", flush=True). Without flush=True, Python buffers the output and text would appear in batches: your transcription would be technically real-time, but visually jerky.

Structuring Processing for Production

Once your application moves beyond a demo, the if/elif loop becomes cumbersome: you want to count fragments, measure duration, hook up a display, or feed a database. Encapsulating the logic in a class solves the problem once and for all. Each event type gets its own dedicated method, and you can add behavior—for example, sending each delta to a web interface—without touching the reception loop.

class GestionnaireTranscription:
    """Handles real-time transcription events."""

    def __init__(self):
        self.texte_complet = []
        self.nb_deltas = 0
        self.debut = None

    async def traiter(self, source_audio):
        """Starts transcription and processes events."""
        import time
        self.debut = time.time()

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

        async for event in client.audio.realtime.transcribe_stream(
            audio_stream=source_audio,
            model="voxtral-mini-transcribe-realtime-3",
            audio_format=audio_format,
        ):
            if isinstance(event, TranscriptionStreamTextDelta):
                self._on_text_delta(event)
            elif isinstance(event, TranscriptionStreamDone):
                self._on_done(event)

        return self.obtenir_resultat()

    def _on_text_delta(self, event):
        """Called for each new text fragment."""
        self.texte_complet.append(event.text)
        self.nb_deltas += 1
        # Live display
        print(event.text, end="", flush=True)

    def _on_done(self, event):
        """Called at the end of transcription."""
        import time
        duree = time.time() - self.debut
        print(f"\n\nFinished in {duree:.1f}s")
        print(f"Fragments received: {self.nb_deltas}")

    def obtenir_resultat(self) -> str:
        """Returns the full transcribed text."""
        return "".join(self.texte_complet)

async def main():
    gestionnaire = GestionnaireTranscription()
    source = source_audio_fichier("audio.pcm")
    texte = await gestionnaire.traiter(source)
    print(f"Result: {len(texte)} characters")

asyncio.run(main())

The nb_deltas counter is not a gimmick: relative to the audio duration, it indicates the model’s emission rate and alerts you if the stream degrades.

Surviving Network Interruptions

A streaming connection stays open for the entire duration of speech, exposing it far more than a few-second offline request. A flickering Wi-Fi, a network switch on a mobile device, or an operator outage can interrupt the stream. The pattern below restarts transcription up to three times, with a delay that increases with each attempt to allow the network time to recover. Beyond that, the exception is propagated: an explicit failure is better than an application that appears to be listening to nothing.

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

MAX_TENTATIVES = 3
DELAI_RECONNEXION = 2  # seconds

async def transcrire_avec_retry(source_audio, tentative=1):
    """Transcription with automatic reconnection."""
    client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
    audio_format = AudioFormat(encoding="pcm_s16le", sample_rate=16000)

    try:
        async for event in client.audio.realtime.transcribe_stream(
            audio_stream=source_audio,
            model="voxtral-mini-transcribe-realtime-3",
            audio_format=audio_format,
        ):
            if isinstance(event, TranscriptionStreamTextDelta):
                print(event.text, end="", flush=True)
            elif isinstance(event, TranscriptionStreamDone):
                print("\n[Finished]")

    except ConnectionError as e:
        if tentative < MAX_TENTATIVES:
            print(f"\nNetwork error (attempt {tentative}/{MAX_TENTATIVES})")
            await asyncio.sleep(DELAI_RECONNEXION * tentative)
            await transcrire_avec_retry(source_audio, tentative + 1)
        else:
            print(f"\nFailed after {MAX_TENTATIVES} attempts: {e}")
            raise

Preparing an Existing File

To test with an MP3 or WAV file you already have on hand, you can automate the conversion to PCM from Python rather than running it manually. The function below delegates the work to FFmpeg, deduces the output name from the input name, and returns the resulting path, ready to pass to source_audio_fichier.

import subprocess

def convertir_en_pcm(chemin_entree: str, chemin_sortie: str = None) -> str:
    """Converts an audio file to 16-bit 16kHz mono PCM via FFmpeg."""
    if chemin_sortie is None:
        chemin_sortie = chemin_entree.rsplit(".", 1)[0] + ".pcm"

    subprocess.run([
        "ffmpeg", "-y",
        "-i", chemin_entree,
        "-f", "s16le",
        "-acodec", "pcm_s16le",
        "-ar", "16000",
        "-ac", "1",
        chemin_sortie
    ], check=True, capture_output=True)

    return chemin_sortie

# Usage
chemin_pcm = convertir_en_pcm("reunion.mp3")
# Then use source_audio_fichier(chemin_pcm)

Two precautions with this function. check=True raises an exception if FFmpeg fails, which is better than discovering an empty PCM file later in the stream; and capture_output=True hides the tool’s output, so a long conversion will make the program appear frozen. For hour-long recordings, plan to print a message before and after the call.

Key Takeaways

• Streaming uses asynchronous generators (async for, yield)
• A 32,000-byte chunk = 1 second of 16-bit mono PCM audio at 16 kHz
• Structure event processing in a class for production
• Implement a retry mechanism to handle network errors
• Use FFmpeg to convert existing audio files to PCM
• Live display uses print(text, end="", flush=True)