Prompt Caching for LLMs: Reducing Latency and Cost at Scale

def kv_cache_size(tokens: int, layers: int, hidden_size: int, kv_heads: int,
                  dtype_bits: int = 16) -> float:
    """Calculate KV cache memory requirements in GB."""
    head_dim = hidden_size // kv_heads
    bytes_per_value = dtype_bits // 8
    total_bytes = 2 * layers * tokens * kv_heads * head_dim * bytes_per_value
    return total_bytes / (1024 ** 3)

# LLaMA-70B: 4096 tokens = ~40GB; 1M tokens = ~2.6TB
print(f"KV cache for 70B model at 4K: {kv_cache_size(4096, 80, 8192, 64):.1f} GB")
print(f"KV cache for 70B model at 1M: {kv_cache_size(1000000, 80, 8192, 64):.1f} GB")

class ProviderPrefixCache:
    """Implement prefix caching across LLM providers."""

    def __init__(self, provider: str = "anthropic"):
        self.provider = provider
        self.cache = {}

    def build_prompt(self, system: str, tools: list, user_query: str) -> dict:
        """Structure prompt with cacheable prefix."""
        return {
            "system": system,
            "tools": tools,
            "messages": [
                {"role": "user", "content": user_query}
            ]
        }

    def estimate_savings(self, prompt: dict, cache_hit: bool) -> dict:
        """Estimate cost and latency savings from caching."""
        system_tokens = self._count_tokens(prompt["system"])
        tools_tokens = self._count_tokens(str(prompt["tools"]))
        query_tokens = self._count_tokens(prompt["messages"][0]["content"])

        total = system_tokens + tools_tokens + query_tokens
        cached = system_tokens + tools_tokens

        return {
            "total_tokens": total,
            "cached_tokens": cached,
            "savings_pct": (cached / total) * 100 if total > 0 else 0,
            "cache_hit": cache_hit
        }

    def _count_tokens(self, text: str) -> int:
        return len(text.split()) * 1.3  # Approximate

import hashlib
import json
import time
from typing import Optional

class ExactResponseCache:
    """Cache complete LLM responses for identical prompts."""

    def __init__(self, ttl: int = 3600):
        self.cache = {}
        self.ttl = ttl
        self.hits = 0
        self.misses = 0

    def _make_key(self, prompt: str, model: str, params: dict) -> str:
        content = json.dumps({"prompt": prompt, "model": model, "params": params}, sort_keys=True)
        return hashlib.sha256(content.encode()).hexdigest()

    def get(self, prompt: str, model: str = "default", **params) -> Optional[str]:
        key = self._make_key(prompt, model, params)
        entry = self.cache.get(key)
        if entry and entry["expires"] > time.time():
            self.hits += 1
            return entry["response"]
        self.misses += 1
        return None

    def set(self, prompt: str, response: str, model: str = "default", **params):
        key = self._make_key(prompt, model, params)
        self.cache[key] = {
            "response": response,
            "expires": time.time() + self.ttl,
            "created": time.time()
        }

    def hit_rate(self) -> float:
        total = self.hits + self.misses
        return self.hits / total if total > 0 else 0.0

import numpy as np
from typing import Tuple

class SemanticResponseCache:
    """Cache responses for semantically similar prompts."""

    def __init__(self, embedding_model, similarity_threshold: float = 0.92):
        self.embedding_model = embedding_model
        self.similarity_threshold = similarity_threshold
        self.cache = []

    def embed(self, text: str) -> np.ndarray:
        return self.embedding_model.encode(text)

    def find_similar(self, prompt: str) -> Tuple[Optional[str], float]:
        prompt_embedding = self.embed(prompt)
        best_match = None
        best_score = 0.0

        for cached_prompt, cached_response, cached_embedding in self.cache:
            similarity = np.dot(prompt_embedding, cached_embedding) / (
                np.linalg.norm(prompt_embedding) * np.linalg.norm(cached_embedding)
            )
            if similarity > best_score:
                best_score = similarity
                best_match = cached_response

        if best_score >= self.similarity_threshold:
            return best_match, best_score
        return None, 0.0

    def store(self, prompt: str, response: str):
        embedding = self.embed(prompt)
        self.cache.append((prompt, response, embedding))

class MultiTierCache:
    """Three-tier caching for maximum cost reduction."""

    def __init__(self, embedding_model, provider_config: dict):
        self.l1 = ExactResponseCache(ttl=300)
        self.l2 = SemanticResponseCache(embedding_model, similarity_threshold=0.95)
        self.l3 = ProviderPrefixCache(provider_config.get("provider", "anthropic"))

    def get(self, prompt: str, **kwargs) -> Tuple[Optional[str], str]:
        result = self.l1.get(prompt, **kwargs)
        if result:
            return result, "L1"

        result, _ = self.l2.find_similar(prompt)
        if result:
            return result, "L2"

        return None, "miss"

    def store(self, prompt: str, response: str, **kwargs):
        self.l1.set(prompt, response, **kwargs)
        self.l2.store(prompt, response)

import anthropic

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-sonnet-4-20260514",
    max_tokens=1024,
    system="You are a legal document analyst...",
    messages=[{"role": "user", "content": "Analyze this contract clause:"}],
    extra_headers={
        "anthropic-beta": "prompt-caching-2025-02-19",
        "anthropic-cache-control": {"type": "ephemeral"}
    }
)

print(f"Cache hit: {response.headers.get('x-cache-hit')}")
print(f"Input tokens: {response.usage.input_tokens}")
print(f"Cached tokens: {response.usage.cache_read_input_tokens}")

from openai import OpenAI

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4.1",
    messages=[
        {"role": "system", "content": "You are a code review assistant..."},
        {"role": "user", "content": "Review this pull request:"}
    ]
)

usage = response.usage
print(f"Input tokens: {usage.prompt_tokens}")
print(f"Cached tokens: {getattr(usage, 'prompt_tokens_details', {}).get('cached_tokens', 0)}")

import google.generativeai as genai

# Create context cache
cache = genai.caching.CachedContent.create(
    model='models/gemini-3.1-pro-001',
    system_instruction="Analyze financial documents...",
    contents=[{"role": "user", "parts": [{"text": "Initial setup context"}]}],
    ttl="3600s"
)

# Use cached context
model = genai.GenerativeModel.from_cached_content(cached_content=cache)
response = model.generate_content("What are the risks in this document?")

# vLLM server configuration
engine_config:
  model: meta-llama/Llama-3.3-70B-Instruct
  max_model_len: 131072
  enable_prefix_caching: true
  gpu_memory_utilization: 0.85
  block_size: 16
  max_num_seqs: 256

Cost Savings = (Cached Prompt Tokens / Total Tokens) × Request Count × Token Price

Example:
- 10,000 requests/hour
- 8,000 token system prompt (cached)
- 500 token user query (unique)
- Anthropic Sonnet 4: $3.00/$0.30 per M tokens

Without caching: 8,500 × 10,000 / 1,000,000 × $3.00 = $255/hour
With caching:   500 × 10,000 / 1,000,000 × $3.00 = $15/hour
              + 8,000 × 10,000 / 1,000,000 × $0.30 = $24/hour
              = $39/hour → 85% cost reduction

System: You are a customer support agent for Acme Corp.
          Our return policy: [static policy text]
          Our shipping policy: [static policy text]

User: [dynamic query]

Layer 1: System prompt (cacheable for days)
Layer 2: Tool/function definitions (cacheable across sessions)
Layer 3: Domain reference docs (cacheable per session)
Layer 4: Conversation history (partially cacheable)
Layer 5: Current user query (never cacheable)

class SessionCache:
    def __init__(self, session_id: str, system_prompt: str):
        self.session_id = session_id
        self.cached_prefix = system_prompt
        self.conversation_history = []

    def add_message(self, role: str, content: str):
        self.conversation_history.append({"role": role, "content": content})

    def build_prompt(self) -> str:
        system = self.cached_prefix
        history = "\n".join(
            f"{m['role']}: {m['content']}"
            for m in self.conversation_history[-10:]
        )
        return f"{system}\n{history}\nUser: "

class vCache:
    """Semantic caching with learned thresholds and error guarantees."""

    def __init__(self, embedding_model, max_error_rate: float = 0.01):
        self.embedding_model = embedding_model
        self.max_error_rate = max_error_rate
        self.thresholds = {}  # prompt_type -> optimal threshold
        self.cache = []
        self.accuracy_log = []

    def find(self, prompt: str, prompt_type: str = "default") -> Tuple[Optional[str], float]:
        threshold = self.thresholds.get(prompt_type, 0.95)
        prompt_embedding = self.embed(prompt)
        best_match = None
        best_score = 0.0

        for entry in self.cache:
            score = self._similarity(prompt_embedding, entry["embedding"])
            if score > best_score:
                best_score = score
                best_match = entry["response"]

        if best_score >= threshold:
            return best_match, best_score
        return None, 0.0

    def log_accuracy(self, prompt: str, cached_response: str, actual_response: str):
        match = cached_response == actual_response
        self.accuracy_log.append(match)

    def update_threshold(self, prompt_type: str):
        recent = self.accuracy_log[-1000:]
        error_rate = 1.0 - (sum(recent) / len(recent)) if recent else 0
        if error_rate > self.max_error_rate:
            self.thresholds[prompt_type] = self.thresholds.get(prompt_type, 0.95) + 0.01

    def _similarity(self, a: np.ndarray, b: np.ndarray) -> float:
        return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))

    def embed(self, text: str) -> np.ndarray:
        return self.embedding_model.encode(text)

class CacheInvalidator:
    """Smart cache invalidation with multiple strategies."""

    def __init__(self, cache):
        self.cache = cache
        this.version = 1

    def invalidate_on_model_update(self, new_version: int):
        if new_version > this.version:
            self.cache.clear()
            this.version = new_version

    def invalidate_stale_entries(self, min_hit_rate: float = 0.01):
        for key, entry in list(self.cache.items()):
            age = time.time() - entry["created"]
            hit_rate = entry.get("hits", 0) / max(entry.get("lookups", 1), 1)
            if age > 86400 and hit_rate < min_hit_rate:
                del self.cache[key]

[BEGIN CACHE ZONE]
System instructions, tool definitions, few-shot examples, reference documents
[END CACHE ZONE]
User-specific query (always unique, never cached)

class CacheMonitor:
    def __init__(self):
        self.metrics = {
            "l1_hits": 0, "l1_misses": 0,
            "l2_hits": 0, "l2_misses": 0,
            "l3_hits": 0, "l3_misses": 0,
            "total_requests": 0,
            "tokens_saved": 0,
            "latency_saved_ms": 0
        }

    def log_request(self, cache_tier: str, hit: bool, tokens_saved: int, latency_saved_ms: int):
        key = f"{cache_tier.lower()}_{'hits' if hit else 'misses'}"
        self.metrics[key] += 1
        self.metrics["total_requests"] += 1
        self.metrics["tokens_saved"] += tokens_saved
        self.metrics["latency_saved_ms"] += latency_saved_ms

    def report(self) -> dict:
        total = self.metrics["l1_hits"] + self.metrics["l1_misses"]
        return {
            "overall_hit_rate": (self.metrics["l1_hits"] + self.metrics["l2_hits"]) / max(total, 1),
            "tokens_saved": self.metrics["tokens_saved"],
            "cost_saved_usd": self.metrics["tokens_saved"] / 1_000_000 * 3.00,
            "latency_saved_seconds": self.metrics["latency_saved_ms"] / 1000
        }

def warm_cache_from_logs(cache, log_file: str, top_n: int = 500):
    """Pre-populate cache from historical request logs."""
    from collections import Counter
    prefix_counts = Counter()
    with open(log_file) as f:
        for line in f:
            prefix = line.strip()[:200]
            prefix_counts[prefix] += 1
    for prefix, _ in prefix_counts.most_common(top_n):
        cache.prefetch(prefix)

class CacheDashboard:
    def generate_report(self, cache_stats: dict) -> dict:
        total = cache_stats["hits"] + cache_stats["misses"]
        hit_rate = cache_stats["hits"] / max(total, 1)
        cost_without_cache = cache_stats["total_tokens"] * 3.00 / 1_000_000
        cost_with_cache = (
            cache_stats["fresh_tokens"] * 3.00 / 1_000_000 +
            cache_stats["cached_tokens"] * 0.30 / 1_000_000
        )
        return {
            "period": cache_stats.get("period", "24h"),
            "total_requests": total,
            "hit_rate": f"{hit_rate:.1%}",
            "tokens_saved": f"{cache_stats['cached_tokens']:,}",
            "cost_without_cache": f"${cost_without_cache:.2f}",
            "cost_with_cache": f"${cost_with_cache:.2f}",
            "savings": f"{(1 - cost_with_cache / max(cost_without_cache, 0.01)):.1%}",
            "avg_latency_hit_ms": cache_stats.get("avg_hit_latency_ms", 0),
            "avg_latency_miss_ms": cache_stats.get("avg_miss_latency_ms", 0),
            "latency_improvement": f"{(1 - cache_stats.get('avg_hit_latency_ms', 0) / max(cache_stats.get('avg_miss_latency_ms', 1), 0.01)):.1%}"
        }