Fehlerbehandlung und Fallbacks

Funktionsaufrufe stellen eine Herausforderung dar, die klassischer Code nicht kennt: Wenn eine Funktion fehlschlägt, gibt es zwei Empfänger für diesen Fehler. Ihre Anwendung, wie üblich, und das Modell, das auf ein Ergebnis wartet, um fortzufahren. Dieses Vergessen führt zu dem verwirrendsten Symptom in der Produktion: ein Gespräch, das abrupt ohne Erklärung abbricht, weil eine Ausnahme auf Serverseite verschluckt wurde und das Modell weiterhin wartet.

Die Regel, die diese gesamte Lektion strukturiert, lautet: geben Sie immer ein function_call_output zurück, auch bei Fehlern. Ein Modell, das informiert wird, dass eine API nicht verfügbar ist, kann eine Alternative vorschlagen, um Klarheit bitten oder die Situation dem Nutzer erklären. Ein Modell, das nichts erhält, kann nichts tun.

Rückgabe von Fehlern an das Modell

Das Format des Fehlers ist ebenso wichtig wie die Übermittlung selbst. Ein bloßes “Error” lehrt das Modell nichts; ein JSON-Objekt, das die Fehlertypen unterscheidet, ob sie vorübergehend oder dauerhaft sind, und die nächsten Schritte, ermöglicht es dem Modell, eine Entscheidung zu treffen. Dies ist kontraintuitiv: Sie verfassen Fehlermeldungen, die von einer vernunftfähigen Maschine gelesen werden sollen, nicht protokolliert für Menschen.

import json
from openai import OpenAI

client = OpenAI()

def execute_function(name: str, arguments: str, call_id: str) -> dict:
    try:
        args = json.loads(arguments)
        result = dispatcher[name](**args)
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps(result)
        }
    except KeyError:
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps({
                "error": "unknown_function",
                "message": f"The function '{name}' does not exist"
            })
        }
    except json.JSONDecodeError:
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps({
                "error": "invalid_arguments",
                "message": "The JSON arguments are malformed"
            })
        }
    except Exception as e:
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps({
                "error": "internal_error",
                "message": str(e)
            })
        }

Wiederholung mit exponentiellem Backoff

Wiederholungen machen nur Sinn für Probleme, die sich bei einem zweiten Versuch lösen könnten: Netzunterbrechungen, Statuscode 429 oder vorübergehende Ausfälle. Das Wiederholen einer Anfrage, die aufgrund ungültiger Authentifizierung oder falscher Parameter abgelehnt wurde, verzögert lediglich den Misserfolg und vervielfacht die Aufrufe. Deshalb filtert der folgende Code explizit Fehler aus – das Filtern erledigt die Arbeit, während Backoff lediglich die Höflichkeit ist, die Sie dem entfernten Dienst schulden.

import time
import random

def execute_with_retry(func, args, max_retries=3, base_delay=1.0):
    """Retry with exponential backoff and jitter."""
    for attempt in range(max_retries + 1):
        try:
            return func(**args)
        except (ConnectionError, TimeoutError) as e:
            if attempt == max_retries:
                raise
            delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
            time.sleep(delay)
        except ValueError:
            # Data error: do not retry
            raise

def execute_function_robust(name, arguments, call_id):
    args = json.loads(arguments)
    try:
        result = execute_with_retry(dispatcher[name], args)
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps(result)
        }
    except Exception as e:
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps({
                "error": type(e).__name__,
                "message": str(e),
                "retries_performed": 3
            })
        }

Zeitlimits pro Funktion

Ein einzelnes Zeitlimit für alle Funktionen ist immer falsch konfiguriert: zu kurz tötet es schwere Verarbeitung, zu lang lässt den Nutzer bei einer einfachen Cache-Abfrage warten. Basieren Sie jeden Wert auf beobachteten Dauern unter normalen Bedingungen, mit einem Puffer – nicht auf dem schlimmst vorstellbaren Fall, was effektiv bedeuten würde, kein Zeitlimit zu haben.

import signal
from functools import wraps

TIMEOUTS = {
    "search_client": 5,
    "generate_report": 30,
    "send_email": 10,
}

def with_timeout(func, timeout_seconds):
    """Executes a function with a timeout."""
    import concurrent.futures
    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
        future = executor.submit(func)
        try:
            return future.result(timeout=timeout_seconds)
        except concurrent.futures.TimeoutError:
            raise TimeoutError(
                f"Function interrupted after {timeout_seconds}s"
            )

def execute_with_timeout(name, arguments, call_id):
    args = json.loads(arguments)
    timeout = TIMEOUTS.get(name, 10)  # 10s by default

    try:
        result = with_timeout(
            lambda: dispatcher[name](**args),
            timeout
        )
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps(result)
        }
    except TimeoutError as e:
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps({
                "error": "timeout",
                "message": str(e),
                "suggestion": "Retry with more specific criteria"
            })
        }

Fallback zwischen Funktionen

Fallback ist ein Kompromiss, kein Reflex: das Bereitstellen von sechs Stunden alten gecachten Daten ist für ein Dashboard besser als nichts, aber für ein Bankkonto weit schlechter. Wenn Sie die Leistung mindern, sagen Sie es – im Ergebnis, das an das Modell zurückgegeben wird, damit es den Nutzer warnen kann, anstatt veraltete Daten als frisch darzustellen.

async def search_product(term: str) -> dict:
    """Search with fallback: primary API -> cache -> local database."""
    # Attempt 1: Catalog API
    try:
        return await api_catalog.search(term)
    except Exception:
        pass

    # Attempt 2: Redis cache
    try:
        cached = await redis_client.get(f"product:{term}")
        if cached:
            result = json.loads(cached)
            result["_source"] = "cache"
            result["_warning"] = "Data may be outdated"
            return result
    except Exception:
        pass

    # Attempt 3: Local SQLite database
    try:
        row = db.execute(
            "SELECT * FROM products WHERE name LIKE ?",
            (f"%{term}%",)
        ).fetchone()
        if row:
            return {
                "name": row["name"],
                "price": row["price"],
                "_source": "local_db",
                "_warning": "Data from last synchronization"
            }
    except Exception:
        pass

    return {
        "error": "unavailable",
        "message": "No data source accessible for this search"
    }

Schalter (Circuit Breaker)

Der Schalter (Circuit Breaker) bewältigt, was Wiederholungen nicht können: einen Dienst, der für längere Zeit ausgefallen ist. Nach N Fehlern schaltet er um und gibt sofort einen Fehler zurück, ohne auch nur zu versuchen, die Verbindung herzustellen, und lässt gelegentlich eine Testanfrage durch, um die Wiederherstellung zu prüfen. Der Gewinn ist zweifach – Ihre Nutzer warten nicht mehr auf ein Timeout bei jeder Anfrage, und der kämpfende Dienst wird nicht durch Ihre Wiederholungen weiter heruntergehalten.

from datetime import datetime, timedelta

class CircuitBreaker:
    def __init__(self, failure_threshold=5, open_duration=60):
        self.threshold = failure_threshold
        self.duration = open_duration
        self.failures = 0
        self.last_failure = None
        self.state = "closed"  # closed, open, half-open

    def can_execute(self) -> bool:
        if self.state == "closed":
            return True
        if self.state == "open":
            if datetime.now() - self.last_failure > timedelta(seconds=self.duration):
                self.state = "half-open"
                return True
            return False
        return True  # half-open: allow one test attempt

    def record_success(self):
        self.failures = 0
        self.state = "closed"

    def record_failure(self):
        self.failures += 1
        self.last_failure = datetime.now()
        if self.failures >= self.threshold:
            self.state = "open"

# One circuit breaker per external service
breakers = {
    "weather_api": CircuitBreaker(failure_threshold=3, open_duration=30),
    "catalog_api": CircuitBreaker(failure_threshold=5, open_duration=60),
}

Vollständiger robuster Zyklus

Das vollständige Muster bringt die fünf oben genannten Mechanismen zusammen. Eine Vorsichtsmaßnahme macht es wirklich sicher, und sie ist leicht zu übersehen: begrenzen Sie die Anzahl der Iterationen. Ein Modell, das in einer Schleife eine Funktion aufruft, deren Ergebnis es nicht mag, ist selten, aber ohne eine Obergrenze würde es Ihr Budget bis zur Erschöpfung verbrennen – und das ist die Art von Vorfall, den Sie erst auf der Rechnung entdecken.

def robust_loop(prompt, tools, max_turns=10):
    messages = prompt
    for turn in range(max_turns):
        response = client.responses.create(
            model="gpt-5.6-terra",
            input=messages,
            tools=tools
        )

        calls = [i for i in response.output if i.type == "function_call"]
        if not calls:
            return response.output_text

        results = []
        for call in calls:
            result = execute_function_robust(
                call.name, call.arguments, call.call_id
            )
            results.append(result)

        messages = response.output + results

    return "Workflow interrupted: turn limit reached"

Kernaussagen

Geben Sie immer ein function_call_output zurück, auch bei Fehlern
Strukturieren Sie Fehler als JSON, damit das Modell sie nutzen kann
Wiederholen Sie nur vorübergehende Fehler mit exponentiellem Backoff
Implementieren Sie Fallbacks zwischen Datenquellen
Verwenden Sie einen Schalter (Circuit Breaker), um fehlerhafte Dienste zu schützen