Rate limiting e logica di retry
Exponential backoff, gestione del codice 429 e utilizzo responsabile delle API
Rate limiting e logica di retry è una lezione AI Agents gratuita su CoddyKit. Questa è la lezione 4 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Agents, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Agents include 4 lezioni in totale.
Che cos'è il rate limiting
Il rate limiting è il meccanismo con cui le API si proteggono dal sovraccarico. Quando l'agente invia troppe richieste troppo rapidamente, l'API restituisce 429 Too Many Requests. I limiti più comuni riguardano il numero di richieste al secondo, al minuto o al giorno.
Ignorare i limiti di frequenza può causare il blocco degli agenti, la revoca delle chiavi API e costi aggiuntivi.
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}')L'intestazione Retry-After
Quando un'API restituisce 429, spesso include un'intestazione Retry-After che indica esattamente quanti secondi attendere prima di ritentare. Rispetti sempre questa intestazione: ignorarla e ritentare immediatamente farà semplicemente ricevere un altro 429.
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()Backoff esponenziale
Il backoff esponenziale è la strategia standard per i nuovi tentativi: attenda più a lungo dopo ogni tentativo fallito. Se il tentativo 1 prevede un'attesa di 2 secondi, il tentativo 2 ne prevede 4, il tentativo 3 ne prevede 8 e così via. In questo modo il carico sul server si riduce progressivamente e il server ha il tempo di riprendersi.
Formula: 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')Aggiunta del jitter al backoff
Se molti agenti ritentano contemporaneamente, come spesso accade dopo una breve interruzione, si riattivano tutti nello stesso momento, creando un thundering herd che viene nuovamente limitato subito dal rate limiting. L'aggiunta del jitter, cioè di un ritardo casuale, distribuisce i nuovi tentativi nel tempo e riduce il carico sul server.
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')La libreria tenacity
tenacity è la libreria Python più utilizzata per la logica di retry. Gestisce il backoff esponenziale, il jitter, il numero massimo di tentativi e le condizioni di arresto personalizzate con una sintassi pulita basata su decoratori. È molto più affidabile dei loop di retry scritti manualmente.
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 con una condizione di retry personalizzata
È possibile configurare tenacity affinché ritenti solo in caso di codici di stato specifici, come 429 e 5xx, e si fermi immediatamente in caso di errori client, 4xx, per i quali ritentare non sarebbe utile. Utilizzi retry_if_result o una funzione richiamabile personalizzata per analizzare la risposta.
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()Gestione proattiva dei limiti di frequenza
La strategia migliore è evitare di raggiungere i limiti di frequenza. Verifichi le intestazioni dei limiti di frequenza in ogni risposta e rallenti quando si avvicina al limite. Molte API includono le intestazioni X-RateLimit-Remaining e 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()Numero massimo di tentativi e rinuncia
La logica di retry deve avere sempre un limite. Ritentare all'infinito può causare errori a cascata, bloccando tutti gli agenti in loop di retry. Dopo max_retries, sollevi un'eccezione finale con il contesto relativo all'errore, così l'agente può registrarlo e passare ad altre attività.
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)Il pattern circuit breaker
Il pattern circuit breaker impedisce all'agente di sovraccaricare un servizio in errore. Dopo una determinata soglia di errori, il circuito si "apre" e tutte le richieste falliscono immediatamente senza raggiungere la rete. Trascorso un periodo di attesa, il circuito prova una richiesta: se va a buon fine, si chiude e riprende il normale funzionamento.
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}')
Accodamento delle richieste per rispettare i limiti
Per gli agenti che effettuano molte chiamate in un batch, utilizzi un token bucket o un semplice throttle basato su pause per rimanere entro i limiti. Calcoli l'intervallo sicuro tra le chiamate in base al rate limit dell'API, ad esempio 60 chiamate al minuto equivalgono a 1 chiamata al secondo.
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 resultsCombinazione della logica di retry con le intestazioni di backoff
Il pattern più robusto combina i tempi di attesa specificati dal server, tramite Retry-After, con il backoff esponenziale come alternativa. Dia sempre priorità alle indicazioni del server quando disponibili: il server sa esattamente quando è possibile ritentare.
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')Verifica rapida: backoff esponenziale
Verifichi la propria comprensione delle strategie di retry.
Riepilogo del rate limiting e dei retry
Ora gli agenti possono gestire i limiti di frequenza in modo corretto:
- 429 Too Many Requests: rispetti l'intestazione
Retry-Aftere attenda prima di ritentare - Backoff esponenziale:
wait = 2^attemptraddoppia il tempo di attesa a ogni tentativo - Jitter: aggiunge casualità per distribuire i nuovi tentativi tra più istanze dell'agente
- tenacity: gestisce tutta la logica di retry tramite decoratori e una configurazione chiara
- Circuit breaker: impedisce di sovraccaricare un servizio in errore dopo il superamento di una soglia
- Throttling proattivo: verifica
X-RateLimit-Remaininge rallenta prima di raggiungere il limite
Domande Frequenti
La lezione «Rate limiting e logica di retry» è gratuita?
Sì — il testo completo di «Rate limiting e logica di retry» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Agents, passa a CoddyKit PRO. Il corso AI Agents include 4 lezioni in totale.
Cosa imparerò in «Rate limiting e logica di retry»?
Exponential backoff, gestione del codice 429 e utilizzo responsabile delle API Eserciti AI Agents con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.
Ho bisogno di esperienza per iniziare AI Agents?
Non è richiesta alcuna esperienza precedente. AI Agents su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 4 di 4.
Quanto tempo richiede la lezione «Rate limiting e logica di retry»?
La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.
Posso scrivere ed eseguire codice in questa lezione AI Agents?
Sì. Ogni lezione AI Agents include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.
Tutte le lezioni di questo corso
- Fondamenti delle REST API per sviluppatori di agenti
- Autenticazione: API key e OAuth
- Gestione delle risposte e degli errori delle API
- Rate limiting e logica di retry