Aller au contenu principal

Optimisation pour la Production

Passer du prototype à la production

Vous savez désormais transcrire de l’audio en mode offline et en temps réel. Mais un prototype fonctionnel est loin d’un système de production fiable. Cette leçon couvre les aspects critiques : gestion d’erreurs, batching, formats optimaux, retry et observabilité.

Gestion d’erreurs robuste

En production, les erreurs sont inévitables : fichiers corrompus, dépassements de quota, timeouts réseau. Voici un wrapper complet :

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 backoff exponentiel

Le délai entre les tentatives double à chaque échec : 2s, 4s, 8s. Cela évite de surcharger l’API lorsqu’elle est temporairement indisponible. Pour les systèmes à fort volume, ajoutez un jitter (variation aléatoire) pour éviter que tous les clients retentent au même moment :

import random

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

Traitement batch de fichiers multiples

Pour transcrire un grand nombre de fichiers, utilisez un pool de workers avec suivi de progression :

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"]
)

Limites de concurrence

L’API Mistral a des limites de requêtes par minute (rate limiting). Avec max_workers=4, vous restez généralement sous ces limites. Si vous recevez des erreurs 429 (Too Many Requests), réduisez le nombre de workers ou ajoutez un délai entre les requêtes.

Formats audio optimaux

Le choix du format audio impacte la taille du fichier (et donc le temps d’upload) et la qualité de la transcription.

Recommandations

  • Pour la meilleure qualité : WAV ou FLAC (sans perte). Le modèle reçoit toute l’information audio.
  • Pour le meilleur compromis : MP3 à 128 kbps ou plus. Qualité suffisante pour la transcription avec une taille raisonnable.
  • Pour minimiser la bande passante : OGG Vorbis ou MP3 à 64 kbps. La qualité reste acceptable pour de la parole.

Prétraitement avec FFmpeg

Normalisez vos fichiers avant la transcription pour des résultats optimaux :

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

La conversion en mono 16 kHz 64 kbps réduit considérablement la taille du fichier sans impact sur la qualité de la transcription. La normalisation du volume (loudnorm) améliore les résultats sur les enregistrements avec un volume variable.

Observabilité et logging

En production, vous devez suivre les performances et les erreurs :

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 de déployer votre système de transcription, vérifiez :

  • Gestion d’erreurs : retry avec backoff exponentiel pour les erreurs temporaires
  • Rate limiting : respecter les limites de l’API, implémenter un throttling si nécessaire
  • Monitoring : logs structurés, métriques de performance, alertes sur les erreurs
  • Sécurité : clé API en variable d’environnement, jamais en dur dans le code
  • Stockage : sauvegarder les transcriptions de manière durable, pas uniquement en mémoire
  • Format audio : normaliser les fichiers en amont (mono, 16 kHz) pour des résultats optimaux
  • Tests : tester avec des fichiers variés (langues, durées, qualité audio, nombre de locuteurs)

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
← Lecon precedenteOffline vs Realtime : Guide de ChoixLecon suivante →Prochaines Étapes : Au-delà de la Transcription