Implémenter la Transcription Temps Réel en Python

Le flux complet en Python asynchrone

Vous allez maintenant écrire une transcription temps réel complète, du premier octet audio jusqu’au fichier texte final. Tout repose sur asyncio : le programme doit lire l’audio et recevoir les événements du modèle en même temps, sans qu’aucune des deux opérations ne bloque l’autre. Un code synchrone classique serait ici condamné à attendre, et la latence que vous avez si soigneusement configurée s’évanouirait dans les temps morts de votre propre programme.

L’architecture d’un programme de streaming

Quelle que soit l’application, la structure reste la même et se lit de gauche à droite. En amont, une source audio : un générateur asynchrone qui produit des chunks d’audio PCM, aujourd’hui un fichier, demain un microphone ou un flux réseau. Au centre, le client Voxtral, c’est-à-dire le SDK, qui pousse ces chunks vers le modèle et vous rend des événements. En aval, le traitement des événements : votre logique métier, celle qui affiche le texte, l’accumule, l’enregistre ou le transmet à un autre service.

Source Audio → [chunks PCM] → SDK Mistral → [événements] → Votre Code

Cette séparation a une vertu pratique : la source audio est interchangeable. Le code de transcription que vous écrivez ci-dessous fonctionnera à l’identique avec le générateur microphone de la leçon suivante, sans qu’une seule ligne ne change.

L’implémentation de base

Deux détails méritent votre attention dans ce premier programme. Le générateur source_audio_fichier découpe le fichier en blocs de 32 000 octets, ce qui correspond exactement à une seconde d’audio PCM 16 bits mono en 16 kHz : 16 000 échantillons multipliés par 2 octets. Le await asyncio.sleep qui suit chaque yield sert à simuler le temps réel ; commentez cette ligne et relancez le programme sur un extrait de quelques minutes, vous verrez le fichier être avalé en une fraction de seconde, sans rien vous apprendre du comportement en streaming. Côté réception, texte_complet accumule les fragments pour que vous disposiez, à la fin, du texte intégral en plus de l’affichage direct.

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


async def source_audio_fichier(chemin: str, taille_chunk: int = 32000):
    """Générateur asynchrone qui lit un fichier PCM par chunks.

    taille_chunk = 32000 octets = 1 seconde d'audio PCM 16-bit mono 16kHz
    (16000 échantillons × 2 octets = 32000 octets)
    """
    with open(chemin, "rb") as f:
        while True:
            chunk = f.read(taille_chunk)
            if not chunk:
                break
            yield chunk
            # Simuler le temps réel (1 chunk = 1 seconde)
            await asyncio.sleep(len(chunk) / 32000)


async def transcrire_realtime(source_audio):
    """Transcrit un flux audio en temps réel avec 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-2602",
        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 terminée ---")

    return "".join(texte_complet)


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

    # Sauvegarder le résultat
    with open("transcription_realtime.txt", "w", encoding="utf-8") as f:
        f.write(texte)
    print(f"\nTexte sauvegardé ({len(texte)} caractères)")


asyncio.run(main())

L’affichage en direct repose sur print(event.text, end="", flush=True). Sans flush=True, Python bufferise la sortie et le texte apparaîtrait par paquets : votre transcription serait techniquement temps réel, mais visuellement saccadée.

Structurer le traitement pour la production

Dès que votre application dépasse la démonstration, la boucle if/elif devient encombrante : vous voulez compter les fragments, mesurer la durée, brancher un affichage, alimenter une base. Encapsuler la logique dans une classe règle le problème une fois pour toutes. Chaque type d’événement obtient sa méthode dédiée, et vous pouvez ajouter un comportement — par exemple envoyer chaque delta à une interface web — sans toucher à la boucle de réception.

class GestionnaireTranscription:
    """Gère les événements de transcription en temps réel."""

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

    async def traiter(self, source_audio):
        """Lance la transcription et traite les événements."""
        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-2602",
            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):
        """Appelé à chaque nouveau fragment de texte."""
        self.texte_complet.append(event.text)
        self.nb_deltas += 1
        # Affichage en direct
        print(event.text, end="", flush=True)

    def _on_done(self, event):
        """Appelé à la fin de la transcription."""
        import time
        duree = time.time() - self.debut
        print(f"\n\nTerminé en {duree:.1f}s")
        print(f"Fragments reçus : {self.nb_deltas}")

    def obtenir_resultat(self) -> str:
        """Retourne le texte complet transcrit."""
        return "".join(self.texte_complet)


async def main():
    gestionnaire = GestionnaireTranscription()
    source = source_audio_fichier("audio.pcm")
    texte = await gestionnaire.traiter(source)
    print(f"Résultat : {len(texte)} caractères")


asyncio.run(main())

Le compteur nb_deltas n’est pas un gadget : rapporté à la durée de l’audio, il vous indique le rythme d’émission du modèle et vous alerte si le flux se dégrade.

Survivre aux coupures réseau

Une connexion en streaming reste ouverte pendant toute la durée de la parole, ce qui l’expose bien plus qu’une requête offline de quelques secondes. Un Wi-Fi qui vacille, un changement de réseau sur un portable, une coupure côté opérateur, et le flux s’interrompt. Le pattern ci-dessous relance la transcription jusqu’à trois fois, avec un délai qui s’allonge à chaque tentative pour laisser au réseau le temps de se rétablir. Au-delà, l’exception est propagée : mieux vaut un échec explicite qu’une application qui semble écouter dans le vide.

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

MAX_TENTATIVES = 3
DELAI_RECONNEXION = 2  # secondes


async def transcrire_avec_retry(source_audio, tentative=1):
    """Transcription avec reconnexion automatique."""
    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-2602",
            audio_format=audio_format,
        ):
            if isinstance(event, TranscriptionStreamTextDelta):
                print(event.text, end="", flush=True)
            elif isinstance(event, TranscriptionStreamDone):
                print("\n[Terminé]")

    except ConnectionError as e:
        if tentative < MAX_TENTATIVES:
            print(f"\nErreur réseau (tentative {tentative}/{MAX_TENTATIVES})")
            await asyncio.sleep(DELAI_RECONNEXION * tentative)
            await transcrire_avec_retry(source_audio, tentative + 1)
        else:
            print(f"\nÉchec après {MAX_TENTATIVES} tentatives : {e}")
            raise

Préparer un fichier existant

Pour tester avec un MP3 ou un WAV que vous avez déjà sous la main, la conversion en PCM peut être automatisée depuis Python plutôt que lancée à la main. La fonction ci-dessous délègue le travail à FFmpeg, déduit le nom de sortie du nom d’entrée et renvoie le chemin obtenu, prêt à être passé à source_audio_fichier.

import subprocess

def convertir_en_pcm(chemin_entree: str, chemin_sortie: str = None) -> str:
    """Convertit un fichier audio en PCM 16-bit 16kHz mono 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

# Utilisation
chemin_pcm = convertir_en_pcm("reunion.mp3")
# Puis utiliser source_audio_fichier(chemin_pcm)

Deux précautions avec cette fonction. check=True fait remonter une exception si FFmpeg échoue, ce qui vaut mieux qu’un fichier PCM vide découvert plus tard dans le flux ; et capture_output=True masque la sortie de l’outil, si bien qu’une conversion longue donnera l’impression d’un programme figé. Sur des enregistrements d’une heure, prévoyez un message avant et après l’appel.

Points clés à retenir

• Le streaming utilise des générateurs asynchrones (async for, yield)
• Un chunk de 32 000 octets = 1 seconde d’audio PCM 16-bit 16 kHz mono
• Structurez le traitement des événements dans une classe pour la production
• Implémentez un mécanisme de retry pour gérer les erreurs réseau
• Utilisez FFmpeg pour convertir les fichiers audio existants en PCM
• L’affichage en direct utilise print(text, end="", flush=True)