Streaming audio en temps réel
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-4o-mini",
input=[{"role": "user", "content": user_message}],
stream=True
)
for event in stream:
if hasattr(event, "delta") and event.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
pcmest 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