Pipeline LLM vers TTS streaming

Connecter un LLM a la synthese vocale en temps reel

Le cas d’usage le plus puissant du WebSocket TTS est le chaining avec un LLM en streaming. Le LLM genere du texte token par token, chaque token est envoye au TTS via WebSocket, et l’audio est joue au fur et a mesure. L’utilisateur entend la reponse pendant que le LLM continue a reflechir.

Architecture du pipeline

Le flux de donnees traverse trois etapes :

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

Chaque etape fonctionne en streaming, ce qui minimise la latence percue. L’utilisateur n’attend pas que le LLM ait fini de generer toute sa reponse pour commencer a l’entendre.

Implementation 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, extra_headers=tts_headers) as tts_ws:

        # Etape 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-3",
                    "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 completes
                        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)

Strategie de buffering

Envoyer chaque token individuellement au TTS n’est pas optimal. Le moteur de synthese produit un meilleur resultat quand il dispose de contexte. Trois strategies de buffering :

Par phrase

Attendez un delimiteur de phrase (., !, ?, retour a 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 qualite et latence.

Par taille

Envoyez quand le buffer atteint une certaine taille (par exemple 200 caracteres) :

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 apres un delai sans nouveau token (utile quand le LLM reflechit) :

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 reel

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

import asyncio

audio_queue = asyncio.Queue()

async def receive_audio(tts_ws):
    """Recoit 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 executant receive_audio et play_audio en parallele, la lecture commence des le premier chunk recu.

Gestion des erreurs dans le pipeline

Le pipeline peut echouer a trois endroits : le LLM, le TTS, ou le reseau. Chaque point doit etre gere :

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

Points cles a retenir

• Le pipeline LLM streaming vers TTS WebSocket offre la latence minimale
• Bufferisez par phrase pour le meilleur compromis qualite/latence
• Utilisez une queue asynchrone pour la lecture audio en temps reel
• Gerez les erreurs a chaque etape du pipeline
• Le texte total est illimite via WebSocket (deltas de 15 000 caracteres max)
• La limite de 50 sessions concurrentes s’applique a l’ensemble de l’equipe