Обработка ошибок и стратегии Retry с экспоненциальной задержкой

30 минут Урок 5

Введение: Иллюзия стабильности и реальность Enterprise

Добро пожаловать на урок, который отделяет любительские скрипты от профессиональных промышленных решений. Когда вы пишете код для хакатона или личного проекта, падение API Gemini из-за перегрузки сети — это просто досадная мелочь. Вы перезапускаете скрипт и идете дальше.

В Enterprise-среде ошибка API может стоить денег, репутации и потерянных клиентов. Представьте, что ваш сервис обрабатывает тысячи запросов в минуту. Сетевой сбой длиной в 2 секунды или внезапный спайк нагрузки (Rate Limit) не должен приводить к падению всего конвейера обработки данных.

В этом уроке мы разберем:

Почему try-except недостаточно.
Какие ошибки Gemini API являются временными (transient), а какие — постоянными.
Математику Exponential Backoff (экспоненциальной задержки) и роль Jitter (случайного разброса).
Как элегантно реализовать это в Python, используя библиотеку tenacity.

Анатомия ошибок: Когда стоит повторять запрос?

Первое правило надежной системы: знай своего врага. Не все ошибки одинаковы. Если вы отправили некорректный JSON или превысили длину контекста, повторная отправка того же самого запроса не даст результата — это безумие. Однако, если сервер Google «моргнул», повтор запроса через секунду может спасти ситуацию.

В контексте Gemini 3 API (и Google Cloud в целом) мы делим ошибки на две категории:

1. Постоянные ошибки (Non-Retriable)

Эти ошибки требуют исправления кода или конфигурации. Механизм Retry здесь вреден.

400 Bad Request: Неверный синтаксис, недопустимый аргумент.
401 Unauthorized / 403 Forbidden: Проблемы с API-ключом или правами доступа.
404 Not Found: Запрашиваемая модель или тюнингованная версия не существует.

2. Временные ошибки (Retriable)

Именно здесь мы применяем стратегии повторов.

429 Too Many Requests (Resource Exhausted): Вы уперлись в квоту (RPM/TPM). Самая частая ошибка при масштабировании.
500 Internal Server Error: Внутренняя ошибка на стороне Google.
503 Service Unavailable: Сервис перегружен или находится на обслуживании.
504 Gateway Timeout: Время ожидания ответа истекло.

Важно: В SDK Google Generative AI эти ошибки часто обернуты в специфические исключения, такие как google.api_core.exceptions.ResourceExhausted или google.api_core.exceptions.ServiceUnavailable.

python

import google.generativeai as genai
from google.api_core import exceptions
import time

# Пример наивного подхода (АНТИПАТТЕРН)
# Никогда не используйте "голый" sleep в продакшене без логики выхода

def naive_retry_request(prompt, model):
    max_retries = 3
    for attempt in range(max_retries):
        try:
            response = model.generate_content(prompt)
            return response.text
        except exceptions.ResourceExhausted:
            print(f"Квота исчерпана. Попытка {attempt + 1} из {max_retries}...")
            time.sleep(2)  # Жесткая задержка - это плохо!
        except Exception as e:
            print(f"Критическая ошибка: {e}")
            break
    return None

Стратегия Exponential Backoff

Почему фиксированная задержка (как time.sleep(2) выше) — это плохо? Представьте, что ваш сервис упал, и 1000 инстансов вашего приложения одновременно получили ошибку 503. Если все они подождут ровно 2 секунды и повторят запрос, они создадут новую волну атаки на сервер (эффект Thundering Herd), снова получат ошибку и снова ударят одновременно.

Решение — Экспоненциальная задержка (Exponential Backoff). Мы увеличиваем время ожидания после каждой неудачи.

Формула:

Delay = Base * (Multiplier ^ Attempt)

Где:

Base: начальная задержка (например, 1 секунда).
Multiplier: множитель (обычно 2).
Attempt: номер попытки (0, 1, 2...).

Пример ряда ожидания: 1с, 2с, 4с, 8с, 16с.

Добавляем Jitter (Дрожание)

Даже с экспонентой есть риск синхронизации потоков. Чтобы «размазать» нагрузку во времени, мы добавляем случайный фактор — Jitter.

Улучшенная формула:

Delay = (Base * 2^Attempt) + Random(0, 1)

Это делает поведение системы более органичным и снижает вероятность повторных коллизий.

python

import random
import time
from google.api_core import exceptions

# Реализация алгоритма Exponential Backoff + Jitter вручную

def generate_with_backoff(model, prompt, max_retries=5, base_delay=1.0):
    attempt = 0
    while attempt < max_retries:
        try:
            # Попытка запроса к Gemini
            return model.generate_content(prompt)
            
        except (exceptions.ResourceExhausted, exceptions.ServiceUnavailable) as e:
            attempt += 1
            if attempt >= max_retries:
                raise e  # Сдаемся после всех попыток
            
            # Вычисляем задержку: 2^attempt + случайное отклонение
            sleep_time = (base_delay * (2 ** (attempt - 1))) + random.uniform(0, 1)
            
            print(f"Ошибка {type(e).__name__}. Ждем {sleep_time:.2f} сек перед попыткой {attempt + 1}")
            time.sleep(sleep_time)
            
    return None

Использование библиотеки Tenacity

Писать циклы while и высчитывать математику вручную в каждом месте вызова API — утомительно и чревато ошибками. В экосистеме Python стандартом де-факто для таких задач является библиотека Tenacity.

Она позволяет вынести логику повторов в декоратор, делая код бизнес-логики чистым и читаемым. Tenacity уже умеет обрабатывать Jitter, максимальное время ожидания и условия остановки.

Ключевые компоненты Tenacity:

retry_if_exception_type(...): указываем, какие именно ошибки ловить.
wait_exponential(...): настройка алгоритма задержки.
stop_after_attempt(...): защита от бесконечных циклов.
before_sleep(...): логирование перед ожиданием (очень полезно для отладки).

python

import google.generativeai as genai
from google.api_core import exceptions
from tenacity import (
    retry,
    stop_after_attempt,
    wait_random_exponential,
    retry_if_exception_type,
    before_sleep_log
)
import logging
import sys

# Настройка логирования
logging.basicConfig(stream=sys.stderr, level=logging.INFO)
logger = logging.getLogger(__name__)

# Декоратор Tenacity - это декларативное описание стратегии отказоустойчивости
@retry(
    # 1. Какие ошибки мы ловим? (ResourceExhausted - 429, ServiceUnavailable - 503)
    retry=retry_if_exception_type(
        (exceptions.ResourceExhausted, exceptions.ServiceUnavailable, exceptions.GoogleAPICallError)
    ),
    
    # 2. Как долго ждать? (Экспонента: от 1с до макс 60с + Jitter)
    # wait_random_exponential автоматически добавляет jitter
    wait=wait_random_exponential(multiplier=1, max=60),
    
    # 3. Когда остановиться? (После 5 попыток)
    stop=stop_after_attempt(5),
    
    # 4. Что делать перед сном? (Логировать)
    before_sleep=before_sleep_log(logger, logging.INFO)
)
def robust_gemini_call(model, prompt):
    """
    Эта функция выглядит просто, но она "бронированная" благодаря декоратору.
    """
    print(f"Отправка запроса: {prompt[:20]}...")
    response = model.generate_content(prompt)
    return response.text

# --- Использование ---
# genai.configure(api_key="YOUR_KEY")
# model = genai.GenerativeModel('gemini-1.5-pro')
# try:
#     result = robust_gemini_call(model, "Объясни квантовую физику")
#     print(result)
# except Exception as e:
#     print(f"Все попытки исчерпаны. Последняя ошибка: {e}")

Скрытые угрозы: Finish Reason и Safety Filters

В работе с Gemini есть особый тип «ошибок», который технически не вызывает Python Exception, но делает ответ бесполезным. Это срабатывание фильтров безопасности.

Если модель решит, что ваш запрос нарушает политики безопасности, она вернет объект response со статусом 200 OK, но без текстового содержимого. Попытка обратиться к response.text вызовет ошибку, но это не сетевая проблема!

Стратегия обработки:

Проверяйте response.prompt_feedback.
Проверяйте response.candidates[0].finish_reason.
Не используйте Retry для блокировок безопасности (Safety Blocks) — модель не передумает через 2 секунды. Здесь нужна логика изменения промпта или уведомления пользователя.

python

from google.generativeai.types import HarmCategory, HarmBlockThreshold

def safe_gemini_handler(response):
    """
    Анализирует ответ на предмет блокировок безопасности.
    """
    try:
        # Пытаемся получить текст. Если сработал фильтр, это может вызвать ValueError
        return response.text
    except ValueError:
        # Анализируем причину
        feedback = response.prompt_feedback
        if feedback:
             print(f"Запрос заблокирован: {feedback}")
             
        if response.candidates:
            reason = response.candidates[0].finish_reason
            print(f"Генерация остановлена. Причина: {reason}")
            # Например: FINISH_REASON_SAFETY
            
        return "[CONTENT BLOCKED BY SAFETY FILTERS]"

Упражнение

Напишите функцию-обертку 'smart_generate', которая объединяет две концепции: <br>1. Использует Tenacity для обработки сетевых ошибок (429, 503).<br>2. Внутри себя проверяет ответ на наличие блокировок безопасности (Safety Filters).<br>3. Если получена блокировка безопасности, функция должна возвращать специальный словарь {'status': 'blocked', 'reason': ...}, а не вызывать исключение и не делать Retry.

Решение:

python
from tenacity import retry, stop_after_attempt, wait_random_exponential, retry_if_exception_type
from google.api_core import exceptions
import google.generativeai as genai

# Определяем стратегию retry только для сетевых ошибок
@retry(
    retry=retry_if_exception_type((exceptions.ResourceExhausted, exceptions.ServiceUnavailable)),
    wait=wait_random_exponential(multiplier=1, max=10),
    stop=stop_after_attempt(3)
)
def _network_safe_call(model, prompt):
    return model.generate_content(prompt)

def smart_generate(model, prompt):
    try:
        # Шаг 1: Сетевой вызов с Retry
        response = _network_safe_call(model, prompt)
        
        # Шаг 2: Проверка на безопасность (Logic handling)
        # Проверяем, есть ли кандидаты и не заблокированы ли они
        if response.candidates and response.candidates[0].finish_reason.name == "SAFETY":
            return {
                'status': 'blocked',
                'reason': 'Safety filters triggered',
                'content': None
            }
            
        # Успешный путь
        return {
            'status': 'success',
            'content': response.text
        }
        
    except Exception as e:
        # Если Retry сдался после всех попыток
        return {
            'status': 'error',
            'reason': str(e),
            'content': None
        }

Вопрос

Почему при реализации стратегии Retry важно использовать Jitter (случайную добавку к времени ожидания)?

Чтобы увеличить общее время ожидания для надежности.

Чтобы запутать сервер и обойти блокировки по IP.

Чтобы предотвратить проблему 'Thundering Herd' (синхронный наплыв запросов после сбоя).

Это требование протокола HTTP/2.

Jitter используется только для шифрования данных, а не для Retry.

Предыдущий урок Следующий урок