Prompt caching with the Anthropic SDK

Open in Colab Run this notebook in Colab

Prompt caching with the Anthropic SDK#

Track 08 - Production · Notebook 01 · Runtime: ~30s LIVE, <1s replay

cache_control on a long system prompt: 5 questions over a fixed doc, measured input tokens, latency, and dollars. LIVE mode hits the real API; without ANTHROPIC_API_KEY, replays recorded responses from _fixtures/.

from llm_systems_cookbook.nb import bootstrap
from llm_systems_cookbook._utils import repo_root

import json
import os
import time
from pathlib import Path

s = bootstrap("08_production_01_claude_sdk_prompt_caching")

LIVE = bool(os.environ.get("ANTHROPIC_API_KEY"))
MODEL = os.environ.get("MODEL_ANTHROPIC", "claude-sonnet-4-6")
FIXTURE = Path(repo_root()) / "notebooks/08_production/_fixtures/01_caching.json"
print(f"mode={'LIVE' if LIVE else 'REPLAY'}  model={MODEL}")

Corpus and questions#

First 25k chars of CURRICULUM_SPEC.md (~7.8k tokens) as the system prompt; five questions about it. The doc is the cacheable prefix; questions are the variable suffix.

DOC = (Path(repo_root()) / "CURRICULUM_SPEC.md").read_text()[:25_000]

QUESTIONS = [
    "Which notebook covers PagedAttention?",
    "What is the prerequisite for the FlashAttention2 Triton notebook?",
    "Which track has the most notebooks in v0.1?",
    "Name the three KV cache compression methods covered in track 05.",
    "Which evaluation notebook teaches Bradley-Terry?",
]
print(f"doc: {len(DOC):,} chars  questions: {len(QUESTIONS)}")

One call function, two modes#

cache_control={"type": "ephemeral"} on a system block marks it as the cached prefix (5-min TTL). The first call returns cache_creation_input_tokens; subsequent calls return cache_read_input_tokens. Reads cost 10% of input, writes cost 125%.

FIXTURES = json.loads(FIXTURE.read_text())

def call(question: str, *, use_cache: bool, fixture_idx: int) -> dict:
    """One Claude call. LIVE: real API. REPLAY: recorded usage."""
    if not LIVE:
        return FIXTURES["cached" if use_cache else "no_cache"][fixture_idx]

    import anthropic
    client = anthropic.Anthropic()

    system_block: dict = {"type": "text", "text": DOC}
    if use_cache:
        system_block["cache_control"] = {"type": "ephemeral"}

    t0 = time.perf_counter()
    resp = client.messages.create(
        model=MODEL,
        max_tokens=200,
        system=[system_block],
        messages=[{"role": "user", "content": question}],
    )
    latency = time.perf_counter() - t0
    u = resp.usage
    return {
        "question": question,
        "text": resp.content[0].text,
        "input_tokens": u.input_tokens,
        "output_tokens": u.output_tokens,
        "cache_creation_input_tokens": getattr(u, "cache_creation_input_tokens", 0) or 0,
        "cache_read_input_tokens": getattr(u, "cache_read_input_tokens", 0) or 0,
        "latency_s": latency,
    }

Pattern A — no caching#

Doc shipped fresh on every call. Each request pays full input rate on ~7.8k tokens.

no_cache = [call(q, use_cache=False, fixture_idx=i) for i, q in enumerate(QUESTIONS)]
for r in no_cache:
    print(f"  in={r['input_tokens']:>5}  out={r['output_tokens']:>3}  {r['latency_s']:.2f}s")

Pattern B — cached system prompt#

First call pays the cache write (1.25x); calls 2-5 read at 0.10x. Same answers, cheaper and faster.

cached = [call(q, use_cache=True, fixture_idx=i) for i, q in enumerate(QUESTIONS)]
for r in cached:
    c, h = r["cache_creation_input_tokens"], r["cache_read_input_tokens"]
    tag = "WRITE" if c else ("READ" if h else "MISS")
    print(f"  {tag:5}  in={r['input_tokens']:>3}  hit={h:>5}  out={r['output_tokens']:>3}  {r['latency_s']:.2f}s")

Cost#

Claude Sonnet 4.6 pricing per million tokens: input \(3.00, output \)15.00, cache write \(3.75, cache read \)0.30. (Anthropic publishes these on the pricing page; override PRICE for other models.)

PRICE = {"in": 3.0, "out": 15.0, "cache_write": 3.75, "cache_read": 0.30}

def cost_usd(rs: list[dict]) -> float:
    micro = (
        sum(r["cache_creation_input_tokens"] for r in rs) * PRICE["cache_write"]
        + sum(r["cache_read_input_tokens"]    for r in rs) * PRICE["cache_read"]
        + sum(r["input_tokens"]               for r in rs) * PRICE["in"]
        + sum(r["output_tokens"]              for r in rs) * PRICE["out"]
    )
    return micro / 1_000_000

cost_a, cost_b = cost_usd(no_cache), cost_usd(cached)
lat_a  = sum(r["latency_s"] for r in no_cache)
lat_b  = sum(r["latency_s"] for r in cached)

print(f"no cache:  ${cost_a:.4f}   {lat_a:.2f}s")
print(f"cached:    ${cost_b:.4f}   {lat_b:.2f}s")
print(f"savings:   {1 - cost_b/cost_a:.0%} cost   {1 - lat_b/lat_a:.0%} wall time")

Visual#

Total cost on the left; per-call latency on the right. Cached pattern’s first call is slightly slower (cache write); the next four pay almost nothing.

import matplotlib.pyplot as plt

fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(9, 3.2))

ax1.bar(["no cache", "cached"], [cost_a * 100, cost_b * 100],
        color=["tab:red", "tab:green"])
ax1.set_ylabel("cost (cents)")
ax1.set_title(f"5 calls × {len(DOC):,}-char doc")
for i, v in enumerate([cost_a * 100, cost_b * 100]):
    ax1.text(i, v, f"{v:.2f}c", ha="center", va="bottom")

ax2.plot(range(1, 6), [r["latency_s"] for r in no_cache], "o-", color="tab:red", label="no cache")
ax2.plot(range(1, 6), [r["latency_s"] for r in cached],   "o-", color="tab:green", label="cached")
ax2.set_xlabel("call #"); ax2.set_ylabel("latency (s)")
ax2.set_title("per-call latency"); ax2.legend(); ax2.grid(alpha=0.3)

fig.tight_layout(); plt.show()

Checks#

s.check(
    "cache_write_on_first_call",
    lambda: cached[0]["cache_creation_input_tokens"] >= 1024,
    msg=f"first cache_creation = {cached[0]['cache_creation_input_tokens']}",
)
s.check(
    "cache_read_on_subsequent_calls",
    lambda: all(r["cache_read_input_tokens"] > 0 for r in cached[1:]),
    msg=f"reads = {[r['cache_read_input_tokens'] for r in cached[1:]]}",
)
s.check(
    "cached_cheaper_by_at_least_50pct",
    lambda: cost_b <= cost_a * 0.5,
    msg=f"cached=${cost_b:.4f}  uncached=${cost_a:.4f}",
)
s.check(
    "cached_post_first_call_30pct_faster",
    lambda: sum(r["latency_s"] for r in cached[1:]) <= sum(r["latency_s"] for r in no_cache[1:]) * 0.7,
    msg=(f"cached tail={sum(r['latency_s'] for r in cached[1:]):.2f}s "
         f"vs uncached tail={sum(r['latency_s'] for r in no_cache[1:]):.2f}s"),
)
s.check(
    "cached_answers_match_uncached_for_first_question",
    lambda: "PagedAttention".lower() in cached[0]["text"].lower(),
    msg=f"answer = {cached[0]['text']!r}",
)

Notes for production#

  • 1-hour cache (beta): pass extra_headers={"anthropic-beta": "extended-cache-ttl-2025-04-11"} and set cache_control={"type": "ephemeral", "ttl": "1h"}. Useful when one agent works for an hour over the same system prompt.

  • Multiple breakpoints: up to 4 cache_control markers per request. Cache the system, the tool definitions, and the long retrieved context independently.

  • Exact-prefix match: a single token change above a breakpoint busts the cache for that breakpoint and everything below it. Keep dynamic content (user query, fresh retrieval) below the cached blocks.

  • Min size: 1024 tokens for Sonnet/Haiku 4.x. Below that, the flag is silently ignored.

  • OpenAI / Gemini: both auto-cache prefixes ≥1024 tokens with no opt-in needed; pricing and behaviour differ but the prompt-design rule (dynamic content goes last) is the same.

s.summary()
s.save()