Aller au contenu principal

Contrôle de la Latence

Le compromis latence vs précision

En transcription temps réel, il existe un compromis fondamental entre la vitesse de réponse et la précision du texte. Plus le modèle attend avant d’émettre du texte, plus il a de contexte pour transcrire correctement — mais plus l’utilisateur attend.

Voxtral Realtime vous donne un contrôle direct sur ce compromis grâce au paramètre target_streaming_delay_ms.

Le paramètre target_streaming_delay_ms

Ce paramètre définit le délai cible en millisecondes entre la réception de l’audio et l’émission du texte transcrit.

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

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
audio_format = AudioFormat(encoding="pcm_s16le", sample_rate=16000)

# Latence basse : réponse rapide, moins précise
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=240,  # 240 ms
):
    if isinstance(event, TranscriptionStreamTextDelta):
        print(event.text, end="", flush=True)

Valeurs typiques

  • 200-300 ms — Ultra-réactif. Idéal pour les assistants vocaux où la réactivité prime. Le texte peut contenir plus de corrections.
  • 500-1000 ms — Bon équilibre. Adapté au sous-titrage en direct et à la dictée.
  • 2000-3000 ms — Haute précision. Le modèle a plus de contexte, le texte est plus propre mais le délai est perceptible.

Le pattern Dual Delay

C’est la technique la plus puissante de Voxtral Realtime. L’idée : exécuter deux flux en parallèle sur le même audio, avec des latences différentes.

  • Flux rapide (ex: 240 ms) — Fournit un feedback immédiat à l’utilisateur
  • Flux lent (ex: 2400 ms) — Produit une transcription plus précise qui remplace progressivement le texte rapide

C’est exactement ce que font les sous-titres en direct de YouTube ou Google Meet : le texte apparaît rapidement puis se corrige automatiquement.

Implémentation

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


async def flux_double_latence(source_audio_rapide, source_audio_lent):
    """Exécute deux transcriptions en parallèle avec des latences différentes."""
    client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
    audio_format = AudioFormat(encoding="pcm_s16le", sample_rate=16000)

    resultats = {"rapide": [], "lent": []}

    async def flux_rapide():
        async for event in client.audio.realtime.transcribe_stream(
            audio_stream=source_audio_rapide,
            model="voxtral-mini-transcribe-realtime-2602",
            audio_format=audio_format,
            target_streaming_delay_ms=240,
        ):
            if isinstance(event, TranscriptionStreamTextDelta):
                resultats["rapide"].append(event.text)
                # Afficher immédiatement (sera remplacé par le flux lent)
                print(f"\r[RAPIDE] {''.join(resultats['rapide'][-10:])}",
                      end="", flush=True)

    async def flux_lent():
        async for event in client.audio.realtime.transcribe_stream(
            audio_stream=source_audio_lent,
            model="voxtral-mini-transcribe-realtime-2602",
            audio_format=audio_format,
            target_streaming_delay_ms=2400,
        ):
            if isinstance(event, TranscriptionStreamTextDelta):
                resultats["lent"].append(event.text)
                # Afficher la version corrigée
                print(f"\n[PRÉCIS] {''.join(resultats['lent'][-10:])}",
                      flush=True)

    # Exécuter les deux flux en parallèle
    await asyncio.gather(flux_rapide(), flux_lent())

    return {
        "rapide": "".join(resultats["rapide"]),
        "lent": "".join(resultats["lent"])
    }

Logique de remplacement dans une interface

Dans une vraie application, vous afficheriez le texte rapide en gris (provisoire) et le remplaceriez par le texte lent en blanc (final) au fur et à mesure :

class AffichageDoubleLatence:
    """Gère l'affichage avec double latence."""

    def __init__(self):
        self.texte_rapide = ""
        self.texte_final = ""
        self.position_finale = 0

    def on_texte_rapide(self, texte: str):
        """Nouveau texte du flux rapide (provisoire)."""
        self.texte_rapide += texte
        self._afficher()

    def on_texte_lent(self, texte: str):
        """Nouveau texte du flux lent (final)."""
        self.texte_final += texte
        self.position_finale = len(self.texte_final)
        self._afficher()

    def _afficher(self):
        """Affiche le texte final + le texte rapide non encore confirmé."""
        texte_provisoire = self.texte_rapide[self.position_finale:]
        affichage = self.texte_final + texte_provisoire
        print(f"\r{affichage}", end="", flush=True)

Choisir la bonne latence pour votre cas d’usage

Cas d’usageLatence recommandéeRaison
Assistant vocal200-300 msL’utilisateur attend une réaction immédiate
Dictée vocale500-800 msÉquilibre entre réactivité et précision
Sous-titres en direct800-1500 msPrécision importante, léger délai acceptable
Transcription conférence1500-3000 msPrécision maximale, le délai n’est pas critique

Mesurer la latence effective

import time

async def mesurer_latence(source_audio, delay_ms):
    """Mesure la latence effective de la transcription."""
    client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
    audio_format = AudioFormat(encoding="pcm_s16le", sample_rate=16000)

    debut = time.time()
    premier_texte = None

    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=delay_ms,
    ):
        if isinstance(event, TranscriptionStreamTextDelta):
            if premier_texte is None:
                premier_texte = time.time()
                latence = (premier_texte - debut) * 1000
                print(f"Premier texte après {latence:.0f} ms "
                      f"(cible : {delay_ms} ms)")
            print(event.text, end="", flush=True)
        elif isinstance(event, TranscriptionStreamDone):
            break

    print(f"\nDurée totale : {time.time() - debut:.1f}s")

Points clés à retenir

  • target_streaming_delay_ms contrôle le compromis latence/précision
  • Valeurs basses (200-300 ms) : réactif mais moins précis
  • Valeurs hautes (2000-3000 ms) : précis mais avec un délai perceptible
  • Le pattern Dual Delay combine deux flux pour avoir réactivité ET précision
  • Le flux rapide donne un feedback immédiat, le flux lent corrige le texte
  • Choisissez la latence selon votre cas d’usage (assistant vocal vs sous-titrage)