The Concept
The Cost Differential
# Approximate pricing per 1M tokens (2026)
MODEL_PRICING = {
# Input / Output per 1M tokens
"gpt-4o": {"input": 2.50, "output": 10.00},
"gpt-4o-mini": {"input": 0.15, "output": 0.60},
"claude-sonnet": {"input": 3.00, "output": 15.00},
"claude-haiku": {"input": 0.25, "output": 1.25},
"llama-3.1-8b": {"input": 0.05, "output": 0.05}, # via Groq/Together
}
# For a typical query (500 input tokens, 200 output tokens):
# gpt-4o: $0.00125 + $0.002 = $0.00325 per query
# gpt-4o-mini: $0.000075 + $0.00012 = $0.000195 per query
# llama-8b: $0.000025 + $0.00001 = $0.000035 per query
# At 10K queries/day:
# All gpt-4o: $32.50/day = $975/month
# All gpt-4o-mini: $1.95/day = $58/month
# All llama-8b: $0.35/day = $10/month
# Smart routing: ~$5/day = $150/month (estimated)
The math is clear: routing 70% of queries to a cheap model and 30% to an expensive one cuts costs by ~80% while maintaining quality on the queries that need it.
Three Routing Strategies
You run a customer support chatbot. 70% of queries are simple (hours, returns, tracking). 30% require complex reasoning (billing disputes, product troubleshooting). You route everything to gpt-4o at $0.00325/query. If you switch to rule-based routing (simple → gpt-4o-mini at $0.000195, complex → gpt-4o at $0.00325), what are your daily savings at 10K queries?
2
Strategy 1: Rule-Based Routing
The simplest approach — classify queries by surface features:
import re
class RuleBasedRouter:
def __init__(self):
self.simple_patterns = [
r"(?i)what are your (hours|business hours)",
r"(?i)how do i (reset|change) (my )?(password|email)",
r"(?i)where is my (order|package|shipment)",
r"(?i)(cancel|return|refund) (my )?(order|item)",
r"(?i)(contact|phone|email|address)",
]
self.complex_indicators = [
r"```", # Code blocks
r"(?i)(explain|analyze|compare|evaluate)",
r"(?i)(error|bug|broken|not working).*(detail|log|trace)",
r"(?i)(calculate|compute|derive|prove)",
]
def route(self, query: str) -> str:
# Check for complex indicators first
for pattern in self.complex_indicators:
if re.search(pattern, query):
return "gpt-4o"
# Check for simple patterns
for pattern in self.simple_patterns:
if re.search(pattern, query):
return "gpt-4o-mini"
# Default: use query length as a heuristic
if len(query.split()) < 20:
return "gpt-4o-mini"
return "gpt-4o" # Safe default: expensive but correct
Pros: Fast, deterministic, no extra model calls, easy to debug. Cons: Brittle — misses edge cases, requires manual rule maintenance, can't capture semantic complexity.
Strategy 2: Cascade Routing (Try Cheap First, Escalate if Needed)
class CascadeRouter:
def __init__(self, cheap_model, expensive_model, confidence_threshold=0.7):
self.cheap_model = cheap_model
self.expensive_model = expensive_model
self.confidence_threshold = confidence_threshold
def route(self, query: str, context: str = "") -> dict:
# Step 1: Try the cheap model
cheap_response = self.cheap_model.generate(
query=query,
context=context,
# Ask the model to rate its own confidence
extra_instruction="After your answer, on a new line, write 'CONFIDENCE: X' where X is 0-1."
)
confidence = self._extract_confidence(cheap_response)
if confidence >= self.confidence_threshold:
return {
"answer": self._strip_confidence(cheap_response),
"model": self.cheap_model.name,
"cost": self.cheap_model.last_cost,
"escalated": False
}
# Step 2: Escalate to expensive model
expensive_response = self.expensive_model.generate(
query=query,
context=context
)
return {
"answer": expensive_response,
"model": self.expensive_model.name,
"cost": self.cheap_model.last_cost + self.expensive_model.last_cost,
"escalated": True
}
def _extract_confidence(self, response: str) -> float:
import re
match = re.search(r'CONFIDENCE:\s*([0-9.]+)', response)
return float(match.group(1)) if match else 0.5
Pros: Never sacrifices quality — if the cheap model is unsure, the expensive model handles it. Self-correcting. Cons: You pay for the cheap model even on escalated queries. The confidence score is a self-assessment, not always reliable.
Strategy 3: Semantic Caching (Eliminate Redundant Calls Entirely)
Before routing to any model, check if you've already answered a similar question:
import hashlib
import numpy as np
class SemanticCache:
def __init__(self, embedder, redis_client, similarity_threshold=0.95):
self.embedder = embedder
self.redis = redis_client
self.threshold = similarity_threshold
def get(self, query: str) -> Optional[dict]:
"""Check if a similar query has been answered before."""
query_embedding = self.embedder.embed(query)
# Search Redis for similar cached queries
# In production: use Redis with vector similarity search (RediSearch)
cached = self.redis.search_vectors(
query_embedding,
top_k=1,
min_similarity=self.threshold
)
if cached:
return {
"answer": cached[0]["answer"],
"model": "cache",
"cost": 0.0,
"cache_hit": True
}
return None
def put(self, query: str, answer: str, model: str, cost: float):
"""Cache a query-answer pair."""
embedding = self.embedder.embed(query)
self.redis.store_vector(
vector=embedding,
metadata={
"query": query,
"answer": answer,
"model": model,
"cost": cost,
"timestamp": time.time()
}
)
Your support chatbot gets 10K queries/day. You add a semantic cache with 0.95 similarity threshold. After a week, you analyze the cache: 30% of queries are cache hits. What's the daily savings if your average pre-cache cost is $11.12/day?
1
Combining All Three: The Production Pattern
class CostAwareLLMService:
def __init__(self, cache, router, cheap_model, expensive_model):
self.cache = cache
self.router = router
self.cheap_model = cheap_model
self.expensive_model = expensive_model
def answer(self, query: str, user_context: str = "") -> dict:
# Layer 1: Semantic cache — free if hit
cached = self.cache.get(query)
if cached:
return cached
# Layer 2: Rule-based routing — pick the right model
model_name = self.router.route(query)
model = self.cheap_model if model_name == "cheap" else self.expensive_model
# Layer 3: Generate with confidence check (cascade)
response = model.generate(query=query, context=user_context)
confidence = self._assess_confidence(query, response)
# Layer 3b: Escalate if cheap model is unconfident
if model == self.cheap_model and confidence < 0.7:
response = self.expensive_model.generate(query=query, context=user_context)
model = self.expensive_model
result = {
"answer": response,
"model": model.name,
"cost": model.last_cost,
"cache_hit": False
}
# Cache the result for future queries
self.cache.put(query, response, model.name, model.last_cost)
return result
def _assess_confidence(self, query: str, response: str) -> float:
"""Heuristic confidence assessment."""
# Low confidence signals
if "I'm not sure" in response or "I don't know" in response:
return 0.2
if len(response) < 50: # Very short answers may be incomplete
return 0.5
if "however" in response.lower() or "but" in response.lower():
return 0.6 # Hedging language
return 0.85 # Default: reasonably confident
Measuring Success: Cost-Per-Quality
The goal isn't just to reduce cost — it's to reduce cost without degrading quality. Track these metrics:
@dataclass
class RoutingMetrics:
total_queries: int
cache_hits: int
cheap_model_calls: int
expensive_model_calls: int
escalations: int # cheap → expensive
total_cost: float
avg_latency_ms: float
user_satisfaction: float # thumbs up/down or CSAT
@property
def cache_hit_rate(self) -> float:
return self.cache_hits / self.total_queries
@property
def cost_per_query(self) -> float:
return self.total_cost / self.total_queries
@property
def escalation_rate(self) -> float:
return self.escalations / self.cheap_model_calls if self.cheap_model_calls else 0
def summary(self) -> str:
return f"""
Queries: {self.total_queries}
Cache hit rate: {self.cache_hit_rate:.1%}
Cheap model: {self.cheap_model_calls} ({self.cheap_model_calls/self.total_queries:.1%})
Expensive model: {self.expensive_model_calls} ({self.expensive_model_calls/self.total_queries:.1%})
Escalation rate: {self.escalation_rate:.1%}
Cost per query: ${self.cost_per_query:.4f}
Avg latency: {self.avg_latency_ms:.0f}ms
User satisfaction: {self.user_satisfaction:.1%}
"""