Cas d'Usage de la Transcription Temps Réel

Applications concrètes du streaming

La transcription temps réel ouvre des possibilités que le mode offline ne peut pas couvrir. Dans cette leçon, vous découvrirez quatre cas d’usage majeurs avec du code fonctionnel pour chacun.

Sous-titrage en direct

Le sous-titrage en direct est le cas le plus visible du temps réel. Que ce soit pour un webinaire, une conférence ou un cours en ligne, les sous-titres apparaissent au fur et à mesure que l’orateur parle.

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


class SousTitrageDirect:
    """Système de sous-titrage en direct avec gestion des lignes."""

    def __init__(self, max_chars_par_ligne=50, max_lignes=2):
        self.max_chars = max_chars_par_ligne
        self.max_lignes = max_lignes
        self.ligne_courante = ""
        self.lignes = []

    async def lancer(self, source_audio):
        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,
            target_streaming_delay_ms=800,
        ):
            if isinstance(event, TranscriptionStreamTextDelta):
                self._ajouter_texte(event.text)
            elif isinstance(event, TranscriptionStreamDone):
                self._afficher_final()

    def _ajouter_texte(self, texte):
        self.ligne_courante += texte

        if len(self.ligne_courante) >= self.max_chars:
            pos = self.ligne_courante.rfind(" ", 0, self.max_chars)
            if pos > 0:
                self.lignes.append(self.ligne_courante[:pos])
                self.ligne_courante = self.ligne_courante[pos + 1:]
            else:
                self.lignes.append(self.ligne_courante[:self.max_chars])
                self.ligne_courante = self.ligne_courante[self.max_chars:]

            if len(self.lignes) > self.max_lignes:
                self.lignes = self.lignes[-self.max_lignes:]

        self._afficher()

    def _afficher(self):
        for ligne in self.lignes:
            print(f"  {ligne}")
        print(f"  {self.ligne_courante}", end="", flush=True)

    def _afficher_final(self):
        print("\n\n[Sous-titrage terminé]")

Assistant vocal interactif

Un assistant vocal doit comprendre l’utilisateur pendant qu’il parle pour pouvoir répondre rapidement. La transcription temps réel est le premier maillon de la chaîne : Speech-to-Text, puis LLM, puis Text-to-Speech.

async def assistant_vocal(source_audio):
    """Transcrit puis répond via un LLM."""
    client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
    audio_format = AudioFormat(encoding="pcm_s16le", sample_rate=16000)

    # Étape 1 : Transcrire la question
    question_parts = []

    async for event in client.audio.realtime.transcribe_stream(
        audio_stream=source_audio,
        model="voxtral-mini-transcribe-realtime-2602",
        audio_format=audio_format,
        target_streaming_delay_ms=300,
    ):
        if isinstance(event, TranscriptionStreamTextDelta):
            question_parts.append(event.text)
            texte_en_cours = "".join(question_parts)
            print(f"\rVous : {texte_en_cours}", end="", flush=True)
        elif isinstance(event, TranscriptionStreamDone):
            break

    question = "".join(question_parts).strip()
    print()

    if not question:
        print("Aucune question détectée.")
        return

    # Étape 2 : Envoyer au LLM pour obtenir une réponse
    print("Assistant : ", end="", flush=True)

    reponse_stream = client.chat.stream(
        model="mistral-large-latest",
        messages=[
            {
                "role": "system",
                "content": "Vous êtes un assistant vocal concis. "
                           "Répondez en 2-3 phrases maximum."
            },
            {"role": "user", "content": question}
        ]
    )

    for chunk in reponse_stream:
        texte = chunk.data.choices[0].delta.content
        if texte:
            print(texte, end="", flush=True)

    print("\n")

Dictée vocale avec sauvegarde

La dictée vocale va au-delà de la simple transcription : elle met en forme le texte et le sauvegarde au fil de l’eau.

class DicteeVocale:
    """Dictée vocale avec sauvegarde automatique."""

    def __init__(self, fichier_sortie="dictee.txt"):
        self.fichier_sortie = fichier_sortie
        self.texte = []
        self.nb_mots = 0

    async def lancer(self, source_audio):
        client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
        audio_format = AudioFormat(encoding="pcm_s16le", sample_rate=16000)

        print("Dictée en cours... (Ctrl+C pour arrêter)\n")

        async for event in client.audio.realtime.transcribe_stream(
            audio_stream=source_audio,
            model="voxtral-mini-transcribe-realtime-2602",
            audio_format=audio_format,
            target_streaming_delay_ms=600,
        ):
            if isinstance(event, TranscriptionStreamTextDelta):
                self.texte.append(event.text)
                self.nb_mots += len(event.text.split())
                print(event.text, end="", flush=True)

                # Sauvegarde automatique tous les 100 mots
                if self.nb_mots % 100 < 5:
                    self._sauvegarder()

            elif isinstance(event, TranscriptionStreamDone):
                break

        self._sauvegarder()
        print(f"\n\nDictée terminée : {self.nb_mots} mots")
        print(f"Fichier : {self.fichier_sortie}")

    def _sauvegarder(self):
        contenu = "".join(self.texte)
        with open(self.fichier_sortie, "w", encoding="utf-8") as f:
            f.write(contenu)

Transcription de conférence avec métadonnées

Pour une conférence longue, vous ajoutez des métadonnées (orateur, horodatage) et exportez en JSON :

import json
from datetime import datetime


async def transcrire_conference(
    source_audio,
    nom_conference: str,
    orateur: str = "Inconnu"
):
    """Transcrit une session de conférence avec métadonnées."""
    client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
    audio_format = AudioFormat(encoding="pcm_s16le", sample_rate=16000)

    session = {
        "conference": nom_conference,
        "orateur": orateur,
        "debut": datetime.now().isoformat(),
        "texte": [],
    }

    print(f"Transcription : {nom_conference}")
    print(f"Orateur : {orateur}")
    print("-" * 50)

    async for event in client.audio.realtime.transcribe_stream(
        audio_stream=source_audio,
        model="voxtral-mini-transcribe-realtime-2602",
        audio_format=audio_format,
        target_streaming_delay_ms=1500,
    ):
        if isinstance(event, TranscriptionStreamTextDelta):
            session["texte"].append(event.text)
            print(event.text, end="", flush=True)
        elif isinstance(event, TranscriptionStreamDone):
            break

    session["fin"] = datetime.now().isoformat()
    session["texte"] = "".join(session["texte"])
    session["nb_mots"] = len(session["texte"].split())

    nom_fichier = f"session_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
    with open(nom_fichier, "w", encoding="utf-8") as f:
        json.dump(session, f, ensure_ascii=False, indent=2)

    print(f"\n\nSession sauvegardée : {nom_fichier}")
    return session

Points clés à retenir

• Le sous-titrage en direct utilise une latence de 800 ms et gère l’affichage ligne par ligne
• L’assistant vocal chaîne transcription temps réel et LLM pour répondre aux questions
• La dictée vocale sauvegarde automatiquement le texte au fil de l’eau
• La transcription de conférence ajoute des métadonnées et exporte en JSON
• Chaque cas d’usage a sa latence optimale : 300 ms (assistant) à 1500 ms (conférence)