Prompt Caching / KV Cache

Every time a transformer model processes a request, it computes Key and Value matrices for every token in the context. This is the most computationally expensive step in inference, and it is repeated in full for every new request — even when 95% of the tokens are identical to the previous one. KV caching solves this by storing those computed matrices and reusing them whenever the same prefix appears again. The savings are dramatic: up to 90% cost reduction on cached tokens, and 70–85% reduction in time-to-first-token for large prefixes.

How KV Caching Works

The cache is prefix-based. When a request arrives, the provider compares the beginning of the context against stored KV states. If the tokens match exactly from position 0 through some point, those states are reused and only the remaining tokens require fresh computation. A single character difference anywhere in the matched region breaks the cache entirely.

KV Cache Mechanism

Provider Implementations

Major providers implement caching differently. Anthropic uses explicit cache_control markers that you add to system message content blocks, giving you precise control over what gets cached and when. OpenAI caches automatically based on prefix matching — no code changes required, though you get less visibility and control. Google’s context caching API takes a more explicit approach similar to Anthropic’s. Self-hosted deployments using vLLM or TGI get the latency benefits of prefix caching without the monetary discount.

Impact: Cost and Latency

The financial case for caching is clearest when you have a large, stable system prompt used across many requests. Consider a 50K-token system prompt serving 1,000 requests:

Cost and Latency Improvements

Basic Implementation

The following shows how to implement prompt caching with Anthropic’s explicit markers. Note that this topic is one of the exceptions to the standard LangChain/provider-agnostic approach — caching requires provider-specific APIs:

from anthropic import Anthropic

client = Anthropic()

SYSTEM_PROMPT = """You are an expert assistant for our
e-commerce platform. Here is our complete product catalog:

[... imagine 50,000 tokens of product data ...]

Use this catalog to answer customer questions accurately.
Always cite specific product IDs when making recommendations.
"""

def query_with_caching(user_question: str) -> str:
    """Query with prompt caching enabled."""
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": SYSTEM_PROMPT,
                "cache_control": {"type": "ephemeral"}
            }
        ],
        messages=[
            {"role": "user", "content": user_question}
        ]
    )

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

    # First call: cache_creation_input_tokens = ~50,000
    #             cache_read_input_tokens = 0
    # Subsequent: cache_creation_input_tokens = 0
    #             cache_read_input_tokens = ~50,000

    return response.content[0].text

Multi-Turn Conversation Caching

In multi-turn conversations, the system prompt and tool definitions remain stable across every turn while the conversation history grows. The key is to structure requests so the stable prefix is always at the beginning and the dynamic conversation history follows it. This way, each turn gets a cache hit on the expensive system content.

Multi-Turn Cache Structure

from anthropic import Anthropic
from dataclasses import dataclass, field
from typing import List, Dict

@dataclass
class CachedConversation:
    """Multi-turn conversation optimized for prompt caching."""

    client: Anthropic
    system_prompt: str
    tool_definitions: str
    model: str = "claude-sonnet-4-20250514"
    messages: List[Dict] = field(default_factory=list)
    total_cache_reads: int = 0
    total_cache_creates: int = 0

    def chat(self, user_message: str) -> str:
        """Send message and get response with caching."""
        self.messages.append({
            "role": "user",
            "content": user_message
        })

        response = self.client.messages.create(
            model=self.model,
            max_tokens=2048,
            system=[
                # Large system prompt - cached
                {
                    "type": "text",
                    "text": self.system_prompt,
                    "cache_control": {"type": "ephemeral"}
                },
                # Tool definitions - cached
                {
                    "type": "text",
                    "text": self.tool_definitions,
                    "cache_control": {"type": "ephemeral"}
                }
            ],
            messages=self.messages
        )

        # Track cache performance
        usage = response.usage
        self.total_cache_reads += usage.cache_read_input_tokens
        self.total_cache_creates += usage.cache_creation_input_tokens

        assistant_message = response.content[0].text
        self.messages.append({
            "role": "assistant",
            "content": assistant_message
        })

        return assistant_message

    def get_cache_stats(self) -> dict:
        """Get cumulative cache statistics."""
        return {
            "total_cache_reads": self.total_cache_reads,
            "total_cache_creates": self.total_cache_creates,
            "estimated_savings": self._calculate_savings()
        }

    def _calculate_savings(self) -> str:
        """Estimate cost savings from caching."""
        # Anthropic pricing: cached reads are 90% cheaper
        if self.total_cache_reads == 0:
            return "0%"
        full_cost = self.total_cache_reads + self.total_cache_creates
        actual_cost = (
            self.total_cache_creates +
            self.total_cache_reads * 0.1  # 90% discount
        )
        savings = (1 - actual_cost / full_cost) * 100
        return f"{savings:.1f}%"

Optimization Strategies

Maximizing cache hit rates requires careful control over prompt structure. Content should be ordered from most stable to least stable, because the cache prefix ends at the first differing byte. Static instructions go first. Reference data and knowledge bases come next. Tool definitions follow — but only when sorted deterministically, since random ordering breaks the cache. Session-specific context and conversation history go last, outside the cached region.

Optimal Prompt Ordering for Caching

from dataclasses import dataclass
from typing import List, Optional
import hashlib
import json

@dataclass
class CacheOptimizedPromptBuilder:
    """Build prompts optimized for cache hit rates."""

    static_instructions: str
    reference_data: str
    tool_definitions: List[dict]

    def build_system_content(
        self,
        session_context: Optional[str] = None
    ) -> List[dict]:
        """Build system content with optimal cache structure."""
        content = []

        # Layer 1: Static instructions (highest cache stability)
        content.append({
            "type": "text",
            "text": self._normalize(self.static_instructions),
            "cache_control": {"type": "ephemeral"}
        })

        # Layer 2: Reference data (changes rarely)
        content.append({
            "type": "text",
            "text": self._normalize(self.reference_data),
            "cache_control": {"type": "ephemeral"}
        })

        # Layer 3: Tool definitions (normalized for consistency)
        # Sort tools alphabetically for deterministic ordering
        sorted_tools = sorted(
            self.tool_definitions,
            key=lambda t: t.get("name", "")
        )
        tools_text = json.dumps(sorted_tools, sort_keys=True)
        content.append({
            "type": "text",
            "text": f"Available tools:\n{tools_text}",
            "cache_control": {"type": "ephemeral"}
        })

        # Layer 4: Session context (if any, no cache control)
        if session_context:
            content.append({
                "type": "text",
                "text": session_context
                # No cache_control = not cached
            })

        return content

    def _normalize(self, text: str) -> str:
        """Normalize text to maximize cache hits."""
        lines = [line.rstrip() for line in text.split('\n')]
        normalized = '\n'.join(lines)
        return normalized

    def get_cache_key(self) -> str:
        """Get a hash representing the cacheable prefix."""
        cacheable = (
            self._normalize(self.static_instructions) +
            self._normalize(self.reference_data) +
            json.dumps(sorted(
                self.tool_definitions,
                key=lambda t: t.get("name", "")
            ), sort_keys=True)
        )
        return hashlib.sha256(cacheable.encode()).hexdigest()[:16]

Monitoring and Evaluation

Without tracking cache metrics, you will not know whether caching is working. Always log cache_read_input_tokens on every response. If cache_creation_input_tokens keeps appearing on requests that should be hitting the cache, you have a cache-breaking pattern in your prompt structure.

A simple debugging technique: compute a SHA-256 hash of your system prompt before each request and log it. If the hash changes between requests that should be identical, you have found your cache-breaking source. Byte-level matching means invisible differences — trailing spaces, different line endings, CRLF vs LF — will silently prevent cache hits.

Common Pitfalls

Best Practices