Contrôle de la Latence

Le compromis latence contre précision

La transcription temps réel repose sur un arbitrage que rien ne permet d’éviter : la vitesse de réponse se paie en précision. Plus le modèle patiente avant d’émettre un mot, plus il dispose de contexte pour trancher entre deux hypothèses acoustiques proches — et plus votre utilisateur attend. Une phrase comme « il a pris la mer » ne se distingue de « il a pris l’amer » qu’à la lumière de ce qui suit ; un modèle pressé produira parfois la mauvaise version.

Voxtral Realtime ne cache pas cet arbitrage derrière un réglage automatique : il vous le confie explicitement, par le paramètre target_streaming_delay_ms. Vous décidez, cas d’usage par cas d’usage, du point d’équilibre.

Le paramètre target_streaming_delay_ms

La valeur exprime en millisecondes le délai cible entre la réception de l’audio et l’émission du texte correspondant. Elle se passe directement à l’appel de streaming, au même titre que le modèle et le format audio :

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)

Trois plages de valeurs couvrent la quasi-totalité des besoins. Entre 200 et 300 millisecondes, la transcription est ultra-réactive : c’est le réglage des assistants vocaux, où l’immédiateté prime et où le texte peut contenir davantage de corrections. Entre 500 et 1 000 millisecondes, vous obtenez un bon équilibre, adapté au sous-titrage en direct comme à la dictée. Au-delà, entre 2 000 et 3 000 millisecondes, le modèle travaille avec beaucoup plus de contexte et rend un texte nettement plus propre, au prix d’un délai que l’utilisateur perçoit clairement.

Le pattern Dual Delay

Faut-il vraiment choisir ? Voxtral Realtime autorise une échappatoire : lancer deux flux en parallèle sur le même audio, avec des latences différentes, et refuser ainsi l’arbitrage. Le flux rapide, réglé par exemple à 240 millisecondes, donne un retour visuel immédiat qui rassure l’utilisateur et l’informe que le système écoute. Le flux lent, réglé à 2 400 millisecondes, produit une transcription plus précise qui vient progressivement remplacer le texte provisoire.

Vous connaissez le résultat sans forcément l’avoir nommé : c’est le comportement des sous-titres en direct de YouTube ou de Google Meet, où le texte s’affiche presque instantanément puis se corrige tout seul quelques instants plus tard. L’implémentation consiste à définir deux coroutines identiques à l’exception du délai, puis à les exécuter conjointement avec asyncio.gather.

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

Remarquez que la fonction reçoit deux sources audio distinctes et non une seule : un générateur asynchrone ne peut être consommé qu’une fois, il vous faut donc dupliquer le flux en amont pour alimenter les deux transcriptions.

Gérer le remplacement à l’écran

Dans un terminal, la double sortie du code précédent suffit à comprendre le mécanisme, mais une véritable interface doit fusionner les deux flux en un texte unique et lisible. La convention établie consiste à afficher le texte provisoire dans une teinte discrète — gris — et le texte confirmé en blanc, la frontière entre les deux avançant à mesure que le flux lent rattrape le flux rapide. La classe ci-dessous matérialise cette frontière avec position_finale : tout ce qui précède vient du flux lent et ne bougera plus, tout ce qui suit vient du flux rapide et reste susceptible d’être corrigé.

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)

Ce confort a un coût qu’il serait malhonnête de passer sous silence : deux flux signifient deux connexions et deux fois le volume audio traité. Réservez donc le Dual Delay aux interfaces où l’utilisateur lit le texte pendant qu’il se produit ; pour un traitement en arrière-plan, un seul flux bien réglé suffit largement.

Choisir la bonne latence pour votre cas d’usage

Mesurer la latence réellement obtenue

La valeur que vous demandez est une cible, pas une garantie : la distance réseau, la charge du service et la taille de vos chunks s’y ajoutent. Avant d’annoncer un chiffre à vos utilisateurs, mesurez-le depuis l’endroit où votre application tournera vraiment. La fonction suivante chronomètre le délai jusqu’au premier fragment reçu et le compare à la cible demandée.

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

Lancez cette mesure avec 240, puis 800, puis 2 400 millisecondes sur le même extrait, et lisez les trois transcriptions côte à côte. L’écart de qualité entre la première et la dernière vous dira, mieux que n’importe quelle recommandation générale, quelle latence votre application peut se permettre.

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)