AI Engineering Academy · Lektion

Timeout-Budgets und sanfte Verschlechterung

Legen Sie auf jeder Ebene Ihrer Pipeline strenge Timeout-Budgets fest und implementieren Sie eine sanfte Verschlechterung, die gecachte oder vereinfachte Antworten ausliefert, wenn das LLM sein Budget überschreitet.

Lektion 4 von 413 Schritte

Timeout-Budgets und sanfte Verschlechterung ist eine kostenlose AI Engineering Academy-Lektion auf CoddyKit. Dies ist Lektion 4 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des AI Engineering Academy-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der AI Engineering Academy-Kurs umfasst insgesamt 4 Lektionen.

Was ist ein Timeout-Budget?

Ein Timeout-Budget ist die maximal verfügbare Gesamtzeit, in der eine Anfrage alle Phasen Ihrer Pipeline durchlaufen und abgeschlossen werden muss. Statt für jeden einzelnen API-Aufruf ein willkürliches Timeout festzulegen, definieren Sie ein End-to-End-Budget für den benutzerseitigen Vorgang und verteilen es auf Retrieval, LLM-Generierung und Nachbearbeitung. So antworten Sie stets innerhalb einer akzeptablen Zeit, selbst wenn einige Phasen langsam sind.

Budget auf Pipeline-Phasen verteilen

Eine typische RAG-Chat-Pipeline besteht aus drei Phasen: Retrieval, LLM-Generierung und Antwortformatierung. Weisen Sie jeder Phase abhängig von ihrer üblichen Dauer und der von Benutzern tolerierten Wartezeit ein Zeitfenster zu. Die verbleibende Reserve ist Ihr Degradationspuffer – wenn eine Phase ihre gesamte Zuteilung verbraucht, beginnen Sie in den nachfolgenden Phasen Abstriche zu machen, damit das Gesamtbudget eingehalten wird.

# Total user-facing SLA: 8000ms
BUDGET_TOTAL_MS = 8000

BUDGET_STAGES = {
    'retrieval':   1500,  # vector search + rerank
    'llm_call':    5500,  # token streaming
    'formatting':  500,   # post-processing
    'slack':       500,   # buffer for overhead
}

assert sum(BUDGET_STAGES.values()) == BUDGET_TOTAL_MS

Budgetverbrauch verfolgen

Verwenden Sie einen BudgetTracker, der die Startzeit erfasst und bei jedem Übergang zwischen Phasen das verbleibende Budget prüft. Überprüfen Sie vor dem Start einer Phase, ob noch ausreichend Budget vorhanden ist. Dadurch können nachgelagerte Phasen angepasst werden: Wenn ein Retrieval-Schritt 1200 ms seines Budgets von 1500 ms benötigt, bleiben nur 300 ms Reserve. Das sollte einen einfacheren LLM-Prompt auslösen oder den Re-Ranking-Schritt überspringen.

import time

class BudgetTracker:
    def __init__(self, total_ms: float):
        self.start = time.perf_counter()
        self.total_ms = total_ms

    def elapsed_ms(self) -> float:
        return (time.perf_counter() - self.start) * 1000

    def remaining_ms(self) -> float:
        return self.total_ms - self.elapsed_ms()

    def check(self, stage: str, required_ms: float = 0) -> bool:
        remaining = self.remaining_ms()
        if remaining < required_ms:
            print(f'Budget exhausted before {stage}: {remaining:.0f}ms left, need {required_ms}ms')
            return False
        return True

Definition der kontrollierten Verschlechterung

Graceful Degradation bedeutet, eine Antwort mit geringerer Qualität, die aber weiterhin nützlich ist, bereitzustellen, wenn die vollständige Pipeline nicht innerhalb des Budgets abgeschlossen werden kann, statt einen Fehler zurückzugeben. Beispiele sind: eine Antwort aus dem Cache zurückgeben, Re-Ranking überspringen, das Kontextfenster kürzen, ein schnelleres, aber weniger genaues Modell verwenden oder eine vorformulierte Fallback-Nachricht zurückgeben. Das Ziel ist immer, dem Benutzer etwas zu liefern statt gar nichts.

# Degradation ladder for a RAG chat endpoint:
# Level 0 (normal):  retrieve 10 chunks + rerank + GPT-4o   -- 8000ms budget
# Level 1 (fast):    retrieve 5 chunks + skip rerank + GPT-4o -- 5000ms budget
# Level 2 (minimal): retrieve 3 chunks + GPT-4o-mini          -- 3000ms budget
# Level 3 (cached):  return semantic cache hit                 -- 100ms
# Level 4 (sorry):   return static 'Try again in a moment'    -- 1ms

Degradationsstufen implementieren

Prüfen Sie an jedem Entscheidungspunkt der Pipeline das verbleibende Budget und wählen Sie die passende Qualitätsstufe. Der folgende Code wählt anhand des verbleibenden Budgets die Retrieval-Tiefe und das Modell aus. Dadurch erhalten Benutzer unter normaler Auslastung die bestmögliche Qualität, während sie bei hoher Latenz weiterhin eine nützliche Antwort statt eines Timeout-Fehlers bekommen.

async def smart_rag_query(question: str, budget_ms: float = 8000) -> str:
    tracker = BudgetTracker(budget_ms)

    # Retrieval stage
    if tracker.remaining_ms() > 5000:
        chunks = await retrieve_and_rerank(question, top_k=10)
    elif tracker.remaining_ms() > 3000:
        chunks = await retrieve(question, top_k=5)  # skip rerank
    elif tracker.remaining_ms() > 1500:
        chunks = await retrieve(question, top_k=3)  # minimal retrieval
    else:
        return await get_cached_or_static(question)

    # LLM stage
    if tracker.remaining_ms() > 4000:
        model = 'gpt-4o'
    else:
        model = 'gpt-4o-mini'  # faster fallback

    timeout = tracker.remaining_ms() / 1000 - 0.5
    return await generate_answer(question, chunks, model, timeout)

Timeouts auf Ebene der API-Aufrufe festlegen

Legen Sie für jeden externen API-Aufruf immer ein explizites Timeout fest. Das OpenAI Python SDK akzeptiert einen Parameter timeout in Sekunden. Legen Sie ihn etwas kürzer als das verbleibende Budget fest, damit Sie Zeit haben, die Ausnahme zu behandeln und vor Ablauf der Gesamtdauer gegebenenfalls eine kontrollierte Verschlechterung einzuleiten. Verlassen Sie sich niemals auf das Standard-Timeout des SDK – für benutzerseitige Anfragen kann es zu lang sein.

async def generate_answer(question: str, chunks: list, model: str, timeout_sec: float) -> str:
    context = '\n\n'.join(chunks)
    prompt = f'Answer using this context:\n{context}\n\nQuestion: {question}'
    try:
        resp = await client.chat.completions.create(
            model=model,
            messages=[{'role': 'user', 'content': prompt}],
            max_tokens=500,
            timeout=max(timeout_sec, 1.0)  # minimum 1 second
        )
        return resp.choices[0].message.content
    except openai.APITimeoutError:
        return 'I was unable to generate a response in time. Please try again.'

Teilweise Streaming-Antworten zurückgeben

Beim Streaming können Sie Teilantworten zurückgeben, die vor Ablauf des Budgets erzeugt wurden. Wenn während des Streamings ein Timeout auftritt, lesen Sie keine neuen Tokens mehr, fügen Sie eine Auslassung oder eine kurze Fortsetzungsaufforderung an und schließen Sie den Stream. Der Benutzer sieht eine sauber abgeschnittene Antwort statt eines leeren Fehlers. Das ist nur beim Streaming möglich – nicht gestreamte Aufrufe sind ganz oder gar nicht erfolgreich.

async def stream_with_budget(question: str, budget_ms: float):
    tracker = BudgetTracker(budget_ms)
    collected = []
    stream = await client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': question}],
        stream=True
    )
    async for chunk in stream:
        if tracker.remaining_ms() < 200:  # 200ms safety margin
            collected.append(' [response truncated]')
            break
        delta = chunk.choices[0].delta.content or ''
        collected.append(delta)
        yield delta
    # Ensure stream is closed even if budget exceeded
    await stream.close()

Semantischer Cache als Degradationsschicht

Ein semantischer Cache eignet sich hervorragend als Degradationsschicht, da er nahezu keine Latenz verursacht. Fragen Sie vor dem Aufruf des LLM Ihren semantischen Cache nach ähnlichen früheren Fragen ab. Wenn ein Cache-Treffer mit hoher Ähnlichkeit (über 0,92 Cosine Similarity) gefunden wird, geben Sie die zwischengespeicherte Antwort sofort zurück. Dadurch werden Antworten beschleunigt und steht sofort ein Fallback zur Verfügung, wenn das LLM langsam oder nicht verfügbar ist.

async def query_with_cache_fallback(question: str, budget_ms: float = 8000) -> str:
    # Try semantic cache first (fast)
    cached = await semantic_cache.lookup(question, threshold=0.92)
    if cached:
        return cached.response

    tracker = BudgetTracker(budget_ms)
    # Try full pipeline
    if tracker.remaining_ms() > 3000:
        try:
            return await smart_rag_query(question, tracker.remaining_ms())
        except Exception:
            pass  # fall through to static response

    # Last resort
    return 'I am experiencing high load right now. Please try again in a moment.'

Degradationsereignisse protokollieren

Protokollieren Sie jedes Herabstufen Ihrer Pipeline auf eine niedrigere Qualitätsstufe als strukturiertes Ereignis. Erfassen Sie die erreichte Degradationsstufe, das in jeder Phase verbleibende Budget und die endgültige Latenz. Die Analyse dieser Protokolle zeigt, wie oft die einzelnen Degradationsstufen ausgelöst werden. So können Sie Budgets abstimmen, Phasen erkennen, die regelmäßig ihr Budget überschreiten, und Investitionen in die Infrastruktur begründen.

import structlog

log = structlog.get_logger()

def log_degradation(level: int, stage: str, remaining_ms: float, total_ms: float):
    log.warning(
        'pipeline_degradation',
        degradation_level=level,
        triggered_at_stage=stage,
        remaining_budget_ms=round(remaining_ms),
        total_budget_ms=total_ms,
        budget_consumed_pct=round((total_ms - remaining_ms) / total_ms * 100)
    )

Benutzererwartungen mit UI-Signalen steuern

Wenn Sie eine degradierte Antwort bereitstellen, weisen Sie den Benutzer darauf hin, dass die Qualität geringer als üblich sein kann. Zeigen Sie in einer Chat-Oberfläche beispielsweise einen dezenten Hinweis wie „Schnellantwortmodus – einige Details sind möglicherweise eingeschränkt.“ Fügen Sie bei einer Extraction API ein Feld degraded: true in die JSON-Antwort ein, damit nachgelagerte Verbraucher degradierte Ergebnisse anders behandeln können. Transparenz bewahrt das Vertrauen der Benutzer, auch während eines Ausfalls.

from pydantic import BaseModel
from typing import Optional

class ChatResponse(BaseModel):
    content: str
    degraded: bool = False
    degradation_level: Optional[int] = None  # 0=full, 1=fast, 2=minimal, 3=cached
    latency_ms: int

# API response when degraded:
# {
#   'content': 'Here is a brief answer...',
#   'degraded': true,
#   'degradation_level': 2,
#   'latency_ms': 2800
# }

Budgetzuteilungen im Laufe der Zeit abstimmen

Die anfänglichen Budgetzuteilungen sind Schätzungen. Analysieren Sie nach einer Woche im Produktivbetrieb mithilfe Ihrer Tracing-Daten die Verteilung der für die einzelnen Phasen benötigten Zeit. Wenn das Retrieval dauerhaft 800 ms statt der eingeplanten 1500 ms benötigt, verteilen Sie diese Reserve auf die LLM-Phase um und ermöglichen Sie mehr Ausgabetokens oder ein größeres Kontextfenster. Die Abstimmung des Budgets ist eine fortlaufende Betriebsaufgabe und keine einmalige Konfiguration.

# Budget tuning based on production p95 data:
ACTUAL_P95 = {
    'retrieval': 780,    # vs budget 1500ms -> 720ms headroom
    'llm_call':  4200,   # vs budget 5500ms -> 1300ms headroom
    'formatting': 120,   # vs budget 500ms  -> 380ms headroom
}

TOTAL_HEADROOM = sum(
    BUDGET_STAGES[k] - ACTUAL_P95[k] for k in ACTUAL_P95
)
print(f'Total headroom: {TOTAL_HEADROOM}ms')
# Reallocate headroom to allow longer LLM responses

Kurze Überprüfung

Testen Sie Ihr Verständnis von Timeout-Budgets und kontrollierter Verschlechterung.

Zusammenfassung der Lektion

In dieser Lektion haben Sie gelernt: Timeout-Budgets verteilen die verfügbare Zeit auf die Pipeline-Phasen, damit Sie stets innerhalb einer akzeptablen Frist antworten, Degradationsstufen liefern bei erschöpfter Zeit schrittweise Antworten geringerer Qualität statt Fehlern, und das Protokollieren von Degradationsereignissen hilft Ihnen, dauerhafte Engpässe zu erkennen und zu beheben. Als Nächstes untersuchen wir das Muster LLM-as-judge zur automatisierten Qualitätsbewertung.

Kostenlos starten

Lerne Python mit einem KI-Tutor — kostenlos

Schreibe und führe echten Code in deinem Browser aus, bekomme sofortige Hilfe von einem 24/7 KI-Tutor und setze dein Lernen im Web oder in der App fort.

Kurse
30
Lektionen
120

Häufig gestellte Fragen

Ist die Lektion „Timeout-Budgets und sanfte Verschlechterung“ kostenlos?

Ja — der vollständige Text von „Timeout-Budgets und sanfte Verschlechterung“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des AI Engineering Academy-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der AI Engineering Academy-Kurs umfasst insgesamt 4 Lektionen.

Was lerne ich in „Timeout-Budgets und sanfte Verschlechterung“?

Legen Sie auf jeder Ebene Ihrer Pipeline strenge Timeout-Budgets fest und implementieren Sie eine sanfte Verschlechterung, die gecachte oder vereinfachte Antworten ausliefert, wenn das LLM sein Budge… Du übst AI Engineering Academy mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.

Brauche ich Erfahrung, um AI Engineering Academy zu starten?

Keine Vorkenntnisse erforderlich. AI Engineering Academy auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 4 von 4.

Wie lange dauert die Lektion „Timeout-Budgets und sanfte Verschlechterung“?

Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.

Kann ich in dieser AI Engineering Academy-Lektion Code schreiben und ausführen?

Ja. Jede AI Engineering Academy-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.

Alle Lektionen in diesem Kurs

  1. LLM-Latenz messen: TTFT und TPOT
  2. Lastverteilung und Strategien mit mehreren Schlüsseln
  3. Fallback-Anbieter und Circuit Breaker
  4. Timeout-Budgets und sanfte Verschlechterung
← Zurück zu AI Engineering Academy