Prompt Caching: Reduce LLM Costs by 90% with Advanced Optimization Techniques

# Typical RAG pattern - expensive without caching
def answer_question(user_query):
    # Retrieve relevant documents (same for similar queries)
    context_docs = vector_db.search(user_query, k=5)

    # Build prompt with context
    prompt = f"""
    Use the following context to answer the question:

    Context:
    {context_docs}  # 3000 tokens - repeated for every similar query!

    Question: {user_query}  # 20 tokens - the only unique part

    Answer:
    """

    # LLM call - charges for ALL tokens every time
    response = llm.generate(prompt)
    return response

# Simplified explanation of KV cache
class LLMWithCache:
    def __init__(self):
        self.kv_cache = {}  # Key-Value cache

    def generate(self, prompt_tokens):
        """Generate with KV cache"""

        # Hash the prompt prefix
        cache_key = self._hash_prefix(prompt_tokens[:-100])

        # Check if prefix is cached
        if cache_key in self.kv_cache:
            # Reuse cached key-value states
            cached_kv = self.kv_cache[cache_key]

            # Only compute attention for new tokens
            output = self._compute_attention(
                new_tokens=prompt_tokens[-100:],
                cached_kv=cached_kv
            )
        else:
            # Full computation - cache the result
            output, kv_states = self._full_attention(prompt_tokens)
            self.kv_cache[cache_key] = kv_states

        return output

# Example: OpenAI automatic caching (1024+ tokens)
import openai

# First request - full cost
response1 = openai.ChatCompletion.create(
    model="gpt-4-turbo",
    messages=[
        {
            "role": "system",
            "content": LONG_SYSTEM_PROMPT  # 2000 tokens - creates cache
        },
        {
            "role": "user",
            "content": "What is machine learning?"  # 5 tokens
        }
    ]
)
# Cost: 2005 tokens × $0.01/1K = $0.02005

# Second request - cache hit!
response2 = openai.ChatCompletion.create(
    model="gpt-4-turbo",
    messages=[
        {
            "role": "system",
            "content": LONG_SYSTEM_PROMPT  # 2000 tokens - CACHED (50% discount)
        },
        {
            "role": "user",
            "content": "What is deep learning?"  # 5 tokens
        }
    ]
)
# Cost: 2000 tokens × $0.005/1K (cached) + 5 × $0.01/1K = $0.01 + $0.00005 = $0.01005
# Savings: 50%

# BAD - Variable content in middle breaks caching
def build_prompt_bad(user_query, context_docs):
    return f"""
    Question: {user_query}  # Variable - breaks cache!

    Use this context:
    {context_docs}  # Static - but comes after variable

    Instructions:
    - Be concise
    - Cite sources
    """

# GOOD - Static first, variable last
def build_prompt_good(user_query, context_docs):
    return f"""
    Instructions:
    - Be concise
    - Cite sources

    Context:
    {context_docs}  # Static - cached

    Question: {user_query}  # Variable - at end, doesn't break cache
    """

from typing import List

class CacheOptimizedPrompt:
    """Build prompts optimized for caching"""

    def __init__(
        self,
        system_instructions: str,
        few_shot_examples: List[dict],
        context_template: str
    ):
        # Static components - cached
        self.static_prefix = self._build_static_prefix(
            system_instructions,
            few_shot_examples,
            context_template
        )

    def _build_static_prefix(
        self,
        instructions: str,
        examples: List[dict],
        template: str
    ) -> str:
        """Build cacheable static prefix"""

        prefix = f"""# Instructions
{instructions}

# Examples
"""
        for i, example in enumerate(examples, 1):
            prefix += f"\nExample {i}:\nQ: {example['question']}\nA: {example['answer']}\n"

        prefix += f"\n# Context Format\n{template}\n"

        return prefix

    def format(self, user_query: str, context: str) -> str:
        """Format complete prompt with caching optimization"""

        # Static prefix (cached) + dynamic suffix (computed)
        return f"""{self.static_prefix}

# Context
{context}

# User Query
{user_query}

# Answer
"""

# Usage
prompt_builder = CacheOptimizedPrompt(
    system_instructions="Answer questions using provided context...",
    few_shot_examples=[
        {"question": "What is X?", "answer": "X is..."},
        {"question": "How does Y work?", "answer": "Y works by..."}
    ],
    context_template="Documents: [...]"
)

# Each call reuses cached prefix
prompt1 = prompt_builder.format("What is AI?", context1)
prompt2 = prompt_builder.format("What is ML?", context2)

import asyncio

# BAD - Cold start for all requests
async def process_batch_cold_start(queries):
    """All requests hit cold cache"""

    tasks = [
        process_query(query)  # Each creates cache independently
        for query in queries
    ]

    # All start simultaneously - no cache hits!
    results = await asyncio.gather(*tasks)
    return results

# First 100 requests: 0% cache hit rate
# Cost: FULL price for all requests

class CacheWarmer:
    def __init__(self, llm_client):
        self.client = llm_client

    async def warm_cache(self, static_prompts: List[str]):
        """Pre-warm cache with static prompts"""

        warm_tasks = []

        for prompt in static_prompts:
            # Make dedicated cache-warming call
            task = self.client.generate(
                prompt=prompt,
                max_tokens=1,  # Minimal generation - just need cache
                metadata={"purpose": "cache_warming"}
            )
            warm_tasks.append(task)

        # Wait for all caches to be created
        await asyncio.gather(*warm_tasks)

        # Cache is now ready for real requests
        print(f"Warmed {len(static_prompts)} cache entries")

    async def process_batch_with_warming(
        self,
        queries: List[str],
        static_prefix: str
    ):
        """Process batch with cache warming"""

        # Step 1: Warm cache (2-4 seconds)
        await self.warm_cache([static_prefix])

        # Step 2: Process all queries (now cache hits!)
        tasks = [
            self.process_query_with_prefix(query, static_prefix)
            for query in queries
        ]

        results = await asyncio.gather(*tasks)

        # 99%+ cache hit rate!
        return results

# Example usage
warmer = CacheWarmer(llm_client)

# Warm cache before parallel processing
await warmer.warm_cache([
    SYSTEM_PROMPT,
    RAG_CONTEXT_TEMPLATE,
    FEW_SHOT_EXAMPLES
])

# Now all parallel requests hit warm cache
results = await warmer.process_batch_with_warming(
    queries=user_queries,
    static_prefix=SYSTEM_PROMPT
)

import hashlib
from typing import Optional

class ExactMatchCache:
    def __init__(self, ttl_seconds: int = 3600):
        self.cache = {}
        self.ttl = ttl_seconds

    def _hash_prompt(self, prompt: str) -> str:
        """Generate cache key from prompt"""
        return hashlib.sha256(prompt.encode()).hexdigest()

    def get(self, prompt: str) -> Optional[str]:
        """Get cached response"""

        cache_key = self._hash_prompt(prompt)

        if cache_key in self.cache:
            entry = self.cache[cache_key]

            # Check TTL
            if time.time() - entry['timestamp'] < self.ttl:
                entry['hits'] += 1
                return entry['response']

        return None

    def set(self, prompt: str, response: str):
        """Cache response"""

        cache_key = self._hash_prompt(prompt)

        self.cache[cache_key] = {
            'response': response,
            'timestamp': time.time(),
            'hits': 0
        }

from sentence_transformers import SentenceTransformer
import numpy as np

class SemanticCache:
    def __init__(
        self,
        similarity_threshold: float = 0.95,
        ttl_seconds: int = 3600
    ):
        self.threshold = similarity_threshold
        self.ttl = ttl_seconds
        self.embedder = SentenceTransformer('all-MiniLM-L6-v2')
        self.cache = []  # List of (embedding, response, metadata)

    def _compute_similarity(
        self,
        query_embedding: np.ndarray,
        cached_embedding: np.ndarray
    ) -> float:
        """Cosine similarity"""
        return np.dot(query_embedding, cached_embedding) / (
            np.linalg.norm(query_embedding) * np.linalg.norm(cached_embedding)
        )

    def get(self, prompt: str) -> Optional[str]:
        """Get cached response for similar prompt"""

        # Embed query
        query_embedding = self.embedder.encode(prompt)

        # Find most similar cached prompt
        best_match = None
        best_similarity = 0

        for entry in self.cache:
            # Check TTL
            if time.time() - entry['timestamp'] > self.ttl:
                continue

            similarity = self._compute_similarity(
                query_embedding,
                entry['embedding']
            )

            if similarity > best_similarity:
                best_similarity = similarity
                best_match = entry

        # Return if above threshold
        if best_match and best_similarity >= self.threshold:
            best_match['hits'] += 1
            return best_match['response']

        return None

    def set(self, prompt: str, response: str):
        """Cache response with embedding"""

        embedding = self.embedder.encode(prompt)

        self.cache.append({
            'prompt': prompt,
            'embedding': embedding,
            'response': response,
            'timestamp': time.time(),
            'hits': 0
        })

class HybridCache:
    def __init__(self):
        self.exact_cache = ExactMatchCache()
        self.semantic_cache = SemanticCache(similarity_threshold=0.92)

    async def get(self, prompt: str) -> Optional[str]:
        """Try exact match first, then semantic"""

        # Fast exact match
        exact_match = self.exact_cache.get(prompt)
        if exact_match:
            return exact_match

        # Fallback to semantic similarity
        semantic_match = self.semantic_cache.get(prompt)
        if semantic_match:
            return semantic_match

        return None

    def set(self, prompt: str, response: str):
        """Cache in both systems"""

        self.exact_cache.set(prompt, response)
        self.semantic_cache.set(prompt, response)

import openai

class OpenAICachedClient:
    def __init__(self, api_key: str):
        self.client = openai.OpenAI(api_key=api_key)

    async def generate_with_caching(
        self,
        system_prompt: str,
        user_query: str,
        cache_key: str
    ):
        """Use OpenAI's automatic caching"""

        # Ensure system prompt is 1024+ tokens for caching
        response = self.client.chat.completions.create(
            model="gpt-4-turbo",
            messages=[
                {
                    "role": "system",
                    "content": system_prompt  # Cached if 1024+ tokens
                },
                {
                    "role": "user",
                    "content": user_query
                }
            ],
            # Use consistent cache_key for same prefix
            extra_body={"prompt_cache_key": cache_key}
        )

        return response.choices[0].message.content

import anthropic

class AnthropicCachedClient:
    def __init__(self, api_key: str):
        self.client = anthropic.Anthropic(api_key=api_key)

    async def generate_with_caching(
        self,
        system_blocks: List[dict],
        user_query: str
    ):
        """Use Anthropic's cache control"""

        response = self.client.messages.create(
            model="claude-3-opus-20240229",
            max_tokens=1024,
            system=[
                {
                    "type": "text",
                    "text": system_blocks[0],
                },
                {
                    "type": "text",
                    "text": system_blocks[1],
                    "cache_control": {"type": "ephemeral"}  # Mark for caching
                }
            ],
            messages=[
                {
                    "role": "user",
                    "content": user_query
                }
            ]
        )

        # Check cache usage
        usage = response.usage
        print(f"Cache creation tokens: {usage.cache_creation_input_tokens}")
        print(f"Cache read tokens: {usage.cache_read_input_tokens}")
        print(f"Regular input tokens: {usage.input_tokens}")

        return response.content[0].text

import boto3

class BedrockCachedClient:
    def __init__(self):
        self.client = boto3.client('bedrock-runtime')

    async def generate_with_caching(
        self,
        system_prompt: str,
        user_query: str,
        cache_id: str
    ):
        """Use Bedrock prompt caching"""

        response = self.client.invoke_model(
            modelId='anthropic.claude-3-sonnet-20240229-v1:0',
            body={
                "anthropic_version": "bedrock-2023-05-31",
                "max_tokens": 1024,
                "system": [
                    {
                        "type": "text",
                        "text": system_prompt,
                        "cache_control": {"type": "ephemeral"}
                    }
                ],
                "messages": [
                    {
                        "role": "user",
                        "content": user_query
                    }
                ]
            }
        )

        return response['content'][0]['text']

from dataclasses import dataclass
from typing import List

@dataclass
class CacheMetrics:
    total_requests: int
    cache_hits: int
    cache_misses: int
    hit_rate: float
    tokens_saved: int
    cost_saved_usd: float

class CacheMonitor:
    def __init__(self):
        self.metrics = {
            "total_requests": 0,
            "cache_hits": 0,
            "cache_misses": 0,
            "tokens_saved": 0,
            "cost_saved": 0.0
        }

    def record_cache_hit(self, tokens_cached: int, cost_per_token: float):
        """Record cache hit"""

        self.metrics["total_requests"] += 1
        self.metrics["cache_hits"] += 1
        self.metrics["tokens_saved"] += tokens_cached
        self.metrics["cost_saved"] += tokens_cached * cost_per_token

    def record_cache_miss(self):
        """Record cache miss"""

        self.metrics["total_requests"] += 1
        self.metrics["cache_misses"] += 1

    def get_metrics(self) -> CacheMetrics:
        """Get current cache metrics"""

        hit_rate = (
            self.metrics["cache_hits"] / self.metrics["total_requests"]
            if self.metrics["total_requests"] > 0
            else 0
        )

        return CacheMetrics(
            total_requests=self.metrics["total_requests"],
            cache_hits=self.metrics["cache_hits"],
            cache_misses=self.metrics["cache_misses"],
            hit_rate=hit_rate,
            tokens_saved=self.metrics["tokens_saved"],
            cost_saved_usd=self.metrics["cost_saved"]
        )

    def optimize_cache_strategy(self):
        """Provide optimization recommendations"""

        metrics = self.get_metrics()

        recommendations = []

        if metrics.hit_rate < 0.5:
            recommendations.append(
                "Low cache hit rate - review prompt structure"
            )

        if metrics.hit_rate > 0.95:
            recommendations.append(
                "Excellent cache performance - consider increasing cache size"
            )

        return recommendations

from collections import OrderedDict
import time

class LRUCacheWithTTL:
    """LRU cache with TTL"""

    def __init__(self, max_size: int = 1000, ttl_seconds: int = 3600):
        self.cache = OrderedDict()
        self.max_size = max_size
        self.ttl = ttl_seconds

    def get(self, key: str):
        """Get from cache, refresh LRU"""

        if key not in self.cache:
            return None

        # Check TTL
        entry = self.cache[key]
        if time.time() - entry['timestamp'] > self.ttl:
            del self.cache[key]
            return None

        # Move to end (most recently used)
        self.cache.move_to_end(key)

        return entry['value']

    def set(self, key: str, value: any):
        """Set cache entry"""

        # Remove if exists (to update timestamp)
        if key in self.cache:
            del self.cache[key]

        # Add new entry
        self.cache[key] = {
            'value': value,
            'timestamp': time.time()
        }

        # Evict oldest if over size
        if len(self.cache) > self.max_size:
            self.cache.popitem(last=False)  # Remove oldest (FIFO)

- 50,000 requests/day
- Avg 3,200 tokens/request (3,000 context + 200 query/response)
- Cost: $0.03/1K tokens
- Daily cost: 50,000 × 3.2 × $0.03 = $4,800/day
- Monthly cost: $144,000

class ProductionRAGWithCaching:
    def __init__(self):
        self.cache = HybridCache()
        self.warmer = CacheWarmer(llm_client)
        self.monitor = CacheMonitor()

    async def startup(self):
        """Warm cache on startup"""

        # Pre-warm with top document contexts
        top_contexts = await self._get_popular_contexts(limit=100)
        await self.warmer.warm_cache(top_contexts)

    async def answer_question(self, user_query: str):
        """Answer with aggressive caching"""

        # Retrieve context
        context = await self.retrieve_context(user_query)

        # Build cache-optimized prompt
        prompt = self._build_prompt(context, user_query)

        # Check semantic cache first
        cached_response = await self.cache.get(prompt)

        if cached_response:
            self.monitor.record_cache_hit(
                tokens_cached=3000,
                cost_per_token=0.00003
            )
            return cached_response

        # Cache miss - call LLM
        self.monitor.record_cache_miss()
        response = await self.llm.generate(prompt)

        # Cache for future
        await self.cache.set(prompt, response)

        return response

- Cache hit rate: 87%
- Tokens saved: 130,500,000
- Cost saved: $3,915
- New monthly cost: $18,720
- Total savings: $125,280/month (87%)