Pipeline LLM vers TTS streaming

Connecter un LLM à la synthèse vocale en temps réel

Le cas d’usage le plus puissant du WebSocket TTS est le chaining avec un LLM en streaming. Le LLM génère du texte token par token, chaque token est envoyé au TTS via WebSocket, et l’audio est joue au fur et à mesure. L’utilisateur entend la réponse pendant que le LLM continue à réfléchir.

Architecture du pipeline

Le flux de données traverse trois étapes :

Utilisateur -> LLM (streaming) -> WebSocket TTS -> Audio (lecture)
   question      tokens texte       audio.delta      haut-parleur

Chaque étape fonctionne en streaming, ce qui minimise la latence perçue. L’utilisateur n’attend pas que le LLM ait fini de générer toute sa réponse pour commencer à l’entendre.

Implémentation en Python

Voici un pipeline complet qui connecte l’API Chat Completions de Grok au TTS WebSocket :

import asyncio
import websockets
import json
import base64
import aiohttp

async def llm_to_speech(api_key, prompt, voice="rex", language="fr"):
    # Etape 1 : Connexion WebSocket TTS
    tts_uri = f"wss://api.x.ai/v1/tts?language={language}&voice={voice}"
    tts_headers = {"Authorization": f"Bearer {api_key}"}

    audio_chunks = []

    async with websockets.connect(tts_uri, additional_headers=tts_headers) as tts_ws:

        # Étape 2 : Appel LLM en streaming
        async with aiohttp.ClientSession() as session:
            async with session.post(
                "https://api.x.ai/v1/chat/completions",
                headers={
                    "Authorization": f"Bearer {api_key}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": "grok-4.5",
                    "messages": [{"role": "user", "content": prompt}],
                    "stream": True
                }
            ) as llm_response:

                # Etape 3 : Chaque token LLM -> delta TTS
                buffer = ""
                async for line in llm_response.content:
                    line = line.decode().strip()
                    if not line.startswith("data: "):
                        continue
                    if line == "data: [DONE]":
                        break

                    chunk = json.loads(line[6:])
                    token = chunk["choices"][0].get("delta", {}).get("content", "")

                    if token:
                        buffer += token
                        # Envoyer par phrases complètes
                        if any(c in token for c in ".!?\n"):
                            await tts_ws.send(json.dumps({
                                "type": "text.delta",
                                "delta": buffer
                            }))
                            buffer = ""

                # Envoyer le reste du buffer
                if buffer:
                    await tts_ws.send(json.dumps({
                        "type": "text.delta",
                        "delta": buffer
                    }))

                await tts_ws.send(json.dumps({"type": "text.done"}))

        # Etape 4 : Collecter l'audio
        async for msg in tts_ws:
            data = json.loads(msg)
            if data["type"] == "audio.delta":
                audio_chunks.append(base64.b64decode(data["delta"]))
            elif data["type"] == "audio.done":
                break

    return b"".join(audio_chunks)

Stratégie de buffering

Envoyer chaque token individuellement au TTS n’est pas optimal. Le moteur de synthèse produit un meilleur résultat quand il dispose de contexte. Trois stratégies de buffering :

Par phrase

Attendez un délimiteur de phrase (., !, ?, retour à la ligne) avant d’envoyer :

if any(c in token for c in ".!?\n"):
    await tts_ws.send(json.dumps({
        "type": "text.delta", "delta": buffer
    }))
    buffer = ""

C’est le meilleur compromis entre qualité et latence.

Par taille

Envoyez quand le buffer atteint une certaine taille (par exemple 200 caractères) :

if len(buffer) >= 200:
    await tts_ws.send(json.dumps({
        "type": "text.delta", "delta": buffer
    }))
    buffer = ""

Plus simple mais peut couper au milieu d’un mot.

Par temporisation

Envoyez après un délai sans nouveau token (utile quand le LLM réfléchit) :

last_token_time = time.time()

# Si plus de 500ms sans token, envoyer le buffer
if time.time() - last_token_time > 0.5 and buffer:
    await tts_ws.send(...)
    buffer = ""

Lecture audio en temps réel

Pour lire l’audio au fur et à mesure (pas seulement le collecter), utilisez une queue asynchrone :

import asyncio

audio_queue = asyncio.Queue()

async def receive_audio(tts_ws):
    """Reçoit les chunks audio et les met en queue."""
    async for msg in tts_ws:
        data = json.loads(msg)
        if data["type"] == "audio.delta":
            audio_bytes = base64.b64decode(data["delta"])
            await audio_queue.put(audio_bytes)
        elif data["type"] == "audio.done":
            await audio_queue.put(None)  # Sentinelle de fin
            break

async def play_audio():
    """Lit les chunks audio depuis la queue."""
    while True:
        chunk = await audio_queue.get()
        if chunk is None:
            break
        # Jouer le chunk (pyaudio, sounddevice, etc.)
        play_chunk(chunk)

En exécutant receive_audio et play_audio en parallèle, la lecture commence dès le premier chunk reçu.

Gestion des erreurs dans le pipeline

Le pipeline peut échouer à trois endroits : le LLM, le TTS, ou le réseau. Chaque point doit être géré :

async def robust_pipeline(api_key, prompt):
    try:
        audio = await llm_to_speech(api_key, prompt)
        return audio
    except aiohttp.ClientError:
        print("Erreur LLM - vérifiez votre connexion")
    except websockets.ConnectionClosed:
        print("Connexion TTS perdue - reconnexion necessaire")
    except json.JSONDecodeError:
        print("Réponse invalide - format inattendu")
    except Exception as e:
        print(f"Erreur inattendue: {e}")
    return None

Points clés à retenir

• Le pipeline LLM streaming vers TTS WebSocket offre la latence minimale
• Bufferisez par phrase pour le meilleur compromis qualité/latence
• Utilisez une queue asynchrone pour la lecture audio en temps réel
• Gérez les erreurs à chaque étape du pipeline
• Le texte total est illimité via WebSocket (deltas de 15 000 caractères max)
• La limite de 50 sessions concurrentes s’applique à l’ensemble de l’équipe