Rate limits van de Gemini API

Rate limit overzicht

De Gemini API begrenst je verbruik op meerdere dimensies tegelijk. Je raakt een 429-fout zodra je ook maar een van deze limieten overschrijdt, ook als je ruim onder de andere zit.

De limieten gelden per project, niet per API-sleutel. Meerdere sleutels binnen hetzelfde project delen dus dezelfde quota.

Limieten per tier en model

De gratis tier limieten verschillen per model. De onderstaande waarden zijn de richtwaarden zoals gepubliceerd in juni 2026. Google past deze regelmatig aan, dus controleer je actuele limieten altijd in Google AI Studio.

De oudere modellen uit de 1.5- en 2.0-serie zijn inmiddels grotendeels uitgefaseerd ten gunste van de 2.5-serie. Gebruik bij voorkeur een actueel modelpunt en check de modelpagina voordat je een model vastpint.

Betaalde tiers liggen fors hoger en zijn afhankelijk van je cumulatieve uitgaven:

De exacte drempelbedragen en de bijbehorende RPM, TPM en RPD kunnen wijzigen. Bekijk je werkelijke tier en limieten in Google AI Studio onder je project.

HTTP 429 opvangen

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

def generate_with_backoff(model, prompt: str, max_retries: int = 5) -> str:
    for attempt in range(max_retries):
        try:
            response = model.generate_content(prompt)
            return response.text
        except google.api_core.exceptions.ResourceExhausted:
            if attempt == max_retries - 1:
                raise

            wait_time = (2 ** attempt) + random.uniform(0, 1)
            print(f"Rate limit bereikt (poging {attempt + 1}). Wacht {wait_time:.1f}s...")
            time.sleep(wait_time)

    return ""

Exponentiële backoff met tenacity

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    wait_random,
    retry_if_exception_type,
    before_sleep_log
)
import logging
import google.api_core.exceptions

logger = logging.getLogger(__name__)

@retry(
    stop=stop_after_attempt(6),
    wait=wait_exponential(multiplier=2, min=4, max=120) + wait_random(0, 2),
    retry=retry_if_exception_type(google.api_core.exceptions.ResourceExhausted),
    before_sleep=before_sleep_log(logger, logging.WARNING)
)
def reliable_generate(model, prompt: str) -> str:
    response = model.generate_content(prompt)
    return response.text

Token rate limiting

Naast RPM raak je ook de TPM-limiet bij grote prompts. Houd je tokenverbruik bij binnen een glijdend venster van een minuut:

import time

class TokenRateLimiter:
    def __init__(self, max_tokens_per_minute: int):
        self.max_tpm = max_tokens_per_minute
        self.tokens_used = 0
        self.window_start = time.time()

    def add_usage(self, tokens_used: int):
        now = time.time()
        if now - self.window_start > 60:
            self.tokens_used = 0
            self.window_start = now

        self.tokens_used += tokens_used
        if self.tokens_used > self.max_tpm:
            sleep_time = 60 - (now - self.window_start)
            if sleep_time > 0:
                print(f"TPM-limiet bereikt, wacht {sleep_time:.1f}s")
                time.sleep(sleep_time)
            self.tokens_used = tokens_used
            self.window_start = time.time()

limiter = TokenRateLimiter(max_tokens_per_minute=240_000)

Stel je interne drempel iets onder je werkelijke limiet in (hier 240.000 bij een TPM van 250.000) zodat je een veiligheidsmarge houdt voor het verschil tussen geschatte en werkelijk afgerekende tokens.

Request queue implementeren

Voor doorlopende workloads is een queue met een vaste minimuminterval tussen requests betrouwbaarder dan losse retries:

import asyncio
from collections import deque
from dataclasses import dataclass

@dataclass
class ApiRequest:
    prompt: str
    future: asyncio.Future

class RateLimitedQueue:
    def __init__(self, rpm: int = 14):
        self.rpm = rpm
        self.queue = deque()
        self.min_interval = 60.0 / rpm

    async def add_request(self, prompt: str) -> str:
        loop = asyncio.get_event_loop()
        future = loop.create_future()
        self.queue.append(ApiRequest(prompt, future))
        return await future

    async def process_queue(self, model):
        while True:
            if self.queue:
                request = self.queue.popleft()
                try:
                    response = await model.generate_content_async(request.prompt)
                    request.future.set_result(response.text)
                except Exception as e:
                    request.future.set_exception(e)
                await asyncio.sleep(self.min_interval)
            else:
                await asyncio.sleep(0.1)

Kies de rpm-waarde net onder je werkelijke limiet, zodat tijdelijke vertragingen je niet alsnog over de grens duwen.

Limieten verhogen

Als je de gratis tier limieten structureel bereikt, heb je twee opties.