Limitación de velocidad y lógica de reintentos
Retroceso exponencial, gestión del código 429 y consumo responsable de las API.
Limitación de velocidad y lógica de reintentos es una lección gratuita de AI Agents en CoddyKit. Esta es la lección 4 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Agents, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Agents incluye 4 lecciones en total.
¿Qué es el límite de solicitudes?
El límite de solicitudes es el mecanismo con el que las API se protegen para no verse desbordadas. Cuando su agente envía demasiadas solicitudes con demasiada rapidez, la API devuelve 429 Too Many Requests. Entre los límites habituales se incluyen solicitudes por segundo, por minuto o por día.
Ignorar los límites de solicitudes provoca agentes bloqueados, claves de API revocadas y cargos adicionales.
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}')El encabezado Retry-After
Cuando una API devuelve 429, suele incluir un encabezado Retry-After que indica exactamente cuántos segundos debe esperar antes de volver a intentarlo. Respete siempre este encabezado: ignorarlo y reintentarlo de inmediato solo hará que reciba otro 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()Espera exponencial
La espera exponencial es la estrategia de reintento estándar: espere más tiempo después de cada intento fallido. Si el intento 1 espera 2 segundos, el intento 2 espera 4, el intento 3 espera 8, etc. Esto reduce progresivamente la carga del servidor y le da tiempo para recuperarse.
Fórmula: 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')Añadir jitter a la espera exponencial
Si muchos agentes reintentan al mismo tiempo (una situación habitual después de una interrupción breve), todos se activan simultáneamente y crean una estampida que vuelve a alcanzar de inmediato el límite de solicitudes. Añadir jitter (un retraso aleatorio) distribuye los reintentos y reduce la carga del servidor.
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 biblioteca tenacity
tenacity es la biblioteca de Python más popular para la lógica de reintentos. Gestiona la espera exponencial, el jitter, el número máximo de reintentos y las condiciones de detención personalizadas mediante una sintaxis limpia basada en decoradores. Es mucho más fiable que crear bucles de reintento 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 condición de reintento personalizada
Puede configurar tenacity para que reintente únicamente con códigos de estado específicos (como 429 y 5xx) y se detenga de inmediato ante errores del cliente (4xx), que no se beneficiarán de un reintento. Utilice retry_if_result o un callable personalizado para inspeccionar la respuesta.
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()Gestión proactiva del límite de solicitudes
La mejor estrategia es evitar alcanzar los límites de solicitudes desde el principio. Compruebe los encabezados del límite de solicitudes en cada respuesta y reduzca la velocidad cuando se acerque al límite. Muchas API incluyen los encabezados X-RateLimit-Remaining y 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()Número máximo de reintentos y abandono
La lógica de reintentos siempre debe tener un límite. Reintentar indefinidamente puede provocar fallos en cascada en los que todos sus agentes quedan atrapados en bucles de reintento. Después de max_retries, genere una excepción final con contexto sobre el fallo para que el agente pueda registrarlo y continuar con otras tareas.
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)El patrón Circuit Breaker
El patrón circuit breaker evita que su agente bombardee un servicio que está fallando. Después de alcanzar un umbral de fallos, el circuito se «abre» y todas las solicitudes fallan de inmediato sin acceder a la red. Tras un periodo de espera, intenta realizar una solicitud; si tiene éxito, el circuito se cierra y se reanuda el funcionamiento normal.
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}')
Encolar solicitudes para mantenerse dentro de los límites
Para los agentes que realizan muchas llamadas en un lote, utilice un token bucket o un limitador sencillo basado en pausas para mantenerse dentro de los límites. Calcule el intervalo seguro entre llamadas según el límite de solicitudes de la API (por ejemplo, 60 llamadas por minuto = 1 llamada por segundo).
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 resultsCombinar la lógica de reintentos con los encabezados de espera
El patrón más sólido combina los tiempos de espera especificados por el servidor (Retry-After) con la espera exponencial como alternativa. Dé siempre prioridad a las indicaciones del servidor cuando estén disponibles: sabe exactamente cuándo puede volver a intentarlo.
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')Comprobación rápida: espera exponencial
Compruebe cuánto ha comprendido sobre las estrategias de reintento.
Resumen de límites de solicitudes y reintentos
Ahora sus agentes pueden gestionar los límites de solicitudes correctamente:
- 429 Too Many Requests: respete el encabezado
Retry-Aftery espere antes de reintentar - Espera exponencial:
wait = 2^attemptduplica la espera en cada reintento - Jitter: añade aleatoriedad para distribuir los reintentos entre varias instancias del agente
- tenacity: gestiona toda la lógica de reintentos mediante decoradores y una configuración clara
- Circuit breaker: evita bombardear un servicio que falla después de alcanzar un umbral
- Limitación proactiva: compruebe
X-RateLimit-Remainingy reduzca la velocidad antes de alcanzar el límite
Preguntas frecuentes
¿La lección «Limitación de velocidad y lógica de reintentos» es gratis?
Sí — el texto completo de «Limitación de velocidad y lógica de reintentos» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Agents, actualiza a CoddyKit PRO. El curso de AI Agents incluye 4 lecciones en total.
¿Qué aprenderé en «Limitación de velocidad y lógica de reintentos»?
Retroceso exponencial, gestión del código 429 y consumo responsable de las API. Practicas AI Agents con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.
¿Necesito experiencia previa para empezar AI Agents?
No se requiere experiencia previa. AI Agents en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 4 de 4.
¿Cuánto tiempo toma la lección «Limitación de velocidad y lógica de reintentos»?
La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.
¿Puedo escribir y ejecutar código en esta lección de AI Agents?
Sí. Cada lección de AI Agents incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.
Todas las lecciones de este curso
- Fundamentos de REST API para desarrolladores de agentes
- Autenticación: claves de API y OAuth
- Gestión de respuestas y errores de API
- Limitación de velocidad y lógica de reintentos