Cache & Optimization Patterns

AI inference is expensive and slow compared to traditional computing. Caching and optimization patterns can reduce costs by 50-90%, cut latency from seconds to milliseconds, and make AI applications economically viable at scale.

Why Caching Matters for AI

Every LLM API call costs money and takes time. A single GPT-4 call might cost $0.03-0.10 and take 2-10 seconds. At 100,000 requests per day, that is $3,000-10,000 daily. Caching identical or similar requests can dramatically reduce these costs while also improving response times.

Exact Match Caching

The simplest caching strategy: if the exact same prompt has been seen before, return the cached response. This works best for deterministic queries like data lookups, fixed templates, and system-generated prompts.

• Hash the prompt: Create a hash of the full prompt (including system message, temperature, and other parameters) as the cache key
• Store the response: Cache the complete API response including usage metadata
• TTL management: Set appropriate time-to-live based on how quickly the underlying data changes
• Limitations: Even tiny differences (a single space, different capitalization) cause cache misses. Only effective for highly repetitive, identical queries

Semantic Caching

Semantic caching goes beyond exact matches by recognizing that differently-worded queries can have the same meaning. It uses embedding similarity to match new queries against cached ones:

1. When a new query arrives, compute its embedding vector
2. Search the cache for stored queries whose embeddings are within a similarity threshold (e.g., cosine similarity > 0.95)
3. If a match is found, return the cached response
4. If no match, call the LLM, store both the query embedding and the response

Semantic Cache with Redis and Embeddings

import json
import hashlib
import time
import numpy as np
from typing import Optional
from dataclasses import dataclass


@dataclass
class CacheEntry:
    query: str
    embedding: list[float]
    response: str
    created_at: float
    ttl: int
    hit_count: int = 0


class SemanticCache:
    """Cache that matches queries by semantic similarity."""

    def __init__(self, similarity_threshold: float = 0.95,
                 max_entries: int = 10000,
                 default_ttl: int = 3600):
        self.threshold = similarity_threshold
        self.max_entries = max_entries
        self.default_ttl = default_ttl
        self.entries: list[CacheEntry] = []
        self.stats = {"hits": 0, "misses": 0}

    def _cosine_similarity(self, a: list[float],
                           b: list[float]) -> float:
        """Compute cosine similarity between two vectors."""
        a_np, b_np = np.array(a), np.array(b)
        return float(
            np.dot(a_np, b_np)
            / (np.linalg.norm(a_np) * np.linalg.norm(b_np))
        )

    def get(self, query: str,
            query_embedding: list[float]) -> Optional[str]:
        """Find a semantically similar cached response."""
        now = time.time()
        best_match = None
        best_similarity = 0.0

        for entry in self.entries:
            # Skip expired entries
            if now - entry.created_at > entry.ttl:
                continue

            similarity = self._cosine_similarity(
                query_embedding, entry.embedding
            )

            if (similarity > self.threshold
                    and similarity > best_similarity):
                best_similarity = similarity
                best_match = entry

        if best_match:
            best_match.hit_count += 1
            self.stats["hits"] += 1
            return best_match.response

        self.stats["misses"] += 1
        return None

    def put(self, query: str, query_embedding: list[float],
            response: str, ttl: int = None):
        """Store a query-response pair in the cache."""
        entry = CacheEntry(
            query=query,
            embedding=query_embedding,
            response=response,
            created_at=time.time(),
            ttl=ttl or self.default_ttl
        )
        self.entries.append(entry)
        self._evict_if_needed()

    def _evict_if_needed(self):
        """Remove expired and least-used entries."""
        now = time.time()
        # Remove expired
        self.entries = [
            e for e in self.entries
            if now - e.created_at <= e.ttl
        ]
        # If still over limit, remove least-hit entries
        if len(self.entries) > self.max_entries:
            self.entries.sort(key=lambda e: e.hit_count)
            self.entries = self.entries[-self.max_entries:]

    def get_hit_rate(self) -> float:
        total = self.stats["hits"] + self.stats["misses"]
        return self.stats["hits"] / max(total, 1)


# Usage example
async def cached_llm_call(cache: SemanticCache,
                          query: str,
                          embed_fn, llm_fn):
    """LLM call with semantic caching."""
    embedding = await embed_fn(query)

    # Check cache
    cached = cache.get(query, embedding)
    if cached:
        return {"response": cached, "cached": True}

    # Cache miss - call LLM
    response = await llm_fn(query)
    cache.put(query, embedding, response)
    return {"response": response, "cached": False}

Prompt Cache with TTL and LRU

For exact match caching, this implementation combines Time-To-Live (TTL) expiration with Least Recently Used (LRU) eviction:

from collections import OrderedDict
import hashlib
import time
import json


class PromptCache:
    """Exact-match prompt cache with TTL and LRU eviction."""

    def __init__(self, max_size: int = 5000,
                 default_ttl: int = 1800):
        self.max_size = max_size
        self.default_ttl = default_ttl
        self.cache: OrderedDict = OrderedDict()
        self.stats = {"hits": 0, "misses": 0, "evictions": 0}

    def _make_key(self, prompt: str, model: str,
                  temperature: float) -> str:
        """Create a deterministic cache key."""
        key_data = json.dumps({
            "prompt": prompt,
            "model": model,
            "temperature": temperature
        }, sort_keys=True)
        return hashlib.sha256(key_data.encode()).hexdigest()

    def get(self, prompt: str, model: str,
            temperature: float = 0.0) -> Optional[str]:
        """Look up a cached response."""
        key = self._make_key(prompt, model, temperature)

        if key not in self.cache:
            self.stats["misses"] += 1
            return None

        entry = self.cache[key]

        # Check TTL
        if time.time() - entry["created"] > entry["ttl"]:
            del self.cache[key]
            self.stats["misses"] += 1
            return None

        # Move to end (most recently used)
        self.cache.move_to_end(key)
        self.stats["hits"] += 1
        return entry["response"]

    def put(self, prompt: str, model: str,
            response: str, temperature: float = 0.0,
            ttl: int = None):
        """Store a response in the cache."""
        key = self._make_key(prompt, model, temperature)
        self.cache[key] = {
            "response": response,
            "created": time.time(),
            "ttl": ttl or self.default_ttl
        }
        self.cache.move_to_end(key)
        self._evict()

    def _evict(self):
        """Remove oldest entries if over capacity."""
        while len(self.cache) > self.max_size:
            self.cache.popitem(last=False)
            self.stats["evictions"] += 1

KV Cache for Transformer Inference

The Key-Value (KV) cache is an internal optimization used during transformer model inference. When generating tokens one at a time, the model computes attention over all previous tokens. Without caching, this requires recomputing the key and value matrices for every previous token at each step. The KV cache stores these computed matrices so they are only computed once:

• How it works: At each generation step, only the new token's key/value vectors need to be computed; all previous tokens' vectors are read from cache
• Memory cost: KV cache memory grows linearly with sequence length and can consume significant GPU memory for long sequences
• Optimizations: Techniques like Multi-Query Attention (MQA) and Grouped-Query Attention (GQA) reduce KV cache memory by sharing key/value heads
• PagedAttention: Used by vLLM, this manages KV cache memory like operating system pages, reducing waste from pre-allocated memory blocks

Prompt Caching (Anthropic, OpenAI)

API-level prompt caching is a feature offered by model providers to reduce costs for requests that share a common prefix (system prompt + context):

• Anthropic prompt caching: Mark parts of the prompt as cacheable with cache_control breakpoints. Cached prefixes cost 90% less on subsequent calls. Minimum 1024 tokens for caching.
• OpenAI prompt caching: Automatic caching of common prefixes across API calls. Cached tokens are billed at 50% discount. Works best with long, shared system prompts.
• Best use cases: RAG applications with large context documents, multi-turn conversations with long system prompts, batch processing with shared instructions

Batch Processing Pattern

Instead of processing requests one at a time, accumulate them and process in batches for better throughput and lower cost:

• Request accumulation: Buffer incoming requests in a queue until a batch threshold (size or time) is reached
• Batch API calls: Many providers offer batch APIs with significant discounts (OpenAI Batch API: 50% off, Anthropic Batches: up to 50% off)
• Parallel processing: Send multiple requests simultaneously, respecting rate limits, to maximize throughput
• Result distribution: Map batch results back to the original requesters

Model Distillation Pattern

Model distillation trains a smaller, faster, cheaper model to mimic the outputs of a larger, more capable model:

1. Generate training data: Run your production prompts through the large model (e.g., Claude Opus, GPT-4) to create a dataset of input-output pairs
2. Fine-tune a small model: Train a smaller model (e.g., GPT-4o-mini, Claude Haiku, Llama 3.1 8B) on this dataset to reproduce the large model's outputs
3. Evaluate quality: Compare the distilled model's outputs against the teacher model on a held-out test set. Accept if quality meets your threshold.
4. Deploy the small model: Replace the large model with the distilled one for cost savings of 10-100x and latency improvements of 3-10x

Quantization for Speed

Quantization reduces model precision from 32-bit or 16-bit floating point to 8-bit or 4-bit integers. This trades a small amount of quality for significant speed and memory improvements:

• INT8 quantization: Reduces model size by ~2x with minimal quality loss (typically <1% accuracy degradation)
• INT4 quantization: Reduces model size by ~4x with moderate quality loss (1-3% accuracy degradation). Enables running 70B models on consumer GPUs.
• GPTQ/AWQ: Post-training quantization methods optimized for LLMs that preserve quality better than naive quantization
• GGUF format: Used by llama.cpp for efficient CPU inference with various quantization levels (Q4_K_M, Q5_K_M, Q8_0)

Cost Optimization Strategies

Cache Invalidation

Cache invalidation - knowing when to discard cached responses - is one of the hardest problems in computing, and AI caching is no exception:

• Time-based (TTL): Simplest approach. Responses expire after a fixed time. Good for data that changes on a known schedule.
• Event-based: Invalidate cache when underlying data changes (database update, document modification, knowledge base refresh)
• Version-based: Tag cached entries with a version of the source data. When the version changes, invalidate all entries from the old version.
• Manual invalidation: Provide admin tools to clear specific cache entries or entire categories when you know the cached data is stale