¡Finalmente solucioné el problema del límite de rate limit de la API de mis bots!

Hola a todos, Marcus aquí de ai7bot.com. Hoy es 17 de marzo de 2026 y he estado lidiando con un problema en particular últimamente que apuesto a que muchos de ustedes han encontrado al construir bots para clientes o incluso solo para sus propios proyectos personales. Se trata de gestionar esos molestos límites de tasa de API, especialmente cuando se trata de servicios de terceros que tienen un talento especial para ser… bueno, tacaños con sus solicitudes.

He visto demasiados bots prometedores estrellarse y quemarse porque alcanzaron un límite de tasa y simplemente se rindieron. O, peor aún, fueron restringidos, lo que llevó a una experiencia de usuario terrible. Mi último cliente, una pequeña startup de comercio electrónico, quería un bot de Telegram que pudiera extraer actualizaciones de inventario en tiempo real y comprobaciones de precios de la API increíblemente antigua y, francamente, bastante frágil de su proveedor. La documentación de la API del proveedor era vaga en cuanto a los límites de tasa, solo mencionando “uso razonable”, que, como todos sabemos, es un eufemismo para “te banearemos si lo miras de forma extraña.”

Mi primer pensamiento fue: “Oh chico, aquí vamos de nuevo.” Pero esta vez, decidí abordarlo de frente, no solo con simples demoras, sino con un enfoque más sofisticado y autogestionado. Quería construir un bot que pudiera manejar estos límites de manera elegante, incluso cuando no sabía exactamente cuáles eran. Así que hoy, vamos a profundizar en la construcción de una cola de solicitudes de API autogestionada para tus bots, centrándonos en una estrategia que se adapta en lugar de simplemente asumir.

El Problema: Límites de Tasa de API Inpredecibles

Piénsalo. Estás construyendo un bot de Telegram que necesita obtener datos de un servicio externo. Tal vez sean precios de acciones, actualizaciones meteorológicas o, en mi caso, inventario de productos. Envías una solicitud, obtienes una respuesta. Fácil, ¿verdad? Hasta que, de repente, comienzas a recibir errores HTTP 429 Demasiadas Solicitudes. O peor, la API simplemente comienza a devolver datos vacíos o respuestas mal formadas sin decirte por qué. Eso es lo que “uso razonable” a menudo traduce en el mundo real.

La API del proveedor de mi cliente era un ejemplo claro. A veces, podía acceder a ella cinco veces por segundo sin problemas. Otras veces, dos solicitudes en un segundo provocarían un tiempo de espera. Era frustrante. No podía codificar un simple time.sleep(1) después de cada solicitud, porque eso haría que el bot fuera dolorosamente lento cuando la API se sentía generosa, y aún así no prevenir problemas cuando estaba de mal humor.

El problema principal es la falta de información de límite de tasa transparente y consistente. Muchas APIs proporcionan los encabezados X-RateLimit-Limit, X-RateLimit-Remaining y X-RateLimit-Reset, y si tienes la suerte de trabajar con una de esas, ¡genial! Puedes implementar un algoritmo de cubo de tokens o cubo con fugas bastante fácilmente. Pero, ¿qué pasa con las APIs que no lo hacen? Aquellas que simplemente arrojan errores, o peor aún, te limitan en silencio. Ahí es donde entra la parte de “autogestionado”.

La Solución: Una Cola de Solicitudes Adaptativa con Retroceso

Mi enfoque consistió en crear una cola de solicitudes central por la que pasarían todas las llamadas a la API. Esta cola no solo contendría solicitudes; gestionaría inteligentemente su temporización según el comportamiento de la API. Los componentes clave son:

• Una cola para mantener las solicitudes pendientes.
• Un mecanismo para rastrear el éxito/fallo de las solicitudes recientes.
• Una demora adaptativa que aumenta ante un fallo y disminuye ante un éxito.
• Un trabajador dedicado que procesa la cola.

Desglosemos cómo implementé esto en Python, que es mi lenguaje predilecto para el desarrollo de bots. Estoy usando la biblioteca asyncio porque es perfecta para operaciones concurrentes sin la sobrecarga de los hilos, lo cual es crucial para un bot receptivo.

Construyendo la Cola de Solicitudes Principal

Primero, necesitamos una forma de almacenar nuestras solicitudes. Cada “solicitud” en nuestra cola será una llamada a una función junto con sus argumentos, y una forma de indicar el resultado de vuelta. Descubrí que envolver la llamada a la API real en una pequeña clase o un functools.partial lo hace más limpio.

import asyncio
import time
import collections

class ApiRequestQueue:
 def __init__(self, api_call_func, initial_delay=0.1, max_delay=5.0, delay_factor=1.5, success_factor=0.9):
 self.api_call_func = api_call_func
 self.queue = asyncio.Queue()
 self.current_delay = initial_delay
 self.max_delay = max_delay
 self.delay_factor = delay_factor # Multiplicador para la demora en caso de fallo
 self.success_factor = success_factor # Multiplicador para la demora en caso de éxito
 self.is_running = False
 self.worker_task = None
 self.last_request_time = 0

 async def _worker(self):
 self.is_running = True
 while self.is_running:
 try:
 task_id, future, func_args, func_kwargs = await self.queue.get()
 
 # Obligando a un mínimo de demora entre solicitudes
 elapsed = time.monotonic() - self.last_request_time
 if elapsed < self.current_delay:
 await asyncio.sleep(self.current_delay - elapsed)

 self.last_request_time = time.monotonic()
 
 try:
 result = await self.api_call_func(*func_args, **func_kwargs)
 future.set_result(result)
 # Si es exitoso, reducir ligeramente la demora, pero no por debajo de la inicial
 self.current_delay = max(self.current_delay * self.success_factor, 0.05) 
 print(f"[{time.monotonic():.2f}] Solicitud {task_id} exitosa. Nueva demora: {self.current_delay:.2f}s")
 except Exception as e:
 future.set_exception(e)
 # Si falla, aumentar la demora significativamente, hasta el máximo
 self.current_delay = min(self.current_delay * self.delay_factor, self.max_delay)
 print(f"[{time.monotonic():.2f}] Solicitud {task_id} falló: {e}. Nueva demora: {self.current_delay:.2f}s")
 finally:
 self.queue.task_done()
 except asyncio.CancelledError:
 print("Tarea del trabajador cancelada.")
 break
 except Exception as e:
 print(f"El trabajador encontró un error no manejado: {e}")
 # No dejar que el trabajador se caiga, mantenerlo funcionando
 await asyncio.sleep(1) # Pequeña pausa para evitar un bucle apretado en caso de error persistente

 async def start(self):
 if not self.is_running:
 self.worker_task = asyncio.create_task(self._worker())
 print("Trabajador de la cola de solicitudes de API iniciado.")

 async def stop(self):
 if self.is_running:
 self.is_running = False
 if self.worker_task:
 self.worker_task.cancel()
 await self.worker_task
 print("Trabajador de la cola de solicitudes de API detenido.")

 async def put(self, *args, **kwargs):
 if not self.is_running:
 raise RuntimeError("Cola no iniciada. Llama a .start() primero.")
 future = asyncio.Future()
 task_id = f"req_{time.time():.4f}" # ID único simple para registrar
 await self.queue.put((task_id, future, args, kwargs))
 return await future

Algunas cosas a tener en cuenta aquí:

• api_call_func: Esta es la función asíncrona real que realiza la llamada a la API. Es crucial que esta función maneje posibles excepciones (como errores de red, HTTP 429, etc.) y las levante para que nuestra cola pueda atraparlas.
• current_delay: Este es el corazón de nuestra estrategia adaptativa. Comienza en un valor pequeño (por ejemplo, 0.1 segundos) y cambia según el éxito o el fracaso.
• delay_factor y success_factor: Estos controlan cuán agresivamente se ajusta la demora. He encontrado que un delay_factor de 1.5 es un buen equilibrio para aumentar la demora en caso de fallo, y un success_factor de 0.9 para reducirla lentamente. Es posible que necesites ajustar estos para tu API específica.
• max_delay: Un límite sobre cuánto puede llegar a ser la demora. No queremos esperar 30 segundos entre solicitudes si la API está temporalmente sobrecargada.
• future: Así es como obtenemos el resultado de la llamada a la API de vuelta al llamador. Cuando una solicitud se pone en la cola, creamos un objeto asyncio.Future. El trabajador luego establece el resultado o la excepción en este futuro, y el llamador awaite esto.

Integrando con la Lógica de Tu Bot

Ahora, ¿cómo usas esto en tu bot? Imaginemos un bot de Telegram simple (usando python-telegram-bot o una biblioteca similar) que necesita obtener detalles de productos de nuestra problemática API de proveedor.

# Suponiendo que tienes una función asíncrona para hacer la llamada real a la API
# Esta función debería lanzar una excepción en caso de errores de API (por ejemplo, 429, 500 o datos mal formados)
async def fetch_product_details_from_supplier_api(product_id: str):
 # Simular una llamada real a la API con limitación de tasa impredecible
 # En un escenario real, esto utilizaría aiohttp o requests_async
 await asyncio.sleep(0.05) # Simular la latencia de red
 
 # Simular un fallo ocasional de la API (por ejemplo, 429 Demasiadas Solicitudes)
 # Ajustar la probabilidad para probar diferentes escenarios
 if random.random() < 0.2: # 20% de probabilidad de fallo
 if random.random() < 0.5:
 raise Exception(f"Error de API: Producto {product_id} - Demasiadas Solicitudes (Simulado 429)")
 else:
 raise Exception(f"Error de API: Producto {product_id} - Error Interno del Servidor (Simulado 500)")
 
 # Simular la recuperación exitosa de datos
 return {"id": product_id, "name": f"Gadget Asombroso {product_id}", "price": random.randint(10, 100)}

# --- Integración del Bot ---
import random
# from telegram import Update
# from telegram.ext import Application, CommandHandler, ContextTypes

# Inicializar nuestra cola adaptativa
product_api_queue = ApiRequestQueue(fetch_product_details_from_supplier_api, initial_delay=0.2)

async def start_queue():
 await product_api_queue.start()

async def stop_queue():
 await product_api_queue.stop()

async def get_product_info_command(product_id: str): # Simplificado para demostración, sería `update: Update, context: ContextTypes.DEFAULT_TYPE`
 # Aquí es donde tu manejador de comandos del bot llamaría a la cola
 try:
 product_data = await product_api_queue.put(product_id)
 # await update.message.reply_text(f"Producto {product_data['name']}: ${product_data['price']}")
 print(f"El bot recibió información para {product_id}: {product_data['name']}")
 return product_data
 except Exception as e:
 # await update.message.reply_text(f"Lo siento, no pudo obtener información del producto en este momento. Por favor, intenta de nuevo más tarde. Error: {e}")
 print(f"El bot falló al obtener información para {product_id}: {e}")
 return None

# Ejemplo de uso (sin configuración real del bot de Telegram por brevedad)
async def main():
 await start_queue()

 # Simular múltiples solicitudes concurrentes de usuarios
 print("\n--- Enviando un estallido de 10 solicitudes ---")
 tasks = []
 for i in range(1, 11):
 tasks.append(get_product_info_command(f"PROD-{i}"))
 
 results = await asyncio.gather(*tasks)
 print(f"\nTodas las solicitudes procesadas. Resultados: {len([r for r in results if r is not None])} exitosas.")

 print("\n--- Enviando otro estallido después de una breve pausa ---")
 await asyncio.sleep(3) # Simular una pausa en la actividad del usuario
 tasks_2 = []
 for i in range(11, 16):
 tasks_2.append(get_product_info_command(f"PROD-{i}"))
 
 await asyncio.gather(*tasks_2)

 await stop_queue()

if __name__ == "__main__":
 asyncio.run(main())

En la función get_product_info_command, en lugar de llamar directamente a fetch_product_details_from_supplier_api, ahora llamamos a await product_api_queue.put(product_id). Esto significa que los manejadores de comandos de nuestro bot no tienen que preocuparse por los límites de tasa; simplemente envían su solicitud a la cola y esperan el resultado. La cola maneja todo el retroceso y la reintentos (aunque para esta implementación específica, solo retrasa las solicitudes posteriores, no vuelve a intentar las que fallaron directamente; podrías añadir un mecanismo de reintento dentro de _worker si fuera necesario).

Mejora: Manejo de Diferentes Tipos de Errores

La API del proveedor de mi cliente era particularmente complicada. A veces devolvía un 429, otras veces un 500, y a veces solo un arreglo JSON vacío si estaba sobrecargada. La implementación actual trata todas las excepciones por igual. Para un sistema más sofisticado, podrías querer diferenciar:

• Errores Temporales (429, 503, tiempos de espera en conexión): Aumentar el retraso, potencialmente reintentar la misma solicitud un par de veces antes de rendirse.
• Errores Permanentes (400, 401, 404): Esto usualmente significa que la solicitud en sí es incorrecta, o que la autenticación falló. No aumentes el retraso; simplemente falla la solicitud específica inmediatamente.

Puedes lograr esto modificando tu api_call_func para capturar códigos de estado HTTP específicos y lanzar diferentes excepciones personalizadas, y luego tu _worker puede tener bloques `except` más granulares.

Para mi cliente actual, dada la inestabilidad general de la API, tratar la mayoría de errores como “necesitamos retroceder” fue la apuesta más segura. Priorizó la estabilidad sobre la identificación inmediata de errores, lo cual fue un buen intercambio en ese escenario específico.

Lecciones Prácticas para Tu Próximo Proyecto de Bot

1. No confíes en “uso razonable”: Asume que cualquier API de terceros sin cabeceras de límites de tasa explícitos fallará bajo carga. Planea para ello desde el primer día.
2. Centrar las llamadas a la API: Dirige todas las solicitudes a una API de terceros específica a través de una sola cola o un servicio dedicado. Esto facilita la gestión de los límites.
3. Implementa retroceso adaptativo: En lugar de retrasos fijos, crea un sistema que reaccione a los fallos de la API desacelerando y acelerando cuando la API es receptiva. El retroceso exponencial es tu amigo aquí.
4. Usa programación asíncrona: Para bots, especialmente, asyncio en Python es invaluable. Permite que tu bot siga siendo receptivo a la entrada del usuario mientras espera a que las llamadas a la API se completen (o se acumulen).
5. Monitorea y registra: Registra cuando tu cola aumenta retrasos y cuando se recupera. Esto te brinda información vital sobre el comportamiento de la API y te ayuda a ajustar tu delay_factor y success_factor. Normalmente conecto los registros de mi bot a Grafana o a una herramienta de monitoreo similar para visualizar estas tendencias.
6. Considera límites por usuario: Si tu bot hace llamadas en nombre de usuarios individuales (por ejemplo, cada usuario tiene su propia clave de API), es posible que necesites una cola separada o un limitador de tasa para cada usuario para evitar que un usuario agote la asignación de otro. Este es un tema más avanzado pero que vale la pena considerar.

Construir bots resilientes trata de anticipar fallos. Cuanto más graciosamente tu bot pueda manejar interrupciones o limitaciones del servicio externo, mejor será la experiencia del usuario, y menos cabello perderás depurando errores “aleatorios”. Esta cola de solicitudes adaptativa me ha ahorrado innumerables dolores de cabeza y ha permitido que el bot de Telegram de mi cliente funcione sin problemas, incluso con la antigua y temperamental API de su proveedor.

Prueba este enfoque en tu próximo proyecto de bot y házmelo saber cómo te va en los comentarios abajo. ¿Hay otras estrategias que uses para abordar los límites impredecibles de la API? Siempre estoy interesado en aprender.

Artículos Relacionados

• Tendencias de IA Conversacional 2026: El Futuro de los Chatbots
• Diseño de Conversación: Creando Diálogos Atractivos y Naturales
• ¿Cómo Funcionan los Chatbots en el Comercio Electrónico?

🕒 Published: March 25, 2026