0Pricing
AI Engineering Academy · Lezione

Budget di timeout e degrado controllato

Imposti budget di timeout rigorosi a ogni livello della pipeline e implementi un degrado controllato che fornisca risposte memorizzate nella cache o semplificate quando l'LLM supera il proprio budget.

Budget di timeout e degrado controllato è una lezione AI Engineering Academy gratuita su CoddyKit. Questa è la lezione 4 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Engineering Academy, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Engineering Academy include 4 lezioni in totale.

Che cos'è un budget di timeout?

Un budget di timeout è il tempo massimo complessivo assegnato al completamento di una richiesta in tutte le fasi della pipeline. Invece di impostare un timeout arbitrario per ogni singola chiamata API, definisce il budget end-to-end per l'operazione visibile all'utente e lo distribuisce tra il recupero, la generazione dell'LLM e le fasi di post-elaborazione. In questo modo può rispondere sempre entro un tempo accettabile, anche quando alcune fasi sono lente.

Distribuzione del budget tra le fasi della pipeline

Una tipica pipeline di chat RAG comprende tre fasi: recupero, generazione dell'LLM e formattazione della risposta. Assegni a ciascuna una porzione di tempo in base alla sua durata abituale e al ritardo che gli utenti sono disposti a tollerare. Il margine rimanente costituisce il buffer di degrado: se una fase utilizza tutta la propria quota, nelle fasi successive inizia a semplificare alcune operazioni per rimanere entro il budget complessivo.

# 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

Monitoraggio del consumo del budget

Utilizzi un BudgetTracker che registri l'ora di inizio e verifichi il budget residuo a ogni passaggio tra le fasi. Prima di avviare una fase, verifichi che sia disponibile budget sufficiente. In questo modo le fasi successive possono adattarsi: un passaggio di recupero che impiega 1200 ms del proprio budget di 1500 ms lascia solo 300 ms di margine, e ciò dovrebbe attivare un prompt LLM più semplice oppure saltare la fase di riordinamento.

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

Definizione del degrado controllato

Il degrado controllato consiste nel fornire una risposta di qualità inferiore ma comunque utile quando la pipeline completa non può terminare entro il budget, invece di restituire un errore. Alcuni esempi sono: restituire una risposta memorizzata nella cache, saltare il riordinamento, troncare la finestra di contesto, utilizzare un modello più veloce ma meno accurato oppure restituire un messaggio di fallback predefinito. L'obiettivo è sempre fornire qualcosa all'utente, anziché nulla.

# 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

Implementazione della scala di degrado

A ogni punto decisionale della pipeline, controlli il budget residuo e scelga il livello di qualità appropriato. Il codice seguente sceglie la profondità del recupero e il modello in base al budget residuo. In questo modo, in condizioni di carico normali gli utenti ottengono la qualità migliore, mentre nei periodi di latenza elevata ricevono comunque una risposta utile invece di un errore di timeout.

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)

Impostazione dei timeout a livello di chiamata API

Imposti sempre timeout espliciti per ogni chiamata API esterna. OpenAI Python SDK accetta un parametro timeout espresso in secondi. Lo imposti a un valore leggermente inferiore al budget residuo, così avrà tempo di gestire l'eccezione e, se necessario, applicare un degrado controllato prima della scadenza complessiva della risposta. Non faccia mai affidamento sul timeout predefinito dell'SDK: potrebbe essere troppo lungo per le richieste rivolte agli utenti.

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

Restituzione di risposte in streaming parziali

Con lo streaming può restituire risposte parziali generate prima dell'esaurimento del budget. Quando si verifica un timeout durante lo streaming, interrompa la lettura di nuovi token, aggiunga un'ellissi o un breve prompt di continuazione e chiuda lo stream. L'utente visualizza una risposta che si interrompe in modo ordinato invece di un errore vuoto. Ciò è possibile solo con lo streaming: le chiamate non in streaming sono operazioni tutto o niente.

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

Cache semantica come livello di degrado

Una cache semantica è un eccellente livello di degrado perché offre una latenza quasi nulla. Prima di chiamare l'LLM, interroghi la cache semantica per cercare domande precedenti simili. Se viene trovata una corrispondenza nella cache con un'elevata similarità (superiore a 0.92 di similarità coseno), restituisca immediatamente la risposta memorizzata. In questo modo accelera le risposte e fornisce un fallback istantaneo quando l'LLM è lento o non disponibile.

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

Registrazione degli eventi di degrado

Ogni volta che la pipeline passa a un livello di qualità inferiore, lo registri come evento strutturato. Includa il livello di degrado raggiunto, il budget residuo in ogni fase e la latenza finale. L'analisi di questi log mostra con quale frequenza viene attivato ciascun livello di degrado, aiutandola a ottimizzare i budget, individuare le fasi che superano costantemente il budget e giustificare gli investimenti nell'infrastruttura.

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

Gestione delle aspettative degli utenti con segnali nell'interfaccia

Quando fornisce una risposta degradata, segnali all'utente che la qualità potrebbe essere inferiore al solito. In un'interfaccia di chat, mostri un indicatore discreto come 'Modalità risposta rapida — alcuni dettagli potrebbero essere limitati.' Per un'API di estrazione, includa un campo degraded: true nella risposta JSON, così i sistemi che la utilizzano possono gestire diversamente i risultati degradati. La trasparenza preserva la fiducia degli utenti anche durante le interruzioni del servizio.

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

Ottimizzazione delle allocazioni del budget nel tempo

Le allocazioni iniziali del budget sono stime. Dopo una settimana di utilizzo in produzione, analizzi la distribuzione del tempo impiegato in ogni fase utilizzando i dati di tracing. Se il recupero richiede costantemente 800 ms invece dei 1500 ms previsti dal budget, riallochi il margine alla fase LLM, consentendo più token di output o una finestra di contesto più ampia. L'ottimizzazione del budget è un'attività operativa continuativa, non una configurazione da eseguire una sola volta.

# 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

Verifica rapida

Verifichi la propria comprensione dei budget di timeout e del degrado controllato.

Riepilogo della lezione

In questa lezione ha appreso che i budget di timeout distribuiscono il tempo tra le fasi della pipeline, così da consentire di rispondere sempre entro una scadenza accettabile; le scale di degrado forniscono risposte progressivamente meno accurate anziché errori quando il tempo si esaurisce; e la registrazione degli eventi di degrado aiuta a individuare e risolvere i colli di bottiglia cronici. Nel prossimo argomento esploreremo il pattern LLM-as-judge per la valutazione automatizzata della qualità.

Domande Frequenti

La lezione «Budget di timeout e degrado controllato» è gratuita?

Sì — il testo completo di «Budget di timeout e degrado controllato» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Engineering Academy, passa a CoddyKit PRO. Il corso AI Engineering Academy include 4 lezioni in totale.

Cosa imparerò in «Budget di timeout e degrado controllato»?

Imposti budget di timeout rigorosi a ogni livello della pipeline e implementi un degrado controllato che fornisca risposte memorizzate nella cache o semplificate quando l'LLM supera il proprio budget. Eserciti AI Engineering Academy con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.

Ho bisogno di esperienza per iniziare AI Engineering Academy?

Non è richiesta alcuna esperienza precedente. AI Engineering Academy su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 4 di 4.

Quanto tempo richiede la lezione «Budget di timeout e degrado controllato»?

La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.

Posso scrivere ed eseguire codice in questa lezione AI Engineering Academy?

Sì. Ogni lezione AI Engineering Academy include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.

Tutte le lezioni di questo corso

  1. Misurare la latenza degli LLM: TTFT e TPOT
  2. Bilanciamento del carico e strategie multi-key
  3. Provider di fallback e circuit breaker
  4. Budget di timeout e degrado controllato
← Torna a AI Engineering Academy