Build Prompt Caching Patterns: System Prompts and Few-Shot Examples 2026

# Use uv for fast, reproducible installs — avoids pip dependency conflicts
uv pip install anthropic>=0.40.0
export ANTHROPIC_API_KEY="sk-ant-..."

import anthropic

client = anthropic.Anthropic()

# A realistic system prompt: role definition + behavior rules + output format
SYSTEM_PROMPT = """You are a senior Python code reviewer with 10 years of experience.
Your job is to identify bugs, performance issues, and security vulnerabilities.

Rules:
- Always cite the specific line number
- Explain the risk level: LOW / MEDIUM / HIGH / CRITICAL
- Suggest the exact fix, not just the problem
- If the code is clean, say so explicitly — don't invent issues

Output format:
## Finding [N]: [SHORT TITLE]
**Line:** [N]
**Risk:** [LEVEL]
**Issue:** [Description]
**Fix:** [Code snippet or instruction]
""" * 3  # Repeat to exceed 1,024-token minimum for this example

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": SYSTEM_PROMPT,
            "cache_control": {"type": "ephemeral"},  # Mark this block as cacheable
        }
    ],
    messages=[
        {"role": "user", "content": "Review this: def get_user(id): return db.query(f'SELECT * FROM users WHERE id={id}')"}
    ],
)

# Check cache performance
usage = response.usage
print(f"Input tokens:          {usage.input_tokens}")
print(f"Cache creation tokens: {usage.cache_creation_input_tokens}")  # 125% cost — first hit
print(f"Cache read tokens:     {usage.cache_read_input_tokens}")       # 10% cost — subsequent hits

# Few-shot examples that teach the model your exact output format
FEW_SHOT = [
    {
        "role": "user",
        "content": "Classify this support ticket: 'My payment failed with error code 4003'"
    },
    {
        "role": "assistant",
        "content": '{"category": "billing", "priority": "high", "error_code": "4003", "requires_human": true}'
    },
    {
        "role": "user",
        "content": "Classify this support ticket: 'How do I export my data to CSV?'"
    },
    {
        "role": "assistant",
        # cache_control on the LAST message in the static prefix
        # Everything before this point will be cached as one block
        "content": [
            {
                "type": "text",
                "text": '{"category": "feature_request", "priority": "low", "error_code": null, "requires_human": false}',
                "cache_control": {"type": "ephemeral"},
            }
        ],
    },
]

def classify_ticket(ticket_text: str) -> dict:
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=256,
        messages=[
            *FEW_SHOT,  # Cached prefix — only processed once per 5-min window
            {"role": "user", "content": f"Classify this support ticket: '{ticket_text}'"},  # Dynamic suffix
        ],
    )
    return response.content[0].text

# First call: cache write (125% cost on FEW_SHOT tokens)
result = classify_ticket("I can't log in, keeps saying invalid password")
print(result)

# Second call: cache hit (10% cost on FEW_SHOT tokens)
result = classify_ticket("Does your API support webhooks?")
print(result)

TOOLS = [
    {
        "name": "search_codebase",
        "description": "Search the codebase for files matching a pattern or containing specific text.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search query or regex pattern"},
                "file_type": {"type": "string", "enum": ["py", "ts", "go", "rs", "all"]},
                "max_results": {"type": "integer", "default": 10}
            },
            "required": ["query"]
        }
    },
    # ... more tools ...
    {
        "name": "run_tests",
        "description": "Execute the test suite for a given module and return results.",
        "input_schema": {
            "type": "object",
            "properties": {
                "module": {"type": "string"},
                "verbose": {"type": "boolean", "default": False}
            },
            "required": ["module"]
        },
        # cache_control on the LAST tool in the array
        "cache_control": {"type": "ephemeral"},
    }
]

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    tools=TOOLS,
    messages=[{"role": "user", "content": "Find all files that import requests and run their tests"}],
)

def chat_with_cache(conversation_history: list, new_message: str) -> str:
    """
    conversation_history: list of {"role": ..., "content": ...} dicts
    The FIRST message in history has cache_control — it marks the end of the static prefix.
    """

    # Mark the first assistant turn as the cache boundary
    # This caches: system prompt + first user message + first assistant response
    # Everything after is dynamic and processed fresh
    cached_prefix = []
    for i, msg in enumerate(conversation_history):
        if i == 0 and msg["role"] == "assistant":
            cached_prefix.append({
                **msg,
                "content": [{"type": "text", "text": msg["content"], "cache_control": {"type": "ephemeral"}}]
                if isinstance(msg["content"], str) else msg["content"]
            })
        else:
            cached_prefix.append(msg)

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        system=[{"type": "text", "text": SYSTEM_PROMPT, "cache_control": {"type": "ephemeral"}}],
        messages=[*cached_prefix, {"role": "user", "content": new_message}],
    )
    return response.content[0].text

import time

def benchmark_cache(prompt_func, label: str, runs: int = 3):
    timings = []
    cache_reads = []

    for i in range(runs):
        start = time.perf_counter()
        response = prompt_func()
        elapsed = time.perf_counter() - start

        timings.append(elapsed)
        cache_reads.append(response.usage.cache_read_input_tokens)
        print(f"  Run {i+1}: {elapsed:.2f}s | cache_read={response.usage.cache_read_input_tokens}")
        time.sleep(1)  # Avoid rate limits

    print(f"\n{label} summary:")
    print(f"  Avg latency:    {sum(timings)/len(timings):.2f}s")
    print(f"  Avg cache read: {sum(cache_reads)/len(cache_reads):.0f} tokens")
    print(f"  Cache active:   {'YES' if cache_reads[-1] > 0 else 'NO — check token minimum'}")

benchmark_cache(
    lambda: client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=128,
        system=[{"type": "text", "text": SYSTEM_PROMPT, "cache_control": {"type": "ephemeral"}}],
        messages=[{"role": "user", "content": "Ping"}],
    ),
    label="System prompt cache"
)