Produktion: Skalierung und Überwachung

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 nicht mehr funktionieren (Ersatz: gpt-realtime-2.1-mini, wobei die OpenAI-TTS-Richtlinien angeben, dass dies nicht durch /v1/audio/speech unterstützt wird: Die Migration der Spracherstellung muss über die Realtime-API erfolgen); gpt-realtime, gpt-4o-realtime und Legacy-Audiomodelle 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 Kostenverwaltung passte zuvor in einen einzigen Satz; sie spiegelt nun das von OpenAI beschriebene Abrechnungsmodell wider (Abrechnung pro Antwort, jede Konversation wird pro Runde zurückgegeben, Audio wird in Tokens gezählt, automatische Caching, Transkription wird separat abgerechnet) und die Trunkierung über retention_ratio, deren Format gegen die openai SDK-Typen Version 3.42.1 verifiziert wurde. Quelle: Kostenoptimierung.

Produktion: Skalierung und Überwachung

Der Übergang von einem Sprachagenten-Prototypen zu einem zuverlässigen, skalierbaren Produktionsservice erfordert eine sorgfältige Aufmerksamkeit für Infrastruktur, Überwachung und Fehlerbehandlung. Diese letzte Lektion behandelt die operativen Grundlagen für den Einsatz eines Sprachagenten unter realen Bedingungen.

Produktionsarchitektur

In der Produktion läuft der Sprachagent hinter einem Lastverteiler und einem Pool von Arbeitern:

Verwaltung gleichzeitiger Sitzungen

Jede WebSocket-Verbindung unterhält eine Realtime-API-Sitzung. Die Anzahl der gleichzeitigen Sitzungen wird durch Ihre OpenAI-Rate Limits und die Kapazität Ihrer Worker begrenzt:

import asyncio
from dataclasses import dataclass

@dataclass
class SessionManager:
    """Verwaltet gleichzeitige Realtime-API-Sitzungen."""
    max_sessions: int = 50
    active_sessions: dict = None
    semaphore: asyncio.Semaphore = None

    def __post_init__(self):
        self.active_sessions = {}
        self.semaphore = asyncio.Semaphore(self.max_sessions)

    async def create_session(self, session_id: str, config: dict):
        """Erstellt eine neue Sitzung mit Steuerung der Parallelität."""
        async with self.semaphore:
            if session_id in self.active_sessions:
                raise ValueError(f"Sitzung {session_id} ist bereits aktiv")

            self.active_sessions[session_id] = {
                "created_at": asyncio.get_event_loop().time(),
                "config": config,
                "status": "active"
            }
            return session_id

    async def close_session(self, session_id: str):
        """Schließt eine Sitzung ordnungsgemäß."""
        if session_id in self.active_sessions:
            del self.active_sessions[session_id]

    @property
    def active_count(self) -> int:
        return len(self.active_sessions)

    @property
    def available_slots(self) -> int:
        return self.max_sessions - self.active_count

session_manager = SessionManager(max_sessions=50)

Strukturierte Protokollierung

Protokollieren Sie jede Interaktion zur Fehleranalyse und Compliance:

import logging
import json
from datetime import datetime

class ConversationLogger:
    """Strukturierte Protokollierung für Sprachkonversationen."""

    def __init__(self, session_id: str):
        self.session_id = session_id
        self.logger = logging.getLogger(f"voice.{session_id}")
        self.events = []

    def log_event(self, event_type: str, data: dict = None):
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "session_id": self.session_id,
            "event_type": event_type,
            "data": data or {}
        }
        self.events.append(entry)
        self.logger.info(json.dumps(entry))

    def log_transcription(self, role: str, text: str):
        self.log_event("transcription", {
            "role": role,
            "text": text
        })

    def log_function_call(self, name: str, args: dict, result: dict, duration_ms: float):
        self.log_event("function_call", {
            "name": name,
            "arguments": args,
            "result": result,
            "duration_ms": round(duration_ms)
        })

    def log_error(self, error_type: str, message: str):
        self.log_event("error", {
            "type": error_type,
            "message": message
        })

    async def save_to_db(self, db_pool):
        """Persistiert das vollständige Protokoll in der Datenbank."""
        async with db_pool.acquire() as conn:
            await conn.execute(
                "INSERT INTO conversation_logs (session_id, events, created_at) "
                "VALUES ($1, $2, NOW())",
                self.session_id, json.dumps(self.events)
            )

Prometheus-Metriken

Stellen Sie Metriken für die Echtzeitüberwachung bereit:

from prometheus_client import Counter, Histogram, Gauge, start_http_server

# Wichtige Metriken
sessions_active = Gauge(
    "voice_sessions_active",
    "Anzahl aktiver Sprachsessions"
)

session_duration = Histogram(
    "voice_session_duration_seconds",
    "Dauer der Sprachsessions",
    buckets=[30, 60, 120, 300, 600, 1800]
)

response_latency = Histogram(
    "voice_response_latency_ms",
    "Antwortlatenz (Ende der Sprache → erstes Audio)",
    buckets=[200, 400, 600, 800, 1000, 1500, 2000, 3000]
)

errors_total = Counter(
    "voice_errors_total",
    "Gesamtzahl der Fehler",
    ["error_type"]
)

tokens_used = Counter(
    "voice_tokens_total",
    "Verbrauchte Tokens",
    ["type"]  # input_audio, output_audio, input_text, output_text
)

# Starten des Metrik-Servers
start_http_server(9090)

Kostenverwaltung

Laut OpenAI-Richtlinie zur Kostenoptimierung wird eine Realtime-Konversationsitzung pro Antwort abgerechnet, in Eingabe- und Ausgabetokens für Text, Audio und Bilder. Drei Mechanismen treiben den Großteil der Rechnung:

• Die gesamte Konversation wird mit jeder Antwort an das Modell gesendet: spätere Runden kosten mehr als frühere.
• Audio wird in Tokens gezählt: 1 Token pro 100 ms Nutzer-Audio, 1 Token pro 50 ms Assistenten-Audio.
• Prompt-Caching wird automatisch angewendet: solange der Anfang der Konversation unverändert bleibt, wird er mit dem reduzierten Tarif für gecachte Tokens abgerechnet. Das Ändern oder Löschen eines älteren Elements „bricht” das Caching ab diesem Punkt an.

Die Eingabetranskription, falls aktiviert, wird separat nach dem Tarif des jeweiligen Modells abgerechnet: gpt-transcribe wird nach Dauer berechnet, ausgelesen vom Event conversation.item.input_audio_transcription.completed. Der Verbrauch pro Antwort erscheint in response.done unter response.usage (total_tokens, input_tokens, output_tokens, plus die Aufschlüsselung für Text, Audio, Bild und Cache).

Um die Kontextgröße – und damit die Kosten späterer Runden – zu begrenzen, akzeptiert die Sitzung eine Trunkierungsregel: Jenseits des Limits werden die ältesten Nachrichten verworfen. Mit retention_ratio unter 1 entfernt der Server etwas mehr als unbedingt notwendig, was die Trunkierungen verteilt und das Caching erhält:

await ws.send(json.dumps({
    "type": "session.update",
    "session": {
        "type": "realtime",
        "truncation": {
            "type": "retention_ratio",
            "retention_ratio": 0.8,
            "token_limits": {"post_instructions": 8000},
        },
    },
}))

Der Kompromiss: Ein Agent, der weniger Kontext behält, vergisst schneller. Ergänzen Sie dies mit Sitzungslimits in Ihrem Code:

class CostController:
    """Steuert Kosten für eine Realtime-Sitzung."""

    def __init__(self, max_duration_s: int = 600, max_tokens: int = 50000):
        self.max_duration_s = max_duration_s
        self.max_tokens = max_tokens
        self.total_tokens = 0
        self.start_time = None

    def start(self):
        self.start_time = asyncio.get_event_loop().time()

    def update_tokens(self, usage: dict):
        self.total_tokens += usage.get("total_tokens", 0)

    @property
    def elapsed_seconds(self) -> float:
        if not self.start_time:
            return 0
        return asyncio.get_event_loop().time() - self.start_time

    @property
    def should_terminate(self) -> bool:
        if self.elapsed_seconds > self.max_duration_s:
            return True
        if self.total_tokens > self.max_tokens:
            return True
        return False

    @property
    def warning_message(self) -> str:
        remaining = self.max_duration_s - self.elapsed_seconds
        if remaining < 60:
            return f"Diese Sitzung endet in {int(remaining)} Sekunden."
        return ""

Fehlerbehandlung in der Produktion

async def production_event_loop(ws, logger, cost_ctrl):
    """Robuste Ereignisschleife für die Produktion."""
    cost_ctrl.start()

    try:
        async for raw in ws:
            event = json.loads(raw)
            t = event["type"]

            # Kostenlimits prüfen
            if cost_ctrl.should_terminate:
                logger.log_event("session_limit_reached")
                await ws.send(json.dumps({
                    "type": "conversation.item.create",
                    "item": {
                        "type": "message",
                        "role": "user",
                        "content": [{"type": "input_text",
                                     "text": "Die Sitzung nähert sich ihrer Dauerbegrenzung."}]
                    }
                }))
                break

            if t == "response.done":
                usage = event["response"].get("usage", {})
                cost_ctrl.update_tokens(usage)
                tokens_used.labels(type="total").inc(usage.get("total_tokens", 0))

            elif t == "error":
                error = event["error"]
                logger.log_error(error["type"], error["message"])
                errors_total.labels(error_type=error["type"]).inc()

                if error["type"] == "rate_limit_error":
                    await asyncio.sleep(1)

    except websockets.exceptions.ConnectionClosed as e:
        logger.log_error("connection_closed", str(e))
    except Exception as e:
        logger.log_error("unexpected", str(e))
    finally:
        session_duration.observe(cost_ctrl.elapsed_seconds)
        sessions_active.dec()

Kernaussagen

• Die Produktionsarchitektur trennt Lastverteiler, asyncio-Worker, Datenbank und Überwachung.
• Ein SessionManager mit Semaphore steuert die Anzahl der gleichzeitigen Sitzungen.
• JSON-basierte strukturierte Protokollierung ermöglicht Fehleranalyse und Compliance (Aufzeichnung von Konversationen).
• Prometheus-Metriken (TTFA, Dauer, Tokens, Fehler) sind für den täglichen Betrieb unerlässlich.
• Der CostController begrenzt Sitzungsdauer und Tokens, um Budgetüberschreitungen zu vermeiden.

Wissen testen

Fünf Fragen, die die gesamte Sprachkette abdecken, vom WebSocket bis zur Produktionsbereitstellung.

1. Warum verwendet die Realtime-API WebSocket statt HTTP-Anfragen?

Antwort: Weil eine persistente, bidirektionale Verbindung es ermöglicht, Mikrofon-Audio zu streamen und Antworten kontinuierlich zu empfangen, ohne die Overhead-Kosten einer Anfrage für jeden Austausch zu zahlen. Dies ist die Voraussetzung für eine geringe Latenz in der Konversation.

2. Was müssen Sie sofort nach `session.created` senden und warum?

Antwort: session.update, sofort: Hier konfigurieren Sie die Stimme, Anweisungen, Tools und das Audioformat. Bis diese Nachricht gesendet wurde, läuft die Sitzung mit Standardwerten.

3. Warum ist ein Proxy-Server in der Produktion unerlässlich?

Antwort: Er verhindert die Offenlegung des API-Schlüssels im Client und ermöglicht die Authentifizierung des Nutzers, die Begrenzung des Verbrauchs und das Protokollieren von Aufrufen – drei Dinge, die unmöglich sind, wenn der Browser direkt mit der API spricht.

4. Was ist der Zweck des `prompt`-Parameters in der Transkriptions-API?

Antwort: Um die Erkennung in Richtung erwarteten Vokabulars zu lenken: Eigennamen, Marken, branchenspezifische Fachbegriffe. Begriffe, die Sie dort auflisten, werden viel genauer transkribiert.

5. Was ist der wichtigste Hebel für die wahrgenommene Latenz bei einem Sprachagenten?

Antwort: End-to-End-Streaming: Audio abzuspielen, sobald die ersten Fragmente eintreffen, anstatt auf die vollständige Antwort zu warten, und Unterbrechungen zu handhaben, sodass der Nutzer jederzeit einlegen kann.