RAG y Circuit Breakers: Cuando Gemini dice basta

May 19, 2026
4 min read
Table of Contents

Hace poco estuve trabajando en SaberPro-AI-Assistant, un sistema de Retrieval-Augmented Generation (RAG) en Python y FastAPI diseñado para resolver dudas de estudiantes universitarios colombianos sobre las pruebas de estado Saber Pro y TyT.

No es un proyecto masivo de escala corporativa. No tiene un presupuesto ilimitado. Se ejecuta sobre contenedores ligeros y utiliza ChromaDB local y la API gratuita de Google Gemini para las respuestas generativas.

Justo por eso me encontré con un problema clásico que se repite constantemente cuando dependes de servicios externos:

Los límites de cuota son reales. Y si no los modelas en tu código, tu experiencia de usuario se romperá en el primer minuto de tráfico real.

El Choque Contra la Pared del Free Tier

La capa gratuita de las APIs de LLMs es un excelente punto de partida, pero tiene reglas del juego bastante estrictas. Por ejemplo, límites severos de solicitudes por minuto (RPM) y solicitudes por día (RPD).

En nuestro entorno, al ejecutar un script automatizado de evaluación (benchmark.py) para medir métricas de recuperación y generación sobre 25 preguntas complejas seguidas, chocamos de frente contra la pared:

google.api_core.exceptions.ResourceExhausted: 429 You exceeded your current quota...
* Quota exceeded for metric: generativelanguage.googleapis.com/generate_content_free_tier_requests, limit: 20, model: gemini-3-flash

En la telemetría del dashboard de Google AI Studio, las gráficas eran claras y apiladas:

  • Gemini 2.5 Flash: 8 / 5 RPM | 24 / 20 RPD (Cuota diaria superada)
  • Gemini 3 Flash: 8 / 5 RPM | 26 / 20 RPD (Cuota diaria superada)

El Fallo Clásico: La Latencia Acumulativa

La primera forma de solucionar esto suele ser un bucle básico de fallback:

  • Si el Modelo A falla con un error, intenta con el Modelo B.
  • Si el Modelo B falla, intenta con el Modelo C.

El problema con este enfoque síncrono clásico es que no tiene memoria.

Si el Modelo A ya agotó su cuota del día, la siguiente solicitud del usuario volverá a intentar llamar al Modelo A, esperará a que el servidor de Google responda con un error de tipo 429 (lo cual toma tiempo de red), y luego pasará al Modelo B.

Al correr nuestro benchmark bajo este esquema, la latencia promedio de generación se disparó a 8.29 segundos. Para el usuario final, el chat se sentía congelado, pesado e inusable, a pesar de que el sistema “técnicamente” no se caía gracias al try-except.

La Solución: Un Circuit Breaker con Memoria (Model Cooldown)

Para solucionar esto de raíz, implementamos un patrón de diseño clásico de sistemas distribuidos: el Circuit Breaker (o Disyuntor), adaptado al consumo de APIs de inteligencia artificial.

La idea es simple: si un modelo te dice que está agotado, no lo vuelvas a molestar durante un tiempo.

Añadimos un registro dinámico de estados de cooldown en nuestro servicio de lenguaje:

services/gemini_service.py
class GeminiLLMService(BaseLLMService):
    def __init__(self, ...):
        # Registro para rastrear períodos de cooldown activos por modelo
        self.model_cooldowns = {}
 
    def _is_on_cooldown(self, model_name: str) -> bool:
        cooldown_until = self.model_cooldowns.get(model_name, 0.0)
        return time.time() < cooldown_until

Antes de realizar cualquier llamada a la API, el servicio filtra la lista de modelos configurados utilizando este helper:

services/gemini_service.py
def _get_active_models(self) -> list:
    models_to_try = [self.model_name] + self.fallback_models
    # Filtrar y omitir de inmediato cualquier modelo que esté en cooldown
    active_models = [m for m in models_to_try if not self._is_on_cooldown(m)]
    
    # Prevención de bloqueos mutuos (Deadlock)
    if not active_models:
        logger.warning("Todos los modelos están en cooldown. Intentando con todos.")
        return models_to_try
        
    return active_models

Y si atrapamos una excepción de tipo ResourceExhausted (HTTP 429), le ponemos una “penalización” de enfriamiento de 60 segundos a ese modelo específico:

services/gemini_service.py
def _apply_cooldown(self, model_name: str, exception: Exception):
    ex_name = type(exception).__name__
    ex_msg = str(exception)
    
    if "ResourceExhausted" in ex_name or "429" in ex_msg or "quota" in ex_msg.lower():
        self.model_cooldowns[model_name] = time.time() + 60.0 # 60 segundos

Los Resultados: Caída del 90% en la Latencia

Los números de la verificación bajo saturación extrema demostraron la efectividad inmediata del patrón:

  • Latencia promedio de generación antes: 8.29 segundos por consulta.
  • Latencia promedio de generación después: 0.93 segundos por consulta.
  • Reducción de latencia: ~88.7% de optimización de tiempo de respuesta.

El sistema dejó de hacer viajes de red inútiles a endpoints agotados. En cuanto un modelo avisaba que su cuota estaba llena, las siguientes consultas pasaban instantáneamente al siguiente modelo disponible de forma transparente para el frontend.

Lecciones para el Mundo Real

Este refactor técnico me dejó un par de conclusiones aplicables a cualquier desarrollo moderno con IA:

  1. La resiliencia no es solo atrapar errores: Es diseñar lógica que aprenda del estado de tus dependencias en tiempo de ejecución.
  2. Las cuotas son parte del flujo: No trates las cuotas límites de las APIs externas como “casos extremos raros”. En producción, son la regla, no la excepción.
  3. Optimización invisible: A veces la mejor optimización de rendimiento no consiste en escribir algoritmos más rápidos, sino en saber cuándo no hacer una petición de red.

Escribir software útil no es solo construir el camino feliz donde las APIs responden en milisegundos y son gratuitas. Es blindar el sistema para que, cuando las cosas fallen afuera, tus usuarios ni siquiera se enteren.