Phase 11: LLM Engineering

El caché, la limitación de tasas y la optimización de costes

Type: Build

Languages: Python

Prerequisites: Phase 11 Lesson 09 (Function Calling)

Time: ~45 minutes

Related:Fase 11 · 15 (Caching de inmediato) esta lección cubre el caché de la capa de aplicación (caché semántico, caché hash exacto, enrutamiento de modelo).

Objetivos de aprendizaje

• Implementar caché semántico que sirve consultas repetidas o similares desde la caché en lugar de hacer una nueva llamada de API
• Calcular los costes por solicitud entre los proveedores e implementar alertas de tarifa y presupuestos de restricción de conocimiento de tokens
• Construir una capa de optimización de costos con compresión rápida, enrutamiento de modelos (caros vs baratos) y almacenamiento en caché de respuesta
• Diseñar una estrategia de caché en niveles utilizando coincidencia exacta, similitud semántica y caché prefijo para diferentes tipos de consultas

El problema

Construye un chatbot RAG, funciona muy bien y a los usuarios les encanta.

Entonces llega la factura.

Los costes del GPT-5 $5 per million input tokens and $15 por millón de producción.$15 input / $75 de salida. Gemini 3 Pro cuesta $1.25 input / $5 de salida. GPT-5-mini es $0.25/$Los precios a continuación son ilustrativos; siempre compruebe la página de precios actual del proveedor.

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

• 10.000 usuarios activos diarios
• 10 consultas por usuario y día
• 1,000 tokens de entrada por consulta (invite del sistema + contexto + mensaje del usuario)
• 500 tokens de salida por respuesta

Daily input cost:10.000 x 10 x 1.000 / 1.000.000 x $2.50 = $250/día

Daily output cost:10.000 x 10 x 500 / 1.000.000 x $10.00 = $500/día

Monthly total: $22,500/month

Eso es sólo el LLM. Añade embebidos, alojamiento de base de datos vectorial, infraestructura. Estás buscando $30,000 al mes para un chatbot.

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

Estás pagando el precio completo por los cálculos redundantes.

El concepto

La anatomía de los costes de una convocatoria de LLM

Cada llamada de API tiene cinco componentes de costo.

graph LR
    A[User Query] --> B[System Prompt<br/>500-2000 tokens]
    A --> C[Retrieved Context<br/>500-4000 tokens]
    A --> D[User Message<br/>50-500 tokens]
    B --> E[Input Cost<br/>$2.50/1M tokens]
    C --> E
    D --> E
    E --> F[Model Processing]
    F --> G[Output Cost<br/>$10.00/1M tokens]

Las instrucciones del sistema son el asesino silencioso.$3.75 per million requests just for that prefix. At 100K requests per day, that is $375 días -- $11,250 al mes -- por un texto que nunca cambia.

El caché de los proveedores: descuentos incorporados

Los tres principales proveedores ofrecen caché rápido en el lado del proveedor en 2026, pero las mecánicas difieren.

Anthropic's approachMarca las secciones de su mensaje con cache_control: {"type": "ephemeral"}. La primera solicitud paga una prima de escritura del 25%. Las solicitudes posteriores con el mismo prefijo obtienen un descuento del 90%.$0.005 normally costs $0.000625 en visitas de caché. Más de 100K solicitudes, que ahorra $437.50 / día.

OpenAI's approachEl precio de la aplicación es de un precio de más de $500.00 por año.

El caché semántico: su capa personalizada

El 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 devolución?" y "¿Cómo devuelvo un artículo?" son cadenas diferentes pero con la misma intención. Una caché semántica incorpora ambas consultas, calcula la similitud cosina y devuelve la respuesta caché si la similitud supera un umbral (normalmente 0.92-0.95).

flowchart TD
    A[User Query] --> B[Embed Query]
    B --> C{Similar query<br/>in cache?}
    C -->|sim > 0.95| D[Return Cached Response]
    C -->|sim < 0.95| E[Call LLM API]
    E --> F[Cache Response<br/>with Embedding]
    F --> G[Return Response]
    D --> G

Los costos de incorporación son insignificantes. OpenAI de texto de incorporación 3-pequeño cuesta $ 0.02 por millón de tokens.

Caching exacto: hash y coincidencia

Para llamadas deterministas (temperatura = 0, mismo modelo, mismo pedido), el caché exacto es más simple y más rápido.

Esto funciona perfectamente para:

• Impulso de sistema + contexto fijo + consultas de usuario idénticas
• Llamadas de función con definiciones idénticas de herramientas
• Procesamiento en lote donde el mismo documento se procesa varias veces

Limitación de tasas: Proteger su presupuesto

La limitación de tasas no se trata sólo de justicia, sino de supervivencia.

Token bucket algorithm:Cada usuario recibe un cubo de N tokens que se reponen a una velocidad R por segundo. Una solicitud consume tokens del cubo. Si el cubo está vacío, la solicitud es rechazada. Esto permite estallar (usar el cubo completo a la vez) mientras se aplica una tasa promedio.

Per-user quotas:fijar límites diarios/mensuales de tokens por nivel de usuario.

Modelo de ruta: modelo adecuado para el trabajo correcto

No todas las consultas necesitan GPT-4o.

"A qué hora cierra la tienda?" no requiere un $10/M-output model. GPT-4o-mini at $La salida de 0.60M la maneja perfectamente. Claude Haiku a $1.25/M la maneja. Un clasificador simple envía consultas baratas a modelos baratos y consultas complejas a modelos caros.

flowchart TD
    A[User Query] --> B[Complexity Classifier]
    B -->|Simple: lookup, FAQ| C[GPT-4o-mini<br/>$0.15/$0.60 per 1M]
    B -->|Medium: analysis, summary| D[Claude Sonnet<br/>$3.00/$15.00 per 1M]
    B -->|Complex: reasoning, code| E[GPT-4o / Claude Opus<br/>$2.50/$10.00+]

Un router bien ajustado ahorra entre un 40% y un 70% en los costos del modelo.

Seguimiento de costos: saber dónde va el dinero

No se puede optimizar lo que no se mide. Registre cada llamada de API con:

• Estampilla de tiempo
• Nombre del modelo
• Tokens de entrada
• Tokens de salida
• La latencia (ms)
• Costo calculado ($)
• Identificación de usuario
• El caché se encuentra/hace falta
• Categoría de solicitud

Estos datos revelan qué características son caras, qué usuarios son consumidores pesados y dónde el almacenamiento en caché tiene más impacto.

Los lotes: descuentos masivos

La API de lotes de OpenAI procesa las solicitudes de manera asíncrona con un descuento del 50%. Envía un lote de hasta 50.000 solicitudes, y los resultados vuelven dentro de las 24 horas.

Utilice el batch para:

• Tratamiento nocturno de documentos
• Clasificación en masa
• Las pruebas de evaluación
• Línea de enriquecimiento de datos

No para: consultas en tiempo real dirigidas al usuario (cuestiones de latencia).

Alertas presupuestarias y interrupciones de circuitos

Si no tienes un interruptor de circuito, el gasto se detiene, y si no tienes un interruptor, un error o un abuso pueden quemar tu presupuesto mensual en horas.

Establezca tres umbrales:

1. Warning(70% del presupuesto): enviar una alerta
2. Throttle(85% del presupuesto): sólo se cambiará a modelos más baratos
3. Stop(95% del presupuesto): rechazar nuevas solicitudes, devolver respuestas almacenadas en caché solamente

La pila de optimización

Aplique estas técnicas en orden. Cada capa se compone con las anteriores.

Una aplicación RAG que aplica capas 1-5 normalmente reduce los costes de $22,500/month to $4000 a 6.000 al mes. Esa es la diferencia entre quemar una pista y construir un negocio.

Ahorros reales: antes y después

Aquí hay una avería real para un chatbot RAG que sirve 10.000 DAU.

El costo de incorporación para el caché semántico ($ 180 / mes) se paga por sí mismo dentro de la primera hora de visitas al caché.

Construye el mismo

Paso 1: Calculadora de costos

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

pythonCopyimport 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: Cache exacto

Hace hash de la solicitud completa y devuelva las respuestas almacenadas en caché para las mismas solicitudes.

pythonCopyclass 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: Cache semántica

Embed las consultas y devuelven las respuestas almacenadas en caché cuando la similitud exceda un umbral.

pythonCopydef 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: Limitación de tasa

Limitar tasa de token con cuotas por usuario.

pythonCopyclass 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: Seguimiento de costes

Registra todas las llamadas y computa los totales de ejecución.

pythonCopyclass 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 6: Modelo de enrutador

Envía las consultas al modelo más barato que pueda manejarlas.

pythonCopySIMPLE_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: ejecuta la demostración

pythonCopydef 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()

Usalo

Caching de datos de la persona

pythonCopy# import anthropic
#
# client = anthropic.Anthropic()
#
# response = client.messages.create(
#     model="claude-sonnet-5",
#     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 llamada se escribe en la caché (premia del 25%). Cada llamada posterior con el mismo prefijo de pedido del sistema se lee desde la caché (descuento del 90%). La caché dura 5 minutos y se restablece el temporizador en cada golpe.

OpenAI Caché automático

pythonCopy# 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 se almacena automáticamente. Cualquier prefijo de 1.024+ tokens que coincida con una solicitud reciente obtiene un descuento del 50%. No se necesitan cambios de código - sólo comprobarprompt_tokens_details.cached_tokensen la respuesta para verificar que está funcionando.

API de lotes de OpenAI

pythonCopy# 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 de lote ofrece un descuento del 50% en todos los tokens. Los resultados llegan dentro de las 24 horas. Perfecto para cargas de trabajo no en tiempo real: evaluaciones, etiquetado de datos, resumen en masa.

Producción Cache semántica con Redis

pythonCopy# 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 la producción, reemplaza el escaneo lineal con un índice vectorial (Redis Vector Search, Pinecone o pgvector). El escaneo lineal funciona para <1,000 entradas.

Envío

Esta lección produceoutputs/prompt-cost-optimizer.md-- un mensaje reutilizable que analiza su solicitud de LLM y recomienda optimizaciones específicas de costos con ahorros proyectados.

También produce outputs/skill-cost-patterns.md-- un marco de decisión para elegir la estrategia de almacenamiento en caché adecuada, configuración de limitación de velocidad y reglas de enrutamiento modelo para su caso de uso.

Los ejercicios

1. Implement LRU eviction for the semantic cache.Replace el desalojo más antiguo con el menos recientemente utilizado. Rastrear el último tiempo de acceso para cada entrada y desalojar la entrada con el tiempo de acceso más antiguo cuando la caché esté lleno. Comparar las tasas de impacto entre las dos estrategias más de 100 consultas.

1. Build a cost projection tool.Dado un registro de llamadas de API (los registros de CostTracker), proyecta el costo mensual basado en el promedio de 7 días posteriores.

1. Implement tiered semantic caching.Utilice dos umbrales de similitud: 0,98 para los hits de alta confianza (retorno inmediato) y 0,90 para los hits de confianza media (retorno con una descarga de responsabilidad: "Basado en una pregunta anterior similar...").

1. Build a model routing classifier.Reemplazar el clasificador basado en palabras clave por uno basado en embebed. Incorporar 50 consultas etiquetadas (simples/medias/complejas), luego clasificar nuevas consultas encontrando el ejemplo etiquetado más cercano. Medir la precisión de clasificación en comparación con un conjunto de pruebas de 20 consultas.

1. Implement a circuit breaker with degradation levels.Con un presupuesto del 70%, registre una advertencia. Con el 85%, cambie automáticamente todo el enrutamiento al modelo más barato (gpt-4o-mini). Con el 95%, solo sirva respuestas almacenadas en caché y rechaza nuevas consultas. Prueba simulado de 1.000 solicitudes contra un presupuesto de $1.00 y verifique cada umbral que se activa correctamente.

Términos clave

Leer más

• Anthropic Prompt Caching Guide-- los documentos oficiales para los marcadores de control de caché explícito de Anthropic, precios y comportamiento de vida útil de caché
• OpenAI Prompt Caching-- La caché automática de OpenAI, cómo verificar los caches de acceso a través de campos de uso, y el prefijo mínimo de longitud
• OpenAI Batch API-- 50% de descuento para el procesamiento asincrono, formato JSONL, ventana de 24 horas de finalización y límites de solicitudes de 50K
• GPTCache-- biblioteca de caché semántico de código abierto que admite múltiples fondos de fondo de incorporación, tiendas vectoriales y políticas de desalojo
• Martian Model Router-- enrutamiento de modelos de producción que selecciona automáticamente el modelo más barato capaz de manejar cada consulta
• Not Diamond-- Router modelo basado en ML que aprende de sus patrones de tráfico para optimizar los compromisos de costo / calidad entre los proveedores
• Helicone-- LLM plataforma de observabilidad con el seguimiento de costos, almacenamiento en caché, limitación de tasas y alertas presupuestarias como una capa proxy
• Dean & Barroso, "The Tail at Scale" (CACM 2013)-- latencia, rendimiento, porcentajes TTFT/TPOT, y solicitudes cubiertas; el modelo de costo detrás de "escolle el modelo más barato que todavía cumpla con P95."
• Kwon et al., "Efficient Memory Management for Large Language Model Serving with PagedAttention" (SOSP 2023)-- el documento vLLM; por qué el KV-cache paged + batching continuo supera a los servidores ingenuos 24x en rendimiento, la capa infra bajo "caching y costo".
• Dao et al., "FlashAttention-2: Faster Attention with Better Parallelism and Work Partitioning" (ICLR 2024)-- reducción de costos a nivel de núcleo ortogonal para solicitar el almacenamiento en caché; leer junto con la descifrado especulativo y GQA para la imagen completa de la curva de costos.

This free lesson is part of the AI Engineering from Scratch curriculum. Read the full explanation, run the lesson code, and verify the result in the interactive reader or from the repository source.

Browse the complete course catalog or open this lesson on GitHub.