Limitation du débit et logique de nouvelle tentative
Temporisation exponentielle, gestion du code 429 et utilisation respectueuse des API.
Limitation du débit et logique de nouvelle tentative est une leçon AI Agents gratuite sur CoddyKit. Ceci est la leçon 4 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage AI Agents, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours AI Agents comprend 4 leçons au total.
Qu’est-ce que la limitation de débit ?
La limitation de débit est le mécanisme par lequel les API se protègent contre la surcharge. Lorsque votre agent envoie trop de requêtes trop rapidement, l’API renvoie 429 Too Many Requests. Les limites courantes sont définies par seconde, par minute ou par jour.
Le non-respect des limites de débit entraîne le blocage des agents, la révocation des clés d’API et des frais supplémentaires.
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’en-tête Retry-After
Lorsqu’une API renvoie 429, elle inclut souvent un en-tête Retry-After qui vous indique exactement combien de secondes attendre avant de réessayer. Respectez toujours cet en-tête : l’ignorer et réessayer immédiatement entraînerait simplement une nouvelle réponse 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()Temporisation exponentielle
La temporisation exponentielle est la stratégie standard de nouvelle tentative : attendez plus longtemps après chaque échec. Si la tentative 1 attend 2 secondes, la tentative 2 attend 4 secondes, la tentative 3 attend 8 secondes, etc. Cela réduit progressivement la charge du serveur et lui laisse le temps de récupérer.
Formule : 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')Ajouter de la gigue à la temporisation
Si de nombreux agents réessaient en même temps (scénario fréquent après une brève panne), ils se réveillent tous simultanément — créant une ruée collective qui déclenche immédiatement une nouvelle limitation de débit. L’ajout de gigue (délai aléatoire) répartit les nouvelles tentatives, ce qui réduit la charge du serveur.
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 bibliothèque tenacity
tenacity est la bibliothèque Python la plus populaire pour gérer les nouvelles tentatives. Elle prend en charge la temporisation exponentielle, la gigue, le nombre maximal de nouvelles tentatives et les conditions d’arrêt personnalisées grâce à une syntaxe claire fondée sur les décorateurs. Elle est bien plus fiable que des boucles de nouvelles tentatives écrites manuellement.
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 avec une condition personnalisée de nouvelle tentative
Vous pouvez apprendre à tenacity à ne réessayer que pour certains codes d’état (comme 429 et 5xx) et à s’arrêter immédiatement en cas d’erreurs côté client (4xx), qui ne bénéficieraient pas d’une nouvelle tentative. Utilisez retry_if_result ou une fonction appelable personnalisée pour examiner la réponse.
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()Gestion proactive des limites de débit
La meilleure stratégie consiste à éviter de dépasser les limites de débit dès le départ. Vérifiez les en-têtes de limitation de débit dans chaque réponse et ralentissez lorsque vous approchez de la limite. De nombreuses API incluent les en-têtes X-RateLimit-Remaining et 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()Nombre maximal de nouvelles tentatives et abandon
La logique de nouvelle tentative doit toujours avoir une limite. Réessayer indéfiniment peut provoquer des défaillances en cascade, au cours desquelles tous vos agents restent bloqués dans des boucles de nouvelles tentatives. Après max_retries, levez une dernière exception contenant le contexte de l’échec afin que l’agent puisse la consigner et passer à une autre tâche.
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)Le modèle du coupe-circuit
Le modèle du coupe-circuit empêche votre agent de marteler un service défaillant. Après un certain nombre d’échecs, le circuit « s’ouvre » et toutes les requêtes échouent immédiatement sans atteindre le réseau. Après une période de refroidissement, il tente une requête ; si celle-ci réussit, le circuit se ferme et le fonctionnement normal reprend.
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}')
Mettre les requêtes en file pour rester dans les limites
Pour les agents qui effectuent de nombreux appels par lot, utilisez un seau de jetons ou un mécanisme de limitation simple fondé sur des pauses pour rester dans les limites. Calculez l’intervalle sûr entre les appels en fonction de la limite de débit de l’API (par exemple, 60 appels par minute = 1 appel par seconde).
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 resultsCombiner la logique de nouvelle tentative avec les en-têtes de temporisation
Le modèle le plus robuste combine les temps d’attente indiqués par le serveur (Retry-After) avec une temporisation exponentielle comme solution de repli. Préférez toujours les indications du serveur lorsqu’elles sont disponibles : il sait exactement quand vous pourrez réessayer.
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')Vérification rapide : temporisation exponentielle
Testez votre compréhension des stratégies de nouvelle tentative.
Récapitulatif de la limitation de débit et des nouvelles tentatives
Vos agents savent désormais gérer les limites de débit de manière fiable :
- 429 Trop de requêtes — respectez l’en-tête
Retry-After; attendez avant de réessayer - Temporisation exponentielle —
wait = 2^attemptdouble le temps d’attente à chaque nouvelle tentative - Gigue — ajoute une part d’aléatoire pour répartir les nouvelles tentatives entre plusieurs instances d’agents
- tenacity — gère toute la logique de nouvelle tentative grâce aux décorateurs et à une configuration claire
- Coupe-circuit — cesse de marteler un service défaillant après un certain nombre d’échecs
- Limitation proactive — vérifiez
X-RateLimit-Remaininget ralentissez avant d’atteindre la limite
Questions Fréquemment Posées
La leçon « Limitation du débit et logique de nouvelle tentative » est-elle gratuite ?
Oui — le texte complet de « Limitation du débit et logique de nouvelle tentative » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours AI Agents, passe à CoddyKit PRO. Le cours AI Agents comprend 4 leçons au total.
Qu'est-ce que j'apprendrai dans « Limitation du débit et logique de nouvelle tentative » ?
Temporisation exponentielle, gestion du code 429 et utilisation respectueuse des API. Tu pratiques AI Agents avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.
Dois-je avoir de l'expérience pour commencer AI Agents ?
Aucune expérience préalable n'est requise. AI Agents sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 4 sur 4.
Combien de temps prend la leçon « Limitation du débit et logique de nouvelle tentative » ?
La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.
Peux-tu écrire et exécuter du code dans cette leçon AI Agents ?
Oui. Chaque leçon AI Agents inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.
Toutes les leçons de ce cours
- Fondamentaux des API REST pour les développeurs d’agents
- Authentification : clés API et OAuth
- Gérer les réponses et les erreurs d’API
- Limitation du débit et logique de nouvelle tentative