Отказоустойчивый роутер для AI API: как избежать простоев

Зачем нужен отказоустойчивый роутер для AI API?

Два месяца назад я смотрел на ошибку 503 от провайдера AI API, пока мои пользователи общались с приложением. Сессия оборвалась, логи были красными, телефон разрывался от гневных сообщений. Тогда я усвоил жёсткий урок: полагаться на один AI API — всё равно что строить дом на одной свае. Если вы используете AI в продакшене, рано или поздно провайдер подведёт. В этой статье я покажу, как построить отказоустойчивый роутер, который автоматически переключается между несколькими провайдерами, использует повторные попытки с экспоненциальной задержкой и circuit breaker, чтобы ваше приложение работало без сбоев.

Проблема: одиночная точка отказа

Моё приложение использовало GPT-4 для генерации ответов в реальном времени. Всё работало отлично, пока у OpenAI не случился частичный сбой. Запросы начали таймаутиться, затем падать. Мой наивный подход — попробовать один раз, показать ошибку — оставил пользователей в тупике. Я лихорадочно пытался переключиться на другого провайдера, но пришлось вручную обновлять код и переразворачивать приложение. Это заняло час. Час простоя.

Мне нужна была система, которая автоматически обрабатывает сбои нескольких AI провайдеров, с запасными вариантами, повторными попытками и, в идеале, балансировкой стоимости. Я не хотел терять качество, но и разоряться на дешёвых моделях тоже не хотел.

Первая попытка: простой fallback

Моя первая попытка была простой: попробовать провайдера A, если не получилось — попробовать B. Я захардкодил список и использовал try-except.

import openai
import anthropic

def generate_response(prompt):
    try:
        return openai.ChatCompletion.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}]
        )
    except:
        try:
            return anthropic.complete(
                prompt=prompt,
                model="claude-v1"
            )
        except:
            raise Exception("Both providers failed")

Это было лучше, чем ничего, но имело серьёзные недостатки:

• Нет повторных попыток для временных ошибок.
• Жёсткая очерёдность fallback — если A упал, B берёт всю нагрузку, но что если B тоже упадёт?
• Нет таймаутов — медленный провайдер может повесить всю систему.
• Нет метрик — я летел вслепую.

Что сработало: взвешенный роутер с circuit breaker

В итоге я написал небольшую Python-библиотеку, которая делает три вещи:

• Взвешенный round-robin — вы назначаете веса провайдерам (например, 3 для GPT-4, 1 для Claude, 1 для бесплатной модели). Запросы распределяются пропорционально, но если провайдер часто падает, его вес временно уменьшается.
• Экспоненциальная задержка с джиттером — повторные попытки с увеличивающейся задержкой, но со случайным разбросом, чтобы избежать эффекта лавины.
• Circuit breaker — если провайдер упал X раз за Y секунд, запросы к нему прекращаются на период охлаждения.

Вот суть подхода, урезанная до основ:

import asyncio
import random
import time
from typing import Dict, List, Callable, Awaitable

class AIProvider:
    def __init__(self, name: str, weight: int,
                 callable: Callable[[str], Awaitable[str]]):
        self.name = name
        self.weight = weight
        self.callable = callable
        self.failures = 0
        self.last_failure_time = 0
        self.circuit_open = False

class MultiProviderRouter:
    def __init__(self, providers: List[AIProvider],
                 circuit_breaker_threshold: int = 3,
                 circuit_breaker_timeout: int = 60):
        self.providers = providers
        self.circuit_breaker_threshold = circuit_breaker_threshold
        self.circuit_breaker_timeout = circuit_breaker_timeout

    def _select_provider(self):
        # Фильтруем провайдеров с открытой цепью
        available = [p for p in self.providers if not p.circuit_open]
        if not available:
            raise RuntimeError("All providers are in circuit breaker mode")
        # Взвешенный случайный выбор
        total_weight = sum(p.weight for p in available)
        r = random.uniform(0, total_weight)
        cumulative = 0
        for p in available:
            cumulative += p.weight
            if r <= cumulative:
                return p
        return available[-1]

    async def call(self, prompt: str, max_retries: int = 3):
        for attempt in range(max_retries):
            provider = self._select_provider()
            try:
                result = await provider.callable(prompt)
                # Успех: сбрасываем счётчик ошибок
                provider.failures = 0
                return result
            except Exception as e:
                provider.failures += 1
                provider.last_failure_time = time.time()
                if provider.failures >= self.circuit_breaker_threshold:
                    provider.circuit_open = True
                    # Планируем сброс цепи через таймаут
                    asyncio.create_task(self._reset_circuit(provider))
                # Экспоненциальная задержка с джиттером
                delay = (2 ** attempt) + random.random()
                await asyncio.sleep(delay)
        raise RuntimeError("All retries exhausted")

    async def _reset_circuit(self, provider):
        await asyncio.sleep(self.circuit_breaker_timeout)
        provider.circuit_open = False
        provider.failures = 0

Чтобы использовать, оберните реальные вызовы API в асинхронные функции:

async def call_openai(prompt: str) -> str:
    # ваша реальная реализация
    ...

async def call_anthropic(prompt: str) -> str:
    ...

# Можно добавить локальную модель или дешёвый fallback
router = MultiProviderRouter([
    AIProvider("openai", weight=3, callable=call_openai),
    AIProvider("anthropic", weight=2, callable=call_anthropic),
    # AIProvider("local", weight=1, callable=call_local_small_model),
])

result = await router.call("Объясни квантовую запутанность как пятилетнему")

Я также добавил метрики: логирую каждый успех/сбой в Prometheus counter и histogram. Это дало реальные данные для настройки весов.

Уроки и компромиссы

• Качество vs стоимость: установив высокий вес для GPT-4, я сохранил качество. Но когда он был медленным, роутер использовал более дешёвые модели, экономя деньги. Компромисс — иногда ответы были хуже во время сбоев.
• Настройка circuit breaker: слишком чувствительный (низкий порог) — и вы будете переключаться слишком часто, теряя контекст. Слишком lenient — будете долго биться в мёртвый провайдер. Я остановился на 3 ошибках за 60 секунд.
• Идемпотентность: роутер не гарантирует exactly-once доставку. Если запрос таймаутнулся, но на самом деле выполнился, ваш downstream может получить дубликат. Обрабатывайте это на своей стороне.
• Отладка усложняется: когда ответ выглядит странно, нужно проверить, какой провайдер его отдал. Я добавил заголовок X-Provider в свои ответы.

Что я сделал бы иначе

В следующий раз я начал бы с простого fallback и метрик, прежде чем строить полноценный роутер. Circuit breaker и веса появились после анализа реальных паттернов сбоев. Также рассмотрел бы использование хостингового сервиса, который делает это за вас — техника та же, строите вы или покупаете.

Сейчас мой роутер обрабатывает более 10 000 запросов в день без ручного вмешательства. Когда один сбой длился 6 часов, пользователи почти ничего не заметили — роутер бесшумно переключился на Anthropic, затем на локальную модель.

Практический вывод

Отказоустойчивость — это не устранение сбоев, а умение переживать их без потерь. Умная стратегия fallback дёшева в реализации и окупается при первом же падении основного API. Не ждите, пока ваш телефон разорвётся от гневных сообщений пользователей.

Прямо сейчас: если ваше приложение использует всего один AI API, добавьте хотя бы простой fallback с повторными попытками. Начните с малого, затем внедрите circuit breaker и веса. Ваши пользователи скажут вам спасибо.