Укрощаем зоопарк API: универсальный Python-клиент для GigaChat, YandexGPT и локальных моделей (v0.5.0)

👋

Если вы пробовали внедрять российские LLM в свои проекты, то наверняка сталкивались с “зоопарком” API. У GigaChat — OAuth2 и свои эндпоинты, у YandexGPT — IAM-токены и gRPC/REST, у локальных моделей через Ollama — третий формат.

В какой-то момент мне надоело писать бесконечные if provider == 'gigachat': ... elif provider == 'yandex': ..., и я решил создать универсальный слой абстракции.

Так появился Multi-LLM Orchestrator — open-source библиотека, которая позволяет работать с разными LLM через единый интерфейс, поддерживает умный роутинг и автоматический fallback (переключение на другую модель при ошибке ^[1]).

Сегодня расскажу, как я её проектировал, с какими сложностями столкнулся при реализации потоковой генерации (Streaming), и как за неделю довел проект до версии v0.5.0 с поддержкой LangChain и 92% покрытия тестами.

Проблема: “Каждый пишет свой велосипед”

Представьте задачу: нужно сделать чат-бота, который использует GigaChat как основную модель, но если Сбер “лежит” или выдает ошибку 500 — незаметно переключается на YandexGPT.

Без абстракции код выглядит примерно так:

async def generate_response(prompt):
    try:
        # Пытаемся GigaChat
        token = await get_gigachat_token()  # OAuth2 логика
        response = await requests.post(..., headers={"Authorization": f"Bearer {token}"})
        return response.json()['choices'][0]['message']['content']
    except Exception:
        # Пытаемся YandexGPT
        response = await requests.post(..., headers={"Authorization": f"Bearer {iam_token}"})
        return response.json()['result']['alternatives'][0]['message']['text']

А теперь добавьте сюда:

• Обработку Rate Limits (429)

• Разные форматы messages

• Разные названия параметров (max_tokens vs maxTokens)

• Streaming (потоковую передачу токенов), где у каждого API свой формат чанков

• Логирование и метрики

Решение: Архитектура Оркестратора

Роутер автоматически переключается между провайдерами при сбоях:

• GigaChat (облако Сбер) — основной провайдер

• YandexGPT (облако Яндекс) — резервный

• Ollama (self-hosted) — локальная альтернатива

Все провайдеры поддерживают потоковую генерацию. Реальные метрики производительности смотрите в разделе “Боевое тестирование” ниже.

Единый интерфейс

Теперь код пользователя выглядит чисто и декларативно:

from orchestrator import Router
from orchestrator.providers import GigaChatProvider, YandexGPTProvider, ProviderConfig

# Конфигурируем роутер
router = Router(strategy="round-robin")

# Добавляем GigaChat
router.add_provider(GigaChatProvider(
    ProviderConfig(name="sber", api_key="...", scope="GIGACHAT_API_PERS")
))

# Добавляем YandexGPT
router.add_provider(YandexGPTProvider(
    ProviderConfig(name="yandex", api_key="...", folder_id="...", model="yandexgpt/latest")
))

# Используем! Если один упадет — роутер сам переключится на следующий
response = await router.route("Привет! Как дела?")

Под капотом: Интеграция и сложности

1. GigaChat и “протухающие” токены

Сбер использует OAuth2 Client Credentials flow. Главная сложность — токен живет 30 минут.

В GigaChatProvider я реализовал автоматическое управление токеном: он хранится в памяти ^[2] и проверяется перед каждым запросом. Если API вернул 401 (токен отозван раньше времени), провайдер сам обновит его и повторит запрос.

2. YandexGPT и заголовки

У Яндекса другая специфика: нужен не только IAM-токен, но и folder_id (идентификатор каталога в облаке), который нужно передавать в заголовке x-folder-id. Пришлось расширить конфигурацию, сохранив обратную совместимость.

class ProviderConfig(BaseModel):
    name: str
    api_key: str | None = None
    folder_id: str | None = None  # Специфично для Yandex
    # ...

3. Самое сложное: Streaming (Потоковая генерация) 🌊

К версии v0.5.0 я добавил поддержку Streaming. Это когда ответ приходит не целиком, а по кусочкам (токенам), как в ChatGPT.

Это оказалось сложнее, чем обычный запрос:

• Разные форматы: GigaChat использует Server-Sent Events (SSE) с префиксом data:, локальная Ollama отдает JSON-объекты, а мок-провайдер просто эмулирует задержки.

• Обработка ошибок в потоке: Что делать, если соединение разорвалось на середине фразы?

  • Мое решение: Если ошибка произошла до первого полученного чанка — роутер делает fallback на другого провайдера. Если текст уже начал печататься — fallback не делается, чтобы не смешивать ответы разных моделей.

Пример использования стриминга:

# Асинхронный генератор с автоматическим fallback
async for chunk in router.route_stream("Напиши короткое стихотворение про Python"):
    print(chunk, end="", flush=True)

Эволюция проекта

Проект быстро развивался на основе моих потребностей ^[3] и фидбека коллег. Что добавилось к текущей версии:

1. Ollama Provider (v0.3.x): Теперь можно бесплатно гонять локальные модели (Llama 3, Mistral) и использовать облачные модели только как fallback, если локальная перегружена.

2. LangChain Integration (v0.4.0): Я написал wrapper MultiLLMOrchestrator, который совместим с BaseLLM. Это позволяет вставить оркестратор в любые цепочки (Chains) и агенты LangChain одной строчкой:

# pip install multi-llm-orchestrator[langchain]
from orchestrator.langchain import MultiLLMOrchestrator

llm = MultiLLMOrchestrator(router=router)
# Теперь 'llm' можно использовать внутри LangChain chains!

3. Streaming Support (v0.5.0): Полная поддержка потоковой генерации для GigaChat с SSE-парсингом и интеграция с LangChain streaming.

Качество кода: Mypy, Ruff и тесты

Я верю, что Open Source должен быть качественным. Поэтому в CI/CD сразу зашил жесткие требования:

• Mypy в режиме --strict. Полная типизация спасла от кучи багов.

• Ruff как линтер.

• Tests Coverage ≈ 92%. Написано 133 теста, включая тесты на SSE-стриминг и моки для проверки rate limits.

Результат на сегодня:

• ✅ 133 теста (включая 18 для LangChain integration)

• ✅ 92% покрытия кода

• ✅ Полностью асинхронная реализация (asyncio/httpx)

• ✅ 4 провайдера: GigaChat, YandexGPT, Ollama, Mock

• ✅ Совместимость с LangChain (streaming included)

Боевое тестирование (Real-world testing)

Тесты на моках — это база, но реальная жизнь интереснее. Перед релизом я провел серию «боевых» тестов с реальными ключами от Сбера и Яндекса.

1. Проверка роутинга (Round-Robin)

Вот лог работы роутера, который балансирует нагрузку между двумя провайдерами. Запросы уходят по очереди:

Как видите, оркестратор корректно чередует запросы, обеспечивая отказоустойчивость.

2. Streaming в действии (New in v0.5.0!)

Самое интересное в новой версии — это потоковая генерация. Теперь ответ не нужно ждать целиком, он приходит по токенам, как в ChatGPT.

Вот как это выглядит при вызове GigaChat:

from orchestrator import Router
from orchestrator.providers import GigaChatProvider, ProviderConfig

router = Router(strategy="round-robin")
config = ProviderConfig(
    name="gigachat",
    api_key="your_key_here",
    model="GigaChat",
    verify_ssl=False  # Для российских сертификатов Сбера
)
router.add_provider(GigaChatProvider(config))

# Streaming: текст появляется постепенно
async for chunk in router.route_stream("Напиши хокку про Python"):
    print(chunk, end="", flush=True)

Результат в консоли:

Текст печатается в реальном времени, слово за словом. Скорость генерации впечатляет.

3. Метрики производительности

Я написал специальный тест (examples/real_tests/test_streaming_real.py), который замеряет ключевые метрики. Вот что он показал при вызове реального GigaChat API:

• TTFT (Time to First Token): 1.4 секунды от отправки запроса до первого слова. Это включает OAuth2-авторизацию и сетевые задержки до серверов Сбера.

• Speed: 137.7 токенов/сек — отличная скорость генерации на реальном API. Для сравнения: это быстрее, чем читает средний человек.

• Total Time: 1.8 секунды на генерацию 248 токенов (полное стихотворение).

4. Автоматический Fallback

Также протестировал сценарий с ошибкой: что будет, если GigaChat недоступен?

Я специально указал неверный API-ключ для GigaChat, и роутер автоматически переключился на YandexGPT до начала стрима. Вот что произошло:

1. Роутер попытался вызвать GigaChat → получил ошибку 401 Unauthorized.

2. До того, как пользователь увидел хоть одно слово, роутер переключился на YandexGPT.

3. Текст начал печататься от YandexGPT, как будто ничего не произошло.

Метрики fallback-теста:

Важно: после того как первый токен уже отправлен — fallback не происходит, чтобы не смешивать ответы разных моделей. Это стандартное поведение ^[4] для streaming API.

Как попробовать?

Проект уже на PyPI. Установка элементарная:

pip install multi-llm-orchestrator

# Для использования с LangChain:
pip install multi-llm-orchestrator[langchain]

Пример с YandexGPT

import asyncio
from orchestrator import Router
from orchestrator.providers import YandexGPTProvider, ProviderConfig

async def main():
    router = Router(strategy="first-available")
    
    config = ProviderConfig(
        name="yandex",
        api_key="ваш_iam_token",
        folder_id="ваш_folder_id",
        model="yandexgpt/latest"
    )
    
    router.add_provider(YandexGPTProvider(config))
    
    try:
        response = await router.route("Расскажи шутку про Python")
        print(response)
    except Exception as e:
        print(f"Все провайдеры недоступны: {e}")

asyncio.run(main())

Планы (Roadmap)

Уже реализовано (v0.5.0):

• ✅ Умный роутинг и Fallback

• ✅ GigaChat, YandexGPT, Ollama, Mock

• ✅ Streaming (потоковая генерация)

• ✅ Интеграция с LangChain (включая streaming)

В планах (v0.6.0+):

• 🛠 Observability: Структурные логи и метрики (latency, error rate) для мониторинга в Prometheus/Grafana

• 🛠 Расширенный роутинг: Учёт латентности, стоимости запросов и качества ответов

• 🛠 YandexGPT streaming: Расширение потоковой генерации на YandexGPT

---

Приглашаю к участию!

Проект полностью открытый (MIT License). Если вам интересно развитие инструментов для российских LLM — залетайте в репозиторий, ставьте звезды ⭐ и кидайте PR-ы!

🔗 GitHub: github.com/MikhailMalorod/Multi-LLM-Orchestrator ^[5]
📦 PyPI: pypi.org/project/multi-llm-orchestrator/ ^[6]

Буду рад любой технической критике в комментариях! 👇