The Problem
AI API costs scale with every token processed. An agent that works beautifully in testing can become financially unsustainable at scale. Three engineering decisions determine whether your AI product is profitable: prompt caching, model tiering, and budget enforcement.
These aren't micro-optimizations — the right combination can reduce costs by 50–80% with no loss in quality.
Prompt caching
Anthropic's prompt caching lets you mark portions of your prompt with cache_control: {"type": "ephemeral"}. When the same cached prefix appears in a subsequent request within the 5-minute TTL, Anthropic charges only 10% of the normal input token cost for those tokens — a 90% discount.
The key rule: stable content must come first. Anthropic caches a prefix of your message — everything up to the first position that changes between requests. Structure your messages to maximize the cached prefix:
[System prompt - 5,000 tokens - CACHE THIS]
[Retrieved documents - 40,000 tokens - CACHE THIS]
[Conversation history - 2,000 tokens - CACHE THIS]
[Current user message - 50 tokens - changes each request]
For a system with a 50,000-token knowledge base injected into every request, caching can save 90% on those 50,000 tokens for any request within the TTL. At $3/MTok for input tokens, that's $0.135 saved per request on the cached portion alone.
You have a 50,000-token system prompt (a large knowledge base). How should you structure messages to maximize cache hits?
Anthropic's prompt cache works positionally: it caches a prefix of the message up to the first position that changes. Put your large, stable content (system prompt, knowledge base, retrieved docs) at the very beginning. Dynamic content (user message, this-session context) goes at the end. This maximizes the cached prefix, giving the maximum 90% discount on stable tokens.
Model tiering
Not every task requires your most capable (and expensive) model. Route tasks by complexity:
| Model | Cost | Best for |
|---|
| Haiku | ~$0.25/MTok | Classification, extraction, routing, simple Q&A |
| Sonnet | ~$3/MTok | Multi-step reasoning, code gen, most agent tasks |
| Opus/frontier | ~$15/MTok | Complex reasoning, novel synthesis, final reports |
Cost difference: Haiku is approximately 20x cheaper than Opus per token. A TaskClassifier routes each subtask to the appropriate tier before executing.
class Complexity(str, Enum):
LOW = "low" # → Haiku
MEDIUM = "medium" # → Sonnet
HIGH = "high" # → Opus
Most agent systems can route 40–60% of tasks to Haiku — classification, routing decisions, simple lookups, formatting — reducing total costs dramatically while maintaining quality where it matters.
Which task should route to Claude Haiku (cheapest model) rather than Claude Sonnet?
Classification is the canonical Haiku use case: structured output, limited options, minimal reasoning required. Routing this to Sonnet or Opus wastes 5-20x the tokens for no quality benefit. Apply model tiering by task type: extraction/classification → Haiku; agent tasks/code → Sonnet; complex reasoning/synthesis → Opus. Most agent systems can route 40-60% of tasks to Haiku.
Token budget enforcement
Budget management has three levels:
Per-session: A single conversation cannot exceed X tokens. Prevents a runaway agent loop from consuming unlimited resources on one session. When the session budget is exceeded, the agent should gracefully summarize what it has done and stop — not error out.
Per-user/day: Limits abuse and cost exposure from any single user. Essential for free-tier products where individual users can't be billed for overages.
Per-agent: Different agent types have different expected costs. A FAQ bot should cost far less than a research synthesis agent. Budget enforcement ensures each agent type stays within its expected range, making unusual spikes visible.
Build
Build it
Tracking real cost: cache reads are not free, cache writes are not cheap
The TokenUsage dataclass is the foundation of cost debugging. It separates four token categories that most engineers collapse into one: regular input tokens, output tokens, cache-creation tokens, and cache-read tokens. The cost calculation is where the pricing asymmetry becomes visible — cache reads cost 10% of normal input, but cache creation costs 25% more than normal input. If your cache hit rate is low, you're paying a premium to write caches that never get read, which is worse than not caching at all. This is the number to check first when costs are higher than expected.
from dataclasses import dataclass
@dataclass
class TokenUsage:
input_tokens: int = 0
output_tokens: int = 0
cache_creation_tokens: int = 0
cache_read_tokens: int = 0
@property
def total_tokens(self) -> int:
return self.input_tokens + self.output_tokens
def cost_usd(self, model: str, cost_per_mtk: dict[str, float]) -> float:
rate = cost_per_mtk.get(model, 3.0)
# Cache reads cost 10% of normal input — the 90% discount
# Cache creation costs 25% MORE than normal input — the write premium
effective_input = (
(self.input_tokens - self.cache_read_tokens) * rate
+ self.cache_read_tokens * rate * 0.1
+ self.cache_creation_tokens * rate * 1.25
) / 1_000_000
output_cost = self.output_tokens * rate / 1_000_000
return effective_input + output_cost
Structuring messages for maximum cache hits
The PromptCacheManager encodes the positional caching rule as code. Stable content (retrieved context, system prompt) gets cache_control: {"type": "ephemeral"} and comes first in the message. Dynamic content (the user's actual question) comes last with no cache marker. The critical implementation detail is that cache_control is set on the content block, not the message — Anthropic caches at the block level within a message, so you can mix cached and uncached content in the same turn.
class PromptCacheManager:
"""Structure messages to maximize Anthropic prompt cache hits."""
def build_cached_messages(
self,
system_prompt: str,
retrieved_context: str,
user_message: str,
) -> tuple[str, list[dict]]:
messages = [
{
"role": "user",
"content": [
# Stable: retrieved context with cache_control — comes first
{
"type": "text",
"text": f"<context>\n{retrieved_context}\n</context>",
"cache_control": {"type": "ephemeral"},
},
# Dynamic: actual user question — no cache_control, comes last
{
"type": "text",
"text": user_message,
},
],
}
]
return system_prompt, messages
def cache_hit_rate(self, usages: list[TokenUsage]) -> float:
"""Calculate cache hit rate across a batch of calls."""
total_input = sum(u.input_tokens for u in usages)
cache_reads = sum(u.cache_read_tokens for u in usages)
if total_input == 0:
return 0.0
return cache_reads / total_input
Model tiering: use the cheapest model to decide which model to use
The ModelRouter uses Haiku itself to classify task complexity — a pattern called "model-as-router." This costs ~10 tokens at Haiku rates (essentially free) and prevents sending a simple classification task to Opus at 60x the cost. The fallback to MEDIUM on parse failure is important: a routing classifier that errors out and blocks the pipeline is worse than one that defaults to a safe middle tier.
from enum import Enum
import anthropic
class Complexity(str, Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
MODEL_TIERS = {
Complexity.LOW: "claude-haiku-4-5",
Complexity.MEDIUM: "claude-sonnet-4-5",
Complexity.HIGH: "claude-opus-4-5",
}
class ModelRouter:
"""Route tasks to the appropriate model tier based on complexity."""
def __init__(self, client: anthropic.Anthropic):
self.client = client
def classify_complexity(self, task: str) -> Complexity:
"""Use Haiku itself to classify — cheap, fast."""
response = self.client.messages.create(
model="claude-haiku-4-5",
max_tokens=10,
system="Classify task complexity. Reply with exactly one word: low, medium, or high.",
messages=[{"role": "user", "content": task}],
)
raw = response.content[0].text.strip().lower()
try:
return Complexity(raw)
except ValueError:
return Complexity.MEDIUM # safe fallback, never block the pipeline
def route(self, task: str) -> str:
complexity = self.classify_complexity(task)
return MODEL_TIERS[complexity]
Budget enforcement: three limits, one check
The TokenBudgetManager checks all three budget levels in a single check_and_consume call before the API request fires. The order matters: session is checked first (cheapest to evaluate, most likely to trip), then per-user, then per-agent. If any limit is exceeded, BudgetExceededError is raised before the API call — not after. A budget check that runs after the call is a report, not a control. The token estimate (len // 4) is deliberately rough; precise counting happens after the response returns, but the pre-check only needs to be close enough to prevent runaway costs.
from dataclasses import dataclass, field
class BudgetExceededError(Exception):
pass
@dataclass
class TokenBudgetManager:
"""Enforce per-session, per-user, per-agent token limits."""
session_limit: int = 100_000
user_daily_limit: int = 500_000
agent_limit: int = 200_000
_session_used: int = field(default=0, init=False)
_user_daily: dict[str, int] = field(default_factory=dict, init=False)
_agent_used: dict[str, int] = field(default_factory=dict, init=False)
def check_and_consume(
self, tokens: int, user_id: str, agent_type: str,
) -> None:
"""Raise BudgetExceededError if any limit is breached — before the API call."""
if self._session_used + tokens > self.session_limit:
raise BudgetExceededError(
f"Session limit reached ({self.session_limit} tokens). "
f"Used: {self._session_used}"
)
user_used = self._user_daily.get(user_id, 0)
if user_used + tokens > self.user_daily_limit:
raise BudgetExceededError(
f"User daily limit reached for {user_id}: "
f"{self.user_daily_limit} tokens"
)
agent_used = self._agent_used.get(agent_type, 0)
if agent_used + tokens > self.agent_limit:
raise BudgetExceededError(
f"Agent type '{agent_type}' limit reached: "
f"{self.agent_limit} tokens"
)
# Only consume if all three checks pass
self._session_used += tokens
self._user_daily[user_id] = user_used + tokens
self._agent_used[agent_type] = agent_used + tokens
Tying it together: the cost-optimized agent
The CostOptimizedAgent wires all three mechanisms into a single run() method. The call order is deliberate: budget check first (cheapest, prevents the API call entirely), then model routing (spends ~10 Haiku tokens to pick the right tier), then the actual API call with cached messages. The cost log records the model used, token breakdown, and per-call cost — this is the data you need when someone asks "why did our API bill triple last month?"
class CostOptimizedAgent:
"""Agent with caching, tiering, and budget enforcement."""
def __init__(self, system_prompt: str, budget: TokenBudgetManager | None = None):
self.client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
self.system_prompt = system_prompt
self.cache_mgr = PromptCacheManager()
self.router = ModelRouter(self.client)
self.budget = budget or TokenBudgetManager()
self.cost_log: list[dict] = []
def run(self, user_message: str, context: str = "",
user_id: str = "default", agent_type: str = "general") -> str:
# 1. Budget check BEFORE the call — rough estimate (4 chars ≈ 1 token)
estimated_tokens = (len(user_message) + len(context)) // 4
self.budget.check_and_consume(estimated_tokens, user_id, agent_type)
# 2. Route to the cheapest model that can handle this task
model = self.router.route(user_message)
# 3. Build cached messages (stable content first, dynamic last)
system, messages = self.cache_mgr.build_cached_messages(
self.system_prompt, context, user_message
)
# 4. Make the call with system-level caching too
response = self.client.messages.create(
model=model,
max_tokens=1024,
system=[{"type": "text", "text": system,
"cache_control": {"type": "ephemeral"}}],
messages=messages,
)
# 5. Log the real cost breakdown
usage = response.usage
token_usage = TokenUsage(
input_tokens=usage.input_tokens,
output_tokens=usage.output_tokens,
cache_creation_tokens=getattr(usage, "cache_creation_input_tokens", 0),
cache_read_tokens=getattr(usage, "cache_read_input_tokens", 0),
)
self.cost_log.append({
"model": model,
"input_tokens": token_usage.input_tokens,
"output_tokens": token_usage.output_tokens,
"cache_read_tokens": token_usage.cache_read_tokens,
"cost_usd": token_usage.cost_usd(model, COST_PER_MTK),
})
return response.content[0].text
def cost_report(self) -> dict:
total_cost = sum(e["cost_usd"] for e in self.cost_log)
total_tokens = sum(e["input_tokens"] + e["output_tokens"] for e in self.cost_log)
cache_reads = sum(e["cache_read_tokens"] for e in self.cost_log)
return {
"calls": len(self.cost_log),
"total_cost_usd": round(total_cost, 4),
"total_tokens": total_tokens,
"cache_hit_rate": f"{100 * cache_reads / max(total_tokens, 1):.1f}%",
"avg_cost_per_call_usd": round(total_cost / max(len(self.cost_log), 1), 4),
}
The full runnable demo is in lesson_08_03_cost_optimization.py (under Code artifacts at the end of this lesson). Run it to see the cost report with cache hit rates:
python code/lesson_08_03_cost_optimization.py
Query 1: What is your return policy?
Answer: Our return policy allows...
Query 2: How do I track my order?
Answer: You can track your order...
Query 3: I was charged twice for my order. Can you help?
Answer: I'm sorry to hear...
=== Cost Report ===
calls: 3
total_cost_usd: 0.0012
total_tokens: 847
cache_hit_rate: 61.2%
avg_cost_per_call_usd: 0.0004
=== Budget Usage ===
session_used: 847
session_limit: 50000
session_pct: 1.7%