Phase 11 - Lesson 11

Caché, Limitación de Tasa y Optimización de Costos

This lesson includes a graded coding exercise that runs in your browser, unlocked with lifetime access.

La mayoría de las startups de IA no mueren por malos modelos. Mueren por una mala economía unitaria. Una sola llamada a GPT-4o cuesta fracciones de centavo. Diez mil usuarios realizando diez llamadas por día cuesta

50 solo en tokens de entrada -- antes de que cobres un solo dólar. Las empresas que sobreviven son las que tratan cada llamada a la API como una transacción financiera, no como una llamada a una función.

Tipo: Build Lenguajes: Python Prerrequisitos: Phase 11 Lesson 09 (Function Calling) Tiempo: ~45 minutos Relacionado: Phase 11 · 15 (Prompt Caching) — esta lección cubre el almacenamiento en caché a nivel de aplicación (caché semántico, caché de hash exacto, enrutamiento de modelos). La Lección 15 cubre el almacenamiento en caché de prompts a nivel de proveedor (cache_control de Anthropic, automático de OpenAI, CachedContent de Gemini). Combina ambos para una reducción de costos del 50-95%.

Objetivos de Aprendizaje

El Problema

Construyes un chatbot de RAG. Funciona de maravilla. A los usuarios les encanta.

Entonces llega la factura.

GPT-5 cuesta $5 por millón de tokens de entrada y

5 por millón de salida. Claude Opus 4.7 cuesta
5 de entrada / $75 de salida. Gemini 3 Pro cuesta
.25 de entrada / $5 de salida. GPT-5-mini cuesta $0.25/
. Los precios a continuación son ilustrativos; siempre verifica la página de precios actual del proveedor.

Aquí está la matemática que mata a las startups:

Costo diario de entrada: 10,000 x 10 x 1,000 / 1,000,000 x

.50 =
50/día Costo diario de salida: 10,000 x 10 x 500 / 1,000,000 x
0.00 = $500/día Total mensual:
2,500/mes

Eso es solo el LLM. Agrega embeddings, alojamiento de bases de datos vectoriales, infraestructura. Estás ante un costo de $30,000/mes por un chatbot.

La parte brutal: el 40-60% de esas consultas son casi duplicadas. Los usuarios hacen las mismas preguntas con palabras ligeramente diferentes. Tu prompt del sistema -- idéntico en cada solicitud -- se factura cada vez. Los documentos de contexto recuperados por RAG se repiten entre usuarios que preguntan sobre el mismo tema.

Estás pagando el precio completo por computación redundante.

El Concepto

La Anatomía de Costos de una Llamada de LLM

Cada llamada a la API tiene cinco componentes de costo.

graph LR
    A[Consulta del Usuario] --> B[Prompt del Sistema<br/>500-2000 tokens]
    A --> C[Contexto Recuperado<br/>500-4000 tokens]
    A --> D[Mensaje del Usuario<br/>50-500 tokens]
    B --> E[Costo de Entrada<br/>
  .50/1M tokens]
    C --> E
    D --> E
    E --> F[Procesamiento del Modelo]
    F --> G[Costo de Saída<br/>
0.00/1M tokens]

Los prompts del sistema son el asesino silencioso. Un prompt del sistema de 1,500 tokens enviado con cada solicitud cuesta $3.75 por millón de solicitudes solo por ese prefijo. Con 100K solicitudes al día, eso equivale a $375/día --

1,250/mes -- por texto que nunca cambia.

Caché del Proveedor: Descuentos Integrados

Los tres proveedores principales ofrecen almacenamiento en caché de prompts en el lado del proveedor en 2026, pero la mecánica difiere. Consulta Phase 11 · 15 para un análisis detallado.

Proveedor Mecanismo Descuento Mínimo Duración del Caché
Anthropic Marcadores explícitos cache_control 90% en aciertos de caché (se paga un 25% extra en la escritura) 1,024 tokens (Sonnet/Opus), 2,048 (Haiku) 5 min por defecto; 1h extendido (2x prima de escritura)
OpenAI Coincidencia automática de prefijo 50% en aciertos de caché 1,024 tokens Mejor esfuerzo de hasta 1 hora
Google Gemini API CachedContent explícita ~75% de reducción (más almacenamiento) 4,096 (Flash) / 32,768 (Pro) TTL configurable por el usuario

El enfoque de Anthropic es explícito. Marcas secciones de tu prompt con cache_control: {"type": "ephemeral"}. La primera solicitud paga una prima de escritura del 25%. Las solicitudes subsiguientes con el mismo prefijo obtienen un 90% de descuento. Un prompt del sistema de 2,000 tokens que normalmente cuesta $0.005 cuesta $0.000625 en aciertos de caché. Con 100K solicitudes, eso ahorra $437.50/día.

El enfoque de OpenAI es automático. Cualquier prefijo de prompt que coincida con una solicitud anterior obtiene un 50% de descuento. No se necesitan marcadores. El compromiso: menor descuento, menor control, pero cero esfuerzo de implementación.

Caché Semántico: Tu Capa Personalizada

El almacenamiento en caché del proveedor solo funciona para prefijos idénticos. El caché semántico maneja el caso más difícil: diferentes consultas con el mismo significado.

"¿Cuál es la política de devoluciones?" y "¿Cómo devuelvo un artículo?" son cadenas diferentes pero con la misma intención. Un caché semántico genera embeddings de ambas consultas, calcula la similitud de coseno y devuelve la respuesta almacenada en caché si la similitud supera un umbral (normalmente 0.92-0.95).

flowchart TD
    A[Consulta del Usuario] --> B[Generar Embedding de Consulta]
    B --> C{¿Consulta similar<br/>en caché?}
    C -->|sim > 0.95| D[Devolver Respuesta Almacenada]
    C -->|sim < 0.95| E[Llamar API del LLM]
    E --> F[Guardar Respuesta en Caché<br/>con Embedding]
    F --> G[Devolver Respuesta]
    D --> G

Los costos de embedding son insignificantes. El text-embedding-3-small de OpenAI cuesta $0.02 por millón de tokens. Verificar el caché no cuesta casi nada en comparación con una llamada completa de LLM.

Caché Exacto: Hash y Coincidencia

Para llamadas deterministas (temperatura=0, mismo modelo, mismo prompt), el caché exacto es más simple y rápido. Genera un hash del prompt completo, verifica el caché y devuélvelo si lo encuentras.

Esto funciona perfectamente para:

Limitación de Tasa: Protegiendo tu Presupuesto

La limitación de tasa no es solo por justicia. Es por supervivencia.

Algoritmo de cubeta de tokens (token bucket): cada usuario obtiene una cubeta de N tokens que se rellena a una tasa R por segundo. Una solicitud consume tokens de la cubeta. Si la cubeta está vacía, la solicitud se rechaza. Esto permite ráfagas (usar toda la cubeta a la vez) mientras se impone una tasa promedio.

Cuotas por usuario: establece límites diarios/mensuales de tokens por nivel de usuario.

Nivel Límite Diario de Tokens Máx Solicitudes/min Acceso a Modelos
Gratis 50,000 10 Solo GPT-4o-mini
Pro 500,000 60 GPT-4o, Claude Sonnet
Enterprise 5,000,000 300 Todos los modelos

Enrutamiento de Modelos: El Modelo Concreto para el Trabajo Concreto

No todas las consultas necesitan GPT-4o.

"¿A qué hora cierra la tienda?" no requiere un modelo de

0/M de salida. GPT-4o-mini a $0.60/M de salida lo maneja perfectamente. Claude Haiku a
.25/M de salida también. Un clasificador simple enruta las consultas baratas a modelos baratos y las consultas complejas a modelos caros.

flowchart TD
    A[Consulta del Usuario] --> B[Clasificador de Complejidad]
    B -->|Simple: búsqueda, FAQ| C[GPT-4o-mini<br/>$0.15/$0.60 por 1M]
    B -->|Media: análisis, resumen| D[Claude Sonnet<br/>$3.00/
5.00 por 1M]
    B -->|Compleja: razonamiento, código| E[GPT-4o / Claude Opus<br/>

  .50/
0.00+]

Un enrutador bien ajustado ahorra entre un 40 y un 70% solo en costos de modelos.

Seguimiento de Costos: Sabe a Dónde se Va el Dinero

No puedes optimizar lo que no mides. Registra cada llamada a la API con:

  • Marca de tiempo
  • Nombre del modelo
  • Tokens de entrada
  • Tokens de salida
  • Latencia (ms)
  • Costo calculado ($)
  • ID de usuario
  • Acierto/fallo de caché (cache hit/miss)
  • Categoría de solicitud

Estos datos revelan qué características son costosas, qué usuarios son grandes consumidores y dónde tiene mayor impacto el caché.

Procesamiento por Lotes (Batching): Descuentos por Volumen

La API Batch de OpenAI procesa solicitudes de forma asíncrona con un 50% de descuento. Envías un lote de hasta 50,000 solicitudes y los resultados regresan en un plazo de 24 horas.

Usa el procesamiento por lotes para:

  • Procesamiento nocturno de documentos
  • Clasificación masiva
  • Ejecuciones de evaluación
  • Pipelines de enriquecimiento de datos

No lo uses para: consultas en tiempo real de cara al usuario (la latencia importa).

Alertas de Presupuesto e Interruptores Automáticos (Circuit Breakers)

Un interruptor automático detiene el gasto cuando alcanzas un límite. Sin uno, un error o abuso puede consumir tu presupuesto mensual en cuestión de horas.

Establece tres umbrales:

  1. Advertencia (70% del presupuesto): envía una alerta
  2. Restricción (85% del presupuesto): cambia solo a modelos más baratos
  3. Detener (95% del presupuesto): rechaza nuevas solicitudes, devuelve solo respuestas en caché

El Stack de Optimización

Aplica estas técnicas en orden. Cada capa se acumula sobre las anteriores.

Capa Técnica Ahorro Típico Esfuerzo de Implementación
1 Caché de prompt del proveedor 30-50% Bajo (agregar marcadores de caché)
2 Caché exacto 10-20% Bajo (hash + dict)
3 Caché semántico 15-30% Medio (embeddings + similitud)
4 Enrutamiento de modelos 40-70% Medio (clasificador)
5 Limitación de tasa Protección de presupuesto Bajo (cubeta de tokens)
6 Compresión de prompt 10-30% Medio (reescribir prompts)
7 Procesamiento por lotes 50% en solicitudes elegibles Bajo (API de lotes)

Una aplicación RAG que aplica las capas 1-5 normalmente reduce los costos de

2,500/mes a $4,000-6,000/mes. Esa es la diferencia entre agotar los fondos de inversión y construir un negocio.

Ahorros Reales: Antes y Después

Aquí hay un desglose real para un chatbot RAG que atiende a 10,000 DAU.

Métrica Antes de la Optimización Después de la Optimización Ahorro
Costo mensual de LLM 2,500 $5,200 77%
Costo promedio por consulta $0.0075 $0.0017 77%
Tasa de aciertos de caché 0% 52% --
Consultas enrutadas a mini 0% 65% --
Latencia P95 2,800ms 900ms (aciertos de caché: 50ms) 68%
Costo mensual de embeddings $0
80
 (nuevo costo)
Costo mensual total 2,500 $5,380 76%

El costo de embedding para el caché semántico (

80/mes) se amortiza en la primera hora de aciertos de caché.

Constrúyelo

Paso 1: Calculadora de Costos

Construye una calculadora de costos de tokens que conozca los precios antiguos/actuales de los modelos principales.

import hashlib
import time
import json
import math
from dataclasses import dataclass, field


MODEL_PRICING = {
    "gpt-4o": {"input": 2.50, "output": 10.00, "cached_input": 1.25},
    "gpt-4o-mini": {"input": 0.15, "output": 0.60, "cached_input": 0.075},
    "gpt-4.1": {"input": 2.00, "output": 8.00, "cached_input": 0.50},
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60, "cached_input": 0.10},
    "gpt-4.1-nano": {"input": 0.10, "output": 0.40, "cached_input": 0.025},
    "o3": {"input": 2.00, "output": 8.00, "cached_input": 0.50},
    "o3-mini": {"input": 1.10, "output": 4.40, "cached_input": 0.55},
    "o4-mini": {"input": 1.10, "output": 4.40, "cached_input": 0.275},
    "claude-opus-4": {"input": 15.00, "output": 75.00, "cached_input": 1.50},
    "claude-sonnet-4": {"input": 3.00, "output": 15.00, "cached_input": 0.30},
    "claude-haiku-3.5": {"input": 0.80, "output": 4.00, "cached_input": 0.08},
    "gemini-2.5-pro": {"input": 1.25, "output": 10.00, "cached_input": 0.3125},
    "gemini-2.5-flash": {"input": 0.15, "output": 0.60, "cached_input": 0.0375},
}


def calculate_cost(model, input_tokens, output_tokens, cached_input_tokens=0):
    if model not in MODEL_PRICING:
        return {"error": f"Unknown model: {model}"}
    pricing = MODEL_PRICING[model]
    non_cached = input_tokens - cached_input_tokens
    input_cost = (non_cached / 1_000_000) * pricing["input"]
    cached_cost = (cached_input_tokens / 1_000_000) * pricing["cached_input"]
    output_cost = (output_tokens / 1_000_000) * pricing["output"]
    total = input_cost + cached_cost + output_cost
    return {
        "model": model,
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "cached_input_tokens": cached_input_tokens,
        "input_cost": round(input_cost, 6),
        "cached_input_cost": round(cached_cost, 6),
        "output_cost": round(output_cost, 6),
        "total_cost": round(total, 6),
    }

Paso 2: Caché Exacto

Genera el hash del prompt completo y devuelve respuestas almacenadas en caché para solicitudes idénticas.

class ExactCache:
    def __init__(self, max_size=1000, ttl_seconds=3600):
        self.cache = {}
        self.max_size = max_size
        self.ttl = ttl_seconds
        self.hits = 0
        self.misses = 0

    def _hash(self, model, messages, temperature):
        key_data = json.dumps({"model": model, "messages": messages, "temperature": temperature}, sort_keys=True)
        return hashlib.sha256(key_data.encode()).hexdigest()

    def get(self, model, messages, temperature=0.0):
        if temperature > 0:
            self.misses += 1
            return None
        key = self._hash(model, messages, temperature)
        if key in self.cache:
            entry = self.cache[key]
            if time.time() - entry["timestamp"] < self.ttl:
                self.hits += 1
                entry["access_count"] += 1
                return entry["response"]
            del self.cache[key]
        self.misses += 1
        return None

    def put(self, model, messages, temperature, response):
        if temperature > 0:
            return
        if len(self.cache) >= self.max_size:
            oldest_key = min(self.cache, key=lambda k: self.cache[k]["timestamp"])
            del self.cache[oldest_key]
        key = self._hash(model, messages, temperature)
        self.cache[key] = {
            "response": response,
            "timestamp": time.time(),
            "access_count": 1,
        }

    def stats(self):
        total = self.hits + self.misses
        return {
            "hits": self.hits,
            "misses": self.misses,
            "hit_rate": round(self.hits / total, 4) if total > 0 else 0,
            "cache_size": len(self.cache),
        }

Paso 3: Caché Semántico

Genera embeddings de las consultas y devuelve respuestas almacenadas en caché cuando la similitud supera un umbral.

def simple_embed(text):
    words = text.lower().split()
    vocab = {}
    for w in words:
        vocab[w] = vocab.get(w, 0) + 1
    norm = math.sqrt(sum(v * v for v in vocab.values()))
    if norm == 0:
        return {}
    return {k: v / norm for k, v in vocab.items()}


def cosine_similarity(a, b):
    if not a or not b:
        return 0.0
    all_keys = set(a) | set(b)
    dot = sum(a.get(k, 0) * b.get(k, 0) for k in all_keys)
    return dot


class SemanticCache:
    def __init__(self, similarity_threshold=0.85, max_size=500, ttl_seconds=3600):
        self.entries = []
        self.threshold = similarity_threshold
        self.max_size = max_size
        self.ttl = ttl_seconds
        self.hits = 0
        self.misses = 0

    def get(self, query):
        query_embedding = simple_embed(query)
        now = time.time()
        best_match = None
        best_sim = 0.0
        for entry in self.entries:
            if now - entry["timestamp"] > self.ttl:
                continue
            sim = cosine_similarity(query_embedding, entry["embedding"])
            if sim > best_sim:
                best_sim = sim
                best_match = entry
        if best_match and best_sim >= self.threshold:
            self.hits += 1
            best_match["access_count"] += 1
            return {"response": best_match["response"], "similarity": round(best_sim, 4), "original_query": best_match["query"]}
        self.misses += 1
        return None

    def put(self, query, response):
        if len(self.entries) >= self.max_size:
            self.entries.sort(key=lambda e: e["timestamp"])
            self.entries.pop(0)
        self.entries.append({
            "query": query,
            "embedding": simple_embed(query),
            "response": response,
            "timestamp": time.time(),
            "access_count": 1,
        })

    def stats(self):
        total = self.hits + self.misses
        return {
            "hits": self.hits,
            "misses": self.misses,
            "hit_rate": round(self.hits / total, 4) if total > 0 else 0,
            "cache_size": len(self.entries),
        }

Paso 4: Limitador de Tasa

Limitador de tasa por cubeta de tokens con cuotas por usuario.

class TokenBucketRateLimiter:
    def __init__(self):
        self.buckets = {}
        self.tiers = {
            "free": {"capacity": 50_000, "refill_rate": 500, "max_requests_per_min": 10},
            "pro": {"capacity": 500_000, "refill_rate": 5_000, "max_requests_per_min": 60},
            "enterprise": {"capacity": 5_000_000, "refill_rate": 50_000, "max_requests_per_min": 300},
        }

    def _get_bucket(self, user_id, tier="free"):
        if user_id not in self.buckets:
            tier_config = self.tiers.get(tier, self.tiers["free"])
            self.buckets[user_id] = {
                "tokens": tier_config["capacity"],
                "capacity": tier_config["capacity"],
                "refill_rate": tier_config["refill_rate"],
                "last_refill": time.time(),
                "request_timestamps": [],
                "max_rpm": tier_config["max_requests_per_min"],
                "tier": tier,
                "total_tokens_used": 0,
            }
        return self.buckets[user_id]

    def _refill(self, bucket):
        now = time.time()
        elapsed = now - bucket["last_refill"]
        refill = int(elapsed * bucket["refill_rate"])
        if refill > 0:
            bucket["tokens"] = min(bucket["capacity"], bucket["tokens"] + refill)
            bucket["last_refill"] = now

    def check(self, user_id, tokens_needed, tier="free"):
        bucket = self._get_bucket(user_id, tier)
        self._refill(bucket)
        now = time.time()
        bucket["request_timestamps"] = [t for t in bucket["request_timestamps"] if now - t < 60]
        if len(bucket["request_timestamps"]) >= bucket["max_rpm"]:
            return {"allowed": False, "reason": "rate_limit", "retry_after_seconds": 60 - (now - bucket["request_timestamps"][0])}
        if bucket["tokens"] < tokens_needed:
            deficit = tokens_needed - bucket["tokens"]
            wait = deficit / bucket["refill_rate"]
            return {"allowed": False, "reason": "token_limit", "tokens_available": bucket["tokens"], "retry_after_seconds": round(wait, 1)}
        return {"allowed": True, "tokens_available": bucket["tokens"]}

    def consume(self, user_id, tokens_used, tier="free"):
        bucket = self._get_bucket(user_id, tier)
        bucket["tokens"] -= tokens_used
        bucket["request_timestamps"].append(time.time())
        bucket["total_tokens_used"] += tokens_used

    def get_usage(self, user_id):
        if user_id not in self.buckets:
            return {"error": "User not found"}
        b = self.buckets[user_id]
        return {
            "user_id": user_id,
            "tier": b["tier"],
            "tokens_remaining": b["tokens"],
            "capacity": b["capacity"],
            "total_tokens_used": b["total_tokens_used"],
            "utilization": round(b["total_tokens_used"] / b["capacity"], 4) if b["capacity"] else 0,
        }

Paso 5: Rastreador de Costos

Registra cada llamada y calcula los totales acumulados.

class CostTracker:
    def __init__(self, monthly_budget=1000.0):
        self.logs = []
        self.monthly_budget = monthly_budget
        self.alerts = []

    def log_call(self, model, input_tokens, output_tokens, cached_input_tokens=0, latency_ms=0, user_id="anonymous", cache_status="miss"):
        cost = calculate_cost(model, input_tokens, output_tokens, cached_input_tokens)
        entry = {
            "timestamp": time.time(),
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cached_input_tokens": cached_input_tokens,
            "latency_ms": latency_ms,
            "cost": cost["total_cost"],
            "user_id": user_id,
            "cache_status": cache_status,
        }
        self.logs.append(entry)
        self._check_budget()
        return entry

    def _check_budget(self):
        total = self.total_cost()
        pct = total / self.monthly_budget if self.monthly_budget > 0 else 0
        if pct >= 0.95 and not any(a["level"] == "stop" for a in self.alerts):
            self.alerts.append({"level": "stop", "message": f"Budget 95% consumed: ${total:.2f}/${self.monthly_budget:.2f}", "timestamp": time.time()})
        elif pct >= 0.85 and not any(a["level"] == "throttle" for a in self.alerts):
            self.alerts.append({"level": "throttle", "message": f"Budget 85% consumed: ${total:.2f}/${self.monthly_budget:.2f}", "timestamp": time.time()})
        elif pct >= 0.70 and not any(a["level"] == "warning" for a in self.alerts):
            self.alerts.append({"level": "warning", "message": f"Budget 70% consumed: ${total:.2f}/${self.monthly_budget:.2f}", "timestamp": time.time()})

    def total_cost(self):
        return round(sum(e["cost"] for e in self.logs), 6)

    def cost_by_model(self):
        by_model = {}
        for e in self.logs:
            m = e["model"]
            if m not in by_model:
                by_model[m] = {"calls": 0, "cost": 0, "input_tokens": 0, "output_tokens": 0}
            by_model[m]["calls"] += 1
            by_model[m]["cost"] = round(by_model[m]["cost"] + e["cost"], 6)
            by_model[m]["input_tokens"] += e["input_tokens"]
            by_model[m]["output_tokens"] += e["output_tokens"]
        return by_model

    def cache_savings(self):
        cache_hits = [e for e in self.logs if e["cache_status"] == "hit"]
        if not cache_hits:
            return {"saved": 0, "cache_hits": 0}
        saved = 0
        for e in cache_hits:
            full_cost = calculate_cost(e["model"], e["input_tokens"], e["output_tokens"])
            saved += full_cost["total_cost"]
        return {"saved": round(saved, 4), "cache_hits": len(cache_hits)}

    def summary(self):
        if not self.logs:
            return {"total_calls": 0, "total_cost": 0}
        total_latency = sum(e["latency_ms"] for e in self.logs)
        cache_hits = sum(1 for e in self.logs if e["cache_status"] == "hit")
        return {
            "total_calls": len(self.logs),
            "total_cost": self.total_cost(),
            "avg_cost_per_call": round(self.total_cost() / len(self.logs), 6),
            "avg_latency_ms": round(total_latency / len(self.logs), 1),
            "cache_hit_rate": round(cache_hits / len(self.logs), 4),
            "cost_by_model": self.cost_by_model(),
            "cache_savings": self.cache_savings(),
            "budget_remaining": round(self.monthly_budget - self.total_cost(), 2),
            "budget_utilization": round(self.total_cost() / self.monthly_budget, 4) if self.monthly_budget > 0 else 0,
            "alerts": self.alerts,
        }

Paso 3: Enrutador de Modelos

Enruta las consultas al modelo más barato que pueda manejarlas.

SIMPLE_KEYWORDS = ["what time", "hours", "address", "phone", "price", "return policy", "hello", "hi", "thanks", "yes", "no"]
COMPLEX_KEYWORDS = ["analyze", "compare", "explain why", "write code", "debug", "architect", "design", "trade-off", "evaluate"]


def classify_complexity(query):
    q = query.lower()
    if len(q.split()) <= 5 or any(kw in q for kw in SIMPLE_KEYWORDS):
        return "simple"
    if any(kw in q for kw in COMPLEX_KEYWORDS):
        return "complex"
    return "medium"


def route_model(query, tier="pro"):
    complexity = classify_complexity(query)
    routing_table = {
        "simple": {"free": "gpt-4.1-nano", "pro": "gpt-4o-mini", "enterprise": "gpt-4o-mini"},
        "medium": {"free": "gpt-4o-mini", "pro": "claude-sonnet-4", "enterprise": "claude-sonnet-4"},
        "complex": {"free": "gpt-4o-mini", "pro": "gpt-4o", "enterprise": "claude-opus-4"},
    }
    model = routing_table[complexity].get(tier, "gpt-4o-mini")
    return {"query": query, "complexity": complexity, "model": model, "tier": tier}

Paso 7: Ejecutar la Demostración

def simulate_llm_call(model, query):
    input_tokens = len(query.split()) * 4 + 500
    output_tokens = 150 + (len(query.split()) * 2)
    latency = 200 + (output_tokens * 2)
    return {
        "model": model,
        "response": f"[Simulated {model} response to: {query[:50]}...]",
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "latency_ms": latency,
    }


def run_demo():
    print("=" * 60)
    print("  Caching, Rate Limiting & Cost Optimization Demo")
    print("=" * 60)

    print("\n--- Model Pricing ---")
    for model, pricing in list(MODEL_PRICING.items())[:6]:
        cost_1k = calculate_cost(model, 1000, 500)
        print(f"  {model}: ${cost_1k['total_cost']:.6f} per 1K in + 500 out")

    print("\n--- Cost Comparison: 100K Requests ---")
    for model in ["gpt-4o", "gpt-4o-mini", "claude-sonnet-4", "claude-haiku-3.5"]:
        cost = calculate_cost(model, 1000 * 100_000, 500 * 100_000)
        print(f"  {model}: ${cost['total_cost']:.2f}")

    print("\n--- Anthropic Cache Savings ---")
    no_cache = calculate_cost("claude-sonnet-4", 2000, 500, 0)
    with_cache = calculate_cost("claude-sonnet-4", 2000, 500, 1500)
    saving = no_cache["total_cost"] - with_cache["total_cost"]
    print(f"  Without cache: ${no_cache['total_cost']:.6f}")
    print(f"  With 1500 cached tokens: ${with_cache['total_cost']:.6f}")
    print(f"  Savings per call: ${saving:.6f} ({saving/no_cache['total_cost']*100:.1f}%)")

    exact_cache = ExactCache(max_size=100, ttl_seconds=300)
    semantic_cache = SemanticCache(similarity_threshold=0.75, max_size=100)
    rate_limiter = TokenBucketRateLimiter()
    tracker = CostTracker(monthly_budget=100.0)

    print("\n--- Exact Cache ---")
    messages_1 = [{"role": "user", "content": "What is the return policy?"}]
    result = exact_cache.get("gpt-4o-mini", messages_1, 0.0)
    print(f"  First lookup: {'HIT' if result else 'MISS'}")
    exact_cache.put("gpt-4o-mini", messages_1, 0.0, "You can return items within 30 days.")
    result = exact_cache.get("gpt-4o-mini", messages_1, 0.0)
    print(f"  Second lookup: {'HIT' if result else 'MISS'} -> {result}")
    result = exact_cache.get("gpt-4o-mini", messages_1, 0.7)
    print(f"  With temp=0.7: {'HIT' if result else 'MISS (non-deterministic, skip cache)'}")
    print(f"  Stats: {exact_cache.stats()}")

    print("\n--- Semantic Cache ---")
    test_queries = [
        ("What is the return policy?", "Items can be returned within 30 days with receipt."),
        ("How do I return an item?", None),
        ("What are your store hours?", "We are open 9am-9pm Monday through Saturday."),
        ("When does the store open?", None),
        ("Tell me about quantum computing", "Quantum computers use qubits..."),
        ("Explain quantum mechanics", None),
    ]
    for query, response in test_queries:
        cached = semantic_cache.get(query)
        if cached:
            print(f"  '{query[:40]}' -> CACHE HIT (sim={cached['similarity']}, original='{cached['original_query'][:40]}')")
        elif response:
            semantic_cache.put(query, response)
            print(f"  '{query[:40]}' -> MISS (stored)")
        else:
            print(f"  '{query[:40]}' -> MISS (no match)")
    print(f"  Stats: {semantic_cache.stats()}")

    print("\n--- Rate Limiting ---")
    for i in range(12):
        check = rate_limiter.check("user_1", 1000, "free")
        if check["allowed"]:
            rate_limiter.consume("user_1", 1000, "free")
        status = "OK" if check["allowed"] else f"BLOCKED ({check['reason']})"
        if i < 5 or not check["allowed"]:
            print(f"  Request {i+1}: {status}")
    print(f"  Usage: {rate_limiter.get_usage('user_1')}")

    print("\n--- Model Routing ---")
    routing_queries = [
        "What time do you close?",
        "Summarize this quarterly earnings report",
        "Analyze the trade-offs between microservices and monoliths",
        "Hello",
        "Write code for a binary search tree with deletion",
    ]
    for q in routing_queries:
        route = route_model(q, "pro")
        print(f"  '{q[:50]}' -> {route['model']} ({route['complexity']})")

    print("\n--- Full Pipeline: Before vs After Optimization ---")
    queries = [
        "What is the return policy?",
        "How do I return something?",
        "What are your hours?",
        "When do you open?",
        "Explain the difference between TCP and UDP",
        "Compare TCP vs UDP protocols",
        "Hello",
        "What is your phone number?",
        "Write a Python function to sort a list",
        "Analyze the pros and cons of serverless architecture",
    ]

    print("\n  [Before: no caching, single model (gpt-4o)]")
    tracker_before = CostTracker(monthly_budget=1000.0)
    for q in queries:
        result = simulate_llm_call("gpt-4o", q)
        tracker_before.log_call("gpt-4o", result["input_tokens"], result["output_tokens"], latency_ms=result["latency_ms"], cache_status="miss")
    before = tracker_before.summary()
    print(f"  Total cost: ${before['total_cost']:.6f}")
    print(f"  Avg cost/call: ${before['avg_cost_per_call']:.6f}")
    print(f"  Avg latency: {before['avg_latency_ms']}ms")

    print("\n  [After: caching + routing + rate limiting]")
    exact_c = ExactCache()
    semantic_c = SemanticCache(similarity_threshold=0.75)
    tracker_after = CostTracker(monthly_budget=1000.0)

    for q in queries:
        messages = [{"role": "user", "content": q}]
        cached = exact_c.get("gpt-4o", messages, 0.0)
        if cached:
            tracker_after.log_call("gpt-4o-mini", 0, 0, latency_ms=5, cache_status="hit")
            continue
        sem_cached = semantic_c.get(q)
        if sem_cached:
            tracker_after.log_call("gpt-4o-mini", 0, 0, latency_ms=15, cache_status="hit")
            continue
        route = route_model(q)
        result = simulate_llm_call(route["model"], q)
        tracker_after.log_call(route["model"], result["input_tokens"], result["output_tokens"], latency_ms=result["latency_ms"], cache_status="miss")
        exact_c.put(route["model"], messages, 0.0, result["response"])
        semantic_c.put(q, result["response"])

    after = tracker_after.summary()
    print(f"  Total cost: ${after['total_cost']:.6f}")
    print(f"  Avg cost/call: ${after['avg_cost_per_call']:.6f}")
    print(f"  Avg latency: {after['avg_latency_ms']}ms")
    print(f"  Cache hit rate: {after['cache_hit_rate']:.0%}")

    if before["total_cost"] > 0:
        savings_pct = (1 - after["total_cost"] / before["total_cost"]) * 100
        print(f"\n  SAVINGS: {savings_pct:.1f}% cost reduction")
        print(f"  Latency improvement: {(1 - after['avg_latency_ms'] / before['avg_latency_ms']) * 100:.1f}% faster")

    print("\n--- Budget Alerts Demo ---")
    alert_tracker = CostTracker(monthly_budget=0.01)
    for i in range(5):
        alert_tracker.log_call("gpt-4o", 5000, 2000, latency_ms=500)
    print(f"  Total spent: ${alert_tracker.total_cost():.6f} / ${alert_tracker.monthly_budget}")
    for alert in alert_tracker.alerts:
        print(f"  ALERT [{alert['level'].upper()}]: {alert['message']}")

    print("\n--- Cost Breakdown by Model ---")
    multi_tracker = CostTracker(monthly_budget=500.0)
    for _ in range(50):
        multi_tracker.log_call("gpt-4o-mini", 800, 200, latency_ms=150)
    for _ in range(30):
        multi_tracker.log_call("claude-sonnet-4", 1500, 500, latency_ms=400)
    for _ in range(10):
        multi_tracker.log_call("gpt-4o", 2000, 800, latency_ms=600)
    for _ in range(10):
        multi_tracker.log_call("claude-opus-4", 3000, 1000, latency_ms=1200)
    breakdown = multi_tracker.cost_by_model()
    for model, data in sorted(breakdown.items(), key=lambda x: x[1]["cost"], reverse=True):
        print(f"  {model}: {data['calls']} calls, ${data['cost']:.6f}, {data['input_tokens']:,} in / {data['output_tokens']:,} out")
    print(f"  Total: ${multi_tracker.total_cost():.6f}")

    print("\n" + "=" * 60)
    print("  Demo complete.")
    print("=" * 60)


if __name__ == "__main__":
    run_demo()

Úsalo

Caché de Prompts de Anthropic

# import anthropic
#
# client = anthropic.Anthropic()
#
# response = client.messages.create(
#     model="claude-sonnet-4-20250514",
#     max_tokens=1024,
#     system=[
#         {
#             "type": "text",
#             "text": "You are a helpful customer support agent for Acme Corp...",
#             "cache_control": {"type": "ephemeral"},
#         }
#     ],
#     messages=[{"role": "user", "content": "What is the return policy?"}],
# )
#
# print(f"Input tokens: {response.usage.input_tokens}")
# print(f"Cache creation tokens: {response.usage.cache_creation_input_tokens}")
# print(f"Cache read tokens: {response.usage.cache_read_input_tokens}")

La primera solicitud paga una prima de escritura del 25%. Las solicitudes subsiguientes con el mismo prefijo obtienen un 90% de descuento. El caché dura 5 minutos y reinicia el temporizador en cada acierto.

Caché Automático de OpenAI

# from openai import OpenAI
#
# client = OpenAI()
#
# response = client.chat.completions.create(
#     model="gpt-4o",
#     messages=[
#         {"role": "system", "content": "You are a helpful customer support agent..."},
#         {"role": "user", "content": "What is the return policy?"},
#     ],
# )
#
# print(f"Prompt tokens: {response.usage.prompt_tokens}")
# print(f"Cached tokens: {response.usage.prompt_tokens_details.cached_tokens}")
# print(f"Completion tokens: {response.usage.completion_tokens}")

OpenAI almacena en caché automáticamente. Cualquier prefijo de prompt de 1,024+ tokens que coincida con una solicitud reciente obtiene un 50% de descuento. No se requieren cambios en el código -- solo verifica prompt_tokens_details.cached_tokens en la respuesta para confirmar que está funcionando.

API Batch de OpenAI

# import json
# from openai import OpenAI
#
# client = OpenAI()
#
# requests = []
# for i, query in enumerate(queries):
#     requests.append({
#         "custom_id": f"request-{i}",
#         "method": "POST",
#         "url": "/v1/chat/completions",
#         "body": {
#             "model": "gpt-4o-mini",
#             "messages": [{"role": "user", "content": query}],
#         },
#     })
#
# with open("batch_input.jsonl", "w") as f:
#     for r in requests:
#         f.write(json.dumps(r) + "\n")
#
# batch_file = client.files.create(file=open("batch_input.jsonl", "rb"), purpose="batch")
# batch = client.batches.create(input_file_id=batch_file.id, endpoint="/v1/chat/completions", completion_window="24h")
# print(f"Batch ID: {batch.id}, Status: {batch.status}")

La API Batch otorga un 50% de descuento plano en todos los tokens. Los resultados llegan en un plazo de 24 horas. Perfecto para cargas de trabajo no en tiempo real: evaluaciones, etiquetado de datos, resumen masivo.

Caché Semántico de Producción con Redis

# import redis
# import numpy as np
# from openai import OpenAI
#
# r = redis.Redis()
# client = OpenAI()
#
# def get_embedding(text):
#     response = client.embeddings.create(model="text-embedding-3-small", input=text)
#     return response.data[0].embedding
#
# def semantic_cache_lookup(query, threshold=0.95):
#     query_emb = np.array(get_embedding(query))
#     keys = r.keys("cache:emb:*")
#     best_sim, best_key = 0, None
#     for key in keys:
#         stored_emb = np.frombuffer(r.get(key), dtype=np.float32)
#         sim = np.dot(query_emb, stored_emb) / (np.linalg.norm(query_emb) * np.linalg.norm(stored_emb))
#         if sim > best_sim:
#             best_sim, best_key = sim, key
#     if best_sim >= threshold and best_key:
#         response_key = best_key.decode().replace("cache:emb:", "cache:resp:")
#         return r.get(response_key).decode()
#     return None

En producción, reemplaza el escaneo lineal con un índice vectorial (Redis Vector Search, Pinecone o pgvector). El escaneo lineal funciona para <1,000 entradas. Más allá de eso, usa ANN (vecino más cercano aproximado) para búsquedas O(log n).

Envíalo

Esta lección produce outputs/prompt-cost-optimizer.md -- un prompt reutilizable que analiza tu aplicación de LLM y recomienda optimizaciones de costos específicas con ahorros proyectados.

También produce outputs/skill-cost-patterns.md -- un marco de decisión para elegir la estrategia de caché adecuada, la configuración del límite de tasa y las reglas de enrutamiento de modelos para tu caso de uso.

Ejercicios

  1. Implementar la política de desalojo LRU para el caché semántico. Reemplaza el desalojo del más antiguo primero (oldest-first) por el de uso menos reciente (least-recently-used). Realiza un seguimiento del tiempo del último acceso para cada entrada y desaloja la entrada con el acceso más antiguo cuando el caché esté lleno. Compara las tasas de acierto entre las dos estrategias durante 100 consultas.

  2. Construir una herramienta de proyección de costos. Dado un registro de llamadas de API (los logs de CostTracker), proyecta el costo mensual basado en el promedio móvil de los últimos 7 dias. Ten en cuenta los patrones de días laborables y fines de semana. Activa una alerta si el costo mensual proyectado supera el presupuesto en más de un 20%.

  3. Implementar caché semántico en niveles. Usa dos umbrales de similitud: 0.98 para aciertos de alta confianza (devolver inmediatamente) y 0.90 para aciertos de confianza media (devolver con una advertencia: "Basado en una pregunta anterior similar..."). Registra de qué nivel proviene cada acierto y mide las diferencias en la satisfacción del usuario.

  4. Construir un clasificador de enrutamiento de modelos. Reemplaza el clasificador basado en palabras clave por uno basado en embeddings. Genera embeddings de 50 consultas etiquetadas (simple/media/compleja), luego clasifica nuevas consultas encontrando el ejemplo etiquetado más cercano. Mide la precisión de la clasificación frente a un conjunto de prueba de 20 consultas.

  5. Implementar un interruptor automático con niveles de degradación. Al 70% del presupuesto, registra una advertencia. Al 85%, cambia automáticamente todo el enrutamiento al modelo más barato (gpt-4o-mini). Al 95%, sirve solo respuestas en caché y rechaza nuevas consultas. Prueba simulando 1,000 solicitudes con un presupuesto de

    .00 y verifica que cada umbral se active correctamente.

Términos Clave

Término Lo que la gente dice Lo que realmente significa
Caché de prompt "Caché del prompt del sistema" Almacenamiento en caché a nivel de proveedor donde los prefijos de prompt repetidos obtienen un descuento (90% Anthropic, 50% OpenAI) -- sin cambios de código para OpenAI, marcadores explícitos para Anthropic
Caché semántico "Caché inteligente" Generar embedding de la consulta, calcular la similitud con consultas anteriores y devolver la respuesta almacenada en caché si la similitud supera un umbral -- captura paráfrases que la coincidencia exacta pierde
Caché exacto "Caché por hash" Generar un hash del prompt completo (modelo + mensajes + temperatura) y devolver la respuesta almacenada en caché para entradas idénticas -- solo funciona para llamadas deterministas con temperature=0
Cubeta de tokens "Limitador de tasa" Un algoritmo donde cada usuario tiene una cubeta de N tokens que se rellena a una tasa R por segundo -- permite ráfagas hasta N mientras impone una tasa promedio de R
Enrutamiento de modelos "Enrutamiento de bajo costo" Usar un clasificador para enviar consultas simples a modelos baratos (GPT-4o-mini, Haiku) y consultas complejas a modelos caros (GPT-4o, Opus) -- ahorra un 40-70% en costos de modelos
Seguimiento de costos "Medición" Registrar cada llamada a la API con modelo, tokens, latencia, costo e ID de usuario para saber exactamente a dónde va el dinero y qué características son costosas
Interruptor automático (Circuit breaker) "Botón de apagado" Degradar automáticamente el servicio (modelos más baratos, solo caché) o detener las solicitudes por completo cuando el gasto se acerca al límite del presupuesto
API Batch "Descuento por lote" Procesamiento asíncrono de OpenAI con un 50% de descuento -- envía hasta 50,000 solicitudes, obtén resultados en 24 horas
Compresión de prompt "Dieta de tokens" Reescribir los prompts del sistema y el contexto para usar menos tokens mientras se preserva el significado -- los prompts más cortos cuestan menos y a menudo funcionan mejor
Tasa de aciertos de caché "Eficiencia del caché" El porcentaje de solicitudes servidas desde el caché en lugar de llamar al LLM -- el 40-60% es típico para los chatbots de producción, ahorra proporcionalmente en costos

Lecturas Adicionales

  • Anthropic Prompt Caching Guide -- la documentación oficial de los marcadores explícitos cache_control de Anthropic, precios y comportamiento del tiempo de vida del caché
  • OpenAI Prompt Caching -- el caché automático de OpenAI, cómo verificar los aciertos de caché a través de los campos de uso y longitudes mínimas de prefijo
  • OpenAI Batch API -- 50% de descuento para procesamiento asíncrono, formato JSONL, ventana de finalización de 24 horas y límites de 50K solicitudes
  • GPTCache -- biblioteca de caché semántico de código abierto que admite múltiples motores de embeddings, almacenes de vectores y políticas de desalojo
  • Martian Model Router -- enrutamiento de modelos en producción que selecciona automáticamente el modelo más barato capaz de manejar cada consulta
  • Not Diamond -- enrutador de modelos basado en ML que aprende de tus patrones de tráfico para optimizar las compensaciones de costo/calidad entre proveedores
  • Helicone -- plataforma de observabilidad de LLM con seguimiento de costos, caché, limitación de tasa y alertas de presupuesto como una capa de proxy
  • Dean & Barroso, "The Tail at Scale" (CACM 2013) -- latencia, rendimiento, percentiles TTFT/TPOT y solicitudes de cobertura; el modelo de costo detrás de "elegir el modelo más barato que aún cumpla con el P95."
  • Kwon et al., "Efficient Memory Management for Large Language Model Serving with PagedAttention" (SOSP 2023) -- el artículo de vLLM; por qué el caché KV paginado + el procesamiento por lotes continuo superan a los servidores ingenuos en un 24× en rendimiento, la capa de infraestructura bajo "caché y costo."
  • Dao et al., "FlashAttention-2: Faster Attention with Better Parallelism and Work Partitioning" (ICLR 2024) -- reducción de costos a nivel de kernel ortogonal al almacenamiento en caché de prompts; leer junto con la decodificación especulativa y GQA para obtener el panorama completo de la curva de costos.

Reading is free. The graded autograder, AI tutor, and verified certificate unlock with

0 lifetime access. Curriculum based on AI Engineering from Scratch by Rohit Ghumare (MIT, used under attribution).