Prompt Caching: Cost Optimization Protocol¶
Overview¶
Prompt caching stores system prompts and repeated context, reducing costs by 90% on cache hits.
Essential for production agents processing similar queries.
How Caching Works¶
Cache Layers¶
class PromptCachingArchitecture:
"""Three levels of caching"""
def __init__(self):
self.layers = {
1: "System Prompt Cache", # Static, reused
2: "Context Cache", # Repeated context
3: "Conversation Cache" # Session history
}
# Example:
# Layer 1: System prompt (cached)
# "You are a helpful assistant..."
# Layer 2: Context (cached)
# "Knowledge base about company X"
# Layer 3: Conversation (cached)
# Previous messages in conversation
# Layer 4: Current request (NOT cached)
# "What is Y?"
Claude Implementation¶
Cache Tokens in Claude¶
from anthropic import Anthropic
class ClaudePromptCaching:
def __init__(self):
self.client = Anthropic()
self.system_prompt = """You are an expert analyst...
[LONG SYSTEM PROMPT - 5000 tokens]"""
self.context = """Company background...
[LONG CONTEXT - 10000 tokens]"""
def query_with_caching(self, user_query: str):
"""Use cache for system + context"""
response = self.client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
system=[
{
"type": "text",
"text": self.system_prompt,
"cache_control": {"type": "ephemeral"}
}
],
messages=[
{
"role": "user",
"content": [
{
"type": "text",
"text": self.context,
"cache_control": {"type": "ephemeral"}
},
{
"type": "text",
"text": user_query # NOT cached
}
]
}
]
)
# Check cache usage
usage = response.usage
print(f"Cache creation: {usage.cache_creation_input_tokens}")
print(f"Cache read: {usage.cache_read_input_tokens}")
print(f"Normal input: {usage.input_tokens}")
return response
Cost Math¶
Real Savings¶
Pricing (Claude 3.5 Sonnet):
Input: $3/1M tokens
Cache creation: $3.75/1M tokens (25% more)
Cache read: $0.30/1M tokens (90% less!)
Scenario: RAG Agent
System prompt: 5,000 tokens
Context: 10,000 tokens
User query: 500 tokens
Tokens reused: 15,000/15,500 = 96.8%
FIRST REQUEST (no cache):
Input: 15,500 tokens × $3/1M = $0.0465
Cache creation: 15,000 × $3.75/1M = $0.05625
Total: $0.10275
SUBSEQUENT REQUESTS (with cache):
Input: 500 tokens × $3/1M = $0.0015
Cache read: 15,000 × $0.30/1M = $0.0045
Total: $0.0060
SAVINGS: 94% per cached request!
1000 cached requests:
Without cache: $102.75
With cache: $0.60 + $6.00 = $6.60
SAVINGS: 93.6% !!!
Cache Invalidation¶
When Cache Breaks¶
class CacheInvalidation:
def __init__(self):
self.cache_version = "v1"
self.cache_ttl = 5 * 60 # 5 minutes
def invalidate_on_content_change(self, new_content: str):
"""Invalidate if content changes"""
content_hash = hash(new_content)
# If content changed, cache is invalid
if content_hash != self.last_hash:
self.cache_version += "_new"
self.last_hash = content_hash
return {"cached": False, "reason": "content_changed"}
def invalidate_on_version_change(self):
"""Invalidate on model/API version change"""
# When using new model
# Cache invalidates automatically
# Claude forces new caching
return {"cached": False, "reason": "model_version"}
def invalidate_on_timeout(self):
"""Cache expires after TTL"""
# Ephemeral cache: 5 minutes
# Cached tokens freed after 5 min
# New cache created on next request
return {"cached": False, "reason": "ttl_expired"}
Production Patterns¶
Pattern 1: Static Context Caching¶
class RAGAgentWithCaching:
"""RAG agent that caches knowledge base"""
def __init__(self, kb_content: str):
self.kb = kb_content # Cache this
self.client = Anthropic()
def answer_question(self, question: str):
"""Each question reuses same KB"""
# First request: Create cache
# Subsequent: Read from cache (90% cheaper)
response = self.client.messages.create(
model="claude-3-5-sonnet",
messages=[
{
"role": "user",
"content": [
{
"type": "text",
"text": f"Knowledge base:\n{self.kb}",
"cache_control": {"type": "ephemeral"}
},
{
"type": "text",
"text": question
}
]
}
]
)
return response
When NOT to Cache¶
class CacheDecision:
@staticmethod
def should_cache(content_size, change_frequency):
"""Decide if caching worthwhile"""
# DON'T CACHE IF:
# - Content < 1000 tokens (too small)
# - Changes every request (cache thrashing)
# - Used less than 2x (not worth setup)
# DO CACHE IF:
# - Content > 5000 tokens
# - Reused 3+ times in 5 min window
# - Stable content
# - High volume
if content_size < 1000:
return False, "too_small"
if change_frequency == "every_request":
return False, "high_churn"
if content_size > 5000:
return True, "high_savings"
return False, "not_worth"
3 Warnings ⚠️¶
Warning 1: Cache Not Guaranteed¶
# ❌ WRONG
# Assume cache always hits
result = query_with_cache(question)
# Might not be cached!
# ✅ RIGHT
# Check actual cache usage
response = query_with_cache(question)
cache_hit = response.usage.cache_read_input_tokens > 0
if cache_hit:
print("Cache hit, 90% cheaper")
else:
print("Cache miss, paid full price")
Warning 2: Premature Optimization¶
# ❌ WRONG
# Cache 500-token contexts
# Cache setup costs more than savings!
# ✅ RIGHT
# Only cache if:
# - Content > 5KB (5000 tokens)
# - Reused multiple times
# - High-volume production
Warning 3: Stale Cache¶
# ❌ WRONG
# Cache knowledge base forever
# Update KB but cache still old
# ✅ RIGHT
# Monitor cache TTL
# Update cache version on KB change
# Verify cache invalidation works
Last Updated: August 9, 2026