AI Engineering Academy · Lección

Presupuestos de tiempo de espera y degradación controlada

Presupuestos de tiempo de espera y degradación controlada es una lección gratuita de AI Engineering Academy en CoddyKit. Esta es la lección 4 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Engineering Academy, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Engineering Academy incluye 4 lecciones en total.

Partes de esta lección aún no han sido traducidas y se muestran en inglés.

What Is a Timeout Budget?

A timeout budget is a maximum total time allocated for a request to complete across all stages of your pipeline. Instead of setting an arbitrary timeout on each individual API call, you define the end-to-end budget for the user-facing operation and distribute it across retrieval, LLM generation, and post-processing steps. This ensures you always respond within an acceptable time, even if some stages are slow.

Distributing Budget Across Pipeline Stages

A typical RAG chat pipeline has three stages: retrieval, LLM generation, and response formatting. Assign a time slice to each based on how long each normally takes and how much slack users tolerate. The remaining slack is your degradation buffer — if any stage uses its full allocation, you start cutting corners in subsequent stages to stay within the overall budget.

# 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

Tracking Budget Consumption

Use a BudgetTracker that records the start time and checks remaining budget at each stage transition. Before starting a stage, verify enough budget remains. This allows downstream stages to adapt — a retrieval step that takes 1200ms of its 1500ms budget leaves only 300ms of slack, which should trigger a simpler LLM prompt or skip the re-ranking step.

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

Graceful Degradation Definition

Graceful degradation means serving a lower-quality but still useful response when the full pipeline cannot complete within the budget, rather than returning an error. Examples include: returning a cached response, skipping re-ranking, truncating the context window, using a faster but less accurate model, or returning a pre-written fallback message. The goal is always to give the user something rather than nothing.

# 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

Implementing the Degradation Ladder

At each pipeline decision point, check the remaining budget and choose the appropriate quality level. The code below chooses retrieval depth and model based on remaining budget. This means under normal load users get the best quality, while during high-latency periods they still get a useful response rather than a timeout error.

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)

Setting Timeouts at the API Call Level

Always set explicit timeouts on every external API call. The OpenAI Python SDK accepts a timeout parameter in seconds. Set it to slightly less than the remaining budget so you have time to handle the exception and potentially degrade gracefully before the overall response deadline. Never rely on the SDK's default timeout — it may be too long for user-facing requests.

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.'

Returning Partial Streaming Responses

With streaming, you can return partial responses that were generated before the budget expired. When a timeout occurs mid-stream, stop reading new tokens, append an ellipsis or a brief continuation prompt, and close the stream. The user sees a response that cuts off cleanly rather than a blank error. This is only possible with streaming — non-streaming calls are all-or-nothing.

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()

Semantic Cache as Degradation Layer

A semantic cache is an excellent degradation layer because it has near-zero latency. Before calling the LLM, query your semantic cache for similar previous questions. If a cache hit with high similarity (above 0.92 cosine similarity) is found, return the cached answer immediately. This both speeds up responses and provides an instant fallback when the LLM is slow or unavailable.

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.'

Logging Degradation Events

Every time your pipeline degrades to a lower quality level, log it as a structured event. Include the degradation level reached, the budget remaining at each stage, and the final latency. Analyzing these logs tells you how often each degradation level is triggered, which helps you tune budgets, identify which stages are consistently over-budget, and justify infrastructure investments.

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)
    )

Setting User Expectations with UI Signals

When serving a degraded response, signal to the user that the quality may be lower than usual. For a chat interface, display a subtle indicator like 'Fast response mode — some details may be limited.' For an extraction API, include a degraded: true field in the JSON response so downstream consumers can handle degraded results differently. Transparency preserves user trust even during outages.

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
# }

Tuning Budget Allocations Over Time

Initial budget allocations are estimates. After running in production for a week, analyze the distribution of time spent in each stage using your tracing data. If retrieval consistently takes 800ms instead of the budgeted 1500ms, reallocate that slack to the LLM stage, allowing more output tokens or a larger context window. Budget tuning is an ongoing operation activity, not a one-time configuration.

# 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

Quick Check

Test your understanding of timeout budgets and graceful degradation.

Lesson Recap

In this lesson you learned: timeout budgets allocate time across pipeline stages so you always respond within an acceptable deadline, degradation ladders serve progressively lower-quality responses rather than errors when time is exhausted, and logging degradation events helps you identify and fix chronic bottlenecks. Next up we explore the LLM-as-judge pattern for automated quality evaluation.

Preguntas frecuentes

¿La lección «Presupuestos de tiempo de espera y degradación controlada» es gratis?

Sí — el texto completo de «Presupuestos de tiempo de espera y degradación controlada» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Engineering Academy, actualiza a CoddyKit PRO. El curso de AI Engineering Academy incluye 4 lecciones en total.

¿Qué aprenderé en «Presupuestos de tiempo de espera y degradación controlada»?

Establezca presupuestos de tiempo de espera estrictos en cada capa de su pipeline e implemente una degradación controlada que sirva respuestas almacenadas en caché o simplificadas cuando el LLM super… Practicas AI Engineering Academy con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.

¿Necesito experiencia previa para empezar AI Engineering Academy?

No se requiere experiencia previa. AI Engineering Academy en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 4 de 4.

¿Cuánto tiempo toma la lección «Presupuestos de tiempo de espera y degradación controlada»?

La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.

¿Puedo escribir y ejecutar código en esta lección de AI Engineering Academy?

Sí. Cada lección de AI Engineering Academy incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.

Todas las lecciones de este curso

Medición de la latencia de los LLM: TTFT y TPOT
Equilibrado de carga y estrategias con varias claves
Proveedores de respaldo y disyuntores
Presupuestos de tiempo de espera y degradación controlada

← Volver a AI Engineering Academy