Graceful Degradation для AI-агентов: архитектура отказоустойчивости от прова до очереди задач

Сломанный агент = сломанный бизнес. И это не драма

Ваш AI-агент успешно работает две недели. Команда счастлива, клиенты довольны. Потом, в 11:47 понедельника, прилетает 500-ка от OpenAI. Rate limit. Через две минуты облако Anthropic сообщает об инциденте. Ещё через пять минут ваш прекрасный агент, который обрабатывает заказы, анализирует поддержку и генерирует контент, превращается в тыкву. Все процессы встают.

Сбой внешнего API - не форс-мажор. Это статистическая неизбежность. Graceful Degradation - это не про "если сломается". Это про "когда сломается". Это архитектурный подход, который гарантирует: система не умрёт, а тихо и изящно понизит качество работы, сохранив ключевую функциональность. Давайте строить не просто агента, а агента, который выживает.

Пять кругов ада: уровни деградации от 0 до 4

Представьте шкалу от полного провала (0) до идеальной работы (4). Наша цель - никогда не падать до нуля. Каждый уровень - это чёткий план действий и набор доступных инструментов.

1 Уровень 0: То, что происходит, когда архитектора не было

Классический сценарий: агент тупо обёрнут в try-except, который ловит ошибку API и возвращает "Something went wrong". Бизнес-процессы останавливаются, деньги не зарабатываются, тимлид нервно курит в корпоративном Slack. Это не архитектура, а её отсутствие.

# КАК НЕ НАДО ДЕЛАТЬ
import openai

def ask_agent(question: str) -> str:
    try:
        response = openai.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": question}]
        )
        return response.choices[0].message.content
    except Exception as e:  # Ловим ВСЁ
        return "Извините, произошла ошибка."  # Бесполезно для пользователя и системы

2 Уровень 1: Включаем запасной аэродром

Первый шаг к устойчивости - цепочка провайдеров. Основной (например, OpenAI GPT-4.5), резервный (Anthropic Claude 3.7), локальный (Qwen2.5 32B на своем железе или через GigaChat API). Ключевой принцип: переключение должно быть автоматическим и основанным на чётких критериях (таймауты, коды ошибок 429, 503).

# Базовая реализация цепочки вызовов
from typing import List, Optional
import httpx
from enum import Enum

class Provider(Enum):
    OPENAI = "openai"
    ANTHROPIC = "anthropic"
    GIGACHAT = "gigachat"
    LOCAL = "local_qwen"

class LLMOrchestrator:
    def __init__(self, providers: List[Provider]):
        self.providers = providers
        self.current_provider_index = 0

    async def complete(self, prompt: str, max_retries: int = 3) -> Optional[str]:
        for attempt in range(max_retries):
            provider = self.providers[self.current_provider_index]
            try:
                # Устанавливаем разумный таймаут
                async with httpx.AsyncClient(timeout=30.0) as client:
                    response = await self._call_provider(client, provider, prompt)
                    if response:
                        return response
            except (httpx.TimeoutException, httpx.HTTPStatusError) as e:
                # Логируем ошибку и переключаемся
                self._log_failure(provider, str(e))
                self.current_provider_index = (self.current_provider_index + 1) % len(self.providers)
                continue
        return None  # Все провайдеры недоступны

3 Уровень 2: Кэш как спасательный круг

Когда все внешние API лежат, кэш становится источником истины. Но не любой кэш. Речь о семантическом кэшировании: сохраняем не точный запрос, а его смысл. Запросы "Как сбросить пароль?" и "Не могу войти в аккаунт" должны возвращать один и тот же заранее сгенерированный ответ. Используйте векторные базы вроде Qdrant или Weaviate для этого.

Этот уровень напрямую связан с архитектурой памяти агента, о которой мы подробно писали в статье "Как спроектировать современного AI-агента". Stateful Memory - это не только для контекста, но и для отказоустойчивости.

4 Уровень 3 и 4: Мониторинг и очередь задач

На уровнях 3 и 4 система уже работает, но мы должны контролировать её состояние. Мониторинг задержек, частоты ошибок, затрат на токены. Если задержки растут (уровень 3) - мы можем начать троттлинг неключевых запросов. Идеальный уровень 4 - это когда мы используем асинхронную очередь (RabbitMQ, Kafka) для обработки запросов. Запросы не теряются, а ждут своей очереди, даже если внешний сервис временно недоступен.

Собираем пазл: продакшен-архитектура за час

Теория - это прекрасно, но давайте посмотрим на схему, которая работает прямо сейчас. Основа - это Orchestrator, который принимает решение на основе здоровья провайдеров.

# Продакшен-готовый Orchestrator (упрощённо)
import asyncio
from dataclasses import dataclass
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor
import redis

@dataclass
class ProviderHealth:
    name: str
    is_healthy: bool
    last_error: Optional[datetime]
    avg_latency: float

class ProductionOrchestrator:
    def __init__(self):
        self.providers = self._init_providers()
        self.cache = redis.Redis(host='localhost', port=6379, decode_responses=True)
        self.health_check_interval = 60  # секунды
        self.fallback_to_cache = True

    async def process_request(self, user_input: str, user_id: str) -> dict:
        """Основной метод обработки запроса с полным циклом деградации."""
        # Шаг 1: Проверка семантического кэша
        if self.fallback_to_cache:
            cached = await self._check_semantic_cache(user_input)
            if cached:
                return {"source": "cache", "response": cached}

        # Шаг 2: Выбор живого провайдера
        healthy_providers = [p for p in self.providers if p.health.is_healthy]
        if not healthy_providers:
            # Все провайдеры мертвы, уровень 0 или 2
            return await self._execute_degradation_plan(user_input, user_id)

        # Шаг 3: Попытка выполнения
        for provider in healthy_providers:
            try:
                response = await provider.generate(user_input)
                # Кэшируем успешный ответ для будущего использования
                await self._store_in_semantic_cache(user_input, response)
                return {"source": provider.name, "response": response}
            except Exception as e:
                self._update_provider_health(provider.name, False)
                continue

        # Шаг 4: Если все попытки провалились
        return await self._execute_degradation_plan(user_input, user_id)

    async def _execute_degradation_plan(self, user_input: str, user_id: str) -> dict:
        """План деградации: от очереди до статичного ответа."""
        # 1. Пытаемся поставить в очередь для асинхронной обработки
        queue_success = await self._enqueue_request(user_input, user_id)
        if queue_success:
            return {"source": "queue", "message": "Ваш запрос принят в обработку."}

        # 2. Возвращаем заготовленный шаблонный ответ
        static_response = self._get_static_fallback(user_input)
        return {"source": "static", "response": static_response}

Три ошибки, которые сведут ваши усилия на нет

1. Игнорирование таймаутов на стороне клиента. Вы установили таймаут 10 секунд для вызова API, но не для TCP-соединения. В результате запрос может висеть минутами. Всегда настраивайте многоуровневые таймауты: connect, read, write, pool.
2. Слепое переключение на резервного провайдера. Вы переключились на Claude, потому что OpenAI вернул 429. Но что, если у Anthropic тоже сейчас проблемы с нагрузкой? Нужен health check, который проверяет не просто "доступен/не доступен", а реальную работоспособность (латентность, качество ответов).
3. Отсутствие метрик для принятия решений. Вы не знаете, как часто срабатывает деградация, сколько денег теряете на резервных провайдерах, какая реальная доступность. Без метрик вы летите вслепую. Встраивайте сбор данных с самого начала, как в Kaggle по AI-агентам от Google.

Что дальше? Эволюция отказоустойчивости

Graceful Degradation - это не статичная настройка, а живой процесс. Следующий шаг - predictive degradation. Система, которая, анализируя метрики задержек от провайдера и мониторинг его статус-страницы, предсказывает сбой за 5-10 минут до его наступления и плавно переключает часть трафика на резервные мощности.

Ещё дальше - адаптивная деградация, где агент сам меняет свою работу в зависимости от доступных ресурсов. Нет доступа к поиску Tavily? Агент использует заранее загруженные датасеты. Упала модель для анализа изображений? Агент просит пользователя описать картинку текстом. Это уровень, на котором система не просто выживает, а демонстрирует интеллект в условиях ограничений.

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

Подписаться на канал