Budżety czasu oczekiwania i kontrolowana degradacja

Czym jest budżet czasu?

Rozdzielanie budżetu między etapy potoku

# 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

Śledzenie wykorzystania budżetu

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

Definicja kontrolowanego obniżania jakości

# 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

Implementowanie drabiny obniżania jakości

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)

Ustawianie limitów czasu na poziomie wywołania API

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

Zwracanie częściowych odpowiedzi strumieniowych

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

Pamięć semantyczna jako warstwa obniżania jakości

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

Rejestrowanie zdarzeń obniżenia jakości

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

Kształtowanie oczekiwań użytkownika za pomocą sygnałów interfejsu

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

Dostrajanie przydziałów budżetu w czasie

# 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

Szybkie sprawdzenie

Podsumowanie lekcji