Echtzeit-Funktionaufrufe

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 (der angegebene Ersatz ist gpt-realtime-2.1-mini, den die OpenAI TTS-Anleitung als nicht unterstützt durch /v1/audio/speech ausweist: Die Migration der Spracherstellung erfolgt über die Realtime API); gpt-realtime, gpt-4o-realtime und die 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.

Echtzeit-Funktionaufrufe

Funktionaufrufe in der Realtime API ermöglichen es Ihrem Sprachagenten, während eines Gesprächs konkrete Aktionen auszuführen: Abfragen einer Datenbank, Aufgeben einer Bestellung oder Prüfen eines Status. Die Mechanik ähnelt der der Responses API, ist jedoch an den WebSocket-Event-Stream angepasst.

Deklaration von Tools in der Sitzung

Sie deklarieren Tools in session.update, genau wie Anweisungen und Sprache. Jedes Tool ist ein JSON-Schema, das seine Parameter beschreibt:

tools = [
    {
        "type": "function",
        "name": "get_weather",
        "description": "Holen Sie sich das aktuelle Wetter für eine gegebene Stadt",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "Name der Stadt"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "Temperatureinheit"
                }
            },
            "required": ["city"]
        }
    },
    {
        "type": "function",
        "name": "book_appointment",
        "description": "Ein Termin buchen",
        "parameters": {
            "type": "object",
            "properties": {
                "date": {"type": "string", "description": "Datum im Format YYYY-MM-DD"},
                "time": {"type": "string", "description": "Zeit im Format HH:MM"},
                "service": {"type": "string", "description": "Art der Dienstleistung"}
            },
            "required": ["date", "time", "service"]
        }
    }
]

await ws.send(json.dumps({
    "type": "session.update",
    "session": {
        "type": "realtime",
        "output_modalities": ["audio"],
        "instructions": "Sie sind ein Sprachassistent. Nutzen Sie die verfügbaren Tools.",
        "audio": {"output": {"voice": "marin"}},
        "tools": tools,
        "tool_choice": "auto"
    }
}))

Der Ablauf der Funktionaufrufe

Wenn das Modell entscheidet, ein Tool aufzurufen, ist die Ereignissequenz wie folgt:

Implementierung des Funktionshandlers

Hier ist ein vollständiges Beispiel für die Handhabung von Funktionaufrufen:

import json

# Registry der verfügbaren Funktionen
async def get_weather(city: str, unit: str = "celsius") -> dict:
    # Rufen Sie eine echte Wetter-API auf
    return {"city": city, "temperature": 22, "unit": unit, "condition": "sunny"}

async def book_appointment(date: str, time: str, service: str) -> dict:
    # Buchungslogik
    return {"confirmed": True, "date": date, "time": time, "reference": "RDV-2026-0042"}

FUNCTION_REGISTRY = {
    "get_weather": get_weather,
    "book_appointment": book_appointment,
}

async def handle_function_call(ws, event):
    """Verarbeitet einen Funktionaufruf und gibt das Ergebnis zurück."""
    response = event["response"]

    for output in response.get("output", []):
        if output["type"] != "function_call":
            continue

        func_name = output["name"]
        call_id = output["call_id"]
        arguments = json.loads(output["arguments"])

        print(f"Rufe Funktion auf: {func_name}({arguments})")

        # Führe die Funktion aus
        func = FUNCTION_REGISTRY.get(func_name)
        if func:
            result = await func(**arguments)
        else:
            result = {"error": f"Unbekannte Funktion: {func_name}"}

        # Senden Sie das Ergebnis zurück an das Modell
        await ws.send(json.dumps({
            "type": "conversation.item.create",
            "item": {
                "type": "function_call_output",
                "call_id": call_id,
                "output": json.dumps(result)
            }
        }))

    # Bitten Sie das Modell, mit dem Ergebnis fortzufahren
    await ws.send(json.dumps({
        "type": "response.create"
    }))

Integration in den Ereignisloop

Der Funktionshandler integriert sich in Ihren Hauptloop:

async def event_loop(ws):
    """Hauptverarbeitungsloop für Ereignisse."""
    async for raw_message in ws:
        event = json.loads(raw_message)
        event_type = event["type"]

        if event_type == "response.done":
            response = event["response"]
            # Prüfen, ob das Modell Funktionen aufgerufen hat
            has_function_calls = any(
                o["type"] == "function_call"
                for o in response.get("output", [])
            )
            if has_function_calls:
                await handle_function_call(ws, event)

        elif event_type == "response.output_audio.delta":
            # Abspielen der Antwort-Audio
            audio_bytes = base64.b64decode(event["delta"])
            player.write(audio_bytes)

        elif event_type == "error":
            print(f"Fehler: {event['error']['message']}")

Mehrere Funktionaufrufe

Das Modell kann entscheiden, mehrere Funktionen in einer einzigen Antwort aufzurufen. Das Feld output von response.done kann mehrere Einträge vom Typ function_call enthalten. Verarbeiten Sie alle von ihnen, bevor Sie response.create aufrufen.

Kernaussagen

• Tools werden in session.update unter Verwendung eines standardmäßigen JSON-Schemas deklariert.
• Argumente kommen über Streaming durch response.function_call_arguments.delta an.
• Nach der Ausführung senden Sie das Ergebnis zurück über conversation.item.create mit der call_id.
• Rufen Sie response.create auf, damit das Modell seine Sprachantwort mit dem Ergebnis formulieren kann.
• Behandeln Sie mehrere Aufrufe: Das Modell kann mehrere Funktionen gleichzeitig aufrufen.