Streaming audio en temps réel

Mise à jour du 2 novembre 2026 — modèles audio. Plusieurs modèles de ce cours sont dépréciés : tts-1, tts-1-hd et gpt-4o-mini-tts s’arrêtent le 6 janvier 2027 (remplaçant indiqué : gpt-realtime-2.1-mini, que le guide TTS d’OpenAI déclare non pris en charge par /v1/audio/speech : la migration de la synthèse vocale passe par l’API Realtime) ; gpt-realtime, gpt-4o-realtime et les anciens modèles audio le 20 janvier 2027 (gpt-realtime-2.1, gpt-realtime-2.1-mini, gpt-audio-1.5) ; whisper-1, gpt-4o-transcribe et /v1/audio/translations le 26 février 2027 (gpt-transcribe, gpt-live-transcribe, v1/realtime/translations). Pour la voix full-duplex, GPT-Live 1 est disponible dans l’API depuis le 10 septembre 2026. Corrigé le 3 novembre 2026 : le pipeline LLM → TTS lisait tout événement porteur d’un champ delta, ce qui inclut par exemple les arguments d’appels de fonctions ou les résumés de raisonnement ; il ne retient plus que response.output_text.delta, le texte de la réponse.

Streaming audio en temps réel

Dans un agent vocal, attendre que la totalité de l’audio soit générée avant de commencer la lecture introduit une latence inacceptable. Le streaming TTS permet de commencer la lecture dès les premiers octets reçus, réduisant le temps de première réponse à quelques centaines de millisecondes.

Streaming avec l’API TTS

L’API TTS supporte nativement le streaming. Au lieu d’attendre le fichier complet, vous recevez l’audio par chunks :

from openai import OpenAI
import pyaudio

client = OpenAI()

def stream_tts_to_speaker(text: str, voice: str = "alloy"):
    """Génère et joue de l'audio en streaming."""
    # Initialiser le lecteur audio
    pa = pyaudio.PyAudio()
    player = pa.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=24000,
        output=True
    )

    try:
        # Créer le flux streaming
        with client.audio.speech.with_streaming_response.create(
            model="tts-1",
            voice=voice,
            input=text,
            response_format="pcm"  # PCM brut pour lecture directe
        ) as response:
            # Lire et jouer chunk par chunk
            for chunk in response.iter_bytes(chunk_size=4096):
                player.write(chunk)

    finally:
        player.stop_stream()
        player.close()
        pa.terminate()

stream_tts_to_speaker("Bonjour, voici une réponse en streaming temps réel.")

Le format pcm est idéal pour le streaming car il ne nécessite aucun décodage : les octets bruts sont directement jouables.

Mesurer la latence

Le temps de première réponse (Time-to-First-Byte, TTFB) est la métrique clé du streaming :

import time

def measure_tts_latency(text: str, voice: str = "alloy") -> dict:
    """Mesure les latences du streaming TTS."""
    start_time = time.time()
    first_byte_time = None
    total_bytes = 0

    with client.audio.speech.with_streaming_response.create(
        model="tts-1",
        voice=voice,
        input=text,
        response_format="pcm"
    ) as response:
        for chunk in response.iter_bytes(chunk_size=4096):
            if first_byte_time is None:
                first_byte_time = time.time()
            total_bytes += len(chunk)

    end_time = time.time()

    return {
        "ttfb_ms": round((first_byte_time - start_time) * 1000),
        "total_ms": round((end_time - start_time) * 1000),
        "total_bytes": total_bytes,
        "audio_duration_s": round(total_bytes / (24000 * 2), 2)
    }

metrics = measure_tts_latency("Bonjour, comment allez-vous ?")
print(f"TTFB: {metrics['ttfb_ms']}ms, Total: {metrics['total_ms']}ms")

Streaming asynchrone

Pour les applications serveur, utilisez le client asynchrone :

from openai import AsyncOpenAI
import asyncio

async_client = AsyncOpenAI()

async def stream_tts_async(text: str, voice: str = "alloy"):
    """Streaming TTS asynchrone pour serveurs."""
    chunks = []

    async with async_client.audio.speech.with_streaming_response.create(
        model="tts-1",
        voice=voice,
        input=text,
        response_format="pcm"
    ) as response:
        async for chunk in response.iter_bytes(chunk_size=4096):
            chunks.append(chunk)
            # Ici, vous enverriez le chunk au client via WebSocket
            yield chunk

# Utilisation dans un serveur FastAPI
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/tts/stream")
async def tts_stream(text: str, voice: str = "alloy"):
    return StreamingResponse(
        stream_tts_async(text, voice),
        media_type="audio/pcm"
    )

Buffer de pré-remplissage

Pour une lecture sans coupure, accumulez un petit buffer avant de commencer la lecture :

async def buffered_tts_playback(
    text: str,
    voice: str = "alloy",
    buffer_size: int = 3  # Nombre de chunks à pré-charger
):
    """Lecture TTS avec buffer de pré-remplissage."""
    pa = pyaudio.PyAudio()
    player = pa.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=24000,
        output=True
    )

    buffer = []
    playback_started = False

    try:
        async with async_client.audio.speech.with_streaming_response.create(
            model="tts-1",
            voice=voice,
            input=text,
            response_format="pcm"
        ) as response:
            async for chunk in response.iter_bytes(chunk_size=4096):
                buffer.append(chunk)

                if not playback_started and len(buffer) >= buffer_size:
                    playback_started = True
                    # Vider le buffer initial
                    for buffered_chunk in buffer:
                        player.write(buffered_chunk)
                    buffer.clear()
                elif playback_started:
                    player.write(chunk)

        # Jouer les chunks restants
        for remaining in buffer:
            player.write(remaining)

    finally:
        player.stop_stream()
        player.close()
        pa.terminate()

Intégration avec un pipeline LLM + TTS

Dans un agent vocal non-Realtime, vous pouvez combiner le streaming LLM et le streaming TTS :

async def llm_to_tts_pipeline(user_message: str):
    """Pipeline : streaming LLM → streaming TTS en temps réel."""
    # Phase 1 : accumuler la réponse LLM par phrases
    sentence_buffer = ""

    stream = client.responses.create(
        model="gpt-5.6-luna",
        input=[{"role": "user", "content": user_message}],
        stream=True
    )

    for event in stream:
        # Seul le texte de la réponse doit être prononcé
        if event.type == "response.output_text.delta":
            sentence_buffer += event.delta

            # Détecter la fin d'une phrase
            if sentence_buffer.rstrip().endswith((".", "!", "?")):
                sentence = sentence_buffer.strip()
                sentence_buffer = ""

                # Phase 2 : streamer cette phrase en TTS
                stream_tts_to_speaker(sentence)

    # Traiter le reste du buffer
    if sentence_buffer.strip():
        stream_tts_to_speaker(sentence_buffer.strip())

Points clés à retenir

• Le streaming TTS réduit le TTFB à quelques centaines de ms en commençant la lecture avant la fin de génération
• Le format pcm est optimal pour le streaming : aucun décodage nécessaire, lecture directe
• Un buffer de pré-remplissage de 2-3 chunks évite les coupures au début de la lecture
• Le client asynchrone est indispensable pour les serveurs gérant plusieurs connexions simultanées
• Le pipeline LLM streaming + TTS streaming permet un agent vocal réactif sans la Realtime API