Prompt Caching Economics: Cut LLM API Costs 90% With Intelligent Cache Architecture

Cost_no_cache = N × T × C_normal

Cost_with_cache = T × C_write + (N-1) × T × C_read

T × C_write + (N-1) × T × C_read = N × T × C_normal
C_write + (N-1) × C_read = N × C_normal
N × (C_normal - C_read) = C_write - C_read
N_breakeven = (C_write - C_read) / (C_normal - C_read)

N_breakeven = ($3.75 - $0.30) / ($3.00 - $0.30)
            = $3.45 / $2.70
            = 1.28

N_breakeven = ($2.50 - $1.25) / ($2.50 - $1.25)
            = 1.0

Cost_with_cache = (100K × $4.50/MTok/hour × H) + (N × 100K × $0.3125/MTok)
                = 0.45H + 0.03125N
 
Cost_no_cache   = N × 100K × $1.25/MTok
                = 0.125N
 
Break-even: 0.45H + 0.03125N = 0.125N
0.45H = 0.09375N
N_breakeven = 4.8 × H

from dataclasses import dataclass
from typing import Optional
 
@dataclass
class CachingPricingModel:
    """Pricing model for a single LLM provider's caching tier."""
    provider: str
    model: str
    normal_input_per_mtok: float      # $/MTok for uncached input
    cache_write_per_mtok: float       # $/MTok for cache write (0 if no premium)
    cache_read_per_mtok: float        # $/MTok for cache hit
    output_per_mtok: float            # $/MTok for output tokens
    storage_per_mtok_per_hour: float  # $/MTok/hour for stored cache (0 if not applicable)
    min_cache_tokens: int             # Minimum tokens before caching applies
    ttl_minutes: int                  # Cache TTL in minutes
 
PRICING = {
    "claude-sonnet-3-5": CachingPricingModel(
        provider="Anthropic",
        model="claude-sonnet-3-5",
        normal_input_per_mtok=3.00,
        cache_write_per_mtok=3.75,
        cache_read_per_mtok=0.30,
        output_per_mtok=15.00,
        storage_per_mtok_per_hour=0.0,
        min_cache_tokens=1024,
        ttl_minutes=5,
    ),
    "claude-haiku-3": CachingPricingModel(
        provider="Anthropic",
        model="claude-haiku-3",
        normal_input_per_mtok=0.25,
        cache_write_per_mtok=0.30,
        cache_read_per_mtok=0.03,
        output_per_mtok=1.25,
        storage_per_mtok_per_hour=0.0,
        min_cache_tokens=2048,
        ttl_minutes=5,
    ),
    "gpt-4o": CachingPricingModel(
        provider="OpenAI",
        model="gpt-4o",
        normal_input_per_mtok=2.50,
        cache_write_per_mtok=2.50,  # No write premium
        cache_read_per_mtok=1.25,
        output_per_mtok=10.00,
        storage_per_mtok_per_hour=0.0,
        min_cache_tokens=1024,
        ttl_minutes=-1,  # Managed by OpenAI
    ),
    "gemini-1-5-pro": CachingPricingModel(
        provider="Google",
        model="gemini-1-5-pro",
        normal_input_per_mtok=1.25,
        cache_write_per_mtok=1.25,  # No write premium
        cache_read_per_mtok=0.3125,
        output_per_mtok=5.00,
        storage_per_mtok_per_hour=4.50,
        min_cache_tokens=32768,
        ttl_minutes=60,
    ),
}
 
def analyze_caching_economics(
    pricing: CachingPricingModel,
    cached_tokens: int,
    uncached_tokens: int,
    output_tokens: int,
    num_calls: int,
    storage_hours: float = 1.0,
) -> dict:
    """
    Calculate costs with and without caching, including break-even point.
 
    Args:
        pricing: Provider pricing model
        cached_tokens: Tokens in the stable cached prefix
        uncached_tokens: Tokens that vary per call (not cached)
        output_tokens: Average output tokens per call
        num_calls: Number of API calls to analyze
        storage_hours: Hours the cache is stored (for Gemini)
 
    Returns:
        Dictionary with cost breakdown and break-even analysis
    """
    if cached_tokens < pricing.min_cache_tokens:
        return {
            "eligible": False,
            "reason": f"Prompt too short: {cached_tokens} < {pricing.min_cache_tokens} token minimum"
        }
 
    mtok = 1_000_000
 
    # Cost without caching
    cost_no_cache = (
        num_calls * (cached_tokens + uncached_tokens) * pricing.normal_input_per_mtok / mtok
        + num_calls * output_tokens * pricing.output_per_mtok / mtok
    )
 
    # Cost with caching: 1 write + (N-1) reads for cached portion
    cache_storage_cost = (
        cached_tokens * pricing.storage_per_mtok_per_hour / mtok * storage_hours
    )
    cost_with_cache = (
        # Cache write (first call)
        cached_tokens * pricing.cache_write_per_mtok / mtok
        # Uncached tokens on all calls at normal price
        + num_calls * uncached_tokens * pricing.normal_input_per_mtok / mtok
        # Cache reads on subsequent calls
        + max(0, num_calls - 1) * cached_tokens * pricing.cache_read_per_mtok / mtok
        # Output tokens (same either way)
        + num_calls * output_tokens * pricing.output_per_mtok / mtok
        # Storage cost (Gemini-specific)
        + cache_storage_cost
    )
 
    savings = cost_no_cache - cost_with_cache
    savings_pct = (savings / cost_no_cache) * 100 if cost_no_cache > 0 else 0
 
    # Break-even calculation
    if pricing.cache_write_per_mtok != pricing.cache_read_per_mtok:
        write_premium = pricing.cache_write_per_mtok - pricing.normal_input_per_mtok
        read_savings = pricing.normal_input_per_mtok - pricing.cache_read_per_mtok
 
        if read_savings > 0:
            breakeven_calls = (
                (write_premium * cached_tokens / mtok + cache_storage_cost)
                / (read_savings * cached_tokens / mtok)
                + 1
            )
        else:
            breakeven_calls = float('inf')
    else:
        # No write premium (OpenAI), every call benefits
        breakeven_calls = 1.0
 
    return {
        "eligible": True,
        "provider": pricing.provider,
        "model": pricing.model,
        "num_calls": num_calls,
        "cached_tokens": cached_tokens,
        "cost_no_cache": round(cost_no_cache, 4),
        "cost_with_cache": round(cost_with_cache, 4),
        "savings_dollars": round(savings, 4),
        "savings_percent": round(savings_pct, 1),
        "breakeven_calls": round(breakeven_calls, 2),
        "cache_storage_cost": round(cache_storage_cost, 4),
    }
 
# Example: 50K system prompt, 500 calls per day
for model_key, pricing in PRICING.items():
    result = analyze_caching_economics(
        pricing=pricing,
        cached_tokens=50_000,
        uncached_tokens=500,
        output_tokens=1_000,
        num_calls=500,
        storage_hours=24.0,
    )
    if result["eligible"]:
        print(f"\n{result['provider']} {result['model']}:")
        print(f"  No cache: ${result['cost_no_cache']:.2f}/day")
        print(f"  With cache: ${result['cost_with_cache']:.2f}/day")
        print(f"  Savings: ${result['savings_dollars']:.2f} ({result['savings_percent']}%)")
        print(f"  Break-even at: {result['breakeven_calls']} calls")

Anthropic claude-sonnet-3-5:
  No cache: $75.00/day
  With cache: $8.44/day
  Savings: $66.56 (88.7%)
  Break-even at: 1.28 calls
 
OpenAI gpt-4o:
  No cache: $62.50/day
  With cache: $34.38/day
  Savings: $28.12 (45.0%)
  Break-even at: 1.0 calls
 
Google gemini-1-5-pro:
  No cache: $31.25/day
  With cache: $14.51/day
  Savings: $16.74 (53.6%)
  Break-even at: 6.8 calls

# BAD: Dynamic content buried inside stable content
def build_prompt_bad(user_query: str, customer_history: str) -> list:
    return [
        {
            "role": "user",
            "content": f"""
You are a helpful support agent for Acme Corp.
 
Customer query: {user_query}
 
Customer history:
{customer_history}
 
Tools available:
- lookup_order(order_id: str) -> dict
- issue_refund(order_id: str, amount: float) -> bool
- escalate_ticket(ticket_id: str, reason: str) -> str
 
Always be polite and solution-focused.
JSON output format: {{"action": "...", "response": "..."}}
"""
        }
    ]
# Problem: user_query appears at position 130 chars in, so everything after it
# (customer_history, tool definitions) cannot be cached

# GOOD: Stable content at top, dynamic content at bottom
def build_prompt_optimal(
    user_query: str,
    customer_history: str,
    model_client,
) -> dict:
    """
    Structure: system (stable) → user (semi-stable history + fresh query)
    Cache boundary: after tool definitions, before conversation history
    """
 
    # Layer 1: Stable system instructions + tools (mark for caching)
    stable_system = """You are a helpful support agent for Acme Corp.
Always be polite and solution-focused. Resolve issues on first contact.
 
## Available Tools
 
### lookup_order
Look up details for an order by ID.
Parameters: order_id (string, required)
Returns: {order_id, status, items, total, created_at}
 
### issue_refund
Process a refund for an order.
Parameters: order_id (string, required), amount (float, required)
Returns: {success: bool, refund_id: string}
 
### escalate_ticket
Escalate to human support.
Parameters: ticket_id (string, required), reason (string, required)
Returns: {escalation_id: string, eta_hours: int}
 
## Output Format
Always respond with valid JSON:
{"action": "tool_call|direct_response", "tool": "...", "parameters": {...}, "response": "..."}"""
 
    # Layer 2: Semi-stable (cache with conversation)
    customer_context = f"""## Customer Account History
{customer_history}
 
## Current Session Start"""
 
    # Layer 3: Dynamic (never cached)
    current_query = f"\n\nCustomer: {user_query}"
 
    # Anthropic: use cache_control breakpoints
    return {
        "system": [
            {
                "type": "text",
                "text": stable_system,
                "cache_control": {"type": "ephemeral"},  # Cache layer 1
            }
        ],
        "messages": [
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": customer_context,
                        "cache_control": {"type": "ephemeral"},  # Cache layer 2
                    },
                    {
                        "type": "text",
                        "text": current_query,
                        # No cache_control → not cached (changes every call)
                    }
                ]
            }
        ]
    }

def build_conversation_with_caching(
    system_prompt: str,
    conversation_history: list[dict],
    new_user_message: str,
    cache_turns_threshold: int = 6,
) -> dict:
    """
    Cache the first N turns, leave the most recent turn uncached.
    Rebuild the cache checkpoint when conversation grows.
    """
    messages = []
 
    # Add conversation history with cache on the last stable checkpoint
    history_len = len(conversation_history)
    cache_up_to = max(0, history_len - 2)  # Cache all but last 2 turns
 
    for i, turn in enumerate(conversation_history):
        is_last_stable = (i == cache_up_to - 1)
 
        if is_last_stable and history_len >= cache_turns_threshold:
            # Mark this as the cache checkpoint
            messages.append({
                "role": turn["role"],
                "content": [
                    {
                        "type": "text",
                        "text": turn["content"],
                        "cache_control": {"type": "ephemeral"},
                    }
                ]
            })
        else:
            messages.append(turn)
 
    # Add new user message (uncached)
    messages.append({"role": "user", "content": new_user_message})
 
    return {
        "system": [
            {
                "type": "text",
                "text": system_prompt,
                "cache_control": {"type": "ephemeral"},
            }
        ],
        "messages": messages
    }

import anthropic
from typing import Any
 
client = anthropic.Anthropic()
 
def query_with_cache(
    large_document: str,
    question: str,
    model: str = "claude-opus-4-6",
) -> dict[str, Any]:
    """
    Document Q&A with prompt caching.
    The document is cached; only the question varies per call.
    """
    response = client.messages.create(
        model=model,
        max_tokens=2048,
        system=[
            {
                "type": "text",
                "text": (
                    "You are a precise document analyst. "
                    "Answer questions based only on the provided document. "
                    "If the answer is not in the document, say 'Not found in document.'"
                ),
                "cache_control": {"type": "ephemeral"},
            },
            {
                "type": "text",
                "text": f"<document>\n{large_document}\n</document>",
                "cache_control": {"type": "ephemeral"},
            },
        ],
        messages=[
            {
                "role": "user",
                "content": question,
                # No cache_control: varies per request
            }
        ],
    )
 
    usage = response.usage
    return {
        "answer": response.content[0].text,
        "cache_creation_tokens": usage.cache_creation_input_tokens,
        "cache_read_tokens": usage.cache_read_input_tokens,
        "uncached_input_tokens": usage.input_tokens,
        "output_tokens": usage.output_tokens,
        "cache_hit": usage.cache_read_input_tokens > 0,
    }
 
# Batch questions against same document
questions = [
    "What are the payment terms?",
    "Who are the authorized signatories?",
    "What is the termination clause?",
    "What are the liability caps?",
]
 
for i, question in enumerate(questions):
    result = query_with_cache(large_document=contract_text, question=question)
    status = "CACHE HIT" if result["cache_hit"] else "CACHE MISS"
    print(f"Q{i+1} [{status}]: read={result['cache_read_tokens']:,} tokens")
# Output:
# Q1 [CACHE MISS]: read=0 tokens      ← first call writes cache
# Q2 [CACHE HIT]:  read=45,230 tokens ← subsequent calls read cache
# Q3 [CACHE HIT]:  read=45,230 tokens
# Q4 [CACHE HIT]:  read=45,230 tokens

from openai import OpenAI
from functools import lru_cache
 
client = OpenAI()
 
# CRITICAL: The system prompt must be byte-for-byte identical across requests.
# Load it once; do not format or modify it at call time.
SYSTEM_PROMPT = open("system_prompt.txt").read()  # Load once at startup
 
def query_openai_with_cache_monitoring(
    user_message: str,
    model: str = "gpt-4o",
) -> dict[str, Any]:
    """
    OpenAI caches automatically. We just need to ensure the system prompt
    is identical across requests (same object reference or string value).
    """
    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_message},
        ],
        max_tokens=1024,
    )
 
    usage = response.usage
    cached_tokens = getattr(
        getattr(usage, "prompt_tokens_details", None),
        "cached_tokens",
        0
    )
 
    return {
        "content": response.choices[0].message.content,
        "total_input_tokens": usage.prompt_tokens,
        "cached_tokens": cached_tokens,
        "uncached_tokens": usage.prompt_tokens - cached_tokens,
        "output_tokens": usage.completion_tokens,
        "cache_ratio": cached_tokens / max(usage.prompt_tokens, 1),
    }

import google.generativeai as genai
from google.generativeai import caching
import datetime
 
genai.configure(api_key="YOUR_API_KEY")
 
def create_gemini_cache(
    large_content: str,
    ttl_minutes: int = 60,
    model: str = "models/gemini-1.5-pro-001",
) -> str:
    """
    Create a named Gemini context cache. Returns the cache name for reuse.
    Minimum 32,768 tokens required.
    """
    cache = caching.CachedContent.create(
        model=model,
        contents=[
            {
                "role": "user",
                "parts": [{"text": large_content}]
            }
        ],
        system_instruction=(
            "You are an expert analyst. Answer questions accurately "
            "and cite specific sections from the provided content."
        ),
        ttl=datetime.timedelta(minutes=ttl_minutes),
    )
    print(f"Cache created: {cache.name}")
    print(f"Token count: {cache.usage_metadata.total_token_count:,}")
    return cache.name
 
def query_with_gemini_cache(cache_name: str, question: str) -> str:
    """Query using an existing Gemini context cache."""
    cache = caching.CachedContent.get(cache_name)
    model = genai.GenerativeModel.from_cached_content(cached_content=cache)
    response = model.generate_content(question)
    return response.text
 
# Usage pattern
cache_name = create_gemini_cache(large_content=corpus_text)  # One-time setup
answers = [query_with_gemini_cache(cache_name, q) for q in questions]

import hashlib
from anthropic import Anthropic
 
client = Anthropic()
 
class CachedRAGPipeline:
    """
    RAG pipeline that caches retrieved context to reduce input token costs.
 
    Strategy: if the same chunks are retrieved for multiple queries (common
    in domain-specific corpora), the second query onward hits the cache.
    """
 
    def __init__(self, retriever, model: str = "claude-sonnet-3-5"):
        self.retriever = retriever
        self.model = model
        self._last_chunk_hash = None
        self._last_chunks = None
 
    def _hash_chunks(self, chunks: list[str]) -> str:
        content = "\n---\n".join(sorted(chunks))
        return hashlib.sha256(content.encode()).hexdigest()[:16]
 
    def query(self, user_question: str, top_k: int = 5) -> dict:
        chunks = self.retriever.retrieve(user_question, top_k=top_k)
        chunk_hash = self._hash_chunks(chunks)
 
        context_text = "\n\n".join([
            f"[Document {i+1}]\n{chunk}"
            for i, chunk in enumerate(chunks)
        ])
 
        # Build messages with cache control on the context
        response = client.messages.create(
            model=self.model,
            max_tokens=1024,
            system=[
                {
                    "type": "text",
                    "text": (
                        "You are a helpful assistant. Answer questions based on "
                        "the provided context. If the context doesn't contain "
                        "the answer, say so clearly."
                    ),
                    "cache_control": {"type": "ephemeral"},
                },
                {
                    "type": "text",
                    "text": f"<context>\n{context_text}\n</context>",
                    "cache_control": {"type": "ephemeral"},
                },
            ],
            messages=[{"role": "user", "content": user_question}],
        )
 
        cache_hit = response.usage.cache_read_input_tokens > 0
        same_chunks = chunk_hash == self._last_chunk_hash
 
        self._last_chunk_hash = chunk_hash
        self._last_chunks = chunks
 
        return {
            "answer": response.content[0].text,
            "cache_hit": cache_hit,
            "same_chunks_as_last": same_chunks,
            "context_tokens": (
                response.usage.cache_read_input_tokens
                or response.usage.cache_creation_input_tokens
            ),
        }

def generate_chunk_contexts_with_caching(
    document: str,
    chunks: list[str],
    model: str = "claude-haiku-3",
) -> list[str]:
    """
    Generate contextual summaries for each chunk.
    Cache the full document (stable across all chunks); only the
    specific chunk content changes per request.
 
    Cost: ~$1.02/MTok total (vs $3.60 without caching)
    because document is cached after first chunk, read for all subsequent.
    """
    contexts = []
 
    for chunk in chunks:
        response = client.messages.create(
            model=model,
            max_tokens=150,
            system=[
                {
                    "type": "text",
                    "text": (
                        "Generate a brief 50-75 word context that situates the "
                        "given chunk within the overall document. Focus on what "
                        "the chunk is about and why it matters in context."
                    ),
                    "cache_control": {"type": "ephemeral"},
                },
                {
                    "type": "text",
                    "text": f"<full_document>\n{document}\n</full_document>",
                    "cache_control": {"type": "ephemeral"},
                    # This is the expensive part: cached after first chunk
                },
            ],
            messages=[
                {
                    "role": "user",
                    "content": (
                        f"Generate context for this chunk:\n\n"
                        f"<chunk>\n{chunk}\n</chunk>"
                    )
                }
            ],
        )
        contexts.append(response.content[0].text)
 
    return contexts

import anthropic
import json
from typing import Callable
 
client = anthropic.Anthropic()
 
TOOLS = [
    {
        "name": "search_web",
        "description": "Search the web for current information on any topic.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search query"},
                "num_results": {"type": "integer", "default": 5}
            },
            "required": ["query"]
        }
    },
    {
        "name": "execute_code",
        "description": "Execute Python code and return the output.",
        "input_schema": {
            "type": "object",
            "properties": {
                "code": {"type": "string"},
                "timeout_seconds": {"type": "integer", "default": 30}
            },
            "required": ["code"]
        }
    },
    # ... more tools
]
 
# Pre-serialize tools once: same bytes on every call
TOOLS_JSON = json.dumps(TOOLS, indent=2)
 
AGENT_SYSTEM_PROMPT = f"""You are a precise research and analysis agent.
 
## Capabilities
You have access to web search and code execution tools.
Use tools judiciously; only when needed to answer the question accurately.
 
## Reasoning Pattern
1. Analyze the user's request
2. Identify what information or computation is needed
3. Use tools to gather/compute
4. Synthesize results into a clear answer
 
## Tools Available
{TOOLS_JSON}
 
## Output Format
Think step by step. Use <thinking> tags for reasoning, then give your final answer."""
 
class CachedAgentLoop:
    """
    Agent loop that maximizes cache utilization across iterations.
 
    Cache layers:
    - Layer 1: System prompt + tool definitions (never changes)
    - Layer 2: Conversation history up to current turn (stable portion)
    - Uncached: Latest user message and pending tool responses
    """
 
    def __init__(self, model: str = "claude-opus-4-6"):
        self.model = model
        self.conversation: list[dict] = []
        self.total_cache_writes = 0
        self.total_cache_reads = 0
        self.total_input_tokens = 0
 
    def _build_messages_with_cache(self, new_user_message: str) -> list[dict]:
        """Add cache control to stable conversation history."""
        messages = []
 
        # Mark all but the last 2 turns as cacheable
        cache_boundary = max(0, len(self.conversation) - 2)
 
        for i, msg in enumerate(self.conversation):
            if i == cache_boundary - 1 and len(self.conversation) >= 4:
                # Cache checkpoint on the last stable turn
                if isinstance(msg["content"], str):
                    messages.append({
                        "role": msg["role"],
                        "content": [
                            {
                                "type": "text",
                                "text": msg["content"],
                                "cache_control": {"type": "ephemeral"},
                            }
                        ]
                    })
                else:
                    messages.append(msg)
            else:
                messages.append(msg)
 
        # New message is never cached
        messages.append({"role": "user", "content": new_user_message})
        return messages
 
    def step(
        self,
        user_message: str,
        tool_executor: Callable[[str, dict], str],
    ) -> str:
        """Run one complete agent turn with tool calls."""
        messages = self._build_messages_with_cache(user_message)
 
        while True:
            response = client.messages.create(
                model=self.model,
                max_tokens=4096,
                system=[
                    {
                        "type": "text",
                        "text": AGENT_SYSTEM_PROMPT,
                        "cache_control": {"type": "ephemeral"},
                    }
                ],
                messages=messages,
            )
 
            # Track cache usage
            self.total_cache_writes += response.usage.cache_creation_input_tokens
            self.total_cache_reads += response.usage.cache_read_input_tokens
            self.total_input_tokens += response.usage.input_tokens
 
            # Check stop reason
            if response.stop_reason == "end_turn":
                final_text = response.content[-1].text
                # Update conversation history for next turn
                self.conversation.append({"role": "user", "content": user_message})
                self.conversation.append({"role": "assistant", "content": final_text})
                return final_text
 
            # Handle tool calls
            tool_calls = [b for b in response.content if b.type == "tool_use"]
            tool_results = []
 
            for tool_call in tool_calls:
                result = tool_executor(tool_call.name, tool_call.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": tool_call.id,
                    "content": result,
                })
 
            # Append assistant response and tool results
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})
 
    def get_cache_stats(self) -> dict:
        total_billable = self.total_cache_writes + self.total_cache_reads + self.total_input_tokens
        return {
            "cache_write_tokens": self.total_cache_writes,
            "cache_read_tokens": self.total_cache_reads,
            "uncached_tokens": self.total_input_tokens,
            "cache_hit_rate": (
                self.total_cache_reads / max(total_billable, 1) * 100
            ),
        }

import time
from dataclasses import dataclass, field
from collections import defaultdict
from typing import Optional
import anthropic
 
@dataclass
class CacheMetrics:
    """Per-call caching metrics."""
    timestamp: float
    endpoint: str
    cache_write_tokens: int
    cache_read_tokens: int
    uncached_input_tokens: int
    output_tokens: int
    latency_ms: float
 
    @property
    def total_input_tokens(self) -> int:
        return self.cache_write_tokens + self.cache_read_tokens + self.uncached_input_tokens
 
    @property
    def cache_hit_rate(self) -> float:
        if self.total_input_tokens == 0:
            return 0.0
        return self.cache_read_tokens / self.total_input_tokens
 
    def cost_anthropic_sonnet(self) -> float:
        """Calculate actual cost in dollars."""
        return (
            self.cache_write_tokens * 3.75e-6
            + self.cache_read_tokens * 0.30e-6
            + self.uncached_input_tokens * 3.00e-6
            + self.output_tokens * 15.00e-6
        )
 
    def counterfactual_cost(self) -> float:
        """Cost if caching were disabled."""
        return (
            self.total_input_tokens * 3.00e-6
            + self.output_tokens * 15.00e-6
        )
 
class CacheMonitor:
    """Production cache efficiency monitor."""
 
    def __init__(self):
        self._metrics: list[CacheMetrics] = []
        self._by_endpoint: dict[str, list[CacheMetrics]] = defaultdict(list)
 
    def record(self, metrics: CacheMetrics) -> None:
        self._metrics.append(metrics)
        self._by_endpoint[metrics.endpoint].append(metrics)
 
    def summary(self, endpoint: Optional[str] = None) -> dict:
        data = self._by_endpoint[endpoint] if endpoint else self._metrics
        if not data:
            return {}
 
        total_calls = len(data)
        total_cost = sum(m.cost_anthropic_sonnet() for m in data)
        counterfactual = sum(m.counterfactual_cost() for m in data)
        avg_hit_rate = sum(m.cache_hit_rate for m in data) / total_calls
 
        return {
            "total_calls": total_calls,
            "avg_cache_hit_rate": f"{avg_hit_rate * 100:.1f}%",
            "total_cost": f"${total_cost:.4f}",
            "counterfactual_cost": f"${counterfactual:.4f}",
            "savings": f"${counterfactual - total_cost:.4f}",
            "savings_pct": f"{(1 - total_cost/counterfactual) * 100:.1f}%" if counterfactual > 0 else "N/A",
            "zero_hit_calls": sum(1 for m in data if m.cache_hit_rate == 0),
        }
 
    def alert_low_cache_rate(self, threshold: float = 0.5) -> list[str]:
        """Return endpoint names with suspiciously low cache hit rates."""
        alerts = []
        for endpoint, metrics in self._by_endpoint.items():
            if len(metrics) < 10:
                continue
            avg_hit = sum(m.cache_hit_rate for m in metrics[-50:]) / min(len(metrics), 50)
            if avg_hit < threshold:
                alerts.append(
                    f"LOW CACHE HIT: {endpoint} averaging {avg_hit*100:.1f}% "
                    f"(threshold: {threshold*100:.0f}%)"
                )
        return alerts

# BAD: Timestamp breaks the cache for everything after it
system = f"""
You are a helpful assistant.
Current time: {datetime.now().isoformat()}  # ← Cache breaker!
User ID: {user_id}                          # ← Cache breaker!
 
[40,000 tokens of instructions and examples follow]
"""
 
# GOOD: Stable system prompt, dynamic context in first user message
system = """
You are a helpful assistant.
[40,000 tokens of instructions and examples]
"""
 
messages = [
    {
        "role": "user",
        "content": f"[Context: {datetime.now().strftime('%Y-%m-%d')}, User: {user_id}]\n\n{actual_question}"
    }
]

# BAD: Dynamic tool filtering
tools = [t for t in ALL_TOOLS if t["name"] in user_permissions[user_id]]
# Different users → different tool lists → no cache hits
 
# GOOD: Cache the full schema; handle authorization in tool execution
tools = ALL_TOOLS  # Always the same bytes
# In tool executor: check permissions before running the tool

# BAD: f-string indentation that varies with content
prompt = f"""
    System: {system_text}
    Examples: {examples_text}
    Instructions: {instructions}
"""
# Leading spaces vary if text wraps, causing cache miss
 
# GOOD: No formatting ambiguity
prompt = f"System: {system_text}\nExamples: {examples_text}\nInstructions: {instructions}"

# BAD: Slow batch with gap > TTL
for record in records:  # 10 min/record
    process(record)  # Always cache miss
    time.sleep(600)
 
# GOOD: Burst processing to keep cache warm
BATCH_SIZE = 50
for i in range(0, len(records), BATCH_SIZE):
    batch = records[i:i + BATCH_SIZE]
    results = [process(r) for r in batch]  # 50 calls in <5 min → cache stays warm
    store_results(results)
    # Optional: small pause between batches if needed

<author-bio>