Streaming temps réel

Pourquoi le streaming change tout

Jusqu’ici, vous avez utilisé la génération synchrone : vous envoyez du texte, vous attendez que toute la synthèse soit terminée, puis vous recevez l’audio complet. Pour des textes courts, c’est parfait. Mais pour un assistant vocal interactif ou une application temps réel, cette attente est inacceptable.

Le streaming résout ce problème : l’audio est envoyé au fur et à mesure de sa génération, chunk par chunk. L’utilisateur entend les premiers mots pendant que le modèle génère la suite.

Le protocole d’événements

Le streaming de Voxtral utilise un protocole basé sur des événements (Server-Sent Events). Chaque chunk reçu est un événement typé :

• speech.audio.delta : contient un fragment d’audio en base64. C’est l’événement principal que vous traiterez
• speech.audio.done : signal de fin — la génération est complète

Code Python : streaming basique

Voici comment activer le streaming en ajoutant simplement stream=True :

import base64
from pathlib import Path
from mistralai.client import Mistral

client = Mistral(api_key="votre-cle-api")

audio_chunks = []

with client.audio.speech.complete(
    model="voxtral-mini-tts-2603",
    input="Le streaming rend les agents vocaux réactifs et naturels.",
    voice_id="votre-voice-id",
    response_format="opus",
    stream=True,
) as stream:
    for event in stream:
        if event.event == "speech.audio.delta":
            chunk = base64.b64decode(event.data.audio_data)
            audio_chunks.append(chunk)
            print(f"Chunk reçu : {len(chunk)} octets")

# Assembler tous les chunks
audio_complet = b"".join(audio_chunks)
Path("streaming-output.opus").write_bytes(audio_complet)
print(f"Audio final : {len(audio_complet)} octets")

Lecture en temps réel

Pour une véritable expérience temps réel, vous ne voulez pas attendre tous les chunks avant de commencer la lecture. Voici comment jouer l’audio au fur et à mesure avec pyaudio :

import base64
import pyaudio
from mistralai.client import Mistral

client = Mistral(api_key="votre-cle-api")

# Configuration audio PCM (le format le plus rapide)
p = pyaudio.PyAudio()
stream_audio = p.open(
    format=pyaudio.paFloat32,
    channels=1,
    rate=24000,
    output=True,
)

with client.audio.speech.complete(
    model="voxtral-mini-tts-2603",
    input="Vous entendez cette phrase pendant que je la génère.",
    voice_id="votre-voice-id",
    response_format="pcm",
    stream=True,
) as stream:
    for event in stream:
        if event.event == "speech.audio.delta":
            audio_data = base64.b64decode(event.data.audio_data)
            stream_audio.write(audio_data)

stream_audio.stop_stream()
stream_audio.close()
p.terminate()

Le format PCM est recommandé pour le streaming temps réel car il n’a pas besoin d’être décodé — les octets bruts sont directement jouables.

Intégration dans une application web

Pour une application web, le pattern classique est d’utiliser un WebSocket côté serveur qui relaie les chunks audio vers le navigateur :

# Côté serveur (exemple avec FastAPI)
from fastapi import FastAPI, WebSocket
import base64
from mistralai.client import Mistral

app = FastAPI()
client = Mistral(api_key="votre-cle-api")

@app.websocket("/ws/tts")
async def tts_websocket(websocket: WebSocket):
    await websocket.accept()
    texte = await websocket.receive_text()

    with client.audio.speech.complete(
        model="voxtral-mini-tts-2603",
        input=texte,
        voice_id="votre-voice-id",
        response_format="pcm",
        stream=True,
    ) as stream:
        for event in stream:
            if event.event == "speech.audio.delta":
                await websocket.send_bytes(
                    base64.b64decode(event.data.audio_data)
                )

    await websocket.close()

Côté navigateur, vous utilisez l’API Web Audio pour jouer les chunks PCM reçus via le WebSocket.

Gestion des chunks

Quelques considérations techniques pour le streaming :

• Taille des chunks : variable, généralement quelques millisecondes d’audio chacun
• Buffer : accumulez 2-3 chunks avant de commencer la lecture pour éviter les micro-coupures
• Format : utilisez PCM pour la latence la plus basse (~0.7s premier token) ou Opus pour un bon compromis bande passante/latence

Points clés à retenir

• Le streaming envoie l’audio chunk par chunk pendant la génération
• Activez-le avec stream=True dans l’appel audio.speech.complete()
• Les événements speech.audio.delta contiennent les fragments audio en base64
• Le format PCM offre la latence la plus basse pour le streaming (~0.7s vs ~2s en MP3)
• Pour une lecture temps réel, utilisez pyaudio ou les WebSockets + Web Audio API