Rate limit – omezení API požadavků

Rate limiting (omezení počtu požadavků) je klíčový koncept při práci s API, včetně AI služeb jako Google Gemini, OpenAI nebo Anthropic Claude. Tento návod vysvětluje, co jsou rate limity, proč existují a jak s nimi efektivně pracovat ve vašich aplikacích.

Co je Rate Limit?

Rate limit je omezení počtu požadavků, které můžete poslat na API v určitém časovém období. Nejčastější typy rate limitů jsou:

• Requests Per Minute (RPM): Počet požadavků za minutu
• Requests Per Day (RPD): Počet požadavků za den
• Tokens Per Minute (TPM): Počet tokenů (jednotek textu) zpracovaných za minutu
• Concurrent Requests: Počet současně zpracovávaných požadavků

Proč existují Rate Limity?

Rate limity slouží několika důležitým účelům:

1. Ochrana infrastruktury

• Zabraňují přetížení serverů
• Distribuují zátěž rovnoměrně mezi uživatele
• Chrání před DoS (Denial of Service) útoky

2. Spravedlivé rozdělení zdrojů

• Zajišťují, že jeden uživatel nemůže monopolizovat službu
• Umožňují poskytovatelům API nabízet různé úrovně služeb podle cenových plánů

3. Kontrola nákladů

• Pomáhají uživatelům předvídat a kontrolovat výdaje
• Zabraňují neočekávaným vysokým účtům při chybách v kódu

4. Kvalita služby

• Udržují nízkou latenci pro všechny uživatele
• Zajišťují konzistentní výkon API

Běžné Rate Limity u AI API

Různí poskytovatelé AI API mají různé limity:

Google Gemini API

• Gemini Pro: 60 RPM (bezplatná vrstva), vyšší limity pro placené účty
• Gemini Pro Vision: 60 RPM
• Tokeny: Omezení na počet vstupních a výstupních tokenů za minutu

OpenAI API

• GPT-3.5 Turbo: 3,500 RPM (placené účty)
• GPT-4: 500 RPM (placené účty)
• Tokeny: Měsíční kvóta tokenů podle tarifu

Anthropic Claude API

• Claude 2: 5 RPM (bezplatná vrstva), až 100+ RPM pro enterprise
• Tokeny: Omezení na počet tokenů za minutu

Jak poznat, že jste narazili na Rate Limit?

Když překročíte rate limit, API obvykle vrátí specifickou chybovou odpověď:

Typické HTTP stavové kódy

• 429 Too Many Requests: Nejčastější kód pro překročení rate limitu
• 403 Forbidden: Někdy používaný při dlouhodobém překračování limitů

Příklad chybové odpovědi

{
  "error": {
    "code": 429,
    "message": "Rate limit exceeded: 60 requests per minute. Please try again later.",
    "status": "RESOURCE_EXHAUSTED"
  }
}

Hlavičky odpovědi

Mnoho API také poskytuje informace o limitech v HTTP hlavičkách:

X-RateLimit-Limit: 60
X-RateLimit-Remaining: 0
X-RateLimit-Reset: 1682675452

Strategie pro práci s Rate Limity

1. Implementace exponenciálního backoff

Exponenciální backoff je technika, kdy při neúspěšném požadavku čekáte postupně delší dobu před opakováním:

import time
import random

def make_api_request_with_backoff(request_func, max_retries=5):
    retries = 0
    while retries < max_retries:
        try:
            return request_func()
        except RateLimitError:
            # Exponenciální backoff s jitter (náhodnost)
            sleep_time = (2 ** retries) + random.random()
            print(f"Rate limit hit. Retrying in {sleep_time:.2f} seconds...")
            time.sleep(sleep_time)
            retries += 1

    raise Exception("Max retries exceeded")

2. Implementace fronty požadavků

Fronta požadavků pomáhá regulovat rychlost odesílání požadavků:

class RequestQueue {
  constructor(requestsPerMinute) {
    this.queue = [];
    this.processing = false;
    this.interval = 60 * 1000 / requestsPerMinute; // ms mezi požadavky
  }

  addRequest(requestFunction) {
    return new Promise((resolve, reject) => {
      this.queue.push({ requestFunction, resolve, reject });
      if (!this.processing) {
        this.processQueue();
      }
    });
  }

  async processQueue() {
    this.processing = true;

    while (this.queue.length > 0) {
      const { requestFunction, resolve, reject } = this.queue.shift();

      try {
        const result = await requestFunction();
        resolve(result);
      } catch (error) {
        reject(error);
      }

      // Čekání před dalším požadavkem
      await new Promise(r => setTimeout(r, this.interval));
    }

    this.processing = false;
  }
}

// Použití
const apiQueue = new RequestQueue(60); // 60 požadavků za minutu

function callAPI(prompt) {
  return apiQueue.addRequest(() => {
    return fetch('https://api.example.com/generate', {
      method: 'POST',
      body: JSON.stringify({ prompt })
    }).then(res => res.json());
  });
}

3. Cachování odpovědí

Cachování pomáhá snížit počet požadavků na API:

import hashlib
import json
import os
import time

class SimpleCache:
    def __init__(self, cache_dir=".cache", ttl=3600):
        self.cache_dir = cache_dir
        self.ttl = ttl  # Time to live v sekundách
        os.makedirs(cache_dir, exist_ok=True)

    def _get_cache_key(self, request_data):
        # Vytvoření unikátního klíče pro požadavek
        data_str = json.dumps(request_data, sort_keys=True)
        return hashlib.md5(data_str.encode()).hexdigest()

    def get(self, request_data):
        key = self._get_cache_key(request_data)
        cache_file = os.path.join(self.cache_dir, key)

        if os.path.exists(cache_file):
            # Kontrola TTL
            file_age = time.time() - os.path.getmtime(cache_file)
            if file_age < self.ttl:
                with open(cache_file, 'r') as f:
                    return json.load(f)

        return None

    def set(self, request_data, response_data):
        key = self._get_cache_key(request_data)
        cache_file = os.path.join(self.cache_dir, key)

        with open(cache_file, 'w') as f:
            json.dump(response_data, f)

# Použití
cache = SimpleCache()

def get_ai_response(prompt):
    request_data = {"prompt": prompt}

    # Zkusit načíst z cache
    cached_response = cache.get(request_data)
    if cached_response:
        print("Cache hit!")
        return cached_response

    # Pokud není v cache, zavolat API
    print("Cache miss, calling API...")
    response = call_ai_api(prompt)

    # Uložit do cache pro příští použití
    cache.set(request_data, response)

    return response

4. Batching (dávkové zpracování)

Spojování více požadavků do jednoho může výrazně snížit počet volání API:

class BatchProcessor {
  constructor(batchSize = 10, processingInterval = 1000) {
    this.queue = [];
    this.batchSize = batchSize;
    this.processingInterval = processingInterval;
    this.timer = null;
  }

  add(item) {
    return new Promise((resolve, reject) => {
      this.queue.push({ item, resolve, reject });

      if (this.queue.length >= this.batchSize) {
        this.processBatch();
      } else if (!this.timer) {
        // Nastavit časovač pro zpracování neúplné dávky
        this.timer = setTimeout(() => this.processBatch(), this.processingInterval);
      }
    });
  }

  async processBatch() {
    if (this.timer) {
      clearTimeout(this.timer);
      this.timer = null;
    }

    if (this.queue.length === 0) return;

    const batch = this.queue.splice(0, this.batchSize);
    const items = batch.map(b => b.item);

    try {
      // Zpracovat celou dávku najednou
      const results = await this.processBatchItems(items);

      // Rozdělit výsledky zpět jednotlivým požadavkům
      batch.forEach((b, index) => {
        b.resolve(results[index]);
      });
    } catch (error) {
      // Při chybě odmítnout všechny požadavky v dávce
      batch.forEach(b => {
        b.reject(error);
      });
    }
  }

  async processBatchItems(items) {
    // Implementace závisí na konkrétním API
    // Například pro API, které podporuje dávkové zpracování:
    const response = await fetch('https://api.example.com/batch', {
      method: 'POST',
      body: JSON.stringify({ items })
    });

    return response.json();
  }
}

5. Monitorování a alerty

Sledování využití API vám pomůže předcházet problémům:

class APIUsageMonitor:
    def __init__(self, limit_per_minute, alert_threshold=0.8):
        self.limit_per_minute = limit_per_minute
        self.alert_threshold = alert_threshold
        self.request_timestamps = []
        self.alert_sent = False

    def record_request(self):
        # Zaznamenat čas požadavku
        current_time = time.time()
        self.request_timestamps.append(current_time)

        # Odstranit staré záznamy (starší než 1 minuta)
        one_minute_ago = current_time - 60
        self.request_timestamps = [t for t in self.request_timestamps if t > one_minute_ago]

        # Zkontrolovat využití
        self.check_usage()

    def check_usage(self):
        current_usage = len(self.request_timestamps)
        usage_percentage = current_usage / self.limit_per_minute

        if usage_percentage >= self.alert_threshold and not self.alert_sent:
            self.send_alert(current_usage, usage_percentage)
            self.alert_sent = True
        elif usage_percentage < self.alert_threshold:
            self.alert_sent = False

    def send_alert(self, current_usage, usage_percentage):
        print(f"ALERT: API usage at {usage_percentage:.1%} ({current_usage}/{self.limit_per_minute})")
        # Zde můžete implementovat odeslání e-mailu, SMS nebo notifikace

Optimalizace nákladů a výkonu

Tipy pro efektivní využití API s rate limity

1. Prioritizujte požadavky: Zpracujte nejdůležitější požadavky jako první
2. Implementujte graceful degradation: Mějte záložní plán, když narazíte na limity
3. Používejte asynchronní zpracování: Nečekejte zbytečně na dokončení požadavků
4. Monitorujte využití: Sledujte, jak se blížíte k limitům
5. Optimalizujte prompty: Kratší a efektivnější prompty šetří tokeny
6. Zvažte různé modely: Menší modely mají často vyšší limity a nižší cenu

Příklad implementace prioritní fronty

import heapq
import time
import threading

class PriorityRequestQueue:
    def __init__(self, requests_per_minute):
        self.queue = []  # Prioritní fronta (heap)
        self.interval = 60 / requests_per_minute  # Interval mezi požadavky v sekundách
        self.lock = threading.Lock()
        self.processing_thread = None

    def add_request(self, request_func, priority=0):
        """Přidá požadavek do fronty. Nižší číslo = vyšší priorita."""
        with self.lock:
            # Přidat do prioritní fronty (heap)
            heapq.heappush(self.queue, (priority, time.time(), request_func))

            # Spustit zpracování, pokud ještě neběží
            if not self.processing_thread or not self.processing_thread.is_alive():
                self.processing_thread = threading.Thread(target=self._process_queue)
                self.processing_thread.daemon = True
                self.processing_thread.start()

    def _process_queue(self):
        """Zpracovává požadavky z fronty s respektováním rate limitu."""
        last_request_time = 0

        while True:
            with self.lock:
                if not self.queue:
                    # Fronta je prázdná, ukončit zpracování
                    return

                # Získat požadavek s nejvyšší prioritou
                _, enqueue_time, request_func = heapq.heappop(self.queue)

            # Dodržet interval mezi požadavky
            current_time = time.time()
            time_since_last_request = current_time - last_request_time

            if time_since_last_request < self.interval:
                # Počkat do uplynutí intervalu
                sleep_time = self.interval - time_since_last_request
                time.sleep(sleep_time)

            # Provést požadavek
            try:
                request_func()
            except Exception as e:
                print(f"Error processing request: {e}")

            # Aktualizovat čas posledního požadavku
            last_request_time = time.time()

Řešení běžných problémů

Problém: Neočekávané 429 chyby

Řešení:

• Implementujte robustní sledování počtu požadavků
• Přidejte jitter (náhodnost) do backoff algoritmu
• Zvažte distribuci požadavků rovnoměrně v čase

Problém: Dlouhé čekací doby při vysokém zatížení

Řešení:

• Implementujte timeout pro požadavky
• Mějte záložní plán pro kritické funkce
• Zvažte použití více API klíčů nebo poskytovatelů

Problém: Neefektivní využití kvót

Řešení:

• Analyzujte vzorce používání a optimalizujte
• Implementujte prediktivní škálování požadavků
• Zvažte přechod na vyšší tarif, pokud pravidelně narážíte na limity

Závěr

Rate limity jsou nezbytnou součástí práce s API, zejména s AI službami, které jsou výpočetně náročné. Správné pochopení a implementace strategií pro práci s těmito limity vám pomůže vytvářet spolehlivější aplikace, optimalizovat náklady a poskytovat lepší uživatelskou zkušenost.

Pamatujte, že rate limity se mohou měnit, proto je důležité pravidelně kontrolovat dokumentaci vašeho poskytovatele API a přizpůsobovat vaši implementaci podle aktuálních podmínek.