0Pricing
AI Agents · Lektion

Rate Limiting und Retry-Logik

Exponential Backoff, Umgang mit 429-Fehlern und rücksichtsvolle API-Nutzung.

Rate Limiting und Retry-Logik ist eine kostenlose AI Agents-Lektion auf CoddyKit. Dies ist Lektion 4 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des AI Agents-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.

Was ist Rate-Limiting?

Rate-Limiting schützt APIs davor, überlastet zu werden. Wenn Ihr Agent zu viele Anfragen in zu kurzer Zeit sendet, gibt die API 429 Too Many Requests zurück. Zu den üblichen Limits gehören Anfragen pro Sekunde, Minute oder Tag.

Wenn Sie Rate-Limits ignorieren, kann dies zu blockierten Agenten, widerrufenen API-Schlüsseln und zusätzlichen Kosten führen.

import requests

response = requests.get(
    'https://api.example.com/data',
    headers={'Authorization': 'Bearer YOUR_KEY'}
)

if response.status_code == 429:
    print('Rate limit exceeded!')
    # Check headers for limit details
    limit = response.headers.get('X-RateLimit-Limit')
    remaining = response.headers.get('X-RateLimit-Remaining')
    reset = response.headers.get('X-RateLimit-Reset')
    print(f'Limit: {limit}, Remaining: {remaining}, Reset: {reset}')

Der Retry-After-Header

Wenn eine API 429 zurückgibt, enthält sie häufig einen Retry-After-Header, der Ihnen genau mitteilt, wie viele Sekunden Sie vor dem nächsten Versuch warten müssen. Beachten Sie diesen Header immer – wenn Sie ihn ignorieren und sofort erneut versuchen, erhalten Sie lediglich einen weiteren 429-Fehler.

import requests
import time

def request_with_retry_after(url, headers):
    response = requests.get(url, headers=headers)

    if response.status_code == 429:
        retry_after = int(response.headers.get('Retry-After', 60))
        print(f'Rate limited. Waiting {retry_after} seconds...')
        time.sleep(retry_after)

        # Retry once after waiting
        response = requests.get(url, headers=headers)

    response.raise_for_status()
    return response.json()

Exponential Backoff

Exponential Backoff ist die standardmäßige Strategie für Wiederholungen: Warten Sie nach jedem fehlgeschlagenen Versuch länger. Wenn Versuch 1 zwei Sekunden wartet, wartet Versuch 2 vier Sekunden, Versuch 3 acht Sekunden usw. Dadurch wird die Serverlast schrittweise reduziert und der Server erhält Zeit, sich zu erholen.

Formel: wait = 2 ** attempt

import requests
import time

def get_with_exponential_backoff(url, headers, max_retries=5):
    for attempt in range(max_retries):
        response = requests.get(url, headers=headers, timeout=(5, 30))

        if response.status_code == 200:
            return response.json()

        if response.status_code in (429, 500, 502, 503):
            wait = 2 ** attempt  # 1, 2, 4, 8, 16 seconds
            print(f'Attempt {attempt+1} failed ({response.status_code}). '
                  f'Waiting {wait}s before retry...')
            time.sleep(wait)
        else:
            response.raise_for_status()  # non-retryable error

    raise Exception(f'Failed after {max_retries} retries')

Jitter zum Backoff hinzufügen

Wenn viele Agenten gleichzeitig einen neuen Versuch starten (ein häufiges Szenario nach einem kurzen Ausfall), werden sie alle gleichzeitig aktiv. Dadurch entsteht eine Thundering-Herd, die sofort wieder ein Rate-Limit auslöst. Das Hinzufügen von Jitter (einer zufälligen Verzögerung) verteilt die neuen Versuche und reduziert die Serverlast.

import requests
import time
import random

def get_with_jittered_backoff(url, headers, max_retries=5):
    for attempt in range(max_retries):
        response = requests.get(url, headers=headers, timeout=(5, 30))

        if response.status_code == 200:
            return response.json()

        if response.status_code in (429, 500, 502, 503):
            base_wait = 2 ** attempt
            # Add random jitter: actual wait is 50%-100% of base
            jitter = random.uniform(0.5, 1.0)
            wait = base_wait * jitter
            print(f'Waiting {wait:.1f}s (attempt {attempt+1})')
            time.sleep(wait)
        else:
            response.raise_for_status()

    raise Exception(f'Failed after {max_retries} retries')

Die tenacity-Bibliothek

tenacity ist die beliebteste Python-Bibliothek für Wiederholungslogik. Sie verarbeitet Exponential Backoff, Jitter, maximale Wiederholungsversuche und benutzerdefinierte Abbruchbedingungen mit einer übersichtlichen Decorator-Syntax. Sie ist wesentlich zuverlässiger als selbst entwickelte Wiederholungsschleifen.

from tenacity import (
    retry, stop_after_attempt, wait_exponential,
    retry_if_exception_type, before_sleep_log
)
import requests
import logging

logger = logging.getLogger(__name__)

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=60),
    retry=retry_if_exception_type(requests.exceptions.HTTPError),
    before_sleep=before_sleep_log(logger, logging.WARNING)
)
def fetch_data(url, headers):
    response = requests.get(url, headers=headers, timeout=(5, 30))
    if response.status_code == 429:
        response.raise_for_status()  # triggers retry
    response.raise_for_status()
    return response.json()

tenacity mit benutzerdefinierter Retry-Bedingung

Sie können tenacity so konfigurieren, dass nur bei bestimmten Statuscodes (z. B. 429 und 5xx) ein neuer Versuch gestartet wird und bei Client-Fehlern (4xx), bei denen eine Wiederholung nicht weiterhilft, sofort abgebrochen wird. Verwenden Sie retry_if_result oder ein benutzerdefiniertes Callable, um die Antwort zu untersuchen.

from tenacity import (
    retry, stop_after_attempt, wait_exponential,
    retry_if_result
)
import requests

def is_retryable_response(response):
    return response.status_code in (429, 500, 502, 503, 504)

@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=2, min=2, max=30),
    retry=retry_if_result(is_retryable_response)
)
def resilient_get(url, headers):
    response = requests.get(url, headers=headers, timeout=(5, 30))
    return response  # retry logic inspects the response object

# Usage
response = resilient_get(
    'https://api.example.com/data',
    {'Authorization': 'Bearer YOUR_KEY'}
)
data = response.json()

Proaktives Rate-Limit-Management

Die beste Strategie besteht darin, Rate-Limits gar nicht erst zu erreichen. Prüfen Sie bei jeder Antwort die Rate-Limit-Header und verlangsamen Sie die Anfragen, wenn Sie sich dem Limit nähern. Viele APIs enthalten die Header X-RateLimit-Remaining und X-RateLimit-Reset.

import requests
import time

class RateLimitAwareClient:
    def __init__(self, base_url, api_key):
        self.base_url = base_url
        self.headers = {'Authorization': f'Bearer {api_key}'}
        self.remaining = 1000  # assume generous limit

    def get(self, path):
        # Proactively slow down if nearly exhausted
        if self.remaining < 10:
            print('Rate limit nearly exhausted, sleeping 5s...')
            time.sleep(5)

        response = requests.get(
            f'{self.base_url}{path}', headers=self.headers
        )

        # Update remaining from response headers
        remaining_str = response.headers.get('X-RateLimit-Remaining')
        if remaining_str:
            self.remaining = int(remaining_str)

        response.raise_for_status()
        return response.json()

Maximale Wiederholungen und das Aufgeben

Die Wiederholungslogik muss immer ein Limit haben. Unbegrenzte Wiederholungen können zu kaskadierenden Fehlern führen, bei denen alle Ihre Agenten in Wiederholungsschleifen feststecken. Lösen Sie nach max_retries eine abschließende Ausnahme mit Kontext zu dem fehlgeschlagenen Vorgang aus, damit der Agent sie protokollieren und mit anderen Aufgaben fortfahren kann.

import requests
import time

class MaxRetriesExceeded(Exception):
    def __init__(self, url, attempts, last_status):
        self.url = url
        self.attempts = attempts
        self.last_status = last_status
        super().__init__(
            f'Failed {url} after {attempts} attempts '
            f'(last status: {last_status})'
        )

def fetch_with_limit(url, headers, max_retries=3):
    last_response = None
    for attempt in range(max_retries):
        last_response = requests.get(url, headers=headers)
        if last_response.status_code == 200:
            return last_response.json()
        time.sleep(2 ** attempt)
    raise MaxRetriesExceeded(url, max_retries, last_response.status_code)

Das Circuit-Breaker-Muster

Das Circuit-Breaker-Muster verhindert, dass Ihr Agent einen fehlerhaften Dienst mit Anfragen überlastet. Nach einer bestimmten Anzahl von Fehlern wird der Circuit „geöffnet“ und alle Anfragen schlagen sofort fehl, ohne das Netzwerk zu verwenden. Nach einer Abkühlphase wird ein einzelner Versuch gestartet. Ist dieser erfolgreich, schließt sich der Circuit und der normale Betrieb wird fortgesetzt.

import time

class CircuitBreaker:
    CLOSED, OPEN, HALF_OPEN = 'closed', 'open', 'half_open'

    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.state = self.CLOSED
        self.failures = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.opened_at = None

    def call(self, func, *args, **kwargs):
        if self.state == self.OPEN:
            if time.time() - self.opened_at > self.recovery_timeout:
                self.state = self.HALF_OPEN
            else:
                raise Exception('Circuit OPEN — service unavailable')
        try:
            result = func(*args, **kwargs)
            self.failures = 0
            self.state = self.CLOSED
            return result
        except Exception as e:
            self.failures += 1
            if self.failures >= self.failure_threshold:
                self.state = self.OPEN
                self.opened_at = time.time()
                print(f'Circuit OPENED after {self.failures} failures')
            raise

# --- demo ---
def flaky():
    raise ValueError('upstream 500')

def works():
    return 'ok'

cb = CircuitBreaker(failure_threshold=3, recovery_timeout=60)
for i in range(3):
    try:
        cb.call(flaky)
    except Exception as e:
        print(f'call {i+1} failed: {e}')
print(f'Breaker state after 3 failures: {cb.state}')
try:
    cb.call(flaky)
except Exception as e:
    print(f'Rejected without calling flaky(): {e}')

Anfragen einreihen, um Limits einzuhalten

Verwenden Sie für Agenten, die viele Aufrufe in einem Batch ausführen, einen Token-Bucket oder eine einfache drosselnde Wartezeit, um die Limits einzuhalten. Berechnen Sie anhand des Rate-Limits der API ein sicheres Intervall zwischen den Aufrufen (z. B. 60 Aufrufe/Minute = 1 Aufruf pro Sekunde).

import requests
import time

def batch_requests(urls, headers, calls_per_minute=60):
    interval = 60.0 / calls_per_minute  # seconds between calls
    results = []

    for i, url in enumerate(urls):
        start = time.time()

        response = requests.get(url, headers=headers, timeout=(5, 30))
        response.raise_for_status()
        results.append(response.json())

        print(f'Processed {i+1}/{len(urls)}')

        # Sleep for remaining time in the interval
        elapsed = time.time() - start
        sleep_time = interval - elapsed
        if sleep_time > 0:
            time.sleep(sleep_time)

    return results

Retry-Logik mit Backoff-Headern kombinieren

Das robusteste Muster kombiniert vom Server vorgegebene Wartezeiten (Retry-After) mit Exponential Backoff als Fallback. Bevorzugen Sie immer die Vorgaben des Servers, sofern sie verfügbar sind – der Server weiß genau, wann der nächste Versuch möglich ist.

import requests
import time
import random

def smart_retry(url, headers, max_retries=5):
    for attempt in range(max_retries):
        response = requests.get(url, headers=headers, timeout=(5, 30))

        if response.status_code == 200:
            return response.json()

        if response.status_code == 429:
            # Use Retry-After if provided, else exponential backoff
            retry_after = response.headers.get('Retry-After')
            if retry_after:
                wait = int(retry_after)
            else:
                wait = (2 ** attempt) + random.uniform(0, 1)
            print(f'429 rate limit. Waiting {wait:.1f}s...')
            time.sleep(wait)

        elif response.status_code >= 500:
            wait = (2 ** attempt) + random.uniform(0, 1)
            print(f'Server error {response.status_code}. Waiting {wait:.1f}s...')
            time.sleep(wait)

        else:
            response.raise_for_status()  # non-retryable

    raise Exception(f'Gave up after {max_retries} attempts')

Kurztest: Exponential Backoff

Testen Sie Ihr Verständnis von Wiederholungsstrategien.

Zusammenfassung: Rate-Limiting und Wiederholungen

Ihre Agenten können jetzt angemessen mit Rate-Limits umgehen:

  • 429 Too Many Requests – beachten Sie den Retry-After-Header und warten Sie vor dem nächsten Versuch.
  • Exponential Backoff – wait = 2^attempt verdoppelt die Wartezeit bei jedem neuen Versuch.
  • Jitter – fügt Zufälligkeit hinzu, um neue Versuche über mehrere Agent-Instanzen zu verteilen.
  • tenacity – übernimmt die gesamte Wiederholungslogik mit Decorators und einer übersichtlichen Konfiguration.
  • Circuit Breaker – verhindert nach einer bestimmten Anzahl von Fehlern weitere Anfragen an einen fehlerhaften Dienst.
  • Proaktives Throttling – prüfen Sie X-RateLimit-Remaining und verlangsamen Sie die Anfragen, bevor das Limit erreicht wird.

Häufig gestellte Fragen

Ist die Lektion „Rate Limiting und Retry-Logik“ kostenlos?

Ja — der vollständige Text von „Rate Limiting und Retry-Logik“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des AI Agents-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.

Was lerne ich in „Rate Limiting und Retry-Logik“?

Exponential Backoff, Umgang mit 429-Fehlern und rücksichtsvolle API-Nutzung. Du übst AI Agents mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.

Brauche ich Erfahrung, um AI Agents zu starten?

Keine Vorkenntnisse erforderlich. AI Agents auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 4 von 4.

Wie lange dauert die Lektion „Rate Limiting und Retry-Logik“?

Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.

Kann ich in dieser AI Agents-Lektion Code schreiben und ausführen?

Ja. Jede AI Agents-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.

Alle Lektionen in diesem Kurs

  1. REST-API-Grundlagen für Agentenentwickler
  2. Authentifizierung: API-Keys und OAuth
  3. API-Responses und Fehler verarbeiten
  4. Rate Limiting und Retry-Logik
← Zurück zu AI Agents