ТЕХНИЧЕСКОЕ ЗАДАНИЕ: Реализовать chaos testing для агента

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

Разработать и выполнить chaos testing для работающего LLM-агента, который вызывает внешнее API. Цель — проверить, как агент реагирует на внезапные сбои API (5xx, таймауты, задержки) и корректно ли применяет механизмы retry (повтор) и fallback (запасной ответ). Полученные результаты оформить в виде краткого отчёта (postmortem-стайл).

Ключевой результат Рабочий скрипт chaos-тестирования, который доказывает, что агент при ошибке API либо повторяет запрос (до N раз), либо возвращает осмысленный fallback, и метрики (количество retry, успешность fallback) собраны в отчёт.

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

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

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

1. Создаём простой HTTP-сервер (на FastAPI или Flask) с одним эндпоинтом /api/weather (или любым).
2. Добавляем в сервер параметры error_rate (вероятность 5xx), timeout_probability (вероятность блокировки ответа на заданное время) и latency_range (случайные задержки).
3. Запускаем сервер локально (localhost:8000). Агент будет обращаться к этому серверу.
4. Для управления ошибками используем environment variables или query-параметры сервера (только для теста).

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

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

Этап 1: Создание базового агента + мок-API (1.5 часа)

Действия

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

   chaos-testing-lab/
   ├── agent/
   │   ├── __init__.py
   │   ├── client.py          # Клиент для вызова API
   │   ├── orchestrator.py    # Бизнес-логика агента
   │   └── config.py          # Настройки: URL, таймауты, max_retries
   ├── mock_api/
   │   └── server.py          # FastAPI сервер с инжекцией ошибок
   ├── tests/
   │   ├── __init__.py
   │   └── test_chaos.py      # Chaos-тесты
   ├── requirements.txt
   └── README.md
   

2. Реализуйте мок-API (mock_api/server.py):

   from fastapi import FastAPI, HTTPException
   import random, time, os
   
   app = FastAPI()
   ERROR_RATE = float(os.getenv("ERROR_RATE", "0.3"))   # 30% ошибок
   TIMEOUT_RATE = float(os.getenv("TIMEOUT_RATE", "0.1")) # 10% таймаутов
   ERROR_SLEEP_MAX = float(os.getenv("ERROR_SLEEP_MAX", "5")) # макс секунд задержки
   
   @app.get("/api/weather")
   async def get_weather(city: str = "London"):
       # Имитация задержки
       if random.random() < TIMEOUT_RATE:
           time.sleep(ERROR_SLEEP_MAX)  # Блокирующий сон (в асинхронном контексте использовать asyncio.sleep)
           raise HTTPException(504, "Gateway Timeout")  # для простоты
   
       # Имитация ошибки 5xx
       if random.random() < ERROR_RATE:
           status = random.choice([500, 502, 503])
           raise HTTPException(status, "Internal Server Error")
   
       # Успешный ответ
       return {"city": city, "temperature": 20, "units": "celsius"}
   

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

3. Реализуйте клиент агента (agent/client.py):

   import httpx
   from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
   from .config import API_URL, MAX_RETRIES, TIMEOUT_SECONDS
   
   class WeatherClient:
       def __init__(self):
           self.client = httpx.AsyncClient(timeout=TIMEOUT_SECONDS)
   
       @retry(
           stop=stop_after_attempt(MAX_RETRIES),
           wait=wait_exponential(multiplier=1, min=1, max=10),
           retry=retry_if_exception_type(
               (httpx.TimeoutException, httpx.HTTPStatusError)
           ),
           reraise=True
       )
       async def get_weather(self, city: str) -> dict:
           response = await self.client.get(f"{API_URL}/api/weather", params={"city": city})
           response.raise_for_status()  # вызовет HTTPStatusError для 4xx/5xx
           return response.json()
   

4. Реализуйте orchestrator с fallback (agent/orchestrator.py):

   from .client import WeatherClient
   
   class WeatherAgent:
       def __init__(self, fallback_response: str = "Сервис временно недоступен, попробуйте позже."):
           self.client = WeatherClient()
           self.fallback = fallback_response
   
       async def handle_request(self, city: str) -> str:
           try:
               data = await self.client.get_weather(city)
               return f"В городе {city} температура {data['temperature']}°C"
           except Exception:
               # Логируем ошибку
               # В реальном chaos-тесте проверяем, что retry исчерпаны
               return self.fallback
   

5. Проверьте, что агент работает без хаоса:

   • Запустите мок-API с ERROR_RATE=0 и TIMEOUT_RATE=0.
   • Напишите простой скрипт, который вызывает агента и выводит результат.

Ожидаемый результат этапа Работающий агент и мок-API, оба запускаются локально.

Этап 2: Добавление хаоса — инжектор ошибок (1 час)

Действия

1. Добавьте в мок-API управление через environment variables (уже сделано).
   Убедитесь, что можно менять параметры на лету (например, перезапустив сервер).

2. Напишите скрипт run_chaos.py, который запускает мок-API с разными комбинациями ERROR_RATE и TIMEOUT_RATE:

   import subprocess
   import pandas as pd
   import asyncio
   from agent.orchestrator import WeatherAgent
   
   CHAOS_CONFIGS = [
       {"error_rate": 0.0, "timeout_rate": 0.0, "label": "без хаоса"},
       {"error_rate": 0.5, "timeout_rate": 0.0, "label": "50% 5xx"},
       {"error_rate": 0.0, "timeout_rate": 0.3, "label": "30% таймаутов"},
       {"error_rate": 0.3, "timeout_rate": 0.2, "label": "смешанный"},
   ]
   
   async def run_test(config):
       # Запускаем мок-API с переменными окружения
       env = {**{k.upper(): str(v) for k, v in config.items() if k in ["error_rate","timeout_rate"]},
              "ERROR_SLEEP_MAX": "5"}
       proc = subprocess.Popen(["uvicorn", "mock_api.server:app", "--host", "127.0.0.1", "--port", "8000"], env=env)
       await asyncio.sleep(2)  # ждём старта
   
       agent = WeatherAgent()
       results = []
       for _ in range(50):  # 50 запросов
           result = await agent.handle_request("London")
           results.append(result)
       proc.terminate()
       return results
   

   Дополните сбором метрик: сколько раз сработал retry, сколько раз был fallback.

3. Добавьте сбор метрик в агента (агент должен логировать каждую попытку retry).
   Используйте logging и tenacity callback:

   import logging
   logger = logging.getLogger(__name__)
   
   def before_retry(retry_state):
       logger.info(f"Retry #{retry_state.attempt_number} after exception: {retry_state.outcome.exception()}")
   
   # В декоратор @retry добавить before=before_retry
   

Ожидаемый результат этапа Возможность запускать мок-API с заданными вероятностями ошибок и собирать логи retry/fallback.

Этап 3: Разработка chaos-тестов (1.5 часа)

Действия

1. Напишите pytest-тесты в tests/test_chaos.py

   • Фикстура, запускающая мок-API с заданными параметрами на время теста.
   • Тест проверяет, что при 0% ошибок агент всегда возвращает успешный результат (без fallback).
   • Тест проверяет, что при 100% ошибок 503 агент после исчерпания retry возвращает fallback.
   • Тест проверяет, что количество retry не превышает MAX_RETRIES.
   • Тест проверяет, что при таймаутах агент делает retry и в итоге fallback (если таймаутов много).
   • Тест проверяет, что fallback не вызывается при успешном ответе.

2. Пример теста (асинхронный):

   @pytest.fixture
   async def agent_with_chaos(request):
       # request.param — dict с настройками хаоса
       params = request.param
       env = {k.upper(): str(v) for k, v in params.items() if k in ["error_rate","timeout_rate"]}
       env["ERROR_SLEEP_MAX"] = "3"
       proc = await asyncio.create_subprocess_exec(
           "uvicorn", "mock_api.server:app",
           "--host", "127.0.0.1", "--port", "8001",
           env=env
       )
       await asyncio.sleep(1.5)
       yield WeatherAgent()
       proc.terminate()
       await proc.wait()
   
   @pytest.mark.parametrize("agent_with_chaos", [
       {"error_rate": 1.0, "timeout_rate": 0.0},
   ], indirect=True)
   async def test_all_errors_leads_to_fallback(agent_with_chaos):
       result = await agent_with_chaos.handle_request("Berlin")
       assert result == "Сервис временно недоступен, попробуйте позже."
   

3. Запустите тесты и убедитесь, что они проходят. Зафиксируйте покрытие сценариев.

Ожидаемый результат этапа Пакет тестов, покрывающий основные сценарии хаоса.

Этап 4: Анализ результатов и написание отчёта (1 час)

Действия

1. Запустите полный прогон chaos-сценариев (из Этапа 2) и соберите статистику:

   • Общее количество запросов, успешных ответов, fallback.
   • Среднее число retry на запрос (для успешных и упавших).
   • Распределение типов ошибок (HTTP status, timeout).

2. Постройте таблицу в отчёте:

   Сценарий	Успех (%)	Fallback (%)	Среднее retry	Макс retry
   Без хаоса	100%	0%	0	0
   50% 5xx	...	...	...	...
   ...	...	...	...	...

3. Напишите краткий postmortem-отчёт (в свободной форме или по шаблону из примера), отвечая на вопросы:

   • Какие ошибки агент пережил без потери качества? (например, редкие 503 с retry)
   • При каких условиях срабатывал fallback?
   • Есть ли риск бесконечного retry при определённых ошибках?
   • Рекомендации по улучшению: увеличить MAX_RETRIES, уменьшить таймаут, добавить circuit breaker.

4. Сохраните отчёт в файл chaos_test_report.md.

Ожидаемый результат этапа Заполненный отчёт с таблицей и выводами.

Этап 5: (Бонус) Интеграция circuit breaker (1 час)

Действия

1. Реализуйте circuit breaker (размыкание цепи) на стороне клиента, используя pybreaker или собственную реализацию.
2. Настройте параметры: failure threshold = 5, recovery timeout = 30 сек.
3. Напишите тест, при котором после 5 ошибок подряд клиент перестаёт вызывать API (открытое состояние) и возвращает fallback без retry.
4. Обновите отчёт, включив раздел о circuit breaker.

Ожидаемый результат этапа Дополнительный код и тесты для circuit breaker.

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

• Агент корректно реализует retry с экспоненциальной задержкой (максимум N раз).
• Агент возвращает fallback-сообщение после исчерпания retry или при невосстанавливаемой ошибке (4xx).
• Мок-API умеет симулировать ошибки 5xx, таймауты, длинные задержки.
• Chaos-тесты (pytest) покрывают минимум 4 сценария: без ошибок, 50% 5xx, 30% таймаутов, смешанный.
• Тесты проверяют, что количество retry не превышает заданный лимит.
• Собран лог событий агента (попытки retry, момент fallback).
• Написан отчёт (markdown) с таблицей метрик по каждому сценарию.
• (Опционально) Реализован circuit breaker и соответствующий тест.

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

Основной артефакт Директория chaos-testing-lab/ со следующей структурой:

chaos-testing-lab/
├── agent/
│   ├── client.py
│   ├── orchestrator.py
│   └── config.py
├── mock_api/
│   └── server.py
├── tests/
│   └── test_chaos.py
├── requirements.txt
├── chaos_test_report.md          # Отчёт с таблицами и выводами
└── README.md                     # Инструкция по запуску

Содержание отчёта

• Цель chaos testing.
• Используемые сценарии и настройки.
• Результаты (таблица метрик).
• Анализ: какие ошибки агент пережил, при каких условиях включился fallback.
• Рекомендации по улучшению устойчивости.

Опциональные доп. результаты

• График распределения retry по сценариям.
• Код circuit breaker.
• Докеризация (Compose для запуска мок-API и агента).

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

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

Примечание: Для первого раза (без опыта с tenacity/chaostoolkit) заложите +2 часа на изучение документации и отладку.

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

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

• Я создал мок-API с возможностью настройки вероятностей ошибок и таймаутов.
• В агенте реализован retry с ограничением по количеству попыток и экспоненциальной задержкой.
• Агент логирует каждую попытку retry (уровень INFO).
• При исчерпании retry возвращается fallback-сообщение.
• Написаны асинхронные pytest-тесты для сценариев: без ошибок, 50% 5xx, 30% таймаутов, смешанный.
• Тесты проверяют как успешный результат, так и факт возврата fallback.
• Собран отчёт с таблицей метрик (успех/fallback/среднее retry) по каждому сценарию.
• Я проверил, что тесты не имеют ложных срабатываний (перезапускал несколько раз).
• README содержит инструкцию по установке зависимостей, запуску мок-API и тестов.
• (Опционально) Реализован circuit breaker и соответствующий тест.