Optimisation pour la Production

Du prototype au service fiable

Vous savez transcrire de l’audio, en différé comme en direct. Faire tourner un service qui traite des centaines de fichiers par jour sans surveillance humaine est un tout autre exercice, parce qu’il rencontre ce que le prototype n’a jamais vu : un fichier corrompu, un quota dépassé, une coupure réseau au milieu d’un envoi, un pic de volume qui déclenche le rate limiting. Cette leçon traite ces cas un par un — gestion d’erreurs, traitement par lots, préparation des fichiers, observabilité.

Une gestion d’erreurs qui distingue le récupérable du définitif

La première règle est de ne pas traiter toutes les erreurs de la même façon. Une coupure réseau, un timeout ou une indisponibilité passagère méritent une nouvelle tentative ; un fichier illisible ou une clé API invalide ne s’arrangeront jamais tout seuls, et les retenter revient à perdre du temps et à polluer vos journaux. Le wrapper ci-dessous matérialise cette distinction avec un tuple d’exceptions temporaires : ce qui figure dans ERREURS_TEMPORAIRES déclenche un retry, tout le reste échoue immédiatement. Dans les deux cas, la fonction renvoie un dictionnaire plutôt que de laisser remonter l’exception, ce qui permet à un traitement par lots de continuer sur les autres fichiers.

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

logger = logging.getLogger("voxtral")

# Exceptions à retenter
ERREURS_TEMPORAIRES = (
    ConnectionError,
    TimeoutError,
    OSError,
)


def transcrire_avec_retry(
    chemin_audio: str,
    max_tentatives: int = 3,
    delai_base: float = 2.0,
    **kwargs
) -> dict:
    """Transcrit un fichier avec retry exponentiel."""
    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-latest"),
                    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:
            # Erreur non récupérable (fichier invalide, auth, etc.)
            logger.error(f"Erreur fatale pour {fichier.name} : {e}")
            return {
                "succes": False,
                "erreur": str(e),
                "fichier": chemin_audio,
                "tentative": tentative,
            }

Le délai entre deux tentatives double à chaque échec — deux secondes, puis quatre, puis huit — afin de ne pas achever une API déjà en difficulté. Ce backoff exponentiel souffre toutefois d’un défaut dès que vos clients sont nombreux : tous ayant échoué au même instant, tous retenteront au même instant, et la vague retombera d’un bloc sur le service. On corrige cela par un jitter, une variation aléatoire ajoutée au délai, qui suffit à désynchroniser les tentatives.

import random

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

Traiter un lot de fichiers sans le surveiller

Transcrire deux cents enregistrements l’un après l’autre est un gaspillage : l’essentiel du temps se passe à attendre le réseau. Un pool de workers mène plusieurs transcriptions de front sans vous faire perdre la progression. La fonction suivante balaie un dossier, retient les extensions audio connues, soumet chaque fichier au wrapper de retry défini plus haut, et écrit la transcription dès qu’elle arrive plutôt qu’à la fin : si le processus tombe à mi-parcours, ce qui est traité est déjà sur disque. Un rapport JSON récapitule succès et échecs, de quoi relancer uniquement ce qui a échoué.

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
):
    """Transcrit tous les fichiers audio d'un dossier en parallèle."""
    dossier = Path(dossier_audio)
    sortie = Path(dossier_sortie)
    sortie.mkdir(parents=True, exist_ok=True)

    # Trouver tous les fichiers audio
    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}"
            )

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

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

    # Sauvegarder le rapport
    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


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

Ne cédez pas à la tentation d’augmenter max_workers pour aller plus vite. L’API Mistral impose des limites de requêtes par minute, et quatre workers vous maintiennent généralement en dessous. Si des erreurs 429 (Too Many Requests) apparaissent dans vos journaux, la réponse n’est pas de retenter plus fort mais de réduire le nombre de workers ou d’introduire un délai entre les requêtes.

Préparer l’audio en amont

Le format des fichiers que vous envoyez influence à la fois le temps d’upload et la qualité du texte obtenu. Si la fidélité prime, restez sur du WAV ou du FLAC, sans perte : le modèle reçoit l’intégralité de l’information audio. Pour un usage courant, un MP3 à 128 kbps ou plus offre le meilleur compromis, avec une qualité amplement suffisante pour la transcription et une taille raisonnable. Et lorsque la bande passante est la contrainte dominante, l’OGG Vorbis ou un MP3 à 64 kbps restent parfaitement acceptables sur de la parole, très différente en cela de la musique.

Plutôt que de subir les formats qui vous arrivent, normalisez-les avant l’envoi. La fonction ci-dessous applique quatre transformations : passage en mono, la stéréo n’apportant rien à la transcription ; rééchantillonnage à 16 kHz, le taux natif de Voxtral ; débit à 64 kbps, suffisant pour la voix ; et loudnorm, qui égalise le volume.

import subprocess


def preparer_audio(chemin_entree: str, chemin_sortie: str = None) -> str:
    """Prépare un fichier audio pour une transcription optimale."""
    if chemin_sortie is None:
        chemin_sortie = chemin_entree.rsplit(".", 1)[0] + "_prep.mp3"

    subprocess.run([
        "ffmpeg", "-y",
        "-i", chemin_entree,
        "-ac", "1",          # Mono (la stéréo n'aide pas la transcription)
        "-ar", "16000",       # 16 kHz (taux natif de Voxtral)
        "-b:a", "64k",        # 64 kbps (suffisant pour la parole)
        "-af", "loudnorm",    # Normalisation du volume
        chemin_sortie
    ], check=True, capture_output=True)

    return chemin_sortie

Le gain est double : la conversion en mono 16 kHz 64 kbps réduit considérablement la taille des fichiers sans dégrader la transcription, et loudnorm améliore nettement les résultats sur les enregistrements au volume irrégulier, typiquement une réunion où un participant est loin du micro.

Savoir ce qui se passe

Un service de transcription sans instrumentation est une boîte noire : le jour où les temps de traitement doublent, vous l’apprendrez par un utilisateur mécontent. Collectez pour chaque appel la taille du fichier, la durée du traitement, le nombre de tentatives et le volume de texte produit, puis écrivez le tout en JSON sur une seule ligne, format directement exploitable par un agrégateur de logs.

import time
import logging
import json
from datetime import datetime

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


def transcrire_avec_metriques(chemin_audio: str, **kwargs) -> dict:
    """Transcrit avec collecte de métriques."""
    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)

    # Log structuré (JSON)
    logger.info(json.dumps(metriques, ensure_ascii=False))

    return resultat

Checklist de mise en production

Avant d’ouvrir votre service au premier utilisateur réel, passez ces sept points en revue :

La dernière ligne est celle que l’on néglige le plus et qui coûte le plus cher. Trois enregistrements propres en français ne vous apprendront rien du comportement réel de votre service : constituez un échantillon volontairement difficile, et vous découvrirez les réglages à ajuster pendant qu’il en est encore temps.

Points clés à retenir

• Implémentez un retry avec backoff exponentiel et jitter pour les erreurs temporaires
• Utilisez un pool de workers pour le traitement batch (4 workers est un bon point de départ)
• Normalisez vos fichiers en mono 16 kHz avec FFmpeg pour des résultats optimaux
• Collectez des métriques structurées pour le monitoring en production
• Respectez les limites de rate limiting de l’API Mistral
• Sauvegardez systématiquement les transcriptions et les rapports de batch