The Problem
A team ships a product on GPT-4 Turbo. Three months later OpenAI deprecates the model version, the bill triples after a pricing change, and the competitor released the same feature using Claude Haiku at 1/20th the cost. This isn't bad luck — it's an engineering failure: the team never built model selection as a system. They hardcoded a single model ID and called it done.
What you'll build
A unified LLM client that wraps multiple providers behind a single interface, plus a model router that picks the right tier for each task type. By the end of this lesson, you can swap models without changing application code.
The modern LLM landscape (2026)
The market has stratified into four tiers. Knowing which tier fits which job is the first skill of an AI engineer.
| Tier | Models | Cost (input/1M tokens) | Best for |
|---|
| Frontier | Claude Opus 5, GPT-5, Gemini 2.5 Pro | 2–15 USD | Complex reasoning, nuanced judgment |
| Workhorse | Claude Haiku 4.5, GPT-5 mini, Gemini 2.5 Flash | 0.10–0.30 USD | High-volume, structured tasks |
| Open-weight | Llama 3.3 70B, Mistral, DeepSeek V3 | ~0 (self-hosted) | Privacy-sensitive, cost-sensitive at scale |
| Edge/Small | Phi-4, Gemma 3 | ~0 | On-device, ultra-low latency |
The key insight: frontier ≠ correct choice. A workhorse model handles 90% of production tasks at 1/20th the cost. You should be able to justify your model choice across five dimensions.
Model names and prices move fast
Tier table last reviewed August 2026. Specific model names and per-token prices change every few months — always check the provider's current pricing page before you commit to a number. The tiering is the durable idea; the names in it are not.
The 5-dimension selection framework
Before picking a model, answer these questions:
- Quality threshold: what's the minimum acceptable accuracy?
- Latency budget: how fast must the response appear? (p95)
- Cost ceiling: what's the max cost per request, or per month?
- Data privacy: can this data leave your infrastructure? (PHI, PII, IP)
- Deployment context: cloud API, self-hosted, or on-device?
The following diagram shows how a request flows through the decision framework before landing on a model tier:
A decision matrix helps when trade-offs conflict:
| Constraint violated | Action |
|---|
| Cost ceiling exceeded | Downgrade tier or add caching |
| Latency too high | Downgrade tier or add streaming |
| Privacy requirement | Self-host open-weight model |
| Quality insufficient | Upgrade tier or add CoT prompt |
You're building a medical note summarization tool that processes 50,000 notes/day. PHI data cannot leave your servers. Which model strategy is correct?
PHI (Protected Health Information) under HIPAA cannot be sent to third-party API providers without a BAA. Self-hosting an open-weight model like Llama 3.3 70B means data never leaves your infrastructure. This is a privacy constraint that overrides all quality and cost considerations.
Calling the APIs — unified pattern
All three major providers follow the same pattern: initialize a client, build a messages array, specify a model ID, and call the completions endpoint. The syntax differs slightly, but the mental model is identical.
OpenAI:
from openai import OpenAI
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Summarize this in one sentence."}],
max_tokens=100,
temperature=0.3,
)
text = response.choices[0].message.content
Anthropic:
import anthropic
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
response = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=100,
messages=[{"role": "user", "content": "Summarize this in one sentence."}],
)
text = response.content[0].text
The abstraction insight: wrap both behind a shared interface and your application code never needs to know which provider it's talking to.
Streaming for long responses
For responses over ~200 tokens, streaming (stream=True in OpenAI, streaming mode in Anthropic) returns tokens as they generate instead of waiting for the full response. This makes your app feel much faster even if total latency is the same.
Error Handling You Must Implement
| Error type | HTTP code | Action |
|---|
| Rate limit | 429 | Exponential backoff + retry (max 3x) |
| Context too long | 400/413 | Truncate input and retry |
| Server error | 500/503 | Retry once, then surface to user |
| Auth failure | 401 | Fail immediately — don't retry |
Your agent handles 95% simple FAQ queries and 5% complex multi-step reasoning. Current cost is $8,000/month on GPT-4o. What's the right fix?
Tiered routing targets the 95% majority with a cheap model (~$0.15/1M tokens vs $2.50/1M) while preserving quality on the 5% that needs it. This typically cuts costs 80–90% with no quality regression on hard cases. Caching helps but doesn't address the root cost driver.
Build
Build it
Before wiring up real API clients, it helps to see the routing and cost-tracking logic in isolation. The mini example below simulates two provider tiers with hardcoded pricing and shows how routing decisions translate directly into cost savings.
# Simulate model routing + cost tracking without any real API calls.
PRICING = {
"workhorse": {"input": 0.15, "output": 0.60}, # $/1M tokens
"frontier": {"input": 2.50, "output": 10.00},
}
def classify(query):
"""Toy classifier: short factual questions are SIMPLE, else COMPLEX."""
complex_markers = ("trade-off", "design", "compare", "explain why")
return "COMPLEX" if any(m in query.lower() for m in complex_markers) else "SIMPLE"
def route(query):
task_type = classify(query)
tier = "frontier" if task_type == "COMPLEX" else "workhorse"
return task_type, tier
def cost(tier, input_tokens, output_tokens):
rates = PRICING[tier]
return (input_tokens * rates["input"] + output_tokens * rates["output"]) / 1_000_000
queries = [
"What is the capital of France?",
"Explain why HNSW beats brute-force search at scale.",
"What does HTTP 429 mean?",
"Compare self-hosting vs. managed APIs for a 10k req/s workload.",
]
total_routed = 0.0
total_frontier_only = 0.0
for q in queries:
task_type, tier = route(q)
c = cost(tier, input_tokens=50, output_tokens=150)
total_routed += c
total_frontier_only += cost("frontier", input_tokens=50, output_tokens=150)
print(f"[{task_type:<7}] -> {tier:<10} ${c:.6f} | {q[:45]}")
savings_pct = (1 - total_routed / total_frontier_only) * 100
print(f"\nTotal routed cost: ${total_routed:.6f}")
print(f"Frontier-only cost: ${total_frontier_only:.6f}")
print(f"Savings from routing: {savings_pct:.1f}%")
The code artifact for this lesson is a unified LLM client with model routing, cost tracking, and retry logic. Key components:
TaskType enum: SIMPLE, COMPLEX, VISION, PRIVATEUnifiedLLMClient: wraps OpenAI and Anthropic behind identical .complete() methodModelRouter: maps task types to appropriate model tiersCostTracker: logs tokens used and estimated cost per call- Retry decorator with exponential backoff
"""
lesson_00_01_llm_apis.py
Unified LLM client with model routing, cost tracking, and retry logic.
Demonstrates the 5-dimension selection framework in code.
Usage:
export OPENAI_API_KEY=sk-...
export ANTHROPIC_API_KEY=sk-ant-...
python lesson_00_01_llm_apis.py
"""
import os
import time
import logging
from enum import Enum
from dataclasses import dataclass, field
from typing import Optional
import openai
import anthropic
logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
logger = logging.getLogger(__name__)
# ─── Task classification ──────────────────────────────────────────────────────
class TaskType(Enum):
SIMPLE = "simple" # FAQ, short extraction, classification
COMPLEX = "complex" # Multi-step reasoning, analysis, code generation
VISION = "vision" # Image understanding
PRIVATE = "private" # PHI/PII — must stay on-prem (not implemented here)
# ─── Cost tracking ────────────────────────────────────────────────────────────
# Approximate 2026 pricing ($/1M tokens)
MODEL_PRICING: dict[str, dict[str, float]] = {
"gpt-4o": {"input": 2.50, "output": 10.00},
"gpt-4o-mini": {"input": 0.15, "output": 0.60},
"claude-sonnet-5": {"input": 3.00, "output": 15.00},
"claude-haiku-4-5-20251001": {"input": 0.25, "output": 1.25},
}
@dataclass
class CallRecord:
model: str
task_type: TaskType
input_tokens: int
output_tokens: int
latency_ms: float
cost_usd: float
@dataclass
class CostTracker:
records: list[CallRecord] = field(default_factory=list)
def record(self, model: str, task_type: TaskType,
input_tokens: int, output_tokens: int, latency_ms: float) -> CallRecord:
pricing = MODEL_PRICING.get(model, {"input": 0.0, "output": 0.0})
cost = (input_tokens * pricing["input"] + output_tokens * pricing["output"]) / 1_000_000
rec = CallRecord(model, task_type, input_tokens, output_tokens, latency_ms, cost)
self.records.append(rec)
return rec
def summary(self) -> dict:
total_cost = sum(r.cost_usd for r in self.records)
total_calls = len(self.records)
by_model: dict[str, float] = {}
for r in self.records:
by_model[r.model] = by_model.get(r.model, 0.0) + r.cost_usd
return {"total_cost_usd": total_cost, "total_calls": total_calls, "cost_by_model": by_model}
# ─── Retry logic ──────────────────────────────────────────────────────────────
def with_retry(fn, max_retries: int = 3, base_delay: float = 1.0):
"""Exponential backoff retry for rate-limit and server errors."""
for attempt in range(max_retries):
try:
return fn()
except (openai.RateLimitError, anthropic.RateLimitError):
if attempt == max_retries - 1:
raise
delay = base_delay * (2 ** attempt)
logger.warning(f"Rate limited. Retrying in {delay:.1f}s (attempt {attempt + 1}/{max_retries})")
time.sleep(delay)
except (openai.APIStatusError, anthropic.APIStatusError) as e:
if hasattr(e, "status_code") and e.status_code == 401:
raise # Auth errors: don't retry
if attempt == max_retries - 1:
raise
time.sleep(base_delay)
return None # unreachable
# ─── Unified client ───────────────────────────────────────────────────────────
class UnifiedLLMClient:
"""Single interface over OpenAI and Anthropic APIs."""
def __init__(self):
self._openai = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
self._anthropic = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
self.tracker = CostTracker()
def complete(
self,
prompt: str,
model: str,
task_type: TaskType = TaskType.SIMPLE,
max_tokens: int = 512,
temperature: float = 0.3,
system: Optional[str] = None,
) -> str:
start = time.perf_counter()
if model.startswith("gpt"):
result = self._call_openai(prompt, model, max_tokens, temperature, system)
elif model.startswith("claude"):
result = self._call_anthropic(prompt, model, max_tokens, temperature, system)
else:
raise ValueError(f"Unknown model provider for model: {model}")
latency_ms = (time.perf_counter() - start) * 1000
rec = self.tracker.record(
model, task_type,
result["input_tokens"], result["output_tokens"], latency_ms
)
logger.info(
f"[{model}] {result['input_tokens']}+{result['output_tokens']} tokens | "
f"${rec.cost_usd:.6f} | {latency_ms:.0f}ms"
)
return result["text"]
def _call_openai(self, prompt: str, model: str, max_tokens: int,
temperature: float, system: Optional[str]) -> dict:
messages = []
if system:
messages.append({"role": "system", "content": system})
messages.append({"role": "user", "content": prompt})
def _call():
return self._openai.chat.completions.create(
model=model, messages=messages,
max_tokens=max_tokens, temperature=temperature,
)
resp = with_retry(_call)
return {
"text": resp.choices[0].message.content,
"input_tokens": resp.usage.prompt_tokens,
"output_tokens": resp.usage.completion_tokens,
}
def _call_anthropic(self, prompt: str, model: str, max_tokens: int,
temperature: float, system: Optional[str]) -> dict:
kwargs = dict(
model=model,
max_tokens=max_tokens,
temperature=temperature,
messages=[{"role": "user", "content": prompt}],
)
if system:
kwargs["system"] = system
def _call():
return self._anthropic.messages.create(**kwargs)
resp = with_retry(_call)
return {
"text": resp.content[0].text,
"input_tokens": resp.usage.input_tokens,
"output_tokens": resp.usage.output_tokens,
}
# ─── Model router ─────────────────────────────────────────────────────────────
ROUTING_TABLE: dict[TaskType, str] = {
TaskType.SIMPLE: "gpt-4o-mini",
TaskType.COMPLEX: "gpt-4o",
TaskType.VISION: "gpt-4o", # vision-capable frontier model
TaskType.PRIVATE: "claude-haiku-4-5-20251001", # placeholder; real use = self-hosted
}
class ModelRouter:
"""Routes tasks to the appropriate model based on task type."""
def __init__(self, client: UnifiedLLMClient):
self.client = client
def classify_and_route(self, prompt: str) -> tuple[TaskType, str]:
"""Classify a query and return (task_type, model_id)."""
classification_prompt = (
"Classify the following query as one of: SIMPLE, COMPLEX\n"
"SIMPLE: FAQ, short extraction, yes/no, single-fact lookup\n"
"COMPLEX: multi-step reasoning, analysis, code generation, comparison\n"
f"Query: {prompt}\n"
"Respond with ONLY the label: SIMPLE or COMPLEX"
)
label = self.client.complete(
classification_prompt,
model="gpt-4o-mini",
task_type=TaskType.SIMPLE,
max_tokens=10,
temperature=0.0,
).strip().upper()
task_type = TaskType.COMPLEX if "COMPLEX" in label else TaskType.SIMPLE
model = ROUTING_TABLE[task_type]
return task_type, model
def run(self, prompt: str, system: Optional[str] = None) -> str:
task_type, model = self.classify_and_route(prompt)
logger.info(f"Routing '{prompt[:60]}...' → {task_type.value} → {model}")
return self.client.complete(prompt, model=model, task_type=task_type, system=system)
# ─── Demo ─────────────────────────────────────────────────────────────────────
def main():
client = UnifiedLLMClient()
router = ModelRouter(client)
test_queries = [
"What is the capital of France?",
"Explain the trade-offs between HNSW and IVF indexing for a 10M-document vector database.",
"What does HTTP 429 mean?",
"Design a rate-limiting strategy for a multi-tenant LLM API that handles 10k req/s at peak.",
]
print("\n" + "="*60)
print("MODEL ROUTER DEMO")
print("="*60)
results = []
for query in test_queries:
task_type, model = router.classify_and_route(query)
answer = client.complete(query, model=model, task_type=task_type, max_tokens=150)
results.append({
"query": query[:50] + "..." if len(query) > 50 else query,
"task_type": task_type.value,
"model": model,
"answer_preview": answer[:80] + "...",
})
print(f"\n{'Query':<55} {'Type':<10} {'Model':<20}")
print("-" * 90)
for r in results:
print(f"{r['query']:<55} {r['task_type']:<10} {r['model']:<20}")
summary = client.tracker.summary()
print(f"\n{'='*60}")
print(f"Total calls: {summary['total_calls']}")
print(f"Total cost: ${summary['total_cost_usd']:.6f}")
print("Cost by model:")
for model, cost in summary["cost_by_model"].items():
print(f" {model}: ${cost:.6f}")
if __name__ == "__main__":
main()