Echtzeit-Audio-Streaming

Update vom 2. November 2026 — Audio-Modelle. Mehrere Modelle in diesem Kurs wurden eingestellt: tts-1, tts-1-hd und gpt-4o-mini-tts werden am 6. Januar 2027 nicht mehr funktionieren (Nachfolger: gpt-realtime-2.1-mini, der laut OpenAI TTS-Anleitung nicht durch /v1/audio/speech unterstützt wird; die Migration der Sprachsynthese erfolgt über die Realtime API); gpt-realtime, gpt-4o-realtime und veraltete Audio-Modelle am 20. Januar 2027 (gpt-realtime-2.1, gpt-realtime-2.1-mini, gpt-audio-1.5); whisper-1, gpt-4o-transcribe und /v1/audio/translations am 26. Februar 2027 (gpt-transcribe, gpt-live-transcribe, v1/realtime/translations). Für die Voll-Duplex-Stimme ist GPT-Live 1 seit dem 10. September 2026 in der API verfügbar. Aktualisiert am 3. November 2026: Die Pipeline LLM → TTS verarbeitete zuvor jedes Event mit einem delta-Feld, einschließlich Funktionsaufruf-Argumenten oder Reaktionszusammenfassungen; sie filtert nun strikt nach response.output_text.delta, dem eigentlichen Antworttext.

In einem Sprachagenten führt das Warten auf die vollständige Generierung der Audiodatei vor dem Start der Wiedergabe zu einer unannehmbaren Latenz. Das TTS-Streaming ermöglicht es, die Wiedergabe sofort zu beginnen, sobald die ersten Bytes eintreffen, und reduziert die Zeit bis zur ersten Antwort auf nur wenige hundert Millisekunden.

Streaming mit der TTS-API

Die TTS-API unterstützt Streaming nativ. Anstatt auf die vollständige Datei zu warten, erhalten Sie das Audio in Blöcken:

from openai import OpenAI
import pyaudio

client = OpenAI()

def stream_tts_to_speaker(text: str, voice: str = "alloy"):
    """Generiere und spiele Audio im Streaming-Modus ab."""
    # Initialisiere den Audioplayer
    pa = pyaudio.PyAudio()
    player = pa.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=24000,
        output=True
    )

    try:
        # Erstelle die Streaming-Antwort
        with client.audio.speech.with_streaming_response.create(
            model="tts-1",
            voice=voice,
            input=text,
            response_format="pcm"  # Rohes PCM für direkte Wiedergabe
        ) as response:
            # Lies und spiele Block für Block ab
            for chunk in response.iter_bytes(chunk_size=4096):
                player.write(chunk)

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

stream_tts_to_speaker("Hallo, hier ist eine Echtzeit-Streaming-Antwort.")

Das pcm-Format ist ideal für Streaming, da es keine Decodierung erfordert: Die rohen Bytes sind sofort bereit für die Wiedergabe.

Messung der Latenz

Die Zeit bis zum ersten Byte (TTFB) ist die entscheidende Kennzahl für die Streaming-Leistung:

import time

def measure_tts_latency(text: str, voice: str = "alloy") -> dict:
    """Misst TTS-Streaming-Latenzen."""
    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("Hallo, wie geht es Ihnen?")
print(f"TTFB: {metrics['ttfb_ms']}ms, Total: {metrics['total_ms']}ms")

Asynchrones Streaming

Für serverseitige Anwendungen verwenden Sie den asynchronen Client:

from openai import AsyncOpenAI
import asyncio

async_client = AsyncOpenAI()

async def stream_tts_async(text: str, voice: str = "alloy"):
    """Asynchrones TTS-Streaming für Server."""
    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)
            # Hier würden Sie den Block über WebSocket an den Client senden
            yield chunk

# Verwendung in einem FastAPI-Server
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"
    )

Puffer-Vorabfüllung

Um eine gleichmäßige Wiedergabe ohne Lücken zu gewährleisten, sammeln Sie einen kleinen Puffer, bevor Sie beginnen:

async def buffered_tts_playback(
    text: str,
    voice: str = "alloy",
    buffer_size: int = 3  # Anzahl der Blöcke zum Vorladen
):
    """TTS-Wiedergabe mit Puffer-Vorabfüllung."""
    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
                    # Spülen Sie den initialen Puffer
                    for buffered_chunk in buffer:
                        player.write(buffered_chunk)
                    buffer.clear()
                elif playback_started:
                    player.write(chunk)

        # Spielen Sie verbleibende Blöcke ab
        for remaining in buffer:
            player.write(remaining)

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

Integration in eine LLM- und TTS-Pipeline

In einem nicht-Realtime-Sprachagenten können Sie das Streaming der LLM mit dem Streaming des TTS kombinieren:

async def llm_to_tts_pipeline(user_message: str):
    """Pipeline: Streaming LLM → Echtzeit-Streaming TTS."""
    # Phase 1: Sammeln der LLM-Antwort nach Satz
    sentence_buffer = ""

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

    for event in stream:
        # Nur Antworttext sollte gesprochen werden
        if event.type == "response.output_text.delta":
            sentence_buffer += event.delta

            # Erkennen des Satzendes
            if sentence_buffer.rstrip().endswith((".", "!", "?")):
                sentence = sentence_buffer.strip()
                sentence_buffer = ""

                # Phase 2: Streamen dieses Satzes an TTS
                stream_tts_to_speaker(sentence)

    # Verarbeiten des verbleibenden Puffers
    if sentence_buffer.strip():
        stream_tts_to_speaker(sentence_buffer.strip())

Kernaussagen

Das TTS-Streaming reduziert die TTFB auf wenige hundert Millisekunden, indem die Wiedergabe vor Abschluss der Generierung beginnt.
Das pcm-Format ist optimal für Streaming: Es ist keine Decodierung erforderlich, die Wiedergabe erfolgt direkt.
Ein Puffer-Vorabfüllung von 2–3 Blöcken verhindert Lücken am Anfang der Wiedergabe.
Der asynchrone Client ist für Server unerlässlich, die mehrere gleichzeitige Verbindungen verwalten.
Die Kombination aus Streaming-LLM und Streaming-TTS ermöglicht einen reaktionsschnellen Sprachagenten ohne die Realtime API.