Running large language models across extensive documents, detailed system instructions, and multi-turn agent conversations quickly inflates API costs and introduces noticeable latency. Re-evaluating hundreds of static context tokens on every request forces the model provider to compute identical attention matrices repeatedly. By implementing prompt caching, applications store pre-computed token states across requests, slashing latency by up to 80% and cutting input token expenses.

How Prompt Caching Works Under the Hood

Modern transformer inference engines spend significant GPU cycles calculating Key-Value (KV) cache states for input tokens. When sequential API requests share an identical prompt prefix, the provider checks an in-memory KV cache rather than recalculating those attention vectors from scratch.

If the prefix matches existing cached blocks, the model immediately jumps to processing only the new appended tokens. This architectural optimization turns heavy document retrieval and multi-step reasoning agents into cost-effective production services.

Design Rules for High Cache Hit Ratios

Because prompt caching relies on exact prefix matching, small architectural mistakes can invalidate your entire cache. Follow these foundational rules when constructing prompts:

• Anchor Static Text Upfront: Place system prompts, developer instructions, and few-shot examples at the absolute start of the message array.
• Keep Dynamic Data at the Tail: Inject user questions, timestamps, and session-specific variables only at the very end of the prompt.
• Mind Token Thresholds: Most providers require a minimum prompt length (typically 1,024 tokens) before activating automatic or explicit caching.
• Maintain Exact Byte Alignment: Any alteration to whitespace, punctuation, or schema ordering breaks the cache match completely.

Implementing Prompt Caching with Python and the Anthropic API

In this example, we build a research assistant that analyzes complex engineering documentation. We mark our large reference context with the cache_control parameter, telling the API to retain those tokens in memory across user turns.

We implement thorough error handling to catch API rate limits and network disconnects, logging cache creation and hit counts directly from the usage response metadata.

from __future__ import annotations

import os
import sys
from typing import Any
import anthropic
from anthropic.types import Message


def query_cached_knowledge_base(
    system_manual: str,
    user_query: str,
    client: anthropic.Anthropic | None = None,
) -> str:
    """
    Query an LLM with prompt caching applied to a large reference manual.
    
    We mark the static system manual with ephemeral cache control. This allows
    consequent user turns to reuse pre-computed attention matrices without paying full token costs.
    """
    if client is None:
        api_key = os.environ.get("ANTHROPIC_API_KEY")
        if not api_key:
            raise ValueError("ANTHROPIC_API_KEY environment variable is missing.")
        client = anthropic.Anthropic(api_key=api_key)

    try:
        # We place the heavy static text in the system array with a cache breakpoint.
        response: Message = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            system=[
                {
                    "type": "text",
                    "text": (
                        "You are a helpful senior systems engineer. "
                        "Answer technical questions accurately based strictly on the manual provided below."
                    ),
                },
                {
                    "type": "text",
                    "text": system_manual,
                    # We set ephemeral cache control so this block is retained in the KV cache.
                    "cache_control": {"type": "ephemeral"},
                },
            ],
            messages=[
                {
                    "role": "user",
                    "content": user_query,
                }
            ],
        )

        # Inspect usage metadata to verify cache behavior.
        usage = response.usage
        cached_read_tokens = getattr(usage, "cache_read_input_tokens", 0)
        cache_creation_tokens = getattr(usage, "cache_creation_input_tokens", 0)
        input_tokens = usage.input_tokens

        print(
            f"[Prompt Cache Metrics] Input: {input_tokens} | "
            f"Cache Created: {cache_creation_tokens} | "
            f"Cache Read (Hit): {cached_read_tokens}"
        )

        # Extract text content safely from the response block.
        first_block = response.content[0]
        if hasattr(first_block, "text"):
            return first_block.text
        return ""

    except anthropic.APIConnectionError as err:
        print(f"Network error connecting to Anthropic API: {err}", file=sys.stderr)
        raise
    except anthropic.RateLimitError as err:
        print(f"Hit provider rate limits. Implement exponential backoff: {err}", file=sys.stderr)
        raise
    except anthropic.APIStatusError as err:
        print(f"API returned status {err.status_code}: {err.message}", file=sys.stderr)
        raise

Monitoring and Measuring Production ROI

After deploying prompt caching, monitor your cache hit percentage through application telemetry. Look specifically at your cache_read_input_tokens metric relative to total input volume.

If your cache hit rate dips below 70%, audit your middleware pipelines. Hidden culprits like randomized tracking IDs, dynamic timestamps, or varying whitespace often silently break cache prefixes.