AI Engineering Academy · Leçon

Budgets de délai et dégradation progressive

Définissez des budgets de délai stricts à chaque couche de votre pipeline et implémentez une dégradation progressive qui fournit des réponses mises en cache ou simplifiées lorsque le LLM dépasse son budget.

Leçon 4 sur 413 étapes

Budgets de délai et dégradation progressive est une leçon AI Engineering Academy gratuite sur CoddyKit. Ceci est la leçon 4 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage AI Engineering Academy, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours AI Engineering Academy comprend 4 leçons au total.

Qu’est-ce qu’un budget de délai d’attente ?

Un budget de délai d’attente est la durée totale maximale allouée à l’exécution d’une demande à travers toutes les étapes de votre chaîne de traitement. Au lieu de définir un délai d’attente arbitraire pour chaque appel d’API, vous définissez le budget de bout en bout de l’opération destinée à l’utilisateur, puis vous le répartissez entre les étapes de récupération, de génération par le LLM et de post-traitement. Vous vous assurez ainsi de toujours répondre dans un délai acceptable, même si certaines étapes sont lentes.

Répartir le budget entre les étapes de la chaîne de traitement

Une chaîne de traitement de discussion RAG classique comporte trois étapes : la récupération, la génération par le LLM et la mise en forme de la réponse. Attribuez une durée à chacune selon sa durée habituelle et le niveau de latence que les utilisateurs sont prêts à accepter. La marge restante constitue votre réserve de dégradation : si une étape utilise toute la durée qui lui est attribuée, vous commencez à simplifier les étapes suivantes pour rester dans le budget global.

# 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

Suivre la consommation du budget

Utilisez un BudgetTracker qui enregistre l’heure de début et vérifie le budget restant à chaque transition entre les étapes. Avant de commencer une étape, vérifiez qu’il reste suffisamment de budget. Les étapes suivantes peuvent ainsi s’adapter : une étape de récupération qui consomme 1200ms sur son budget de 1500ms ne laisse que 300ms de marge, ce qui devrait déclencher une invite LLM plus simple ou l’omission de l’étape de reclassement.

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

Définition de la dégradation progressive

La dégradation progressive consiste à fournir une réponse de qualité moindre, mais toujours utile, lorsque la chaîne de traitement complète ne peut pas s’achever dans le budget imparti, plutôt que de renvoyer une erreur. Exemples : renvoyer une réponse mise en cache, ignorer le reclassement, tronquer la fenêtre de contexte, utiliser un modèle plus rapide mais moins précis ou renvoyer un message de repli pré-écrit. L’objectif est toujours de donner quelque chose à l’utilisateur plutôt que rien.

# 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

Mettre en œuvre l’échelle de dégradation

À chaque point de décision de la chaîne de traitement, vérifiez le budget restant et choisissez le niveau de qualité approprié. Le code ci-dessous choisit la profondeur de récupération et le modèle en fonction du budget restant. Ainsi, dans des conditions normales, les utilisateurs bénéficient de la meilleure qualité, tandis que pendant les périodes de forte latence, ils obtiennent tout de même une réponse utile plutôt qu’une erreur de délai d’attente.

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)

Définir les délais d’attente au niveau des appels d’API

Définissez toujours des délais d’attente explicites pour chaque appel d’API externe. Le SDK Python OpenAI accepte un paramètre timeout exprimé en secondes. Définissez-le à une valeur légèrement inférieure au budget restant afin de disposer du temps nécessaire pour gérer l’exception et éventuellement appliquer une dégradation progressive avant l’échéance globale de la réponse. Ne vous fiez jamais au délai d’attente par défaut du SDK : il peut être trop long pour les demandes destinées aux utilisateurs.

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

Renvoyer des réponses partielles en continu

Avec la diffusion en continu, vous pouvez renvoyer des réponses partielles générées avant l’expiration du budget. Lorsqu’un délai d’attente survient au milieu de la diffusion, cessez de lire de nouveaux jetons, ajoutez des points de suspension ou une brève invite de continuation, puis fermez le flux. L’utilisateur voit une réponse qui s’interrompt proprement plutôt qu’une erreur vide. Cela n’est possible qu’avec la diffusion en continu : les appels sans diffusion sont entièrement réussis ou entièrement échoués.

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

Le cache sémantique comme couche de dégradation

Un cache sémantique constitue une excellente couche de dégradation, car sa latence est presque nulle. Avant d’appeler le LLM, interrogez votre cache sémantique pour rechercher des questions précédentes similaires. Si une correspondance du cache présentant une forte similarité (supérieure à 0.92 selon la similarité cosinus) est trouvée, renvoyez immédiatement la réponse mise en cache. Cela accélère les réponses tout en fournissant un repli instantané lorsque le LLM est lent ou indisponible.

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

Consigner les événements de dégradation

Chaque fois que votre chaîne de traitement passe à un niveau de qualité inférieur, consignez-le comme un événement structuré. Incluez le niveau de dégradation atteint, le budget restant à chaque étape et la latence finale. L’analyse de ces journaux vous indique à quelle fréquence chaque niveau de dégradation est déclenché. Elle vous aide ainsi à ajuster les budgets, à repérer les étapes systématiquement en dépassement de budget et à justifier les investissements dans l’infrastructure.

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

Définir les attentes des utilisateurs avec des indicateurs d’interface

Lorsque vous fournissez une réponse dégradée, indiquez à l’utilisateur que sa qualité peut être inférieure à la normale. Dans une interface de discussion, affichez un indicateur discret tel que « Mode de réponse rapide — certains détails peuvent être limités. » Pour une API d’extraction, ajoutez un champ degraded: true dans la réponse JSON afin que les consommateurs en aval puissent traiter différemment les résultats dégradés. La transparence préserve la confiance des utilisateurs, même pendant les interruptions de service.

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

Ajuster la répartition du budget au fil du temps

Les répartitions initiales du budget sont des estimations. Après une semaine d’utilisation en production, analysez la répartition du temps consacré à chaque étape à l’aide de vos données de traçage. Si la récupération prend systématiquement 800ms au lieu des 1500ms prévus, réattribuez cette marge à l’étape LLM afin d’autoriser davantage de jetons de sortie ou une fenêtre de contexte plus large. L’ajustement du budget est une activité d’exploitation continue, et non une configuration ponctuelle.

# 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

Vérification rapide

Vérifiez votre compréhension des budgets de délai d’attente et de la dégradation progressive.

Récapitulatif de la leçon

Dans cette leçon, vous avez appris que les budgets de délai d’attente répartissent le temps entre les étapes de la chaîne de traitement afin que vous répondiez toujours dans un délai acceptable, que les échelles de dégradation fournissent des réponses progressivement moins qualitatives plutôt que des erreurs lorsque le temps est épuisé, et que la consignation des événements de dégradation vous aide à repérer et à corriger les goulots d’étranglement persistants. Nous allons maintenant étudier le modèle du LLM en tant qu’évaluateur pour l’évaluation automatisée de la qualité.

Gratuit pour commencer

Apprends Python avec un tuteur IA — gratuit

Écris et exécute du vrai code dans ton navigateur, obtiens de l'aide instantanée d'un tuteur IA disponible 24h/24, et reprends là où tu t'es arrêté sur le web ou dans l'app.

Cours
30
Leçons
120

Questions Fréquemment Posées

La leçon « Budgets de délai et dégradation progressive » est-elle gratuite ?

Oui — le texte complet de « Budgets de délai et dégradation progressive » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours AI Engineering Academy, passe à CoddyKit PRO. Le cours AI Engineering Academy comprend 4 leçons au total.

Qu'est-ce que j'apprendrai dans « Budgets de délai et dégradation progressive » ?

Définissez des budgets de délai stricts à chaque couche de votre pipeline et implémentez une dégradation progressive qui fournit des réponses mises en cache ou simplifiées lorsque le LLM dépasse son… Tu pratiques AI Engineering Academy avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.

Dois-je avoir de l'expérience pour commencer AI Engineering Academy ?

Aucune expérience préalable n'est requise. AI Engineering Academy sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 4 sur 4.

Combien de temps prend la leçon « Budgets de délai et dégradation progressive » ?

La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.

Peux-tu écrire et exécuter du code dans cette leçon AI Engineering Academy ?

Oui. Chaque leçon AI Engineering Academy inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.

Toutes les leçons de ce cours

  1. Mesurer la latence des LLM : TTFT et TPOT
  2. Équilibrage de charge et stratégies multi-clés
  3. Fournisseurs de secours et disjoncteurs
  4. Budgets de délai et dégradation progressive
← Retour à AI Engineering Academy