Phase 11: LLM Engineering

Le caching, la limitation des taux et l'optimisation des coûts

Type: Build

Languages: Python

Prerequisites: Phase 11 Lesson 09 (Function Calling)

Time: ~45 minutes

Related:La phase 11 · 15 (Cachage rapide) cette leçon couvre le caching de la couche d'application (cache sémantique, cache hash exact, routage de modèle).

Objectifs d'apprentissage

• Implémenter la mise en cache sémantique qui sert des requêtes répétées ou similaires du cache au lieu de faire un nouvel appel API
• Calculer les coûts par demande entre les fournisseurs et mettre en œuvre des alertes budgétaires et des limites de taux de reconnaissance des jetons
• Construire une couche d'optimisation des coûts avec compression rapide, routage de modèle (chare contre bon marché) et mise en cache de réponse
• Conceptez une stratégie de mise en cache à niveau en utilisant la correspondance exacte, la similitude sémantique et la mise en cache préfixe pour différents types de requêtes

Le problème

Vous construisez un chatbot RAG, il fonctionne magnifiquement, les utilisateurs l'adorent.

Puis la facture arrive.

Coûts du GPT-5 $5 per million input tokens and $15 pour un million de produits.$15 input / $75 sorties. Les prix du Gemini 3 Pro $1.25 input / $5 sortie. GPT-5-mini est $0.25/$2. Les prix ci-dessous sont illustratifs; consultez toujours la page actuelle des prix du fournisseur.

Voici les mathématiques qui tuent les startups:

• 10 000 utilisateurs actifs quotidiens
• 10 requêtes par utilisateur par jour
• 1 000 jetons d'entrée par requête (interrogatoire système + contexte + message utilisateur)
• 500 jetons de sortie par réponse

Daily input cost:10 000 x 10 x 1 000 / 1 000 000 x $2.50 = $250/jour

Daily output cost:10 000 x 10 x 500 / 1 000 000 x $10.00 = $500/jour

Monthly total: $22,500/month

Il y a des intégrations, des hébergements de bases de données vectorielles, des infrastructures.

La partie brutale: 40 à 60% de ces requêtes sont presque dupliquées. Les utilisateurs posent les mêmes questions avec des mots légèrement différents. Votre demande de système - identique à chaque demande - est facturée à chaque fois. Les documents contextuels récupérés par RAG se répètent à travers les utilisateurs qui posent des questions sur le même sujet.

Vous payez le prix total pour les calculs redondants.

Le concept

L'anatomie des coûts d'un appel à la LLM

Chaque appel d'API a cinq composantes de coûts.

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]

Les instructions système sont le tueur silencieux.$3.75 per million requests just for that prefix. At 100K requests per day, that is $375 $ par jour, 11 250 $ par mois, pour un texte qui ne change jamais.

Les services de caching: réductions intégrées

Les trois principaux fournisseurs proposent une mise en cache rapide du côté du fournisseur en 2026, mais les mécanismes diffèrent.

Anthropic's approachVous marquez des sections de votre demande avec cache_control: {"type": "ephemeral"}La première demande est payée avec une prime d'écriture de 25%, les demandes ultérieures avec le même préfixe obtiennent une réduction de 90%.$0.005 normally costs $0,000625 sur les visites de cache, plus de 100 000 demandes, ce qui économise 437,50 $ par jour.

OpenAI's approachLes commandes de la carte de crédit sont automatiques. Tout préfixe prompt qui correspond à une demande précédente obtient une réduction de 50%. Aucun marqueur n'est nécessaire.

Le caching sémantique: votre couche personnalisée

Le caching fournisseur ne fonctionne que pour les préfixes identiques.

"Quelle est la politique de retour?" et "Comment retourner un article?" sont des chaînes différentes mais identiques. Un cache sémantique intègre les deux requêtes, calcule la similitude cosine et renvoie la réponse en cache si la similitude dépasse un seuil (généralement 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

Les coûts d'intégration sont négligeables. L'intégration de texte 3-small d'OpenAI coûte 0,02 $ par million de jetons.

Cachage exact: Hash et correspondance

Pour les appels déterministes (température = 0, même modèle, même prompt), le caching exact est plus simple et plus rapide.

Ça fonctionne parfaitement pour:

• Prompte système + contexte fixe + requêtes utilisateur identiques
• Appel à fonction avec les mêmes définitions d'outil
• Traitement par lots où le même document est traité plusieurs fois

Limiter les tarifs: protéger votre budget

La limitation des taux ne concerne pas seulement l'équité, mais la survie.

Token bucket algorithm:chaque utilisateur reçoit un seau de N jetons qui se remplit à un rythme R par seconde. Une demande consomme des jetons du seau. Si le seau est vide, la demande est rejetée. Cela permet des explosions (utiliser le seau complet à la fois) tout en appliquant un taux moyen.

Per-user quotas:définir des limites quotidiennes/ménaux de jetons par niveau d'utilisateur.

Le modèle de routage: le bon modèle pour le bon travail

Toutes les requêtes n'ont pas besoin de GPT-4o.

" À quelle heure ferme-t-on le magasin ? " n'exige pas de faire une demande.$10/M-output model. GPT-4o-mini at $La sortie de 0,60 / M le gère parfaitement. Claude Haiku à 1,25 $ / M le gère. Un classifiateur simple rote des requêtes bon marché à des modèles bon marché et des requêtes complexes à des modèles coûteux.

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 routeur bien ajusté permet d'économiser 40 à 70% sur les coûts du modèle.

Suivre les coûts: savoir où va l'argent

Vous ne pouvez pas optimiser ce que vous ne mesurez pas.

• Temps de l'année
• Nom du modèle
• Les jetons d'entrée
• Les jetons de sortie
• La latence (ms)
• Coût calculé ($)
• Identifiant de l'utilisateur
• Accès/défaut de cache
• Catégorie de demande

Ces données révèlent quelles fonctionnalités sont chères, quels utilisateurs sont de gros consommateurs et où le caching a le plus d'impact.

Les lots: réductions en vrac

L'API de lot d'OpenAI traite les demandes de manière asynchrone à 50% de réduction. Vous soumettez un lot de 50 000 demandes, et les résultats reviennent dans les 24 heures.

Utiliser le batchage pour:

• Traitement des documents par nuit
• Classification en vrac
• Les cours d'évaluation
• Les pipelines d'enrichissement de données

Pas pour: requêtes en temps réel auxquelles l'utilisateur est confronté (matières de latence).

Alertes budgétaires et interruptions de circuits

Un disjoncteur arrête de dépenser quand vous atteignez une limite.

Définir trois seuils:

1. Warning(70% du budget): envoyer une alerte
2. Throttle(85% du budget): seulement des modèles moins chers
3. Stop(95% du budget): rejet des nouvelles demandes, retour seulement des réponses cachées

Le piquet d'optimisation

Appliquez ces techniques dans l'ordre.

Une application RAG appliquant des couches 1 à 5 réduit généralement les coûts de $22,500/month to $4000 à 6000 par mois, c'est la différence entre brûler une piste et construire une entreprise.

Réelle épargne: avant et après

Voici une vraie panne pour un chatbot RAG qui sert 10 000 DAU.

Le coût d'intégration du caching sémantique (180 $ par mois) se paie à lui-même dans la première heure des visites du cache.

Faites-le

Étape 1: Calculateur de coûts

Construisez une calculatrice de coût de jeton qui connaît les prix actuels des principaux modèles.

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),
    }

Étape 2: Cache exacte

Hacher le prompt complet et retourner les réponses cachées pour les demandes identiques.

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),
        }

Étape 3: Cache sémantique

Embed requêtes et retourner les réponses cachées lorsque la similitude dépasse un seuil.

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),
        }

Étape 4: Limite de taux

Limiteur de taux de jetons avec quotas par utilisateur.

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,
        }

Étape 5: Tracker des coûts

Enregistrez chaque appel et comptez les totaux en cours d'exécution.

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,
        }

Étape 6: Modèle routeur

Retourner les requêtes au modèle le moins cher qui peut les gérer.

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}

Étape 7: Exécutez la démo

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

Utilisez-le

Le caching des instantanés

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}")

Le premier appel est écrit dans le cache (25% de prime). Chaque appel ultérieur avec le même préfixe de prompt système est lu dans le cache (90% de réduction). Le cache dure 5 minutes et réinitialise le minuterie à chaque coup.

Cachage automatique OpenAI

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 cache automatiquement. Tout préfixe prompt de 1.024+ jetons qui correspond à une demande récente obtient une réduction de 50%. Aucun changement de code nécessaire - il suffit de vérifierprompt_tokens_details.cached_tokensdans la réponse pour vérifier qu'il fonctionne.

API de lot 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}")

L'API de lot offre une réduction de 50% sur tous les jetons. Les résultats arrivent dans les 24 heures. Parfait pour les charges de travail non en temps réel: évaluations, étiquetage de données, résumé en vrac.

Production Cache sémantique avec 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 production, remplacer l'analyse linéaire par un index vectoriel (Redis Vector Search, Pinecone ou pgvector).

La faire partir

Cette leçon produit outputs/prompt-cost-optimizer.md-- une demande réutilisable qui analyse votre demande de LLM et recommande des optimisations spécifiques des coûts avec des économies prévues.

Il produit aussi outputs/skill-cost-patterns.md-- un cadre de décision pour choisir la bonne stratégie de mise en cache, la configuration de limitation de taux et les règles de routage de modèle pour votre cas d'utilisation.

Exercices

1. Implement LRU eviction for the semantic cache.Remplacez le premier évacuation le plus ancien par le moins récemment utilisé. Suivez le dernier temps d'accès pour chaque entrée et évacuez l'entrée avec le plus ancien temps d'accès lorsque le cache est plein. Comparer les taux de succès entre les deux stratégies sur 100 requêtes.

1. Build a cost projection tool.En fonction du registre des appels d'API (les logs CostTracker), prévoir le coût mensuel en fonction de la moyenne de 7 jours.

1. Implement tiered semantic caching.Utilisez deux seuils de similitude: 0,98 pour les hits à haute confiance (retour immédiat) et 0,90 pour les hits à moyenne confiance (retour avec une exclusion de responsabilité: "Sur la base d'une question antérieure similaire..."). Suivez le niveau de chaque hit et mesurez les différences de satisfaction des utilisateurs.

1. Build a model routing classifier.Remplacez le classifiateur basé sur des mots clés par un classifiateur basé sur l'intégration. Embed 50 requêtes étiquetées (simple/médium/complexe), puis classifiez de nouvelles requêtes en trouvant l'exemple étiqueté le plus proche. Mesurez la précision de la classification par rapport à un ensemble de test de 20 requêtes.

1. Implement a circuit breaker with degradation levels.À 70% de budget, enregistrer un avertissement. À 85%, passer automatiquement tout le routage au modèle le moins cher (gpt-4o-mini). à 95%, servir seulement des réponses en cache et rejeter de nouvelles requêtes.

Les termes clés

Pour en savoir plus

• Anthropic Prompt Caching Guide-- les documents officiels pour les marqueurs explicites de contrôle cache, les prix et le comportement de la durée de vie du cache d'Anthropic
• OpenAI Prompt Caching-- La mise en cache automatique d'OpenAI, comment vérifier les hits de cache via les champs d'utilisation, et les longueurs minimales de préfixes
• OpenAI Batch API-- 50% de réduction pour le traitement asynchrone, format JSONL, fenêtre de finition de 24 heures et limites de demande de 50K
• GPTCache-- bibliothèque de mise en cache sémantique open source qui prend en charge plusieurs arrière-plans intégrés, magasins vectoriels et politiques d'évacuation
• Martian Model Router-- routage du modèle de production qui sélectionne automatiquement le modèle le moins cher capable de traiter chaque requête
• Not Diamond-- Un routeur de modèle basé sur le système de gestion de données qui apprend de vos schémas de trafic pour optimiser les compromis coûts/qualité entre les fournisseurs
• Helicone-- Plateforme d'observabilité de la LLM avec suivi des coûts, mise en cache, limitation des tarifs et alertes budgétaires en tant que couche proxy
• Dean & Barroso, "The Tail at Scale" (CACM 2013)-- latence, débit, TTFT/TPOT percentiles, et les demandes couvertes; le modèle de coût derrière "choisir le modèle le moins cher qui répond toujours à P95. "
• Kwon et al., "Efficient Memory Management for Large Language Model Serving with PagedAttention" (SOSP 2023)-- le document vLLM; pourquoi le KV-cache en page + le batchage continu bat des serveurs naïfs 24x sur le débit, l'infrarouge sous "caching et coût".
• Dao et al., "FlashAttention-2: Faster Attention with Better Parallelism and Work Partitioning" (ICLR 2024)-- réduction des coûts au niveau du noyau orthogonale pour la mise en cache; lire à côté du décoding spéculatif et GQA pour la courbe de coûts complète.

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.