Optimization for Production

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 on October 31, 2026 and retired on 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 free-form (phrases with spaces). Sources: Migrate from Voxtral 2 · Speech to Text.

From Prototype to Reliable Service

You know how to transcribe audio, both asynchronously and in real time. Running a service that processes hundreds of files daily without human supervision is a different exercise entirely, because it encounters what a prototype never sees: a corrupted file, an exceeded quota, a network drop mid-upload, or a volume spike that triggers rate limiting. This lesson addresses these cases one by one—error handling, batch processing, file preparation, and observability.

Error Handling That Distinguishes Recoverable from Fatal

The first rule is not to treat all errors the same way. A network drop, a timeout, or a temporary outage deserves a retry; an unreadable file or an invalid API key will never fix itself, and retrying them just wastes time and pollutes your logs. The wrapper below implements this distinction with a tuple of temporary exceptions: anything in ERREURS_TEMPORAIRES triggers a retry, while everything else fails immediately. In both cases, the function returns a dictionary instead of raising an exception, allowing batch processing to continue with the remaining files.

import os
import time
import logging
from pathlib import Path
from mistralai.client import Mistral

logger = logging.getLogger("voxtral")

# Exceptions to retry
ERREURS_TEMPORAIRES = (
    ConnectionError,
    TimeoutError,
    OSError,
)

def transcrire_avec_retry(
    chemin_audio: str,
    max_tentatives: int = 3,
    delai_base: float = 2.0,
    **kwargs
) -> dict:
    """Transcribes a file with exponential retry."""
    client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
    fichier = Path(chemin_audio)

    for tentative in range(1, max_tentatives + 1):
        try:
            with open(chemin_audio, "rb") as f:
                response = client.audio.transcriptions.complete(
                    model=kwargs.get("model", "voxtral-mini-transcribe-3"),
                    file={"content": f, "file_name": fichier.name},
                    **{k: v for k, v in kwargs.items() if k != "model"}
                )

            return {
                "succes": True,
                "texte": response.text,
                "fichier": chemin_audio,
                "tentative": tentative,
            }

        except ERREURS_TEMPORAIRES as e:
            if tentative < max_tentatives:
                delai = delai_base * (2 ** (tentative - 1))
                logger.warning(
                    f"Tentative {tentative}/{max_tentatives} échouée "
                    f"pour {fichier.name} : {e}. "
                    f"Retry dans {delai:.0f}s"
                )
                time.sleep(delai)
            else:
                logger.error(
                    f"Échec définitif pour {fichier.name} "
                    f"après {max_tentatives} tentatives : {e}"
                )
                return {
                    "succes": False,
                    "erreur": str(e),
                    "fichier": chemin_audio,
                    "tentative": tentative,
                }

        except Exception as e:
            # Non-recoverable error (invalid file, auth, etc.)
            logger.error(f"Erreur fatale pour {fichier.name} : {e}")
            return {
                "succes": False,
                "erreur": str(e),
                "fichier": chemin_audio,
                "tentative": tentative,
            }

The delay between attempts doubles with each failure—two seconds, then four, then eight—to avoid hammering an API that is already struggling. However, this exponential backoff has a flaw when you have many clients: if they all fail at the same instant, they will all retry at the same instant, and the wave will hit the service all at once. We fix this with jitter, a random variation added to the delay, which is enough to desynchronize the attempts.

import random

delai = delai_base * (2 ** (tentative - 1))
jitter = random.uniform(0, delai * 0.25)
time.sleep(delai + jitter)

Processing a Batch of Files Without Supervision

Transcribing two hundred recordings one after another is wasteful: most of the time is spent waiting on the network. A pool of workers handles multiple transcriptions concurrently without losing track of progress. The function below scans a folder, keeps known audio extensions, submits each file to the retry wrapper defined above, and writes the transcription as soon as it arrives rather than at the end: if the process crashes halfway through, what has been processed is already on disk. A JSON report summarizes successes and failures, allowing you to restart only what failed.

import os
import json
import logging
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed

logger = logging.getLogger("voxtral-batch")

def traiter_batch(
    dossier_audio: str,
    dossier_sortie: str,
    max_workers: int = 4,
    **kwargs_transcription
):
    """Transcribes all audio files in a folder in parallel."""
    dossier = Path(dossier_audio)
    sortie = Path(dossier_sortie)
    sortie.mkdir(parents=True, exist_ok=True)

    # Find all audio files
    extensions = {".mp3", ".wav", ".flac", ".ogg", ".m4a", ".webm"}
    fichiers = [
        f for f in dossier.iterdir()
        if f.suffix.lower() in extensions
    ]

    if not fichiers:
        logger.warning(f"Aucun fichier audio dans {dossier_audio}")
        return []

    logger.info(f"Traitement de {len(fichiers)} fichiers avec {max_workers} workers")

    resultats = []

    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        futures = {
            pool.submit(
                transcrire_avec_retry,
                str(f),
                **kwargs_transcription
            ): f
            for f in fichiers
        }

        for i, future in enumerate(as_completed(futures), 1):
            fichier = futures[future]
            resultat = future.result()
            resultats.append(resultat)

            status = "OK" if resultat["succes"] else "ERREUR"
            logger.info(
                f"[{i}/{len(fichiers)}] {status} — {fichier.name}"
            )

            # Save the transcription
            if resultat["succes"]:
                nom_sortie = fichier.stem + ".txt"
                (sortie / nom_sortie).write_text(
                    resultat["texte"], encoding="utf-8"
                )

    # Final report
    succes = sum(1 for r in resultats if r["succes"])
    echecs = len(resultats) - succes
    logger.info(f"Terminé : {succes} succès, {echecs} échecs")

    # Save the report
    rapport_path = sortie / "rapport_batch.json"
    with open(rapport_path, "w", encoding="utf-8") as f:
        json.dump(resultats, f, ensure_ascii=False, indent=2)

    return resultats

# Usage
resultats = traiter_batch(
    dossier_audio="enregistrements/",
    dossier_sortie="transcriptions/",
    max_workers=4,
    diarize=True,
    timestamp_granularities=["segment"]
)

Do not give in to the temptation to increase max_workers to go faster. The Mistral API imposes limits on requests per minute, and four workers usually keep you below that threshold. If 429 errors (Too Many Requests) appear in your logs, the answer is not to retry harder but to reduce the number of workers or introduce a delay between requests.

Preparing Audio Upstream

The format of the files you send influences both upload time and the quality of the resulting text. If fidelity is paramount, stick with WAV or FLAC, lossless formats: the model receives the full audio information. For common use, MP3 at 128 kbps or higher offers the best compromise, with quality more than sufficient for transcription and a reasonable file size. And when bandwidth is the dominant constraint, OGG Vorbis or MP3 at 64 kbps remain perfectly acceptable for speech, which is very different from music in this regard.

Rather than accepting whatever formats arrive, normalize them before sending. The function below applies four transformations: conversion to mono, since stereo adds nothing to transcription; resampling to 16 kHz, Voxtral’s native rate; bitrate set to 64 kbps, sufficient for voice; and loudnorm, which equalizes volume.

import subprocess

def preparer_audio(chemin_entree: str, chemin_sortie: str = None) -> str:
    """Prepares an audio file for optimal transcription."""
    if chemin_sortie is None:
        chemin_sortie = chemin_entree.rsplit(".", 1)[0] + "_prep.mp3"

    subprocess.run([
        "ffmpeg", "-y",
        "-i", chemin_entree,
        "-ac", "1",          # Mono (stereo does not help transcription)
        "-ar", "16000",       # 16 kHz (Voxtral native rate)
        "-b:a", "64k",        # 64 kbps (sufficient for speech)
        "-af", "loudnorm",    # Volume normalization
        chemin_sortie
    ], check=True, capture_output=True)

    return chemin_sortie

The gain is twofold: converting to mono 16 kHz 64 kbps significantly reduces file sizes without degrading transcription, and loudnorm markedly improves results on recordings with irregular volume, typically a meeting where one participant is far from the microphone.

Knowing What Is Happening

A transcription service without instrumentation is a black box: the day processing times double, you will learn about it from an unhappy user. Collect the file size, processing duration, number of attempts, and volume of text produced for each call, then write it all as JSON on a single line, a format directly usable by a log aggregator.

import time
import logging
import json
from datetime import datetime

logger = logging.getLogger("voxtral-prod")

def transcrire_avec_metriques(chemin_audio: str, **kwargs) -> dict:
    """Transcribes with metrics collection."""
    debut = time.time()
    taille_fichier = Path(chemin_audio).stat().st_size

    resultat = transcrire_avec_retry(chemin_audio, **kwargs)

    duree_traitement = time.time() - debut

    metriques = {
        "timestamp": datetime.now().isoformat(),
        "fichier": chemin_audio,
        "taille_octets": taille_fichier,
        "duree_traitement_s": round(duree_traitement, 2),
        "succes": resultat["succes"],
        "tentatives": resultat["tentative"],
    }

    if resultat["succes"]:
        metriques["nb_caracteres"] = len(resultat["texte"])
        metriques["nb_mots"] = len(resultat["texte"].split())
        ratio = taille_fichier / duree_traitement / 1024
        metriques["debit_ko_par_s"] = round(ratio, 1)

    # Structured log (JSON)
    logger.info(json.dumps(metriques, ensure_ascii=False))

    return resultat

Production Checklist

Before opening your service to the first real user, review these seven points:

The last line is the one most often neglected and the most expensive. Three clean recordings in French will teach you nothing about the real behavior of your service: build a deliberately difficult sample, and you will discover the settings to adjust while there is still time.

Key Takeaways

• Implement retry with exponential backoff and jitter for temporary errors
• Use a worker pool for batch processing (4 workers is a good starting point)
• Normalize your files to mono 16 kHz with FFmpeg for optimal results
• Collect structured metrics for production monitoring
• Respect Mistral API rate limiting limits
• Systematically save transcriptions and batch reports