Latenz und Echtzeit-Optimierung

Aktualisiert am 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 eingestellt (Nachfolger: gpt-realtime-2.1-mini, der laut OpenAI-TTS-Leitfaden nicht durch /v1/audio/speech unterstützt wird; die Migration der Sprachsynthese erfolgt über die Realtime API); gpt-realtime, gpt-4o-realtime und Legacy-Audiodateien 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 Voll-Duplex-Stimmen ist GPT-Live 1 seit dem 10. September 2026 in der API verfügbar.

Korrigiert am 3. November 2026. Die den einzelnen Schritten zugewiesenen Dauern („VAD 300–700 ms”, „Netzwerk 20–100 ms”, „Modell 200–800 ms”) und Wahrnehmungsschwellen („über 500 ms”) besaßen keine Quellen: OpenAI veröffentlicht solche Werte nicht, und die Lektion lehrt Sie nun, sie zu messen. Die Anweisung, den Server „in derselben Region wie die OpenAI-Server (US East)” zu platzieren, wurde aufgrund fehlender Quellen entfernt. Dokumentierte Ergänzungen: semantic_vad, WARP zur Beschleunigung der WebRTC-Sitzungsaufstellung. Quellen: Sprachaktivitätserkennung · WebRTC mit WARP · Kostenoptimierung.

Latenz ist der größte Feind eines Sprachagents: die Stille nach dem Ende einer Frage ist auf dem Telefon oder in einer Anwendung sofort wahrnehmbar. Diese Lektion erläutert die Quellen der Latenz, wie man sie misst und Techniken zur Reduzierung.

Anatomie der Latenz

Die Gesamtlatenz eines Sprachagents ist die Summe mehrerer Komponenten:

VAD

Warten auf die Stille, die den Turn beendet (`silence_duration_ms`)

Netzwerk

Rundreisen zwischen Benutzer, Ihrem Server und der API

Modell

Kontextverarbeitung, dann erstes Audiofragment

Je nach Architektur fügen Sie die Tools hinzu, die der Agent aufruft, sowie die Sitzungssetup-Zeit. Keine dieser Dauern ist fest: Sie hängen von Ihren Einstellungen, der Gesprächslänge und dem Netzwerkpfad ab. Der einzige Weg, zu wissen, wohin die Zeit geht, besteht darin, jeden Schritt zu messen, wie das unten gezeigte Monitor-System.

Optimierung der VAD-Erkennung

Mit server_vad kann die Antwort nicht beginnen, bevor die Stille, definiert durch silence_duration_ms, endet: dies ist daher eine direkte Einstellung für die wahrgenommene Latenz, die gegen das Risiko abgewogen werden muss, den Benutzer zu unterbrechen. Die untenstehenden Werte sind Ausgangspunkte zum Testen, keine Empfehlungen von OpenAI:

# Aggressive Konfiguration (schnelle Antwort, Risiko des Abhakens des Benutzers)
fast_vad = {
    "type": "server_vad",
    "threshold": 0.6,           # Höhere Schwelle = weniger falsch-positive Ergebnisse
    "prefix_padding_ms": 200,
    "silence_duration_ms": 400  # Schnelle Reaktion
}

# Patient Konfiguration (weniger Unterbrechungen, mehr Latenz)
patient_vad = {
    "type": "server_vad",
    "threshold": 0.4,
    "prefix_padding_ms": 400,
    "silence_duration_ms": 800
}

# Alternative: semantische VAD, die das Satzende basierend auf Wörtern bewertet;
# „high" reagiert so schnell wie möglich, „low" lässt den Benutzer zu Ende sprechen
semantic_vad = {"type": "semantic_vad", "eagerness": "high"}

# Adaptive Konfiguration basierend auf Kontext
def get_vad_config(context: str) -> dict:
    configs = {
        "menu_ivr": {"threshold": 0.6, "silence_duration_ms": 400},
        "conversation": {"threshold": 0.5, "silence_duration_ms": 600},
        "dictation": {"threshold": 0.3, "silence_duration_ms": 1200},
    }
    config = configs.get(context, configs["conversation"])
    return {
        "type": "server_vad",
        "prefix_padding_ms": 300,
        **config
    }

Optimierung der Systemanweisungen

Knappe Anweisungen reduzieren den zu verarbeitenden Kontext. Denken Sie auch daran, dass die Realtime API Prompt-Caching automatisch anwendet: stabile Anweisungen und Tools, die oben platziert werden und während einer Sitzung nie geändert werden, werden von Turn zu Turn wiederverwendet.

# Ausführliche Anweisungen (mehr Tokens zur Verarbeitung)
bad_instructions = """
Sie sind ein intelligenter und kompetenter Sprachassistent, der
für Corsen AI arbeitet, ein französisches Unternehmen, das sich
auf künstliche Intelligenz spezialisiert. Sie müssen Benutzerfragen
professionell, knapp und auf Französisch beantworten.
Sie haben Zugriff auf verschiedene Tools, um Termine abzufragen,
Verfügbarkeit zu prüfen und Reservierungen zu tätigen.
Zögern Sie nicht, nach Klärung zu fragen, wenn die Anfrage unklar ist.
"""

# Optimierte Anweisungen (gleiche Wirkung, weniger Tokens)
good_instructions = """
Corsen AI Sprachassistent. Französisch, knapp, professionell.
Verwenden Sie Tools für Termine und Reservierungen.
Fragen Sie bei Bedarf nach Klärung.
"""

Reduzierung der Netzwerklatenz

Messung der Rundreise

Platzieren Sie Ihren WebSocket-Relay so nah wie möglich bei Ihren Benutzern und messen Sie den Pfad, anstatt ihn anzunehmen. Ein einfaches Ereignis, wie das Leeren des Eingabepuffers, reicht aus, um eine Rundreise zu messen:

import time

async def measure_roundtrip(ws):
    """Misst die Netzwerklatenz zur Realtime API."""
    start = time.monotonic()
    await ws.send(json.dumps({"type": "input_audio_buffer.clear"}))
    while json.loads(await ws.recv())["type"] != "input_audio_buffer.cleared":
        pass  # andere Ereignisse ignorieren, die dazwischen eintreffen
    roundtrip_ms = (time.monotonic() - start) * 1000
    print(f"Netzwerklatenz: {roundtrip_ms:.0f} ms")
    return roundtrip_ms

Keepalive und Wiederverbindung

import asyncio
import websockets

async def robust_connection(api_key: str, max_retries: int = 5):
    """Robuste Verbindung mit automatischer Wiederverbindung."""
    retry_count = 0

    while retry_count < max_retries:
        try:
            async with websockets.connect(
                REALTIME_URL,
                additional_headers={
                    "Authorization": f"Bearer {api_key}"
                },
                ping_interval=20,     # Ping alle 20s
                ping_timeout=10,       # Ping Timeout 10s
                close_timeout=5        # Schließ-Timeout 5s
            ) as ws:
                retry_count = 0  # Zurücksetzen bei erfolgreicher Verbindung
                yield ws

        except websockets.exceptions.ConnectionClosed:
            retry_count += 1
            wait = min(2 ** retry_count, 30)
            print(f"Wiederverbindung in {wait}s...")
            await asyncio.sleep(wait)

Eine Wiederverbindung öffnet eine neue Sitzung: das vorherige Gespräch ist nicht mehr vorhanden, und eine Sitzung dauert ohnehin maximal 60 Minuten. Speichern Sie daher Ihre eigenen Aufzeichnungen der Austausche (Transkripte, Tool-Ergebnisse) und injizieren Sie das Wesentliche nach einer Wiederverbindung mit conversation.item.create erneut.

Beschleunigung der WebRTC-Aufstellung: WARP

Auf der Browser- oder nativen App-Seite dokumentiert OpenAI WARP (WebRTC Abridged Roundtrip Protocol), das die WebRTC-Sitzungsaufstellungszeit reduziert, indem es drei Transportoptimierungen kombiniert: DTLS 1.3, SPED und SNAP. Dies sind IETF-Entwürfe, die von aktuellen Versionen von libwebrtc unterstützt werden; Sie können die von Ihrem Client unterstützten schrittweise aktivieren. Der Leitfaden WebRTC mit WARP erläutert die Konfiguration.

Optimierung des Funktionsaufrufs

Funktionsaufrufe fügen Latenz hinzu. Optimieren Sie deren Ausführung:

import asyncio

async def execute_functions_parallel(function_calls: list) -> list:
    """Führt Funktionsaufrufe parallel aus."""
    tasks = []
    for call in function_calls:
        task = execute_single_function(call["name"], call["arguments"])
        tasks.append(task)

    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

# Cache für häufige Funktionen
from functools import lru_cache

@lru_cache(maxsize=100)
def get_cached_data(key: str) -> dict:
    """Cacht Ergebnisse häufiger Anfragen."""
    # Datenbank- oder API-Abfrage
    return {"data": "..."}

Latenzüberwachung

Integrieren Sie ein Messsystem, um Engpässe zu identifizieren:

import time
from dataclasses import dataclass, field

@dataclass
class LatencyMetrics:
    speech_end_time: float = 0
    response_start_time: float = 0
    first_audio_time: float = 0
    response_complete_time: float = 0
    function_call_times: list = field(default_factory=list)

    @property
    def time_to_first_audio(self) -> float:
        if self.speech_end_time and self.first_audio_time:
            return (self.first_audio_time - self.speech_end_time) * 1000
        return 0

    @property
    def total_response_time(self) -> float:
        if self.speech_end_time and self.response_complete_time:
            return (self.response_complete_time - self.speech_end_time) * 1000
        return 0

    def report(self) -> dict:
        return {
            "ttfa_ms": round(self.time_to_first_audio),
            "total_ms": round(self.total_response_time),
            "function_calls_ms": [round(t * 1000) for t in self.function_call_times]
        }

class LatencyTracker:
    def __init__(self):
        self.current = LatencyMetrics()
        self.history = []

    def on_event(self, event_type: str):
        now = time.time()

        if event_type == "input_audio_buffer.speech_stopped":
            self.current.speech_end_time = now

        elif event_type == "response.output_audio.delta":
            if not self.current.first_audio_time:
                self.current.first_audio_time = now

        elif event_type == "response.done":
            self.current.response_complete_time = now
            self.history.append(self.current.report())
            self.current = LatencyMetrics()

    def average_ttfa(self) -> float:
        if not self.history:
            return 0
        return sum(h["ttfa_ms"] for h in self.history) / len(self.history)

Kernaussagen

Die Gesamtlatenz addiert sich aus VAD, Netzwerk, Modell und Tools: messen Sie jeden Schritt, OpenAI veröffentlicht keine typischen Dauern
silence_duration_ms steuert direkt das Warten vor der Antwort; semantic_vad und dessen eagerness sind der andere Hebel
Knappe Systemanweisungen reduzieren die Modellverarbeitungszeit
Parallele Ausführung von Funktionsaufrufen und Caching reduzieren die Anwendungs-Latenz
Ein Überwachungssystem mit TTFA (Time-to-First-Audio) ist für die kontinuierliche Optimierung unerlässlich