Как вы спроектируете систему, которая может переключаться между разными LLM провайдерами без даунтайма?

Краткий тезис

Ключевая идея — слой абстракции над каждым LLM провайдером, единый интерфейс для генерации ответов, комбинированный с механизмами health check’ов, fallback’ов и динамической маршрутизации. Такая архитектура позволяет переключать провайдеров без остановки сервиса, избегает lock-in|vendor lock-in и повышает отказоустойчивость. Простейшая реализация занимает около двух часов кода, но экономит часы переписывания интеграций в будущем.

1. Базовые понятия

LLM провайдер — сервис, предоставляющий API для вызова языковых моделей (OpenAI, Anthropic, Google, локальные self-hosted модели через vLLM).

Downtime (даунтайм) — период, когда система не может обрабатывать запросы пользователей. В контексте переключения провайдеров мы стремимся к zero‑downtime: ни один запрос не должен быть потерян или задержан на время замены поставщика.

Vendor lock‑in — жёсткая привязка к одному провайдеру, когда смена требует переписывания значительных частей кодовой базы.

Абстракция (abstraction layer) — прослойка, скрывающая детали реализации конкретного провайдера за единым интерфейсом.

Health check — периодическая проверка доступности и корректности работы провайдера.

Fallback chain — последовательность провайдеров, которые вызываются, если предыдущий недоступен или вернул ошибку.

Feature flag — конфигурационный флаг, позволяющий включать/выключать функциональность без перезапуска приложения.

2. Зачем проектировать переключение?

Причины проектировать гибкую систему переключения:

• Отказоустойчивость – при сбое одного провайдера запросы автоматически направляются к другому.
• Снижение затрат – можно маршрутизировать дешёвые модели на простые запросы, дорогие — на сложные.
• Покрытие разных требований – у моделей разный максимальный контекст (например, Anthropic поддерживает 200K токенов, OpenAI – 128K).
• Юридические/комплаенс – данные некоторых клиентов нельзя отправлять за пределы определённой юрисдикции (используется self‑hosted провайдер).
• Эксперименты – A/B тестирование моделей без переписывания кода.

3. Абстрактный интерфейс: LLMProvider

Основа архитектуры — единый интерфейс, который должны реализовывать все адаптеры провайдеров.

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional, List

@dataclass
class LLMResponse:
    text: str
    model: str
    provider: str
    tokens_used: int
    latency_ms: float

class LLMProvider(ABC):
    @abstractmethod
    async def generate(
        self,
        prompt: str,
        system_prompt: Optional[str] = None,
        temperature: float = 0.7,
        max_tokens: int = 1024,
    ) -> LLMResponse:
        ...

Такой интерфейс позволяет в будущем легко добавлять новых провайдеров, не меняя остальной код.

4. Реализации для конкретных провайдеров

Каждый адаптер наследует LLMProvider и реализует специфику API.

import openai

class OpenAIProvider(LLMProvider):
    def __init__(self, api_key: str, model: str = "gpt-4o"):
        self.client = openai.AsyncOpenAI(api_key=api_key)
        self.model = model

    async def generate(self, prompt, system_prompt=None, temperature=0.7, max_tokens=1024):
        messages = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        messages.append({"role": "user", "content": prompt})
        start = time.time()
        response = await self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=temperature,
            max_tokens=max_tokens,
        )
        return LLMResponse(
            text=response.choices[0].message.content,
            model=self.model,
            provider="openai",
            tokens_used=response.usage.total_tokens,
            latency_ms=(time.time() - start) * 1000,
        )

Аналогично создаются AnthropicProvider, SelfHostedProvider (через vLLM / TGI). Главное — единый возвращаемый тип LLMResponse.

5. Health checks

Механизм проверки здоровья провайдера позволяет быстро определять недоступность и не отправлять к нему запросы.

from functools import wraps
import asyncio

class HealthCheckedProvider:
    def __init__(self, provider: LLMProvider, check_interval: int = 30):
        self.provider = provider
        self._healthy = True
        self._last_check = 0.0
        self._check_interval = check_interval

    async def is_healthy(self) -> bool:
        # Проверяем раз в check_interval секунд
        if time.time() - self._last_check < self._check_interval:
            return self._healthy
        try:
            # Лёгкий запрос (простой prompt)
            response = await self.provider.generate("ping", max_tokens=1)
            self._healthy = response.text is not None
        except Exception:
            self._healthy = False
        self._last_check = time.time()
        return self._healthy

    async def generate(self, *args, **kwargs):
        if not await self.is_healthy():
            raise ProviderUnavailableError(f"{self.provider.__class__.__name__} is unhealthy")
        return await self.provider.generate(*args, **kwargs)

Можно запускать health check в фоновом цикле для каждого провайдера.

6. Fallback chain (цепочка отката)

При недоступности основного провайдера запрос автоматически перенаправляется на следующий по списку.

class FallbackRouter:
    def __init__(self, providers: List[HealthCheckedProvider]):
        self.providers = providers

    async def generate_with_fallback(self, *args, **kwargs):
        last_exception = None
        for provider in self.providers:
            try:
                return await provider.generate(*args, **kwargs)
            except (ProviderUnavailableError, Exception) as e:
                last_exception = e
                continue
        raise AllProvidersFailedError("All providers failed") from last_exception

Конфигурация цепочки задаётся через список приоритетов: OpenAI, Anthropic, SelfHosted].

7. Dynamic routing (динамическая маршрутизация)

Fallback — это пассивное переключение (при ошибке). Dynamic routing — активный выбор провайдера на основе свойств запроса.

Правила маршрутизации

Реализация

class Router:
    def __init__(self, provider_pool: Dict[str, HealthCheckedProvider]):
        self.pool = provider_pool

    async def select_provider(self, context_len: int, data_sensitivity: str, speed_requirement: str):
        if data_sensitivity == "high":
            return self.pool["self_hosted"]
        if context_len > 100000:
            return self.pool["anthropic"]
        if speed_requirement == "high":
            return self.pool["groq"]
        return self.pool["openai"]

8. Конфигурация через feature flags

Feature flags позволяют менять поведение роутинга без перезапуска и деплоя.

Пример использования (через LaunchDarkly или простой файл конфигурации):

# config.yaml
providers:
  openai:
    enabled: true
    fallback_order: 1
    model: gpt-4o
  anthropic:
    enabled: true
    fallback_order: 2
    model: claude-3-5-sonnet-20241022
  self_hosted:
    enabled: true
    fallback_order: 3
    url: http://localhost:8000

fallback_enabled: true
dynamic_routing_enabled: true

Код при старте загружает конфиг и строит роутер. При изменении флага (например, отключение OpenAI из-за цены) сервис может обновить флаги в рантайме через watch и перестроить цепочку.

9. Обеспечение zero‑downtime

• Graceful degradation – если ни один провайдер не доступен, возвращаем fallback‑ответ (кэшированный или заглушку).
• Retry с экспоненциальной задержкой – временные сбои провайдера не приводят к немедленной смене.
• Connection pooling – не создаём нового клиента на каждый запрос, используем пул.
• Pre‑warming – для self‑hosted моделей держим «тёплый» экземпляр.
• State‑less роутер – роутер не хранит состояние запроса, легко масштабируется.

10. Пример кода – собираем всё вместе

# main.py
import asyncio
from typing import Dict

class LLMGateway:
    def __init__(self, config: dict):
        self.providers = self._init_providers(config)
        self.fallback_router = FallbackRouter(list(self.providers.values()))
        self.dynamic_router = Router(self.providers)

    def _init_providers(self, config) -> Dict[str, HealthCheckedProvider]:
        providers = {}
        if config["openai"]["enabled"]:
            providers["openai"] = HealthCheckedProvider(OpenAIProvider(config["openai"]["api_key"]))
        # ... аналогично для anthropic, self_hosted
        return providers

    async def generate(self, prompt, context_len=0, data_sensitivity="low", speed="normal"):
        if config["dynamic_routing_enabled"]:
            provider = await self.dynamic_router.select_provider(
                context_len=context_len,
                data_sensitivity=data_sensitivity,
                speed_requirement=speed
            )
            return await provider.generate(prompt)
        else:
            return await self.fallback_router.generate_with_fallback(prompt)

Плюсы полная абстракция, смена провайдера — это только обновление конфига.

Минусы необходимо поддерживать совместимость форматов ответов (не все модели возвращают одинаковый уровень детализации).

11. Альтернативные подходы

• LLM Gateway (OpenRouter, Portkey) – готовые облачные сервисы‑посредники. Минус: дополнительная точка отказа и затраты.
• Service‑mesh (Envoy + Lua) – можно перекладывать логику роутинга на sidecar‑прокси. Сложнее, но масштабируемее.
• Event‑driven архитектура – публикация запроса в очередь (Kafka) и подписка нескольких провайдеров, ответ от первого успешного. Но latency выше.

12. Сравнение подходов

Рекомендация: начинать со слоя абстракции в коде, при росте числа провайдеров и требований к масштабированию добавлять gateway или service‑mesh.

Пет‑проект для закрепления

Задача Разработать простой Multi‑LLM Gateway с поддержкой OpenAI, Anthropic и локальной модели через Ollama.

Инструменты

• Python 3.11+, FastAPI, pytest, Docker, Ollama (для локальной модели).

Шаги:

1. Создать абстрактный класс LLMProvider с методом generate.
2. Реализовать OpenAIProvider и AnthropicProvider (используя официальные SDK).
3. Реализовать OllamaProvider для локальной модели (например, llama3).
4. Написать HealthChecker, проверяющий доступность каждого провайдера раз в 30 секунд.
5. Реализовать FallbackRouter с последовательностью: OpenAI → Anthropic → Ollama.
6. Создать FastAPI endpoint POST /chat/completion, принимающий запрос и возвращающий ответ.
7. Добавить feature flag: openai_enabled, anthropic_enabled – при выключении флага маршрутизация меняется без перезапуска.
8. Покрыть тестами: unit‑тесты для каждого адаптера (mock API), интеграционные тесты для fallback с имитацией отказа.
9. Запустить в Docker, проверить переключение путём отключения интернета у одного провайдера.

Ожидаемый результат

• Сервис отвечает на запросы, даже если один из провайдеров недоступен.
• В логах видно, какой провайдер был выбран и почему.
• Изменение feature flags мгновенно влияет на роутинг (без restart).

Связь с другими вопросами

Навигация

• Предыдущий: 88
• Следующий: 90
• Индекс: 00. Индекс разборов