Aller au contenu principal

Implémenter la Transcription Temps Réel en Python

Le flux complet en Python asynchrone

Dans cette leçon, vous allez implémenter une transcription temps réel complète en Python. Le code utilise asyncio pour gérer le flux audio et la réception des événements de manière non bloquante.

Architecture du programme

Un programme de transcription temps réel se compose de trois parties :

  1. Source audio — Un générateur asynchrone qui produit des chunks d’audio PCM
  2. Client Voxtral — Le SDK qui envoie l’audio et reçoit les événements
  3. Traitement des événements — Votre logique pour afficher, stocker ou analyser le texte
Source Audio → [chunks PCM] → SDK Mistral → [événements] → Votre Code

Implémentation de base

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())

Gestion avancée des événements

En production, vous voudrez structurer le traitement des événements :

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())

Gestion des erreurs réseau

Le streaming est sensible aux interruptions réseau. Voici un pattern de reconnexion :

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

Convertir un fichier audio existant

Pour tester avec un fichier MP3 ou WAV existant, convertissez-le d’abord en PCM :

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)

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)
← Lecon precedenteIntroduction à la Transcription Temps RéelLecon suivante →Capturer le Microphone avec PyAudio