Bidirektionales Audio-Streaming
Auf dieser Seite
Aktualisiert am 2. November 2026 — Audio-Modelle. Mehrere Modelle in diesem Kurs wurden eingestellt:
tts-1,tts-1-hdundgpt-4o-mini-ttswerden am 6. Januar 2027 nicht mehr funktionieren (die angegebene Alternative istgpt-realtime-2.1-mini, die laut OpenAI TTS-Leitfaden nicht durch/v1/audio/speechunterstützt wird: Die Migration der Sprachsynthese erfolgt über die Realtime API);gpt-realtime,gpt-4o-realtimeund veraltete Audio-Modelle am 20. Januar 2027 (gpt-realtime-2.1,gpt-realtime-2.1-mini,gpt-audio-1.5);whisper-1,gpt-4o-transcribeund/v1/audio/translationsam 26. Februar 2027 (gpt-transcribe,gpt-live-transcribe,v1/realtime/translations). Für die bidirektionale Sprachübertragung steht GPT-Live 1 seit dem 10. September 2026 in der API zur Verfügung.
Bidirektionales Audio-Streaming
Bidirektionales Audio-Streaming ist das Herzstück der Realtime API. Sie senden kontinuierlich Audio-Schnipsel vom Mikrofon und das Modell gibt kontinuierlich Antwort-Audio-Schnipsel zurück. Diese beiden Streams sind unabhängig und können gleichzeitig arbeiten – das ist es, was das Gespräch natürlich wirken lässt.
PCM16-Audioformat
Die Realtime API verwendet das Format 24 kHz Mono 16-Bit PCM. Dies ist ein rohes, unkomprimiertes Format: Jede Abtastung ist eine 16-Bit vorzeichenbehaftete Ganzzahl. Diese Wahl priorisiert niedrige Latenz gegenüber der Bandbreiteneffizienz.
Audio-Schnipsel werden Base64-codiert innerhalb von JSON-Events gesendet. Jeder Schnipsel repräsentiert typischerweise 20 bis 100 ms Audio.
Audio-Senden: Mikrofon-Erfassung
So erfassen Sie das Mikrofon und streamen das Audio:
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):
"""Erfassen Sie das Mikrofon und senden Sie Schnipsel an die Realtime API."""
loop = asyncio.get_event_loop()
audio_queue = asyncio.Queue()
def audio_callback(indata, frames, time_info, status):
if status:
print(f"Audiowarnung: {status}")
# In int16 umwandeln und in die Warteschlange einreihen
pcm_data = (indata[:, 0] * 32767).astype(np.int16)
loop.call_soon_threadsafe(audio_queue.put_nowait, pcm_data)
# Erfassung starten
stream = sd.InputStream(
samplerate=SAMPLE_RATE,
channels=1,
dtype="float32",
blocksize=CHUNK_SIZE,
callback=audio_callback
)
with stream:
print("Mikrofon aktiv – sprechen Sie jetzt...")
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
}))Der serverseitige Audio-Puffer
Schnipsel, die über input_audio_buffer.append gesendet werden, sammeln sich in einem Puffer auf dem Server an. Dieser Puffer wird auf zwei Arten verarbeitet:
- Automatischer VAD-Modus (
server_vad): Der Server erkennt, wann Sie aufhören zu sprechen, und gibt den Puffer automatisch frei. - Manueller Modus: Sie senden explizit
input_audio_buffer.commit, um die Verarbeitung auszulösen.
# Manuell den Audio-Puffer freigeben
await ws.send(json.dumps({
"type": "input_audio_buffer.commit"
}))
# Oder den Puffer löschen, ohne ihn zu verarbeiten
await ws.send(json.dumps({
"type": "input_audio_buffer.clear"
}))Empfang von Antwort-Audio
Antwort-Audio erreicht in Schnipseln über response.output_audio.delta-Events. Sie müssen diese decodieren und kontinuierlich abspielen:
import pyaudio
async def play_response_audio(ws):
"""Empfangen und Abspielen von Antwort-Audio."""
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":
# Den Base64-Audio-Schnipsel decodieren
audio_bytes = base64.b64decode(event["delta"])
player.write(audio_bytes)
elif event["type"] == "response.output_audio.done":
print("Antwort-Audio abgeschlossen")
elif event["type"] == "response.output_audio_transcript.delta":
# Text-Transkription der Antwort
print(event["delta"], end="", flush=True)
finally:
player.stop_stream()
player.close()
pa.terminate()Vollständige Anwendung: Sprachgespräch
So fügen Sie Erfassung und Wiedergabe für ein vollständiges Gespräch zusammen:
async def conversation_vocale():
headers = {
"Authorization": f"Bearer {OPENAI_API_KEY}"
}
async with websockets.connect(URL, additional_headers=headers) as ws:
# Warten auf session.created
await ws.recv()
# Sitzung konfigurieren
await ws.send(json.dumps({
"type": "session.update",
"session": {
"type": "realtime",
"output_modalities": ["audio"],
"instructions": "Sie sind ein Sprachassistent. Antworten Sie auf Englisch.",
"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
# Erfassung und Wiedergabe parallel starten
await asyncio.gather(
stream_microphone(ws),
play_response_audio(ws)
)
asyncio.run(conversation_vocale())Verwaltung gleichzeitiger Streams
Die Herausforderung beim bidirektionalen Streaming besteht darin, dass Sie gleichzeitig senden und empfangen, und zwar über denselben WebSocket. asyncio.gather startet beide Koroutinen parallel, sie teilen sich jedoch denselben WebSocket-Objekt. Die websockets-Bibliothek bewältigt diese Konfunktionalität korrekt.
Kernaussagen
- Audio wird als 24 kHz Mono 16-Bit PCM übertragen, Base64-codiert in JSON-Events
- Schnipsel sind typischerweise 20 bis 100 ms Audio für einen guten Kompromiss zwischen Latenz und Overhead
- Der serverseitige Audio-Puffer sammelt Schnipsel an, bis er freigegeben wird (automatisch im VAD-Modus)
- Die Wiedergabe der Antwort erfolgt Schnipsel für Schnipsel über
response.output_audio.delta asyncio.gatherermöglicht gleichzeitige Erfassung und Wiedergabe über denselben WebSocket