Как вы обрабатываете rate limiting от LLM провайдеров (OpenAI, Anthropic)?

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

Rate limiting — механизм API-провайдера, ограничивающий количество запросов в единицу времени. Для production-системы обязательно реализовывать многоуровневую защиту: batching (объединение запросов), retry с экспоненциальной задержкой (tenacity), очередь задач]] с контролем частоты (Redis/Celery), адаптивный мониторинг остатка лимита и автоматическое переключение на другого провайдера. Ключевой принцип: собственный rate limiter на стороне клиента надёжнее, чем пассивное ожидание отказов API.

1. Терминология: что такое rate limiting и почему это проблема

Rate limiting — политика API, ограничивающая количество вызовов в секунду (RPS) или в минуту (RPM), часто с окнами (фиксированными или скользящими). Провайдеры (OpenAI, Anthropic) возвращают коды HTTP 429 (Too Many Requests) при превышении лимита.

Проблема для production:

• Неравномерная нагрузка — пачки запросов от пользователей могут превысить RPS.
• Разные лимиты — у OpenAI разграничение по моделям (gpt-4, gpt-3.5-turbo), у Anthropic — по tier аккаунта.
• Динамические заголовки — x-ratelimit-remaining, Service Unavailable|Service Unavailable|retry-after нужно считывать.

Термин "окно" (window) — интервал времени, в течение которого подсчитывается количество запросов. Фиксированное окно (каждую минуту) может допускать всплески на границе; скользящее окно (sliding window) более справедливо.

2. Request batching: объединение запросов

Batching — группировка нескольких независимых запросов в один вызов API (например, через endpoint /v1/chat/completions с массивом messages). Это снижает число вызовов и, следовательно, потребление RPS.

Пример с OpenAI Python SDK

import openai

# Batching через n (количество вариантов) — не то же самое, что группировка разных запросов.
# Для батчинга нескольких пользовательских запросов используем:
responses = [openai.ChatCompletion.create(model="gpt-4", messages=msgs) for msgs in batch]

Но на уровне SDK нет встроенного агрегатора. Используйте библиотеку openai.batch (асинхронный) или очередь для накопления.

3. Retry with exponential backoff (библиотека tenacity)

Экспоненциальная задержка (exponential backoff) — при получении 429 или 5xx ошибки ждём всё дольше: сначала 1с, потом 2с, 4с, 8с... часто с джиттером (случайным разбросом) для предотвращения "стадного" повторного запроса.

Библиотека tenacity — де-факто стандарт в Python.

from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type
import openai
from openai import RateLimitError

@retry(
    wait=wait_exponential(multiplier=1, min=2, max=60),
    stop=stop_after_attempt(5),
    retry=retry_if_exception_type(RateLimitError)
)
def call_llm(messages: list) -> str:
    response = openai.ChatCompletion.create(model="gpt-4", messages=messages)
    return response.choices[0].message.content

Важное дополнение считываем retry-after из заголовков (если провайдер его присылает) и ждём максимум из экспоненциальной задержки и retry-after.

Расшифровка wait_exponential вычисляет wait = multiplier * (2 ^ attempt_number) + random(0, jitter). Это предотвращает повторные отказы.

4. Очередь задач (Redis, Celery, или native)

Очередь задач — буфер между приложением и API, позволяющий контролировать скорость отправки запросов.

Пример с Celery и rate limit:

from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379')

@app.task(rate_limit='10/m')  # 10 задач в минуту
def call_llm_task(messages):
    return call_llm(messages)

Celery использует token bucket для rate limiting внутри воркера. Но это ограничение на воркер; для глобального лимита нужно использовать Redis lock или sliding window.

5. Мониторинг приближения к лимиту

Проактивный мониторинг — считываем заголовки ответа API и планируем следующую отправку, не дожидаясь 429.

Заголовки OpenAI (пример):

• x-ratelimit-remaining-requests
• x-ratelimit-remaining-tokens
• x-ratelimit-reset-requests (оставшееся время)

Реализация middleware на HTTP-клиенте (например, httpx или aiohttp) вычитывает эти заголовки и обновляет счётчик в Redis. При приближении к лимиту (например, < 10% от максимума) искусственно замедляем отправку.

import time
from redis import Redis

redis_client = Redis()

def update_limits(response):
    remaining = int(response.headers.get('x-ratelimit-remaining-requests', 0))
    reset_after = float(response.headers.get('x-ratelimit-reset-requests', 60))
    redis_client.set('openai_remaining', remaining)
    redis_client.set('openai_reset_after', reset_after)

def should_throttle():
    remaining = int(redis_client.get('openai_remaining') or 0)
    if remaining < 5:  # критически мало
        wait_time = float(redis_client.get('openai_reset_after') or 60)
        time.sleep(wait_time)
        return True
    return False

6. Автоматическое переключение провайдера (OpenAI → Anthropic → Groq)

Failover strategy — если один провайдер исчерпал лимит или недоступен, система автоматически выбирает другого.

Архитектура «router»

1. Попытка #1: OpenAI с retry (тенейсити).
2. При получении 429 и исчерпании всех retry → переключить на Anthropic.
3. Если и Anthropic недоступен → Groq (бесплатный вариант для лёгких задач).

Реализация

PROVIDERS = [OpenAIProvider, AnthropicProvider, GroqProvider]

def call_with_failover(messages):
    for Provider in PROVIDERS:
        try:
            provider = Provider()
            return provider.call(messages)
        except RateLimitError:
            continue
    raise Exception("All providers exhausted")

Важно учитывать разную стоимость и качество моделей. Можно ранжировать провайдеров: сначала дешёвые, затем дорогие, или наоборот.

7. Rate limiter на своей стороне (leaky bucket / token bucket)

Свой rate limiter — механизм, который ограничивает собственные вызовы, не дожидаясь API. Почему это важно: если ваша система отправляет 1000 запросов в секунду, а лимит провайдера — 100, вы получите 900 отказов. Лучше заранее "дозировать" трафик.

Алгоритмы

Пример на Python (token bucket):

import time

class TokenBucket:
    def __init__(self, rate: float, capacity: int):
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity
        self.last_time = time.monotonic()

    def consume(self) -> bool:
        now = time.monotonic()
        elapsed = now - self.last_time
        self.last_time = now
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        if self.tokens >= 1:
            self.tokens -= 1
            return True
        return False

Использование с asyncio: await asyncio.sleep(0.1) если токенов нет.

Акцент "Свой rate limiter важнее, чем надеяться на API". Он даёт предсказуемость и уменьшает нагрузку на сеть.

8. Сравнение подходов и best practices

Best practices

• Комбинировать минимум 3 слоя: свой limiter → retry → failover.
• Использовать tenacity для retry, redis для хранения состояния.
• Логировать все 429 и статистику по задержкам.
• Тестировать под нагрузкой (locust, k6) чтобы убедиться, что система не превышает лимиты.

9. Пет-проект для закрепления

Задача Разработать микросервис-прокси для LLM API, который защищает от rate limiting и реализует failover.

Инструменты Python, FastAPI, Redis, tenacity, openai и anthropic SDK, docker-compose.

Шаги:

1. Создать FastAPI-приложение с единственной рукой /chat.
2. Завести Redis-клиент для хранения счетчиков лимита каждого провайдера.
3. Реализовать TokenBucket на 10 запросов/с для исходящих вызовов.
4. Обернуть вызовы в retry (tenacity: 3 попытки, exponential backoff 1–10s).
5. При исчерпании retry — переключиться на второго провайдера (Anthropic), аналогично.
6. Добавить эндпоинт /status, выводящий текущие лимиты из Redis.
7. Написать тесты (pytest) с mock-объектами, которые отдают 429.

Ожидаемый результат Сервис корректно обрабатывает до 12 запросов/с, при этом при превышении лимита OpenAI автоматически направляет часть запросов к Anthropic без потери.

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

Навигация

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