Transcription en temps réel

La transcription en temps réel convertit en texte l’audio en direct pendant que la personne parle encore. Utilisez-la dans les applications agentiques : transcrivez le flux du microphone ou du téléphone d’un utilisateur, puis transmettez la transcription à un modèle qui génère la réponse de l’assistant.

Cette page explique le déroulement d’une session du début à la fin, sa configuration, la lecture de ses événements et le compromis entre vitesse et précision.

Une session correspond à une connexion WebSocket qui passe par quatre étapes, toujours dans le même ordre. Votre application envoie l’audio, et l’API renvoie le texte pendant que l’audio continue d’arriver.

Lisez de haut en bas : chaque flèche représente un message, envoyé par votre application à gauche ou par l’API à droite.

À chaque étape, votre application procède comme suit :

Si l’API rejette un message, elle envoie à la place un événement error à cette étape. Consultez Problèmes courants.

Une session est une connexion WebSocket vers l’endpoint temps réel, avec le modèle dans la chaîne de requête :

wss://api.mistral.ai/v1/audio/transcriptions/realtime?model=voxtral-mini-transcribe-realtime-3

wss://api.mistral.ai/v1/audio/transcriptions/realtime?model=voxtral-mini-transcribe-realtime-3

La connexion s’authentifie avec votre clé API Mistral, envoyée sous la forme Authorization: Bearer <votre clé API>. Pour créer une clé, ouvrez Clés API.

Avec le SDK Python ou TypeScript, connect() ouvre la connexion et rend la main une fois la session créée :

import asyncio
import os

from mistralai.client import Mistral

async def main():
    client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
    async with await client.audio.realtime.connect(
        model="voxtral-mini-transcribe-realtime-3",
    ) as connection:
        print(connection.session.model_dump_json(indent=2))

asyncio.run(main())

import asyncio
import os

from mistralai.client import Mistral

async def main():
    client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
    async with await client.audio.realtime.connect(
        model="voxtral-mini-transcribe-realtime-3",
    ) as connection:
        print(connection.session.model_dump_json(indent=2))

asyncio.run(main())

Envoyez vos paramètres dans un message session.update, avant le premier fragment audio. La session répond par session.updated, qui reprend chaque paramètre qu’elle applique :

{
  "type": "session.update",
  "session": {
    "audio_format": { "encoding": "pcm_s16le", "sample_rate": 24000 },
    "target_streaming_delay_ms": 480,
    "language": "en",
    "context_bias": ["Northwind", "Quillara", "Alder Creek Design", "Rotterdam"]
  }
}

{
  "type": "session.update",
  "session": {
    "audio_format": { "encoding": "pcm_s16le", "sample_rate": 24000 },
    "target_streaming_delay_ms": 480,
    "language": "en",
    "context_bias": ["Northwind", "Quillara", "Alder Creek Design", "Rotterdam"]
  }
}

Le script suivant envoie ces paramètres, transmet un fichier WAV en streaming par fragments de 100 ms et affiche la transcription au fur et à mesure de son arrivée.

import asyncio
import base64
import json
import os
import wave

import websockets

MODEL = "voxtral-mini-transcribe-realtime-3"
URL = f"wss://api.mistral.ai/v1/audio/transcriptions/realtime?model={MODEL}"
HEADERS = {"Authorization": f"Bearer {os.environ['MISTRAL_API_KEY']}"}
AUDIO_FILE = "customer_call.wav"  # 16-bit mono PCM WAV

async def send_audio(ws, path):
    with wave.open(path, "rb") as wav:
        while chunk := wav.readframes(wav.getframerate() // 10):
            await ws.send(json.dumps({"type": "input_audio.append", "audio": base64.b64encode(chunk).decode()}))
    await ws.send(json.dumps({"type": "input_audio.end"}))

async def main():
    with wave.open(AUDIO_FILE, "rb") as wav:
        sample_rate = wav.getframerate()

    async with websockets.connect(URL, additional_headers=HEADERS) as ws:
        print(json.loads(await ws.recv())["type"])  # session.created

        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "audio_format": {"encoding": "pcm_s16le", "sample_rate": sample_rate},
                "target_streaming_delay_ms": 480,
                "language": "en",
                "context_bias": ["Northwind", "Quillara", "Alder Creek Design", "Rotterdam"],
            },
        }))
        sender = asyncio.create_task(send_audio(ws, AUDIO_FILE))

        async for message in ws:
            event = json.loads(message)
            if event["type"] == "session.updated":
                print("session.updated:", json.dumps(event["session"]))
            elif event["type"] == "transcription.text.delta":
                print(event["text"], end="", flush=True)
            elif event["type"] == "transcription.done":
                print("\ntranscription.done")
                break
            elif event["type"] == "error":
                print("error:", event["error"]["message"])
                break
        sender.cancel()

asyncio.run(main())

import asyncio
import base64
import json
import os
import wave

import websockets

MODEL = "voxtral-mini-transcribe-realtime-3"
URL = f"wss://api.mistral.ai/v1/audio/transcriptions/realtime?model={MODEL}"
HEADERS = {"Authorization": f"Bearer {os.environ['MISTRAL_API_KEY']}"}
AUDIO_FILE = "customer_call.wav"  # 16-bit mono PCM WAV

async def send_audio(ws, path):
    with wave.open(path, "rb") as wav:
        while chunk := wav.readframes(wav.getframerate() // 10):
            await ws.send(json.dumps({"type": "input_audio.append", "audio": base64.b64encode(chunk).decode()}))
    await ws.send(json.dumps({"type": "input_audio.end"}))

async def main():
    with wave.open(AUDIO_FILE, "rb") as wav:
        sample_rate = wav.getframerate()

    async with websockets.connect(URL, additional_headers=HEADERS) as ws:
        print(json.loads(await ws.recv())["type"])  # session.created

        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "audio_format": {"encoding": "pcm_s16le", "sample_rate": sample_rate},
                "target_streaming_delay_ms": 480,
                "language": "en",
                "context_bias": ["Northwind", "Quillara", "Alder Creek Design", "Rotterdam"],
            },
        }))
        sender = asyncio.create_task(send_audio(ws, AUDIO_FILE))

        async for message in ws:
            event = json.loads(message)
            if event["type"] == "session.updated":
                print("session.updated:", json.dumps(event["session"]))
            elif event["type"] == "transcription.text.delta":
                print(event["text"], end="", flush=True)
            elif event["type"] == "transcription.done":
                print("\ntranscription.done")
                break
            elif event["type"] == "error":
                print("error:", event["error"]["message"])
                break
        sender.cancel()

asyncio.run(main())

session.update accepte les champs suivants. N’envoyez que ceux que vous voulez modifier : les autres gardent la valeur par défaut indiquée dans session.created.

Tout ce que l’API envoie arrive sous forme de message JSON avec un champ type. Lisez d’abord le type, puis les champs de l’événement correspondant :

Chaque transcription.text.delta ajoute du texte après celui déjà envoyé. Un delta ne corrige jamais le texte précédent : votre application se contente donc d’ajouter, il n’y a rien à remplacer.

• Affichez chaque delta dès son arrivée. Il est définitif dès que vous le recevez.
• Ajoutez-le tel quel. Les deltas contiennent leurs propres espaces et leur ponctuation, comme " for" ou "." : n’ajoutez pas de séparateurs.
• Enregistrez le text de transcription.done comme transcription de référence.

Quand l’audio s’arrête, envoyez input_audio.end. L’API envoie les derniers deltas, puis transcription.done avec la transcription complète. Fermez la connexion une fois cet événement reçu.

Le modèle utilise un délai configurable pour contrôler le temps qu’il attend entre le moment où un mot est prononcé et celui où il est transcrit :

• Un délai plus court transcrit la parole plus tôt, mais avec moins de précision.
• Un délai plus long donne plus de contexte au modèle : le texte est plus précis, mais arrive plus tard.

target_streaming_delay_ms définit ce délai pour la session. Il accepte tout entier de 80 à 1200, ou 2400. Toute autre valeur renvoie une erreur. Commencez par 480 ms, qui équilibre latence et précision. Le SDK Python peut le définir avec update_session(), en même temps que le format audio.

Les lecteurs voient les mots tout de suite, et le texte qu’ils conservent est le plus précis. Les sous-titres en direct et la prise de notes en profitent le plus.

La plupart des problèmes se traduisent par un événement error accompagné d’un message. Voici ceux propres aux sessions en direct :

Pour les autres erreurs, consultez Résolution des problèmes de reconnaissance vocale.