Advanced Prompt Caching Patterns for Claude Code

Overview

Prompt caching is a critical optimization technique that can reduce Claude API costs by up to 90% and latency by up to 85%. This guide covers advanced implementation patterns, cost analysis, and real-world examples for maximizing the efficiency of Claude Code applications.

Key Benefits

• 90% Cost Reduction: Cache reads cost $0.30/MTokvs$3.00/MTok for standard input
• 85% Latency Reduction: Skip redundant processing for cached content
• Improved Throughput: Cache read tokens don’t count against rate limits (Claude 3.7 Sonnet)
• Better User Experience: Faster responses for repetitive queries

Pricing Structure (2025)

Base Pricing

• Claude 4 Opus: $15/MTokinput,$75/MTok output
• Claude 4 Sonnet: $3.00/MTokinput,$15.00/MTok output
• Claude 3.5 Haiku: $0.80/MTokinput,$4.00/MTok output

Cache Pricing

• Cache Write: $3.75/MTok (25% premium over base)
• Cache Read: $0.30/MTok (90% discount)
• Break-even: 2-3 uses of cached content

Implementation Patterns

1. Basic Python Implementation

import anthropic
from typing import List, Dict, Any
import time
 
class CachedClaudeClient:
    def __init__(self, api_key: str):
        self.client = anthropic.Anthropic(api_key=api_key)
        self.cache_metrics = {
            "cache_writes": 0,
            "cache_reads": 0,
            "total_saved": 0.0
        }
    
    def create_cached_message(
        self, 
        system_prompt: str,
        context: str,
        user_query: str,
        model: str = "claude-3-5-sonnet-20241022"
    ) -> Dict[str, Any]:
        """
        Create a message with intelligent caching of system prompt and context.
        """
        start_time = time.time()
        
        messages = [
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": system_prompt,
                        "cache_control": {"type": "ephemeral"}  # Cache system instructions
                    },
                    {
                        "type": "text", 
                        "text": context,
                        "cache_control": {"type": "ephemeral"}  # Cache large context
                    },
                    {
                        "type": "text",
                        "text": user_query  # Don't cache variable user input
                    }
                ]
            }
        ]
        
        response = self.client.messages.create(
            model=model,
            messages=messages,
            max_tokens=1000
        )
        
        # Track metrics
        self._update_metrics(response)
        
        return {
            "response": response,
            "latency": time.time() - start_time,
            "cache_metrics": self.cache_metrics
        }
    
    def _update_metrics(self, response):
        """Update cache metrics based on response."""
        if hasattr(response, 'usage'):
            if response.usage.cache_creation_input_tokens > 0:
                self.cache_metrics["cache_writes"] += 1
            if response.usage.cache_read_input_tokens > 0:
                self.cache_metrics["cache_reads"] += 1
                # Calculate savings (90% discount on cached tokens)
                tokens_saved = response.usage.cache_read_input_tokens
                cost_saved = (tokens_saved / 1_000_000) * 2.70  # $3.00 - $0.30
                self.cache_metrics["total_saved"] += cost_saved

2. TypeScript Implementation with Monitoring

import Anthropic from '@anthropic-ai/sdk';
 
interface CacheMetrics {
  cacheWrites: number;
  cacheReads: number;
  totalSaved: number;
  avgLatency: number;
}
 
interface CacheControl {
  type: 'ephemeral';
  ttl?: number; // Optional TTL in seconds
}
 
class CachedClaudeClient {
  private client: Anthropic;
  private metrics: CacheMetrics = {
    cacheWrites: 0,
    cacheReads: 0,
    totalSaved: 0,
    avgLatency: 0
  };
  private latencies: number[] = [];
 
  constructor(apiKey: string) {
    this.client = new Anthropic({ apiKey });
  }
 
  async createCachedMessage(
    systemPrompt: string,
    context: string,
    userQuery: string,
    model: string = 'claude-3-5-sonnet-20241022'
  ) {
    const startTime = Date.now();
    
    const message = await this.client.messages.create({
      model,
      messages: [{
        role: 'user',
        content: [
          {
            type: 'text',
            text: systemPrompt,
            cache_control: { type: 'ephemeral' }
          },
          {
            type: 'text',
            text: context,
            cache_control: { type: 'ephemeral' }
          },
          {
            type: 'text',
            text: userQuery
          }
        ]
      }],
      max_tokens: 1000
    });
 
    const latency = Date.now() - startTime;
    this.updateMetrics(message, latency);
 
    return {
      response: message,
      latency,
      metrics: this.getMetrics()
    };
  }
 
  private updateMetrics(response: any, latency: number) {
    this.latencies.push(latency);
    this.metrics.avgLatency = 
      this.latencies.reduce((a, b) => a + b, 0) / this.latencies.length;
 
    if (response.usage?.cache_creation_input_tokens > 0) {
      this.metrics.cacheWrites++;
    }
    
    if (response.usage?.cache_read_input_tokens > 0) {
      this.metrics.cacheReads++;
      const tokensSaved = response.usage.cache_read_input_tokens;
      const costSaved = (tokensSaved / 1_000_000) * 2.70;
      this.metrics.totalSaved += costSaved;
    }
  }
 
  getMetrics(): CacheMetrics {
    return { ...this.metrics };
  }
}

3. Multi-Agent Cost Optimization Pattern

from dataclasses import dataclass
from typing import Optional, List
import hashlib
 
@dataclass
class AgentContext:
    """Shared context for multi-agent systems."""
    system_instructions: str
    shared_knowledge: str
    task_history: List[str]
    
    def get_cache_key(self) -> str:
        """Generate a unique cache key for this context."""
        content = f"{self.system_instructions}:{self.shared_knowledge}"
        return hashlib.md5(content.encode()).hexdigest()
 
class MultiAgentCacheManager:
    """Manages caching across multiple Claude agents."""
    
    def __init__(self, client: anthropic.Anthropic):
        self.client = client
        self.context_cache = {}
        self.cache_usage = {}
        
    def register_agent_context(self, agent_id: str, context: AgentContext):
        """Register shared context for an agent."""
        cache_key = context.get_cache_key()
        self.context_cache[agent_id] = {
            "key": cache_key,
            "context": context,
            "last_used": time.time()
        }
        
    def create_agent_message(
        self, 
        agent_id: str, 
        query: str,
        model: str = "claude-3-5-sonnet-20241022"
    ):
        """Create message with cached agent context."""
        if agent_id not in self.context_cache:
            raise ValueError(f"Agent {agent_id} not registered")
            
        context_info = self.context_cache[agent_id]
        context = context_info["context"]
        
        # Build message with cached components
        messages = [{
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": context.system_instructions,
                    "cache_control": {"type": "ephemeral"}
                },
                {
                    "type": "text",
                    "text": context.shared_knowledge,
                    "cache_control": {"type": "ephemeral"}
                },
                {
                    "type": "text",
                    "text": f"Task History:\n" + "\n".join(context.task_history[-5:])
                },
                {
                    "type": "text",
                    "text": query
                }
            ]
        }]
        
        response = self.client.messages.create(
            model=model,
            messages=messages,
            max_tokens=2000
        )
        
        # Update usage tracking
        context_info["last_used"] = time.time()
        self._track_usage(agent_id, response)
        
        return response
    
    def _track_usage(self, agent_id: str, response):
        """Track cache usage per agent."""
        if agent_id not in self.cache_usage:
            self.cache_usage[agent_id] = {
                "total_requests": 0,
                "cache_hits": 0,
                "tokens_saved": 0,
                "cost_saved": 0.0
            }
        
        usage = self.cache_usage[agent_id]
        usage["total_requests"] += 1
        
        if hasattr(response, 'usage') and response.usage.cache_read_input_tokens > 0:
            usage["cache_hits"] += 1
            usage["tokens_saved"] += response.usage.cache_read_input_tokens
            usage["cost_saved"] += (response.usage.cache_read_input_tokens / 1_000_000) * 2.70
    
    def get_savings_report(self) -> Dict[str, Any]:
        """Generate cost savings report across all agents."""
        total_saved = sum(u["cost_saved"] for u in self.cache_usage.values())
        total_requests = sum(u["total_requests"] for u in self.cache_usage.values())
        total_hits = sum(u["cache_hits"] for u in self.cache_usage.values())
        
        return {
            "total_cost_saved": f"${total_saved:.2f}",
            "cache_hit_rate": f"{(total_hits/total_requests*100):.1f}%" if total_requests > 0 else "0%",
            "per_agent_savings": {
                agent_id: {
                    "cost_saved": f"${usage['cost_saved']:.2f}",
                    "hit_rate": f"{(usage['cache_hits']/usage['total_requests']*100):.1f}%"
                }
                for agent_id, usage in self.cache_usage.items()
            }
        }

4. Intelligent Cache Invalidation Pattern

interface CacheEntry {
  content: string;
  checksum: string;
  created: number;
  lastUsed: number;
  useCount: number;
  ttl: number;
}
 
class SmartCacheManager {
  private cache: Map<string, CacheEntry> = new Map();
  private readonly DEFAULT_TTL = 300; // 5 minutes
  private readonly EXTENDED_TTL = 3600; // 1 hour
  
  constructor(private client: Anthropic) {
    // Start cleanup interval
    setInterval(() => this.cleanup(), 60000); // Every minute
  }
 
  async cachedRequest(
    key: string,
    content: string,
    userQuery: string,
    options: {
      ttl?: number;
      autoExtend?: boolean;
    } = {}
  ) {
    const checksum = this.generateChecksum(content);
    const existingEntry = this.cache.get(key);
    
    // Check if content has changed (invalidate if so)
    if (existingEntry && existingEntry.checksum !== checksum) {
      console.log(`Cache invalidated for key: ${key} - content changed`);
      this.cache.delete(key);
    }
    
    // Determine if we should use cache
    const shouldCache = this.shouldCache(content);
    const ttl = options.ttl || this.DEFAULT_TTL;
    
    const messages = [{
      role: 'user' as const,
      content: [
        {
          type: 'text' as const,
          text: content,
          ...(shouldCache && { cache_control: { type: 'ephemeral' as const } })
        },
        {
          type: 'text' as const,
          text: userQuery
        }
      ]
    }];
    
    const response = await this.client.messages.create({
      model: 'claude-3-5-sonnet-20241022',
      messages,
      max_tokens: 2000
    });
    
    // Update cache entry
    if (shouldCache) {
      const entry = this.cache.get(key) || {
        content,
        checksum,
        created: Date.now(),
        lastUsed: Date.now(),
        useCount: 0,
        ttl
      };
      
      entry.lastUsed = Date.now();
      entry.useCount++;
      
      // Auto-extend TTL for frequently used entries
      if (options.autoExtend && entry.useCount > 5) {
        entry.ttl = this.EXTENDED_TTL;
      }
      
      this.cache.set(key, entry);
    }
    
    return response;
  }
  
  private shouldCache(content: string): boolean {
    // Only cache content > 1024 tokens (rough estimate)
    return content.length > 4096; // ~1024 tokens
  }
  
  private generateChecksum(content: string): string {
    // Simple checksum for demo - use crypto.subtle in production
    return content.split('').reduce((a, b) => {
      a = ((a << 5) - a) + b.charCodeAt(0);
      return a & a;
    }, 0).toString();
  }
  
  private cleanup() {
    const now = Date.now();
    for (const [key, entry] of this.cache.entries()) {
      if (now - entry.lastUsed > entry.ttl * 1000) {
        this.cache.delete(key);
        console.log(`Cache expired for key: ${key}`);
      }
    }
  }
  
  getCacheStats() {
    const entries = Array.from(this.cache.values());
    const totalUses = entries.reduce((sum, e) => sum + e.useCount, 0);
    const avgUseCount = entries.length > 0 ? totalUses / entries.length : 0;
    
    return {
      entries: entries.length,
      totalUses,
      avgUseCount: avgUseCount.toFixed(2),
      memoryUsage: entries.reduce((sum, e) => sum + e.content.length, 0)
    };
  }
}

Best Practices

1. Structure Prompts for Caching

# GOOD: Static content first, variable content last
messages = [
    {
        "role": "user",
        "content": [
            # Cache large, stable content
            {"type": "text", "text": system_prompt, "cache_control": {"type": "ephemeral"}},
            {"type": "text", "text": documentation, "cache_control": {"type": "ephemeral"}},
            # Don't cache variable content
            {"type": "text", "text": user_query}
        ]
    }
]
 
# BAD: Mixed static and variable content
messages = [
    {
        "role": "user", 
        "content": system_prompt + user_query  # Can't cache effectively
    }
]

2. Monitor Cache Performance

def analyze_cache_efficiency(metrics: Dict[str, Any]) -> Dict[str, Any]:
    """Analyze cache performance and suggest optimizations."""
    hit_rate = metrics["cache_hits"] / metrics["total_requests"]
    avg_savings_per_request = metrics["total_saved"] / metrics["total_requests"]
    
    recommendations = []
    if hit_rate < 0.5:
        recommendations.append("Consider caching more stable content")
    if avg_savings_per_request < 0.01:  # Less than $0.01 saved per request
        recommendations.append("Increase size of cached content")
    
    return {
        "hit_rate": f"{hit_rate * 100:.1f}%",
        "avg_savings": f"${avg_savings_per_request:.3f}",
        "recommendations": recommendations
    }

3. Cache Warming Strategy

async def warm_cache(client: CachedClaudeClient, contexts: List[Dict[str, str]]):
    """Pre-warm cache with common contexts."""
    for context in contexts:
        try:
            await client.create_cached_message(
                system_prompt=context["system"],
                context=context["data"],
                user_query="Cache warming - please acknowledge.",
                model="claude-3-5-haiku-20241022"  # Use cheaper model for warming
            )
            print(f"✓ Warmed cache for: {context['name']}")
        except Exception as e:
            print(f"✗ Failed to warm cache for {context['name']}: {e}")

Cost Analysis Example

def calculate_roi(
    requests_per_day: int,
    avg_input_tokens: int,
    cache_hit_rate: float = 0.8,
    days: int = 30
) -> Dict[str, float]:
    """Calculate ROI for implementing prompt caching."""
    
    # Costs per million tokens
    STANDARD_COST = 3.00  # $3.00 per MTok
    CACHE_WRITE_COST = 3.75  # $3.75 per MTok (first write)
    CACHE_READ_COST = 0.30  # $0.30 per MTok (subsequent reads)
    
    total_requests = requests_per_day * days
    tokens_per_request = avg_input_tokens / 1_000_000
    
    # Without caching
    cost_without_cache = total_requests * tokens_per_request * STANDARD_COST
    
    # With caching
    cache_writes = requests_per_day  # Assume daily cache refresh
    cache_reads = total_requests * cache_hit_rate
    uncached_requests = total_requests * (1 - cache_hit_rate)
    
    cost_with_cache = (
        (cache_writes * tokens_per_request * CACHE_WRITE_COST) +
        (cache_reads * tokens_per_request * CACHE_READ_COST) +
        (uncached_requests * tokens_per_request * STANDARD_COST)
    )
    
    savings = cost_without_cache - cost_with_cache
    roi_percentage = (savings / cost_without_cache) * 100
    
    return {
        "cost_without_cache": cost_without_cache,
        "cost_with_cache": cost_with_cache,
        "total_savings": savings,
        "roi_percentage": roi_percentage,
        "break_even_days": (cache_writes * tokens_per_request * 
                           (CACHE_WRITE_COST - STANDARD_COST)) / 
                          (savings / days) if savings > 0 else float('inf')
    }
 
# Example calculation
roi = calculate_roi(
    requests_per_day=1000,
    avg_input_tokens=5000,
    cache_hit_rate=0.8
)
print(f"Monthly savings: ${roi['total_savings']:.2f}")
print(f"ROI: {roi['roi_percentage']:.1f}%")
print(f"Break-even: {roi['break_even_days']:.1f} days")

Common Pitfalls

1. Over-Caching

• Don’t cache small prompts (< 1024 tokens)
• Avoid caching rapidly changing content
• Monitor cache miss rates

2. Cache Key Collisions

• Use content hashing for cache keys
• Include version identifiers
• Implement proper invalidation

3. Memory Management

• Implement cache size limits
• Use LRU eviction policies
• Monitor memory usage

Conclusion

Prompt caching is essential for cost-effective Claude Code applications. With proper implementation, you can achieve:

• 90% cost reduction on repeated queries
• 85% latency improvement
• Better scalability for multi-agent systems
• Improved user experience

Start with basic caching for system prompts and gradually implement more sophisticated patterns based on your usage patterns and requirements.

References

• Anthropic Prompt Caching Documentation
• Claude API Pricing
• Anthropic Cookbook Examples