Streaming audio bidirectionnel
Sur cette page
Mise à jour du 2 novembre 2026 — modèles audio. Plusieurs modèles de ce cours sont dépréciés :
tts-1,tts-1-hdetgpt-4o-mini-ttss’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-realtimeet 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-transcribeet/v1/audio/translationsle 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.
Streaming audio bidirectionnel
Le streaming audio bidirectionnel est le cœur de la Realtime API. Vous envoyez des chunks audio du microphone en continu, et le modèle vous renvoie des chunks audio de réponse en continu. Les deux flux sont indépendants et peuvent opérer simultanément — c’est ce qui rend la conversation naturelle.
Format audio PCM16
La Realtime API utilise le format PCM 16-bit mono à 24 kHz. C’est un format brut, sans compression : chaque échantillon est un entier signé sur 16 bits. Ce choix privilégie la faible latence à la bande passante.
Les chunks audio sont envoyés encodés en Base64 dans les événements JSON. Chaque chunk représente typiquement 20 à 100 ms d’audio.
Envoyer de l’audio : capture du microphone
Voici comment capturer le microphone et envoyer l’audio en streaming :
import asyncio
import sounddevice as sd
import numpy as np
import base64
import json
SAMPLE_RATE = 24000
CHUNK_DURATION_MS = 100
CHUNK_SIZE = int(SAMPLE_RATE * CHUNK_DURATION_MS / 1000)
async def stream_microphone(ws):
"""Capture le micro et envoie les chunks à la Realtime API."""
loop = asyncio.get_event_loop()
audio_queue = asyncio.Queue()
def audio_callback(indata, frames, time_info, status):
if status:
print(f"Audio warning: {status}")
# Convertir en int16 et mettre en file
pcm_data = (indata[:, 0] * 32767).astype(np.int16)
loop.call_soon_threadsafe(audio_queue.put_nowait, pcm_data)
# Démarrer la capture
stream = sd.InputStream(
samplerate=SAMPLE_RATE,
channels=1,
dtype="float32",
blocksize=CHUNK_SIZE,
callback=audio_callback
)
with stream:
print("Microphone actif — parlez...")
while True:
pcm_data = await audio_queue.get()
audio_b64 = base64.b64encode(pcm_data.tobytes()).decode("utf-8")
await ws.send(json.dumps({
"type": "input_audio_buffer.append",
"audio": audio_b64
}))Le buffer audio côté serveur
Les chunks envoyés via input_audio_buffer.append s’accumulent dans un buffer côté serveur. Ce buffer est traité de deux manières :
- Mode VAD automatique (
server_vad) : le serveur détecte quand vous arrêtez de parler et commit automatiquement le buffer - Mode manuel : vous envoyez
input_audio_buffer.commitexplicitement pour déclencher le traitement
# Commit manuel du buffer audio
await ws.send(json.dumps({
"type": "input_audio_buffer.commit"
}))
# Ou vider le buffer sans le traiter
await ws.send(json.dumps({
"type": "input_audio_buffer.clear"
}))Recevoir l’audio de réponse
Les réponses audio arrivent en chunks via des événements response.output_audio.delta. Vous devez les décoder et les jouer en continu :
import pyaudio
async def play_response_audio(ws):
"""Reçoit et joue l'audio de réponse."""
pa = pyaudio.PyAudio()
player = pa.open(
format=pyaudio.paInt16,
channels=1,
rate=SAMPLE_RATE,
output=True,
frames_per_buffer=CHUNK_SIZE
)
try:
async for raw_message in ws:
event = json.loads(raw_message)
if event["type"] == "response.output_audio.delta":
# Décoder le chunk audio Base64
audio_bytes = base64.b64decode(event["delta"])
player.write(audio_bytes)
elif event["type"] == "response.output_audio.done":
print("Réponse audio terminée")
elif event["type"] == "response.output_audio_transcript.delta":
# Transcription textuelle de la réponse
print(event["delta"], end="", flush=True)
finally:
player.stop_stream()
player.close()
pa.terminate()Application complète : conversation vocale
Voici comment assembler capture et lecture pour une conversation complète :
async def conversation_vocale():
headers = {
"Authorization": f"Bearer {OPENAI_API_KEY}"
}
async with websockets.connect(URL, additional_headers=headers) as ws:
# Attendre session.created
await ws.recv()
# Configurer la session
await ws.send(json.dumps({
"type": "session.update",
"session": {
"type": "realtime",
"output_modalities": ["audio"],
"instructions": "Vous êtes un assistant vocal. Répondez en français.",
"audio": {
"input": {
"format": {"type": "audio/pcm", "rate": 24000},
"turn_detection": {
"type": "server_vad",
"threshold": 0.5,
"silence_duration_ms": 700
}
},
"output": {
"format": {"type": "audio/pcm", "rate": 24000},
"voice": "marin"
}
}
}
}))
await ws.recv() # session.updated
# Lancer capture et lecture en parallèle
await asyncio.gather(
stream_microphone(ws),
play_response_audio(ws)
)
asyncio.run(conversation_vocale())Gestion du flux concurrent
La difficulté du bidirectionnel est que vous envoyez et recevez simultanément sur le même WebSocket. asyncio.gather lance les deux coroutines en parallèle, mais elles partagent le même objet WebSocket. La bibliothèque websockets gère cette concurrence correctement.
Points clés à retenir
- L’audio est transmis en PCM 16-bit mono 24 kHz, encodé en Base64 dans les événements JSON
- Les chunks font typiquement 20 à 100 ms d’audio pour un bon compromis latence/overhead
- Le buffer audio côté serveur accumule les chunks jusqu’au commit (automatique en mode VAD)
- La lecture de réponse se fait chunk par chunk via
response.output_audio.delta asyncio.gatherpermet de capturer et jouer simultanément sur le même WebSocket