Каскадный роутинг LLM: Как сделать отказоустойчивого AI-бота на Python и бесплатных API

Когда делаешь свой пет-проект в сфере ИИ, на старте логично минимизировать расходы. Создавая DayLog AI (умный Telegram-бот, который анализирует твое состояние и выдает персональные инсайты), я изначально собрал связку из бесплатных ключей: Mistral, Groq и Google Gemini.

На этапе проектирования стало очевидно, что рассчитывать только на один API нельзя. Как только пойдет реальный трафик, бесплатный Mistral упрется в лимит запросов. Знакомая ошибка `429 Too Many Requests`. Google тоже периодически подвисает на несколько секунд и отваливается по таймауту.

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

Использовать тяжелые фреймворки вроде LangChain или LiteLLM ради одного роутинга не было смысла. Слишком много абстракций и лишних зависимостей. Вместо этого я написал легковесную обертку на базе `aiohttp` и `tenacity`.

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

Чаще всего на бесплатных ключах мы сталкиваемся с лимитами. Первое побуждение — обернуть запрос в `try/except` и добавить `asyncio.sleep(3)`.

Это плохая практика. Допустим, API вернул 429 ошибку для десяти параллельных запросов. Все таски засыпают на три секунды. Просыпаются одновременно. И снова отправляют запросы единым батчем. Rate Limit пробивается мгновенно, и вы опять получаете отказ. Данные не отправлены.

Более грамотный подход — Exponential Backoff с Jitter-ом. Случайный разброс времени позволяет избежать синхронных повторных попыток. Для этого я использую библиотеку `tenacity`.

Обертка над вызовом API (на примере Mistral):

@retry( wait=wait_exponential_jitter(initial=2, max=10), stop=stop_after_attempt(3), retry=retry_if_exception_type(aiohttp.ClientResponseError), reraise=True, ) @observe(as_type="generation") async def _call_mistral(self, model: str, messages: list[dict], is_json: bool = False) -> dict: url = "https://api.mistral.ai/v1/chat/completions" headers = { "Authorization": f"Bearer {self.mistral_api_key}", "Content-Type": "application/json", } # ... сборка payload ... async with self.session.post(url, headers=headers, json=payload) as response: response.raise_for_status() return await response.json()

Если возникает ошибка сети, скрипт берет паузу. Изначально небольшую. С каждой неудачной попыткой время ожидания экспоненциально увеличивается, достигая максимума в 10 секунд. Добавляется случайный шум (jitter). Это рассинхронизирует повторные запросы от разных пользователей. Если сервер не ответил за три попытки, прокидываем ошибку дальше.

(Декоратор `@observe` взят из библиотеки Langfuse — применяю его для трейсинга).

Повторные запросы помогают при кратковременных проблемах. Но как быть, если дневной лимит исчерпан или провайдер полностью недоступен?

Нужен каскад. Логика функции `get_ai_response` достаточно прямолинейна:

Обращаюсь к основной модели `mistral-small-2506`.
Если недоступна — переключаюсь на резервную `ministral-14b-2512` внутри инфраструктуры Mistral.
Если весь Mistral не отвечает — бесшовно перевожу запрос на `gemini-3.1-flash-lite` от Google.

async def get_ai_response(user_text: str, lang: str = "ru") -> str | None: # ... сборка контекста ... try: # Попытка 1: Основная модель Mistral response = await ai_router._call_mistral("mistral-small-2506", messages) return response['choices'][0]['message']['content'] except Exception as e: logger.warning("MAIN_MISTRAL_MODEL недоступна ({}), фоллбэк → Ministral 14B", e) try: # Попытка 2: Резерв внутри Mistral response = await ai_router._call_mistral("ministral-14b-2512", messages) return response['choices'][0]['message']['content'] except Exception as e2: logger.warning("Mistral недоступен ({}), фоллбэк → Gemini", e2) try: # Попытка 3: Экстренный переход на инфраструктуру Google response = await ai_router._call_google("gemini-3.1-flash-lite", messages) # Обработка фильтров безопасности if "candidates" not in response and "promptFeedback" in response: raise SafetyBlockError("Safety Blocked") return response['candidates'][0]['content']['parts'][0]['text'] except SafetyBlockError: raise except Exception as e3: logger.error("Все AI-провайдеры недоступны: {}", e3) return None

Пользователь не замечает смену моделей. Ответ просто приходит на пару секунд позже.

Небольшой нюанс для Free Tier: чтобы экономить лимиты основного провайдера, извлечение JSON-метрик я полностью вынес на API Groq (используется модель Qwen 3.6). Так нагрузка балансируется между разными вендорами.

При работе с несколькими API всегда возникают специфические проблемы.

В коде выше есть проверка `promptFeedback` для Google. Gemini строго фильтрует запросы. Если пользователь пишет в логе, что "готов убить за чашку кофе", алгоритм может воспринять это как угрозу. В результате вместо нормального ответа или HTTP-ошибки возвращается пустой массив `candidates`. Без ручной проверки парсер упадет. Пришлось вынести это в отдельное исключение `SafetyBlockError`.

Вторая проблема — работа со структурированными ответами. Даже при включенном `response_format="json_object"` модель иногда возвращает некорректный JSON. Поэтому на бэкенде настроена жесткая валидация. Если нейросеть оценивает уровень стресса на "10" при максимуме "5", значение принудительно обрезается (clamp) до допустимого диапазона. Недостающие поля заполняются значениями по умолчанию.

В общем, мой самописный подход вряд ли потянет на идеальный энтерпрайз, но для соло-проектов это работает надежно (наверное, xd). Роутер уже спокойно пережил несколько падений API, и ни один лог не потерялся.

Вся эта схема сейчас крутится на проде моего бота. Суть простая: даже на бесплатных тарифах можно жить. Пару десятков строк с Exponential Backoff, Jitter и простеньким каскадом закрывают проблему с 429 ошибками и таймаутами.

Если кто-то решал похожую задачу — залетайте в комменты. Интересно посмотреть, кто как изворачивается с лимитами, и стоит ли всё-таки тащить в проект условный LiteLLM.

P.S. Бот, для которого писался этот роутер — DayLog AI. Можете потыкать, как он анализирует логи дней и выдает инсайты. А за новостями разработки и бэкенд-внутрянкой проекта заходите в канал бота. Буду рад видеть!

Каскадный роутинг LLM: Как сделать отказоустойчивого AI-бота на Python и бесплатных API

Умные повторные попытки (Защита от микросбоев)

Что здесь происходит:

Каскадный роутер (Фоллбэк при глобальных сбоях)

Нюансы интеграции: Цензура и валидация JSON