AI Engineering Academy · Урок

Бюджеты времени ожидания и плавное ухудшение качества

Урок 4 из 413 шагов

«Бюджеты времени ожидания и плавное ухудшение качества» — бесплатный урок AI Engineering Academy на CoddyKit. Это урок 4 из 4. Ты можешь прочитать весь урок бесплатно ниже — а потом практиковать его прямо в браузере с встроенным редактором кода и ИИ-репетитором 24/7. Это часть пути обучения AI Engineering Academy, и твой прогресс синхронизируется между веб-версией и приложением CoddyKit. Курс AI Engineering Academy содержит 4 уроков всего.

Что такое бюджет тайм-аута

Бюджет тайм-аута — это максимальное общее время, выделенное на выполнение запроса на всех этапах Вашего конвейера. Вместо установки произвольного тайм-аута для каждого отдельного вызова API Вы задаёте сквозной бюджет для операции, с которой взаимодействует пользователь, и распределяете его между извлечением данных, генерацией LLM и последующей обработкой. Это гарантирует, что Вы всегда ответите за приемлемое время, даже если некоторые этапы выполняются медленно.

Распределение бюджета между этапами конвейера

Типичный конвейер чата с RAG состоит из трёх этапов: извлечение данных, генерация LLM и форматирование ответа. Назначьте каждому этапу временной интервал с учётом обычной длительности и того, сколько ожидания готовы терпеть пользователи. Оставшийся запас времени — это Ваш буфер снижения качества: если какой-либо этап использует весь выделенный ему ресурс, на последующих этапах нужно упростить обработку, чтобы уложиться в общий бюджет.

# 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

Отслеживание расходования бюджета

Используйте BudgetTracker, который записывает время начала и проверяет оставшийся бюджет при каждом переходе между этапами. Перед началом этапа убедитесь, что бюджета достаточно. Это позволяет последующим этапам адаптироваться: если извлечение данных занимает 1200ms из выделенных ему 1500ms, остаётся лишь 300ms запаса, и это должно вызвать переход к более простому запросу LLM или пропуск этапа повторного ранжирования.

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

Определение плавного снижения качества

Плавное снижение качества означает предоставление ответа более низкого качества, но всё ещё полезного, когда полный конвейер не может завершиться в пределах бюджета, вместо возврата ошибки. Например, можно вернуть кэшированный ответ, пропустить повторное ранжирование, сократить контекст, использовать более быструю, но менее точную модель или вернуть заранее написанное резервное сообщение. Цель всегда состоит в том, чтобы дать пользователю хоть что-то, а не оставить его без ответа.

# 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

Реализация последовательности снижения качества

В каждой точке принятия решения конвейером проверяйте оставшийся бюджет и выбирайте подходящий уровень качества. Приведённый ниже код выбирает глубину извлечения и модель в зависимости от оставшегося бюджета. Благодаря этому при обычной нагрузке пользователи получают наилучшее качество, а в периоды высокой задержки — полезный ответ, а не ошибку тайм-аута.

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)

Установка тайм-аутов на уровне вызовов API

Всегда устанавливайте явные тайм-ауты для каждого внешнего вызова API. Пакет OpenAI для Python принимает параметр timeout в секундах. Установите его немного меньше оставшегося бюджета, чтобы у Вас было время обработать исключение и при необходимости плавно снизить качество до наступления общего предельного времени ответа. Никогда не полагайтесь на тайм-аут по умолчанию в пакете: для запросов, с которыми взаимодействует пользователь, он может оказаться слишком большим.

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

Возврат частичных ответов в потоковом режиме

При потоковой передаче можно возвращать частичные ответы, сгенерированные до истечения бюджета. Если в середине потока происходит тайм-аут, прекратите чтение новых токенов, добавьте многоточие или краткое приглашение продолжить и закройте поток. Пользователь увидит ответ, который аккуратно обрывается, а не пустую ошибку. Это возможно только при потоковой передаче: вызовы без потоковой передачи выполняются полностью или не возвращают ничего.

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

Семантический кэш как уровень снижения качества

Семантический кэш — отличный уровень снижения качества, поскольку он практически не добавляет задержки. Перед вызовом LLM выполните в семантическом кэше поиск похожих предыдущих вопросов. Если найдено совпадение с высокой степенью сходства (выше 0.92 по косинусному сходству), немедленно верните кэшированный ответ. Это одновременно ускоряет ответы и обеспечивает мгновенный резервный вариант, когда LLM работает медленно или недоступна.

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

Регистрация событий снижения качества

Каждый раз, когда Ваш конвейер переходит на более низкий уровень качества, регистрируйте это как структурированное событие. Указывайте достигнутый уровень снижения качества, оставшийся бюджет на каждом этапе и итоговую задержку. Анализ этих записей показывает, как часто срабатывает каждый уровень снижения качества, помогает настроить бюджеты, определить этапы, которые постоянно превышают бюджет, и обосновать вложения в инфраструктуру.

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

Формирование ожиданий пользователей с помощью сигналов интерфейса

При предоставлении ответа со сниженным качеством сообщите пользователю, что его качество может быть ниже обычного. В интерфейсе чата можно показать ненавязчивый индикатор вроде «Быстрый режим ответа — некоторые сведения могут быть ограничены». Для API извлечения добавьте в ответ JSON поле degraded: true, чтобы последующие потребители могли обрабатывать результаты со сниженным качеством иначе. Прозрачность помогает сохранять доверие пользователей даже во время сбоев.

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

Настройка распределения бюджета со временем

Первоначальное распределение бюджета является оценочным. После недели работы в производственной среде проанализируйте распределение времени между этапами с помощью данных трассировки. Если извлечение данных стабильно занимает 800ms вместо заложенных в бюджете 1500ms, перераспределите этот запас на этап LLM, чтобы разрешить больше выходных токенов или увеличить размер контекста. Настройка бюджета — это постоянная эксплуатационная задача, а не разовая конфигурация.

# 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

Быстрая проверка

Проверьте, насколько хорошо Вы поняли бюджеты тайм-аутов и плавное снижение качества.

Ответь в интерактивном уроке

Итоги урока

В этом уроке Вы узнали, что бюджеты тайм-аутов распределяют время между этапами конвейера, чтобы Вы всегда отвечали до приемлемого предельного срока, последовательности снижения качества при исчерпании времени постепенно предоставляют ответы более низкого качества вместо ошибок, а регистрация событий снижения качества помогает выявлять и устранять хронические узкие места. Далее Вы познакомитесь с шаблоном LLM в роли оценщика для автоматической проверки качества.

Можно начать бесплатно

Изучай Python с ИИ-репетитором — бесплатно

Пиши и запускай код прямо в браузере, получай мгновенную помощь от ИИ-репетитора 24/7 и продолжи учиться на сайте или в приложении.

Курсы: 30
Уроки: 120

Начать учиться бесплатно Посмотреть полный курс

Часто задаваемые вопросы

Урок «Бюджеты времени ожидания и плавное ухудшение качества» бесплатный?

Да — полный текст урока «Бюджеты времени ожидания и плавное ухудшение качества» бесплатно доступен здесь в веб-версии. Чтобы практиковать его интерактивно (встроенный редактор кода и ИИ-репетитор 24/7) и разблокировать остальной курс AI Engineering Academy, подпишись на CoddyKit PRO. Курс AI Engineering Academy содержит 4 уроков всего.

Чему я научусь в уроке «Бюджеты времени ожидания и плавное ухудшение качества»?

Установите жёсткие бюджеты времени ожидания на каждом уровне конвейера и реализуйте плавное ухудшение качества, при котором при превышении LLM своего бюджета выдаются кэшированные или упрощённые отве… Ты практикуешь AI Engineering Academy с помощью реального кода, который запускаешь прямо в браузере, и ИИ-репетитор 24/7 отвечает на твои вопросы во время урока.

Нужен ли мне опыт, чтобы начать AI Engineering Academy?

Предыдущий опыт не требуется. AI Engineering Academy на CoddyKit структурирован для всех уровней — от новичков до продвинутых, поэтому ты можешь начать отсюда или с самого начала и учиться в своем темпе. Это урок 4 из 4.

Сколько времени занимает урок «Бюджеты времени ожидания и плавное ухудшение качества»?

Большинство уроков CoddyKit занимают около 5–10 минут. Каждый из них компактный и интерактивный, поэтому ты постоянно делаешь прогресс и продолжаешь с того же места в веб-версии и приложении.

Можно ли писать и запускать код в этом уроке AI Engineering Academy?

Да. Каждый урок AI Engineering Academy включает встроенный редактор кода, поэтому ты пишешь и запускаешь реальный код прямо в браузере и получаешь моментальную обратную связь от AI — локальная установка не требуется.

Все уроки этого курса

Измерение задержки LLM: TTFT и TPOT
Балансировка нагрузки и стратегии с несколькими ключами
Резервные поставщики и автоматические размыкатели
Бюджеты времени ожидания и плавное ухудшение качества

← Назад к AI Engineering Academy