Cost & Performance Engineering

# routing.py - one classification task, three models, real prices

import anthropic

client = anthropic.Anthropic()

PRICES = {   # USD per 1M tokens, list price, August 2026
    "claude-haiku-4-5": (1.00, 5.00),
    "claude-sonnet-5":  (2.00, 10.00),
    "claude-opus-5":    (5.00, 25.00),
}

REVIEWS = [
    "Shipping took three weeks and the box was crushed. Never again.",
    "Exactly what I needed, arrived early, works perfectly.",
    "It's fine. Does the job, nothing special.",
]

REPEATS = 3        # how much Opus thinks varies per call, so do not trust one sample

print(f"{'model':<20}{'in':>6}{'out':>6}{'cost/1k reqs':>14}   answers")
for model in PRICES:
    total_in = total_out = calls = 0
    answers = []
    for _ in range(REPEATS):
        for review in REVIEWS:
            r = client.messages.create(
                model=model, max_tokens=2000,
                messages=[{"role": "user", "content":
                    "Classify sentiment as positive, negative, or neutral. "
                    f"Reply with one word only.\n\n{review}"}],
            )
            total_in += r.usage.input_tokens
            total_out += r.usage.output_tokens
            calls += 1
            # Opus 5 thinks by default, so content[0] may be a ThinkingBlock.
            answers.append(next(b.text for b in r.content if b.type == "text").strip().lower())

    inp, out = PRICES[model]
    per_req = (total_in * inp + total_out * out) / 1_000_000 / calls
    print(f"{model:<20}{total_in:>6}{total_out:>6}{per_req * 1000:>14.4f}$   "
          f"{sorted(set(answers))}")

# effort.py - four levels, four samples each, on one Opus 5 question

import statistics
import time

import anthropic

client = anthropic.Anthropic()

PROMPT = ("A team of 5 engineers maintains a Python monolith with 40 minute CI. "
          "Should they split it into services? Answer in under 120 words.")

IN_PRICE, OUT_PRICE = 5.00, 25.00        # Opus 5, USD per 1M tokens
SAMPLES = 4                              # one sample per level is noise, not a measurement

print(f"{'effort':<9}{'thinking':>9}{'output':>8}{'cost/1k reqs':>14}{'latency':>9}")
for effort in ("low", "medium", "high", "max"):
    thinking, output, sent, latency = [], [], [], []
    for _ in range(SAMPLES):
        t0 = time.monotonic()
        r = client.messages.create(
            model="claude-opus-5", max_tokens=8000,
            output_config={"effort": effort},
            messages=[{"role": "user", "content": PROMPT}],
        )
        latency.append(time.monotonic() - t0)
        thinking.append(r.usage.output_tokens_details.thinking_tokens)
        output.append(r.usage.output_tokens)
        sent.append(r.usage.input_tokens)

    med_out = statistics.median(output)
    per_1k = (statistics.median(sent) * IN_PRICE + med_out * OUT_PRICE) / 1_000
    print(f"{effort:<9}{statistics.median(thinking):>9.0f}{med_out:>8.0f}"
          f"{per_1k:>13.4f}${statistics.median(latency):>8.1f}s")

# cache_math.py - what a cached prefix is worth at each request count

PREFIX = 4447     # tokens in the cached prefix (measured in Lesson 17)
RATE   = 1.00     # Haiku input, $ per 1M tokens


def uncached(n):
    return n * PREFIX * RATE / 1_000_000


def cached(n, write_mult=1.25):     # 1.25x for the 5 minute TTL
    return (PREFIX * RATE * write_mult
            + (n - 1) * PREFIX * RATE * 0.1) / 1_000_000


print(f"{'requests':>9}{'uncached':>11} {'cached 5m':>11} {'saving':>9}")
for n in (1, 2, 3, 10, 100):
    u, c = uncached(n), cached(n)
    print(f"{n:>9}{u:>11.5f}${c:>11.5f}${(1 - c / u) * 100:>9.0f}%")

# batch.py - submit, poll, and collect results by custom_id

import time

import anthropic
from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request

client = anthropic.Anthropic()

reviews = [
    "Shipping took three weeks and the box was crushed. Never again.",
    "Exactly what I needed, arrived early, works perfectly.",
    "It's fine. Does the job, nothing special.",
]

batch = client.messages.batches.create(requests=[
    Request(
        custom_id=f"r-{i}",
        params=MessageCreateParamsNonStreaming(
            model="claude-haiku-4-5", max_tokens=10,
            messages=[{"role": "user", "content":
                "Classify sentiment as positive, negative, or neutral. "
                f"Reply with one word only.\n\n{t}"}],
        ),
    )
    for i, t in enumerate(reviews)
])
print("batch id     :", batch.id)
print("status       :", batch.processing_status)

while True:                      # poll until done
    b = client.messages.batches.retrieve(batch.id)
    if b.processing_status == "ended":
        break
    time.sleep(10)

print("final status :", b.processing_status)
print("counts       :", b.request_counts)

# Results arrive in ANY order. Key by custom_id, never by position.
for res in client.messages.batches.results(batch.id):
    answer = res.result.message.content[0].text.strip()
    print(" ", res.custom_id, "->", res.result.type, "|", answer)

# streaming.py - time to first token against time to full answer

import statistics
import time

import anthropic

client = anthropic.Anthropic()

Q = "Explain what a database index is, in about 150 words."
SAMPLES = 3        # network latency is noisy, so take a median

blocking, first_token, streamed = [], [], []
for _ in range(SAMPLES):
    # Non-streaming: the user sees nothing until the whole answer is ready.
    t0 = time.monotonic()
    client.messages.create(model="claude-haiku-4-5", max_tokens=400,
                           messages=[{"role": "user", "content": Q}])
    blocking.append(time.monotonic() - t0)

    # Streaming: measure time to the FIRST visible token.
    t0 = time.monotonic()
    first = None
    with client.messages.stream(model="claude-haiku-4-5", max_tokens=400,
                                messages=[{"role": "user", "content": Q}]) as stream:
        for _chunk in stream.text_stream:
            if first is None:
                first = time.monotonic() - t0
        streamed.append(time.monotonic() - t0)
    first_token.append(first)

b = statistics.median(blocking)
f = statistics.median(first_token)
s = statistics.median(streamed)

print(f"non-streaming: nothing visible for {b:.2f}s, then the whole answer")
print(f"streaming    : first token at {f:.2f}s, complete at {s:.2f}s")
print(f"perceived improvement: {b / f:.1f}x faster to first visible output")

# cost_dashboard.py - a wrapper that prices every call as it happens

import collections

import anthropic

client = anthropic.Anthropic()

PRICES = {"claude-haiku-4-5": (1.00, 5.00)}   # USD per 1M tokens

spend = collections.defaultdict(float)


def tracked(model, **kwargs):
    r = client.messages.create(model=model, **kwargs)
    u = r.usage
    inp, out = PRICES[model]

    # Cached reads are ~0.1x, cache writes 1.25x at the default TTL.
    dollars = (
        u.input_tokens * inp
        + (u.cache_read_input_tokens or 0) * inp * 0.1
        + (u.cache_creation_input_tokens or 0) * inp * 1.25
        + u.output_tokens * out
    ) / 1_000_000

    spend[model] += dollars
    return r


for question in ["Name one benefit of an index.", "Name one cost of an index."]:
    r = tracked("claude-haiku-4-5", max_tokens=60,
                messages=[{"role": "user", "content": question}])
    u = r.usage
    print(f"prompt tokens: uncached={u.input_tokens:4} "
          f"cache_write={u.cache_creation_input_tokens or 0:4} "
          f"cache_read={u.cache_read_input_tokens or 0:4} "
          f"out={u.output_tokens:4}")

for model, dollars in spend.items():
    print(f"{model}: ${dollars:.6f} over {len(spend)} tracked model(s)")