ТЕХНИЧЕСКОЕ ЗАДАНИЕ: Реализовать circuit breaker для LLM API

1. Цель задачи

Научиться проектировать и внедрять паттерн Circuit Breaker для защиты LLM API от каскадных отказов и перегрузок. Реализовать логику автоматического размыкания цепи при превышении порога ошибок (>50%) за 10-секундное окно, а также корректный fallback-механизм для поддержания функциональности потребителя API.

Ключевой результат Работающий circuit breaker с конфигурируемыми параметрами, который при превышении 50% ошибок за 10 секунд размыкает цепь и переключает вызовы на fallback-обработчик.

2. Исходные данные

Перед началом необходимо иметь:

Если нет реального LLM API — симулируем:

1. Запускаем локальный HTTP-сервер на Flask/FastAPI, который эмулирует LLM эндпоинт.
2. Сервер настраивается так, чтобы генерировать 50%+ ошибок (500, 503, таймауты) при каждом втором запросе или по расписанию.
3. Пишем простой клиент, который делает запросы к этому серверу с использованием circuit breaker.

3. Технологический стек

4. Этапы выполнения

Этап 1: Проектирование и подготовка окружения (30 минут)

Действия

1. Создать структуру проекта:

   • circuit_breaker/
     • __init__.py
     • core.py — логика breaker
     • fallback.py — fallback-обработчики
     • config.py — Pydantic-схема параметров
     • metrics.py — Prometheus метрики
     • stub_server.py — симулятор LLM API (если нет реального)
     • tests/test_breaker.py

2. Определить параметры circuit breaker (через Pydantic):

   class BreakerConfig(BaseSettings):
       failure_threshold: int = 5          # кол-во ошибок за окно
       recovery_timeout: int = 30          # сек до перехода в half-open
       window_size: int = 10               # окно в секундах
       failure_percentage: float = 50.0    # порог в % (будем вычислять)
       max_retries: int = 2                # попытки до открытия цепи
   

3. Создать stub-сервер на FastAPI, который с вероятностью 60% возвращает 500:

   # stub_server.py
   import random
   from fastapi import FastAPI, HTTPException
   import asyncio
   
   app = FastAPI()
   
   @app.get("/v1/chat/completions")
   async def chat():
       await asyncio.sleep(0.1)
       if random.random() < 0.6:
           raise HTTPException(500, "Simulated server error")
       return {"choices": [{"message": {"content": "stub answer"}}]}
   

Ожидаемый результат этапа Проектная папка, конфигурация, работающий stub-сервер (если используем симуляцию).

Этап 2: Реализация логики circuit breaker (1.5–2 часа)

Действия

1. Реализовать класс CircuitBreaker с тремя состояниями: CLOSED, OPEN, HALF_OPEN.

2. Использовать скользящее окно для подсчёта доли ошибок:

   • Хранить временные метки успешных и неуспешных вызовов за последние window_size секунд.
   • При каждом вызове вычислять: error_rate = errors / total * 100.
   • Если error_rate > failure_percentage → перейти в OPEN.

3. Реализовать переходы:

   • CLOSED → OPEN: при превышении порога ошибок.
   • OPEN → HALF_OPEN: после recovery_timeout секунд (один пробный запрос).
   • HALF_OPEN → CLOSED: если пробный запрос успешен.
   • HALF_OPEN → OPEN: если пробный запрос снова неудачен.

4. Добавить потокобезопасность (через asyncio.Lock).

5. Пример скелета:

   class CircuitBreaker:
       STATES = ("CLOSED", "OPEN", "HALF_OPEN")
   
       def __init__(self, config: BreakerConfig):
           self.config = config
           self.state = "CLOSED"
           self.history: list[tuple[float, bool]] = []  # (timestamp, success?)
           self.last_open_time = 0.0
           self.lock = asyncio.Lock()
   
       async def call(self, func, *args, **kwargs):
           # 1. Проверить состояние
           # 2. Если OPEN и таймаут истёк → HALF_OPEN
           # 3. Выполнить func (с таймаутом)
           # 4. Обновить историю
           # 5. Проверить процент ошибок
           # 6. Вернуть результат или выбросить CircuitBreakerError
   

6. Реализовать декоратор/контекстный менеджер для удобного применения.

Ожидаемый результат этапа Готовый класс CircuitBreaker с корректной логикой состояний и скользящим окном.

Этап 3: Fallback-механизм (30 минут)

Действия

1. Создать функцию fallback:

   async def fallback_response(original_request: dict) -> dict:
       # Возвращает заглушку или альтернативный ответ
       return {"choices": [{"message": {"content": "Service temporarily unavailable. Please try later."}}]}
   

2. Интегрировать fallback в вызов: при CircuitBreakerError или при открытом состоянии сразу вызывать fallback, а не бросать исключение пользователю.

3. Опционально: реализовать приоритетный fallback (например, локальная модель → кэш → заглушка).

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

   breaker = CircuitBreaker(config)
   async def safe_llm_call(prompt: str):
       try:
           return await breaker.call(llm_api, prompt=prompt)
       except CircuitBreakerError:
           return await fallback_response(prompt)
   

Ожидаемый результат этапа Fallback-функция, вызываемая при размыкании цепи.

Этап 4: Мониторинг и метрики (30 минут)

Действия

1. Добавить Prometheus-метрики:

   • llm_requests_total{status="success|error|fallback"}
   • llm_circuit_breaker_state{state="closed|open|half_open"}
   • llm_request_duration_seconds (гистограмма)

2. Экспонировать метрики через /metrics (использовать prometheus_client + starlette).

3. Настроить логирование переходов и событий (например, logger.info("Circuit breaker opened")).

Ожидаемый результат этапа Метрики доступны по /metrics, состояния фиксируются в логах.

Этап 5: Тестирование и отладка (1 час)

Действия

1. Написать модульные тесты для CircuitBreaker:

   • Проверить переход по порогу 50% ошибок за 10 с.
   • Проверить восстановление после таймаута.
   • Проверить, что fallback вызывается только при открытом состоянии.

2. Написать интеграционный тест со stub-сервером:

   • Запустить stub-сервер с 60% ошибок.
   • Выполнить 20 запросов через circuit breaker с порогом 50%.
   • Убедиться, что после ~10 запросов цепь разомкнулась и начали приходить fallback-ответы.

3. Запустить ручное тестирование: увеличить порог до 80% -> цепь не должна размыкаться.

Ожидаемый результат этапа Все тесты проходят, circuit breaker работает согласно спецификации.

5. Критерии приемки (Definition of Done)

• Класс CircuitBreaker реализован и поддерживает состояния CLOSED, OPEN, HALF_OPEN.
• При превышении 50% ошибок за последние 10 секунд цепь переходит в OPEN.
• После перехода в OPEN все запросы немедленно перенаправляются на fallback без вызова реального API.
• Через заданный recovery_timeout (по умолчанию 30 с) выполняется пробный запрос (HALF_OPEN).
• Успешный пробный запрос переводит цепь обратно в CLOSED.
• Неуспешный пробный запрос возвращает цепь в OPEN.
• Fallback-функция возвращает корректный ответ (заглушку).
• Prometheus-метрики экспонируются: количество вызовов, ошибок, fallback, текущее состояние.
• Написаны минимум 3 модульных и 1 интеграционный тест (все проходят).
• Код покрыт обработкой исключений и таймаутов (не более 2 секунд на запрос).

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

• Файл/артефакт Python-пакет circuit_breaker с реализацией, тестами, конфигурацией и примером использования.
• Содержимое ключевого файла (core.py): основной класс CircuitBreaker со всей логикой.
• Дополнительные результаты
  • README.md с инструкцией по запуску, конфигурации и тестированию.
  • Скрипт stress_test.py, который эмулирует нагрузку и демонстрирует работу breaker.
  • Лог файл с примерами переходов состояний.

7. Возможные сложности и их решение

8. Бюджет времени (оценка)

Примечание для первого раза с учётом изучения библиотек и отладки рекомендуется выделить 6-7 часов.

9. Связанные вопросы из базы знаний

10. Чек-лист самопроверки

• Я понимаю разницу между состояниями CLOSED, OPEN, HALF_OPEN и корректно их реализовал.
• Я проверил, что при 40% ошибок (ниже порога) цепь остаётся закрытой.
• Я убедился, что после размыкания все вызовы идут в fallback без доступа к реальному API.
• Я протестировал сценарий, когда пробный запрос в HALF_OPEN успешен — цепь закрывается.
• Я экспонировал метрики и проверил, что состояние breaker отображается в Prometheus (если настроен).