LLM Prompt Caching: How to Cut Your Claude API Costs by 80%

import anthropic
 
client = anthropic.Anthropic()
 
 
def create_devops_assistant_response(user_question: str) -> str:
    """
    DevOps assistant with cached system context.
    The large system prompt is cached — only processed once per 5-minute window.
    """
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": """You are an expert DevOps engineer assistant specializing in:
- Kubernetes (CKA-level knowledge, troubleshooting, production patterns)
- AWS (EKS, EC2, RDS, IAM, networking, cost optimization)
- Terraform (modules, state management, best practices)
- CI/CD (GitHub Actions, ArgoCD, GitOps patterns)
- Observability (Prometheus, Grafana, Loki, OpenTelemetry)
- Security (RBAC, network policies, secrets management, DevSecOps)
 
When answering:
1. Give specific, actionable answers with exact commands
2. Reference real-world patterns from production environments
3. Point out common pitfalls when relevant
4. Use concrete examples over abstract explanations
 
[... your full 2000+ token system prompt here ...]""",
                "cache_control": {"type": "ephemeral"}  # Cache this block
            }
        ],
        messages=[
            {"role": "user", "content": user_question}
        ]
    )
 
    # Check cache status in response
    usage = response.usage
    print(f"Input tokens: {usage.input_tokens}")
    print(f"Cache read tokens: {getattr(usage, 'cache_read_input_tokens', 0)}")
    print(f"Cache write tokens: {getattr(usage, 'cache_creation_input_tokens', 0)}")
 
    return response.content[0].text

def multi_turn_with_caching(
    conversation_history: list[dict],
    new_user_message: str,
    system_prompt: str
) -> tuple[str, list[dict]]:
    """
    Handle multi-turn conversation with caching on history.
    Caches all previous messages so only new ones are processed fresh.
    """
    # Build messages with cache on all but last turn
    messages = []
 
    # Add conversation history with cache breakpoint at the end
    if conversation_history:
        for i, msg in enumerate(conversation_history):
            if i == len(conversation_history) - 1:
                # Cache breakpoint after last history message
                messages.append({
                    "role": msg["role"],
                    "content": [
                        {
                            "type": "text",
                            "text": msg["content"],
                            "cache_control": {"type": "ephemeral"}
                        }
                    ]
                })
            else:
                messages.append(msg)
 
    # Add new user message (not cached — it changes every turn)
    messages.append({"role": "user", "content": new_user_message})
 
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=1024,
        system=[{
            "type": "text",
            "text": system_prompt,
            "cache_control": {"type": "ephemeral"}
        }],
        messages=messages
    )
 
    assistant_reply = response.content[0].text
 
    # Update history
    updated_history = conversation_history + [
        {"role": "user", "content": new_user_message},
        {"role": "assistant", "content": assistant_reply}
    ]
 
    return assistant_reply, updated_history

def rag_with_cached_context(
    retrieved_docs: list[str],
    user_question: str
) -> str:
    """
    RAG query where retrieved documents are cached.
    Multiple questions about the same docs only process them once.
    """
    context = "\n\n---\n\n".join(retrieved_docs)
 
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": "Answer questions based on the provided documentation context.",
                "cache_control": {"type": "ephemeral"}
            },
            {
                "type": "text",
                "text": f"## Documentation Context\n\n{context}",
                "cache_control": {"type": "ephemeral"}  # Cache the retrieved docs
            }
        ],
        messages=[
            {"role": "user", "content": user_question}  # Only question changes
        ]
    )
 
    return response.content[0].text
 
 
# Example: Multiple questions about the same runbook
runbook_docs = [load_runbook("payment-api"), load_runbook("postgres")]
 
# First question — writes to cache
answer1 = rag_with_cached_context(runbook_docs, "How do I restart the payment service?")
 
# Second question — reads from cache (10x cheaper)
answer2 = rag_with_cached_context(runbook_docs, "What are the postgres backup procedures?")

import logging
 
logger = logging.getLogger(__name__)
 
 
def call_with_cache_metrics(messages: list, system: list, model: str = "claude-sonnet-5") -> str:
    response = client.messages.create(
        model=model,
        max_tokens=1024,
        system=system,
        messages=messages
    )
 
    usage = response.usage
    cache_read = getattr(usage, "cache_read_input_tokens", 0)
    cache_write = getattr(usage, "cache_creation_input_tokens", 0)
    regular_input = usage.input_tokens - cache_read - cache_write
 
    cache_hit_rate = cache_read / usage.input_tokens if usage.input_tokens > 0 else 0
 
    logger.info({
        "event": "llm_call",
        "model": model,
        "regular_input_tokens": regular_input,
        "cache_read_tokens": cache_read,
        "cache_write_tokens": cache_write,
        "cache_hit_rate": f"{cache_hit_rate:.1%}",
        "output_tokens": usage.output_tokens,
    })
 
    return response.content[0].text