Phase 11 - Lesson 11

Cache, Limitação de Taxa e Otimização de Custos

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

A maioria das startups de IA não morre devido a modelos ruins. Elas morrem devido a uma economia unitária ruim. Uma única chamada ao GPT-4o custa frações de centavo. Dez mil usuários fazendo dez chamadas por dia custam

50 apenas em tokens de entrada -- antes mesmo de você cobrar um único dólar. As empresas que sobrevivem são aquelas que tratam cada chamada de API como uma transação financeira, e não como uma chamada de função.

Tipo: Build Linguagens: Python Pré-requisitos: Phase 11 Lesson 09 (Function Calling) Tempo: ~45 minutos Relacionado: Phase 11 · 15 (Prompt Caching) — esta lição cobre o cache na camada de aplicação (cache semântico, cache de hash exato, roteamento de modelo). A Lição 15 cobre o cache de prompt na camada do provedor (cache_control da Anthropic, automático da OpenAI, CachedContent do Gemini). Combine ambos para uma redução de custo de 50-95%.

Objetivos de Aprendizado

O Problema

Você constrói um chatbot de RAG. Ele funciona perfeitamente. Os usuários adoram.

Então chega a fatura.

O GPT-5 custa $5 por milhão de tokens de entrada e

5 por milhão de tokens de saída. O Claude Opus 4.7 custa
5 de entrada / $75 de saída. O Gemini 3 Pro custa
.25 de entrada / $5 de saída. O GPT-5-mini custa $0.25/
. Os preços abaixo são ilustrativos; sempre verifique a página de preços atual do provedor.

Aqui está a matemática que mata as startups:

Custo de entrada diário: 10.000 x 10 x 1.000 / 1.000.000 x

.50 =
50/dia Custo de saída diário: 10.000 x 10 x 500 / 1.000.000 x
0.00 = $500/dia Total mensal:
2.500/mês

Isso é apenas o LLM. Adicione embeddings, hospedagem de banco de dados vetorial, infraestrutura. Você está olhando para $30.000/mês para um chatbot.

A parte brutal: 40-60% dessas consultas são quase duplicadas. Os usuários fazem as mesmas perguntas com palavras ligeiramente diferentes. Seu prompt do sistema -- idêntico em todas as requisições -- é cobrado toda vez. Os documentos de contexto recuperados por RAG se repetem entre usuários que perguntam sobre o mesmo assunto.

Você está pagando o preço total por computação redundante.

O Conceito

A Anatomia de Custos de uma Chamada de LLM

Cada chamada de API tem cinco componentes de custo.

graph LR
    A[Consulta do Usuário] --> B[Prompt do Sistema<br/>500-2000 tokens]
    A --> C[Contexto Recuperado<br/>500-4000 tokens]
    A --> D[Mensagem do Usuário<br/>50-500 tokens]
    B --> E[Custo de Entrada<br/>
  .50/1M tokens]
    C --> E
    D --> E
    E --> F[Processamento do Modelo]
    F --> G[Custo de Saída<br/>
0.00/1M tokens]

Prompts do sistema são o assassino silencioso. Um prompt do sistema de 1.500 tokens enviado a cada requisição custa $3.75 por milhão de requisições apenas para esse prefixo. Com 100K requisições por dia, isso representa $375/dia --

1.250/mês -- para um texto que nunca muda.

Cache do Provedor: Descontos Embutidos

Todos os três principais provedores oferecem cache de prompt no lado do provedor em 2026, mas o funcionamento difere. Veja Phase 11 · 15 para uma análise detalhada.

Provedor Mecanismo Desconto Mínimo Duração do Cache
Anthropic Marcadores explícitos cache_control 90% em acertos de cache (paga 25% extra na gravação) 1.024 tokens (Sonnet/Opus), 2.048 (Haiku) Padrão de 5 min; estendido para 1h (2x prêmio de gravação)
OpenAI Correspondência automática de prefixo 50% em acertos de cache 1.024 tokens Melhor esforço de até 1 hora
Google Gemini API CachedContent explícita ~75% de redução (além do armazenamento) 4.096 (Flash) / 32.768 (Pro) TTL configurável pelo usuário

A abordagem da Anthropic é explícita. Você marca seções do seu prompt com cache_control: {"type": "ephemeral"}. A primeira requisição paga um prêmio de gravação de 25%. As requisições subsequentes com o mesmo prefixo recebem um desconto de 90%. Um prompt de sistema de 2.000 tokens que normalmente custa $0.005 custa $0.000625 em acertos de cache. Com 100K requisições, isso economiza $437.50/dia.

A abordagem da OpenAI é automática. Qualquer prefixo de prompt que corresponda a uma requisição anterior recebe um desconto de 50%. Não são necessários marcadores. O trade-off: menor desconto, menor controle, mas zero esforço de implementação.

Cache Semântico: Sua Camada Personalizada

O cache do provedor funciona apenas para prefixos idênticos. O cache semântico lida com o caso mais difícil: consultas diferentes com o mesmo significado.

"Qual é a política de devolução?" e "Como faço para devolver um item?" são strings diferentes, mas com a mesma intenção. Um cache semântico gera embeddings de ambas as consultas, calcula a similaridade de cosseno e retorna a resposta em cache se a similaridade exceder um limite (geralmente de 0,92 a 0,95).

flowchart TD
    A[Consulta do Usuário] --> B[Gerar Embedding da Consulta]
    B --> C{Consulta semelhante<br/>no cache?}
    C -->|sim > 0.95| D[Retornar Resposta em Cache]
    C -->|sim < 0.95| E[Chamar API do LLM]
    E --> F[Salvar Resposta no Cache<br/>com Embedding]
    F --> G[Retornar Resposta]
    D --> G

Os custos de embedding são insignificantes. O text-embedding-3-small da OpenAI custa $0.02 por milhão de tokens. Verificar o cache custa quase nada em comparação com uma chamada completa de LLM.

Cache Exato: Hash e Correspondência

Para chamadas determinísticas (temperature=0, mesmo modelo, mesmo prompt), o cache exato é mais simples e rápido. Faça o hash do prompt completo, verifique o cache e retorne se encontrado.

Isso funciona perfeitamente para:

Limitação de Taxa: Protegendo seu Orçamento

A limitação de taxa não é apenas uma questão de justiça. É uma questão de sobrevivência.

Algoritmo de balde de tokens (token bucket): cada usuário recebe um balde de N tokens que se enche a uma taxa R por segundo. Uma requisição consome tokens do balde. Se o balde estiver vazio, a requisição é rejeitada. Isso permite picos de uso (usar todo o balde de uma vez) enquanto impõe uma taxa média de requisições.

Cotas por usuário: defina limites diários/mensais de tokens por nível de usuário.

Nível Limite Diário de Tokens Máximo de Requisições/min Acesso a Modelos
Gratuito 50.000 10 Apenas GPT-4o-mini
Pro 500.000 60 GPT-4o, Claude Sonnet
Corporativo 5.000.000 300 Todos os modelos

Roteamento de Modelos: O Modelo Certo para o Trabalho Certo

Nem toda consulta precisa do GPT-4o.

"A que horas a loja fecha?" não requer um modelo de

0/milhão de saída. O GPT-4o-mini a $0.60/milhão de saída resolve perfeitamente. O Claude Haiku a
.25/milhão de saída também. Um classificador simples direciona consultas baratas para modelos baratos e consultas complexas para modelos caros.

flowchart TD
    A[Consulta do Usuário] --> B[Classificador de Complexidade]
    B -->|Simples: busca, FAQ| C[GPT-4o-mini<br/>$0.15/$0.60 por 1M]
    B -->|Média: análise, resumo| D[Claude Sonnet<br/>$3.00/
5.00 por 1M]
    B -->|Complexa: raciocínio, código| E[GPT-4o / Claude Opus<br/>

  .50/
0.00+]

Um roteador bem ajustado economiza de 40% a 70% apenas em custos de modelos.

Acompanhamento de Custos: Saiba para Onde Vai o Dinheiro

Você não pode otimizar o que não mede. Registre cada chamada de API com:

  • Timestamp
  • Nome do modelo
  • Tokens de entrada
  • Tokens de saída
  • Latência (ms)
  • Custo calculado ($)
  • ID do usuário
  • Acerto/erro de cache (cache hit/miss)
  • Categoria da requisição

Esses dados revelam quais recursos são caros, quais usuários são grandes consumidores e onde o cache tem o maior impacto.

Processamento em Lote (Batching): Descontos em Lote

A Batch API da OpenAI processa requisições de forma assíncrona com 50% de desconto. Você envia um lote de até 50.000 requisições e os resultados retornam em até 24 horas.

Use o processamento em lote para:

  • Processamento noturno de documentos
  • Classificação em lote
  • Execuções de avaliação
  • Pipelines de enriquecimento de dados

Não use para: consultas em tempo real voltadas ao usuário (onde a latência importa).

Alertas de Orçamento e Disjuntores (Circuit Breakers)

Um disjuntor interrompe os gastos quando você atinge um limite. Sem ele, um bug ou abuso pode consumir seu orçamento mensal em poucas horas.

Defina três limites:

  1. Aviso (70% do orçamento): envia um alerta
  2. Restrição (85% do orçamento): muda apenas para modelos mais baratos
  3. Bloqueio (95% do orçamento): rejeita novas requisições, retornando apenas respostas em cache

A Stack de Otimização

Aplique essas técnicas em ordem. Cada camada se soma às anteriores.

Camada Técnica Economia Típica Esforço de Implementação
1 Cache de prompt do provedor 30-50% Baixo (adicionar marcadores de cache)
2 Cache exato 10-20% Baixo (hash + dict)
3 Cache semântico 15-30% Médio (embeddings + similaridade)
4 Roteamento de modelos 40-70% Médio (classificador)
5 Limitação de taxa Proteção de orçamento Baixo (balde de tokens)
6 Compressão de prompt 10-30% Médio (reescrever prompts)
7 Processamento em lote 50% nas elegíveis Baixo (API de lote)

Um aplicativo de RAG que aplica as camadas 1 a 5 normalmente reduz os custos de

2.500/mês para $4.000-$6.000/mês. Essa é a diferença entre queimar todo o seu caixa e construir um negócio sustentável.

Economia Real: Antes e Depois

Aqui está uma análise real para um chatbot de RAG que atende 10.000 DAU.

Métrica Antes da Otimização Depois da Otimização Economia
Custo mensal de LLM 2.500 $5.200 77%
Custo médio por consulta $0.0075 $0.0017 77%
Taxa de acerto do cache 0% 52% --
Consultas roteadas para o mini 0% 65% --
Latência P95 2.800ms 900ms (acertos de cache: 50ms) 68%
Custo mensal de embeddings $0
80
 (novo custo)
Custo mensal total 2.500 $5.380 76%

O custo de embedding para o cache semântico (

80/mês) se paga logo na primeira hora de acertos de cache.

Construa

Passo 1: Calculadora de Custos

Construa uma calculadora de custo de tokens que conheça os preços antigos/atuais dos principais modelos.

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

Passo 2: Cache Exato

Gere o hash do prompt completo e retorne as respostas em cache para requisições 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),
        }

Passo 3: Cache Semântico

Gere embeddings das consultas e retorne respostas em cache quando a similaridade exceder um limite.

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

Passo 4: Limitador de Taxa

Limitador de taxa por balde de tokens com cotas por usuário.

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

Passo 5: Rastreador de Custos

Registre cada chamada e calcule os totais 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,
        }

Passo 6: Roteador de Modelos

Roteie consultas para o modelo mais barato que possa lidar com elas.

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}

Passo 7: Execute a Demonstração

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

Use

Cache de Prompt da 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}")

A primeira chamada grava no cache (prêmio de 25%). Cada chamada subsequente com o mesmo prefixo de prompt de sistema lê do cache (desconto de 90%). O cache dura 5 minutos e reinicia o temporizador a cada acerto.

Cache Automático da 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}")

A OpenAI faz o cache automaticamente. Qualquer prefixo de prompt com mais de 1.024 tokens que corresponda a uma requisição recente recebe 50% de desconto. Nenhuma alteração no código é necessária -- basta verificar prompt_tokens_details.cached_tokens na resposta para confirmar o funcionamento.

Batch API da 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}")

A Batch API oferece um desconto fixo de 50% em todos os tokens. Os resultados chegam em até 24 horas. Perfeito para cargas de trabalho que não são em tempo real: avaliações, rotulagem de dados, resumos em lote.

Cache Semântico em Produção com 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

Em produção, substitua a busca linear por um índice vetorial (Redis Vector Search, Pinecone ou pgvector). A busca linear funciona para <1.000 entradas. Além disso, use ANN (approximate nearest neighbor) para busca O(log n).

Publique

Esta lição produz outputs/prompt-cost-optimizer.md -- um prompt reutilizável que analisa seu aplicativo de LLM e recomenda otimizações de custos específicas com projeções de economia.

Também produz outputs/skill-cost-patterns.md -- uma estrutura de decisão para escolher a estratégia de cache correta, configuração de limite de taxa e regras de roteamento de modelos para seu caso de uso.

Exercícios

  1. Implementar a remoção por LRU para o cache semântico. Substitua a remoção baseada na entrada mais antiga (oldest-first) por menos recentemente usada (least-recently-used). Rastreie a hora do último acesso para cada entrada e remova a entrada com o acesso mais antigo quando o cache estiver cheio. Compare as taxas de acerto entre as duas estratégias ao longo de 100 consultas.

  2. Construir uma ferramenta de projeção de custos. A partir de um registro de chamadas de API (los logs do CostTracker), projete o custo mensal com base na média dos últimos 7 dias. Considere padrões de dias úteis e finais de semana. Dispare um alerta se o custo mensal projetado exceder o orçamento em mais de 20%.

  3. Implementar cache semântico em camadas. Use dois limites de similaridade: 0.98 para acertos de alta confiança (retornar imediatamente) e 0.90 para acertos de média confiança (retornar com um aviso: "Com base em uma pergunta anterior semelhante..."). Rastreie de qual camada veio cada acerto e meça as diferenças de satisfação do usuário.

  4. Construir um classificador de roteamento de modelos. Substitua o classificador baseado em palavras-chave por um baseado em embeddings. Gere embeddings de 50 consultas rotuladas (simples/médio/complexo) e, em seguida, classifique as novas consultas encontrando o exemplo rotulado mais próximo. Meça a precisão da classificação em relação a um conjunto de testes de 20 consultas.

  5. Implementar um disjuntor com níveis de degradação. Com 70% do orçamento, registre um aviso. Com 85%, altere automaticamente todo o roteamento para o modelo mais barato (gpt-4o-mini). Com 95%, forneça apenas respostas em cache e rejeite novas consultas. Teste simulando 1.000 requisições contra um orçamento de

    .00 e verifique se cada limite é disparado corretamente.

Termos-Chave

Termo O que as pessoas dizem O que realmente significa
Cache de prompt "Cachear o prompt do sistema" Cache a nível de provedor no qual prefixos de prompt repetidos recebem desconto (90% Anthropic, 50% OpenAI) -- sem alterações no código para a OpenAI, marcadores explícitos para a Anthropic
Cache semântico "Cache inteligente" Gerar embedding da consulta, calcular a similaridade com consultas anteriores e retornar a resposta em cache se a similaridade exceder um limite -- captura paráfrases que a correspondência exata perde
Cache exato "Cache por hash" Gerar hash do prompt completo (modelo + mensagens + temperatura) e retornar a resposta em cache para entradas idênticas -- funciona apenas para chamadas determinísticas com temperature=0
Balde de tokens "Limitador de taxa" Um algoritmo no qual cada usuário tem um balde de N tokens que se enche a uma taxa R por segundo -- permite picos de uso de até N enquanto impõe uma taxa média de R
Roteamento de modelos "Roteamento econômico" Usar um classificador para enviar consultas simples para modelos baratos (GPT-4o-mini, Haiku) e consultas complexas para modelos caros (GPT-4o, Opus) -- economiza de 40% a 70% nos custos de modelo
Acompanhamento de custos "Medição" Registrar cada chamada de API com modelo, tokens, latência, custo e ID do usuário para saber exatamente para onde vai o dinheiro e quais recursos são caros
Disjuntor (Circuit breaker) "Botão de emergência" Degradar automaticamente o serviço (modelos mais baratos, apenas respostas em cache) ou parar totalmente as requisições quando os gastos se aproximam do limite do orçamento
Batch API "Desconto em lote" Processamento assíncrono da OpenAI com 50% de desconto -- envie até 50.000 requisições, obtenha os resultados em até 24 horas
Compressão de prompt "Dieta de tokens" Reescrever prompts do sistema e contexto para usar menos tokens mantendo o significado -- prompts mais curtos custam menos e frequentemente performam melhor
Taxa de acertos do cache "Eficiência de cache" A porcentagem de requisições atendidas pelo cache em vez de chamar o LLM -- 40-60% é típico para chatbots em produção, economiza proporcionalmente nos custos

Leituras Adicionais

  • Anthropic Prompt Caching Guide -- a documentação oficial dos marcadores explícitos cache_control da Anthropic, preços e comportamento do tempo de vida do cache
  • OpenAI Prompt Caching -- o cache automático da OpenAI, como verificar os acertos de cache usando os campos de uso e os comprimentos mínimos de prefixo
  • OpenAI Batch API -- 50% de desconto para processamento assíncrono, formato JSONL, janela de conclusão de 24 horas e limite de 50.000 requisições
  • GPTCache -- biblioteca de cache semântico de código aberto compatível com múltiplos backends de embeddings, bancos de dados vetoriais e políticas de remoção
  • Martian Model Router -- roteamento de modelos em produção que seleciona automaticamente o modelo mais barato capaz de lidar com cada consulta
  • Not Diamond -- roteador de modelos baseado em aprendizado de máquina que aprende com os seus padrões de tráfego para otimizar os trade-offs de custo/qualidade entre provedores
  • Helicone -- plataforma de observabilidade de LLMs com acompanhamento de custos, cache, limite de taxa e alertas de orçamento como uma camada de proxy
  • Dean & Barroso, "The Tail at Scale" (CACM 2013) -- latência, throughput, percentis TTFT/TPOT e requisições protegidas; o modelo de custo por trás de "escolher o modelo mais barato que ainda atenda ao P95."
  • Kwon et al., "Efficient Memory Management for Large Language Model Serving with PagedAttention" (SOSP 2023) -- o artigo do vLLM; por que o cache KV paginado + processamento em lotes contínuo superam servidores comuns em 24x no throughput, a camada de infraestrutura sob "cache e custos".
  • Dao et al., "FlashAttention-2: Faster Attention with Better Parallelism and Work Partitioning" (ICLR 2024) -- redução de custo em nível de kernel ortogonal ao cache de prompt; leia junto com decodificação especulativa e GQA para obter o panorama completo da curva de custos.

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