You deployed a proof-of-concept LLM feature last quarter. It worked beautifully in testing. Then it hit production, and your finance team started asking questions about a line item in your cloud bill that looks like a typo. You're not alone. Teams that build with LLM APIs routinely see costs balloon 5–20x between prototype and production, not because they made bad technical decisions, but because nobody sat down to engineer the cost architecture the same way they engineered the feature itself.
The core insight here is deceptively simple: LLM APIs charge you for tokens, and tokens are everywhere — in your system prompts, your retrieved context chunks, your function-calling scaffolding, and your model's verbose completions. When you multiply that token count by the number of requests per day, per user, and per workflow, small inefficiencies compound into serious budget problems. But this isn't purely a cost-cutting exercise. Done correctly, cost optimization also improves latency, reduces hallucination surface area, and forces you to be clearer about what you actually need from the model.
By the end of this lesson, you'll be able to look at any LLM-powered production system and identify where money is leaking, implement concrete fixes at the token, model, and caching levels, and build monitoring instrumentation so you catch regressions before the next billing cycle.
What you'll learn:
You should be comfortable calling LLM APIs programmatically (OpenAI, Anthropic, or equivalent) and have at least one production or near-production system you're thinking about. Familiarity with Python is assumed for the code examples. You don't need deep ML knowledge — this is about systems engineering, not model internals.
Before you can optimize, you need a mental model of what the pricing structure actually means in practice.
Every major LLM provider (OpenAI, Anthropic, Google, Mistral) charges by the token, typically split into input tokens and output tokens. Output tokens usually cost 3–5x more than input tokens because generation is computationally more expensive than processing. As of mid-2025, a rough but useful benchmark: flagship models like GPT-4o or Claude 3.5 Sonnet run somewhere in the $2–15 per million token range for inputs and $6–60 per million for outputs. Smaller models in the same families cost 10–20x less.
A "token" is roughly 0.75 words in English, or about 4 characters. That sounds small, but consider a customer support system that processes 10,000 tickets per day. If each request involves a 1,000-token system prompt, 500 tokens of retrieved FAQ context, 200 tokens of user message, and 300 tokens of response, you're looking at 20 million tokens daily — just for a mid-sized support tool. At $5 per million input tokens, that's $75,000 per month before you've added a single fancy feature.
Here's a quick way to estimate your current spend accurately:
import tiktoken
from dataclasses import dataclass
from typing import Optional
@dataclass
class CostEstimate:
input_tokens: int
output_tokens: int
input_cost_usd: float
output_cost_usd: float
total_cost_usd: float
def estimate_cost(
messages: list[dict],
completion_text: str,
model: str = "gpt-4o",
input_price_per_million: float = 2.50,
output_price_per_million: float = 10.00,
) -> CostEstimate:
"""
Estimate the cost of a single API call.
Prices default to gpt-4o as of mid-2025 — update as needed.
"""
enc = tiktoken.encoding_for_model(model)
# Count input tokens across all messages
input_tokens = 0
for msg in messages:
# Each message has a small overhead for role/formatting
input_tokens += 4
input_tokens += len(enc.encode(msg.get("content", "")))
input_tokens += 2 # reply priming
output_tokens = len(enc.encode(completion_text))
input_cost = (input_tokens / 1_000_000) * input_price_per_million
output_cost = (output_tokens / 1_000_000) * output_price_per_million
return CostEstimate(
input_tokens=input_tokens,
output_tokens=output_tokens,
input_cost_usd=input_cost,
output_cost_usd=output_cost,
total_cost_usd=input_cost + output_cost,
)
# Example: a customer support interaction
messages = [
{"role": "system", "content": "You are a helpful support agent for AcmeCorp..."},
{"role": "user", "content": "Why was I charged twice for my subscription last month?"},
]
completion = "I understand your concern about the duplicate charge. Let me look into this for you..."
estimate = estimate_cost(messages, completion)
print(f"Input tokens: {estimate.input_tokens}")
print(f"Output tokens: {estimate.output_tokens}")
print(f"Total cost: ${estimate.total_cost_usd:.6f}")
print(f"Daily cost at 10k requests: ${estimate.total_cost_usd * 10_000:.2f}")
Run this against your actual prompts before you do anything else. The numbers often surprise people.
Tip: Always use
tiktoken(for OpenAI models) or the equivalent tokenizer for your provider to count tokens programmatically before requests, not just to estimate retroactively. Building this into a middleware layer gives you precise data instead of guesses.
Token optimization has three distinct targets: your system prompt, your context window, and your completion. Each requires a different approach.
System prompts are the easiest place to find waste because they're static and you send them with every single request. Even a 200-token reduction in a system prompt translates to 200 fewer input tokens times your entire daily request volume.
The most common problem is verbose, conversational system prompts written when the feature was new and nobody was thinking about scale. Here's a real before/after from a document summarization tool:
Before (~380 tokens):
You are an expert document summarization assistant. Your job is to help users
understand complex documents by creating clear, accurate summaries. When you
receive a document, please read it carefully and thoroughly. Then, create a
summary that captures all of the key points. Make sure to include the main
arguments, any important data or statistics, key conclusions, and
recommendations if there are any. Your summary should be well-organized and
easy to read. Please write in a professional tone. Format the summary with
a brief overview paragraph first, then bullet points for the key details.
Do not include your own opinions or analysis — just summarize what is in
the document. If the document is technical, make sure to explain any jargon
in plain language. Always double-check that your summary is accurate and
does not misrepresent the information in the original document.
After (~85 tokens):
Summarize the document below. Format: one overview paragraph, then bullet
points for key details. Tone: professional, neutral. Explain jargon in
plain language. Do not add opinions or analysis.
The second version performs comparably on most documents because LLMs are not humans — they don't benefit from motivational framing or reminders that they should "read carefully." You're spending tokens on instructions the model implicitly follows anyway.
The compression process:
If you're running a RAG system, your retrieved chunks are often your largest cost driver. A naive implementation fetches 5–10 chunks of 500 tokens each and stuffs them all in the context whether they're relevant or not.
The right approach is a two-stage retrieval strategy: retrieve more than you need, then rerank and truncate before sending to the expensive model.
from typing import List, Tuple
import numpy as np
def rerank_and_truncate_chunks(
query: str,
chunks: List[dict], # Each: {"text": str, "score": float, "source": str}
max_context_tokens: int = 1500,
min_relevance_score: float = 0.72,
) -> List[dict]:
"""
Filter retrieved chunks by relevance score and hard token budget.
Call this before building your prompt, not after.
"""
enc = tiktoken.get_encoding("cl100k_base")
# Filter by minimum relevance
relevant = [c for c in chunks if c["score"] >= min_relevance_score]
# Sort by relevance descending
relevant.sort(key=lambda x: x["score"], reverse=True)
selected = []
token_count = 0
for chunk in relevant:
chunk_tokens = len(enc.encode(chunk["text"]))
if token_count + chunk_tokens <= max_context_tokens:
selected.append(chunk)
token_count += chunk_tokens
else:
# If this chunk would exceed budget, try to fit a truncated version
remaining = max_context_tokens - token_count
if remaining > 100: # Only worth including if we have meaningful space
tokens = enc.encode(chunk["text"])[:remaining]
truncated_text = enc.decode(tokens)
selected.append({**chunk, "text": truncated_text, "truncated": True})
break
return selected
def format_context_block(chunks: List[dict]) -> str:
"""Build a compact context string from selected chunks."""
parts = []
for i, chunk in enumerate(chunks, 1):
source = chunk.get("source", f"Document {i}")
truncation_note = " [truncated]" if chunk.get("truncated") else ""
parts.append(f"[{source}{truncation_note}]\n{chunk['text']}")
return "\n\n".join(parts)
Another often-overlooked technique: chunk compression before storage. If you're storing raw document text, consider running a one-time summarization pass on each chunk during ingestion to reduce average chunk size by 40–60%. You pay the compression cost once; you save on every query thereafter.
Output tokens cost more and are harder to predict. Two concrete controls:
1. Use max_tokens to set a hard ceiling. This is obvious but often skipped in production because it feels risky. The right approach is to set it based on your actual output distribution, not a conservative overestimate.
import json
from collections import Counter
def analyze_output_distribution(response_log_path: str) -> dict:
"""
Read a JSONL log of past completions and find the 95th percentile
output length so you can set max_tokens intelligently.
"""
lengths = []
with open(response_log_path, "r") as f:
for line in f:
record = json.loads(line)
lengths.append(record["output_tokens"])
lengths.sort()
n = len(lengths)
return {
"p50": lengths[int(n * 0.50)],
"p90": lengths[int(n * 0.90)],
"p95": lengths[int(n * 0.95)],
"p99": lengths[int(n * 0.99)],
"max_observed": lengths[-1],
"recommended_max_tokens": int(lengths[int(n * 0.95)] * 1.15), # 15% buffer
}
2. Instruct the model to be concise explicitly. "Respond in 2–3 sentences" or "answer in under 100 words" consistently reduces output length without the abrupt cutoffs that max_tokens can cause if set too aggressively.
Warning: Never set
max_tokenswithout analyzing your actual output distribution first. Setting it too low will silently truncate responses, leading to degraded quality that's hard to debug. Set it to the 95th percentile of observed output length plus a safety margin.
One of the highest-leverage cost strategies is simply not using your most expensive model for every request. This isn't about cutting corners — it's about recognizing that most of your request volume consists of tasks that don't require frontier model capability.
A model router evaluates each incoming request and assigns it to a model tier based on estimated complexity. Here's a practical implementation:
from enum import Enum
from dataclasses import dataclass
import re
class ModelTier(Enum):
NANO = "gpt-4o-mini" # ~$0.15/M input — classification, simple Q&A
STANDARD = "gpt-4o" # ~$2.50/M input — reasoning, summarization
FRONTIER = "o3" # ~$10+/M input — complex reasoning, code gen
@dataclass
class RoutingDecision:
tier: ModelTier
reasoning: str
estimated_input_tokens: int
def route_request(
user_message: str,
system_context: str,
has_code: bool = False,
requires_json_output: bool = False,
) -> RoutingDecision:
"""
Rule-based router. In production, you might replace or augment
the rules with a small classifier model for higher accuracy.
"""
enc = tiktoken.get_encoding("cl100k_base")
input_tokens = len(enc.encode(user_message + system_context))
message_lower = user_message.lower()
# Signals that require frontier model
frontier_signals = [
has_code and len(user_message) > 500,
"debug" in message_lower and has_code,
"architect" in message_lower,
"multi-step" in message_lower,
input_tokens > 8000, # Large context needs best comprehension
]
# Signals that work fine with nano/mini
nano_signals = [
re.match(r'^(yes|no|what is|define|how do i|when did)', message_lower),
len(user_message) < 100 and not has_code,
"classify" in message_lower,
"categorize" in message_lower,
"translate" in message_lower and len(user_message) < 300,
]
if any(frontier_signals):
return RoutingDecision(
tier=ModelTier.FRONTIER,
reasoning="Complex reasoning or large context detected",
estimated_input_tokens=input_tokens,
)
elif any(nano_signals):
return RoutingDecision(
tier=ModelTier.NANO,
reasoning="Simple, short-form request suitable for mini model",
estimated_input_tokens=input_tokens,
)
else:
return RoutingDecision(
tier=ModelTier.STANDARD,
reasoning="Moderate complexity, standard model appropriate",
estimated_input_tokens=input_tokens,
)
A rule-based router like this is a starting point. In production, you'll want to validate it against your actual task distribution. A common pattern is to run requests through both your routed model and the flagship model on a 5% sample, compare outputs using an LLM-as-judge approach, and use that data to tune your routing thresholds.
For tasks where quality is non-negotiable but most requests are simple, use a waterfall: try a cheap model first, evaluate the output, and escalate to a more capable model only if the output fails validation.
import openai
from typing import Optional
def waterfall_completion(
messages: list[dict],
validator_fn, # callable: (str) -> bool
max_escalations: int = 1,
) -> tuple[str, str]:
"""
Try gpt-4o-mini first. If the output fails validation,
escalate to gpt-4o. Returns (completion_text, model_used).
"""
client = openai.OpenAI()
model_sequence = [
("gpt-4o-mini", {"max_tokens": 500}),
("gpt-4o", {"max_tokens": 800}),
]
for i, (model, kwargs) in enumerate(model_sequence[:max_escalations + 1]):
response = client.chat.completions.create(
model=model,
messages=messages,
**kwargs,
)
completion = response.choices[0].message.content
if validator_fn(completion):
return completion, model
if i < len(model_sequence) - 1:
print(f"Output from {model} failed validation, escalating...")
# Return last attempt even if it failed validation
return completion, model
# Example validator for a structured data extraction task
def is_valid_json_with_required_fields(text: str) -> bool:
try:
data = json.loads(text)
return all(k in data for k in ["customer_id", "issue_type", "priority"])
except json.JSONDecodeError:
return False
The waterfall pattern works especially well for structured output tasks, classification, and extraction where correctness is binary and verifiable. It's less useful for open-ended generation where "valid" is subjective.
Tip: Track your escalation rate. If more than 15–20% of requests escalate to the expensive model, your validator is too strict or your mini-model genuinely can't handle the task. If less than 2% escalate, you may not need the waterfall at all — just use the cheap model.
Caching is the highest-ROI optimization available to most production systems because it eliminates API calls entirely. There are two distinct caching strategies and they solve different problems.
For deterministic use cases — report generation, data extraction from fixed templates, FAQ responses — many requests are genuinely identical or near-identical. An exact-match cache keyed on a hash of the prompt will serve these responses instantly for zero API cost.
import hashlib
import json
import time
import redis
from typing import Optional
class LLMCache:
def __init__(self, redis_url: str = "redis://localhost:6379", ttl_seconds: int = 86400):
self.redis = redis.from_url(redis_url)
self.ttl = ttl_seconds
def _cache_key(self, messages: list[dict], model: str) -> str:
"""
Generate a deterministic cache key from the full prompt.
Include model because the same prompt on different models
produces different outputs.
"""
payload = json.dumps({"messages": messages, "model": model}, sort_keys=True)
return f"llm:exact:{hashlib.sha256(payload.encode()).hexdigest()}"
def get(self, messages: list[dict], model: str) -> Optional[str]:
key = self._cache_key(messages, model)
cached = self.redis.get(key)
if cached:
self.redis.incr(f"llm:cache:hits")
return cached.decode()
self.redis.incr(f"llm:cache:misses")
return None
def set(self, messages: list[dict], model: str, response: str) -> None:
key = self._cache_key(messages, model)
self.redis.setex(key, self.ttl, response)
def cache_stats(self) -> dict:
hits = int(self.redis.get("llm:cache:hits") or 0)
misses = int(self.redis.get("llm:cache:misses") or 0)
total = hits + misses
return {
"hits": hits,
"misses": misses,
"hit_rate": hits / total if total > 0 else 0,
"estimated_savings_pct": hits / total * 100 if total > 0 else 0,
}
The catch with exact-match caching: even minor variations in user input create cache misses. A user typing "What's your return policy?" versus "what is your return policy?" are different strings. This is where semantic caching comes in.
Semantic caching uses embedding similarity to match new queries against previously answered questions. If the semantic distance is below a threshold, serve the cached response.
import numpy as np
from openai import OpenAI
from dataclasses import dataclass
@dataclass
class CachedEntry:
query: str
embedding: np.ndarray
response: str
timestamp: float
hit_count: int = 0
class SemanticCache:
def __init__(
self,
similarity_threshold: float = 0.95,
max_entries: int = 10_000,
embedding_model: str = "text-embedding-3-small",
):
self.threshold = similarity_threshold
self.max_entries = max_entries
self.embedding_model = embedding_model
self.entries: list[CachedEntry] = []
self.client = OpenAI()
def _get_embedding(self, text: str) -> np.ndarray:
response = self.client.embeddings.create(
input=text,
model=self.embedding_model,
)
return np.array(response.data[0].embedding)
def _cosine_similarity(self, a: np.ndarray, b: np.ndarray) -> float:
return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
def lookup(self, query: str) -> Optional[tuple[str, float]]:
"""
Returns (cached_response, similarity_score) if found, else None.
"""
if not self.entries:
return None
query_embedding = self._get_embedding(query)
best_score = -1
best_entry = None
for entry in self.entries:
score = self._cosine_similarity(query_embedding, entry.embedding)
if score > best_score:
best_score = score
best_entry = entry
if best_score >= self.threshold:
best_entry.hit_count += 1
return best_entry.response, best_score
return None
def store(self, query: str, response: str) -> None:
embedding = self._get_embedding(query)
entry = CachedEntry(
query=query,
embedding=embedding,
response=response,
timestamp=time.time(),
)
self.entries.append(entry)
# Evict least-used entries if over capacity
if len(self.entries) > self.max_entries:
self.entries.sort(key=lambda e: (e.hit_count, e.timestamp))
self.entries = self.entries[len(self.entries) // 4:] # Remove bottom 25%
Warning: The similarity threshold is the most critical parameter in semantic caching. Set it too low (e.g., 0.85) and you'll serve wrong answers to semantically adjacent but meaningfully different questions. Start at 0.95 and lower it only after reviewing cache hit samples manually. In high-stakes domains like healthcare, legal, or financial queries, consider not using semantic caching at all, or adding a human-in-the-loop review step for cache population.
Major providers now offer native prompt caching that happens at the infrastructure level. OpenAI automatically caches prompts over 1,024 tokens at a 50% discount on cached input tokens. Anthropic's prompt caching is explicit — you mark cache breakpoints in your prompt and pay a small upfront cost per cache write, then get 90% discounts on subsequent cache hits.
To maximize provider-level caching:
def build_cacheable_prompt(
static_system_prompt: str,
static_context: str, # e.g., a company's full FAQ document
dynamic_user_message: str,
) -> list[dict]:
"""
Structure your messages so static content comes first and is
consistent across requests. This maximizes provider-side cache hits.
For Anthropic with explicit caching, you'd add cache_control markers.
For OpenAI, just keeping the prefix consistent triggers automatic caching.
"""
return [
{
"role": "system",
# Static content first — this prefix will be cached
"content": f"{static_system_prompt}\n\n<knowledge_base>\n{static_context}\n</knowledge_base>",
},
{
"role": "user",
# Dynamic content last — only this changes between requests
"content": dynamic_user_message,
},
]
The key principle: stable content first, variable content last. If your system prompt changes every request because you're injecting dynamic values into it, you're defeating provider-level caching. Restructure so the static portion is a clean prefix.
Let's put the pieces together into a middleware class you can actually deploy. This wraps any LLM call with caching, model routing, token counting, and cost logging.
import logging
import time
from dataclasses import dataclass, asdict
from openai import OpenAI
logger = logging.getLogger(__name__)
@dataclass
class LLMCallRecord:
request_id: str
timestamp: float
model_requested: str
model_used: str
input_tokens: int
output_tokens: int
cost_usd: float
latency_ms: float
cache_hit: bool
cache_type: Optional[str] # "exact", "semantic", "provider", or None
class CostAwareLLMClient:
MODEL_PRICING = {
"gpt-4o-mini": {"input": 0.15, "output": 0.60}, # per million tokens
"gpt-4o": {"input": 2.50, "output": 10.00},
"o3": {"input": 10.00, "output": 40.00},
}
def __init__(
self,
exact_cache: LLMCache,
semantic_cache: SemanticCache,
enable_routing: bool = True,
cost_alert_threshold_usd: float = 0.10, # Alert if single call exceeds this
):
self.client = OpenAI()
self.exact_cache = exact_cache
self.semantic_cache = semantic_cache
self.enable_routing = enable_routing
self.alert_threshold = cost_alert_threshold_usd
self.call_records: list[LLMCallRecord] = []
self.enc = tiktoken.get_encoding("cl100k_base")
def _count_tokens(self, messages: list[dict]) -> int:
count = 0
for msg in messages:
count += 4 + len(self.enc.encode(msg.get("content", "")))
return count + 2
def _calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
pricing = self.MODEL_PRICING.get(model, self.MODEL_PRICING["gpt-4o"])
return (
(input_tokens / 1_000_000) * pricing["input"] +
(output_tokens / 1_000_000) * pricing["output"]
)
def complete(
self,
messages: list[dict],
preferred_model: str = "gpt-4o",
force_model: Optional[str] = None,
user_query_for_semantic_cache: Optional[str] = None,
request_id: Optional[str] = None,
**openai_kwargs,
) -> tuple[str, LLMCallRecord]:
import uuid
request_id = request_id or str(uuid.uuid4())[:8]
start_time = time.time()
# 1. Check exact cache
cached = self.exact_cache.get(messages, preferred_model)
if cached:
record = LLMCallRecord(
request_id=request_id,
timestamp=start_time,
model_requested=preferred_model,
model_used="cache",
input_tokens=self._count_tokens(messages),
output_tokens=len(self.enc.encode(cached)),
cost_usd=0.0,
latency_ms=(time.time() - start_time) * 1000,
cache_hit=True,
cache_type="exact",
)
self.call_records.append(record)
return cached, record
# 2. Check semantic cache
if user_query_for_semantic_cache:
result = self.semantic_cache.lookup(user_query_for_semantic_cache)
if result:
cached_response, similarity = result
record = LLMCallRecord(
request_id=request_id,
timestamp=start_time,
model_requested=preferred_model,
model_used="cache",
input_tokens=self._count_tokens(messages),
output_tokens=len(self.enc.encode(cached_response)),
cost_usd=0.0,
latency_ms=(time.time() - start_time) * 1000,
cache_hit=True,
cache_type="semantic",
)
logger.info(f"Semantic cache hit (similarity={similarity:.3f}) for request {request_id}")
self.call_records.append(record)
return cached_response, record
# 3. Route to appropriate model
if force_model:
model = force_model
elif self.enable_routing and user_query_for_semantic_cache:
decision = route_request(
user_message=user_query_for_semantic_cache,
system_context=messages[0].get("content", "") if messages else "",
)
model = decision.tier.value
else:
model = preferred_model
# 4. Make the actual API call
input_tokens = self._count_tokens(messages)
response = self.client.chat.completions.create(
model=model,
messages=messages,
**openai_kwargs,
)
completion = response.choices[0].message.content
output_tokens = response.usage.completion_tokens
cost = self._calculate_cost(model, input_tokens, output_tokens)
latency_ms = (time.time() - start_time) * 1000
if cost > self.alert_threshold:
logger.warning(
f"High-cost call: ${cost:.4f} for request {request_id} "
f"({input_tokens} in / {output_tokens} out on {model})"
)
record = LLMCallRecord(
request_id=request_id,
timestamp=start_time,
model_requested=preferred_model,
model_used=model,
input_tokens=input_tokens,
output_tokens=output_tokens,
cost_usd=cost,
latency_ms=latency_ms,
cache_hit=False,
cache_type=None,
)
# 5. Populate caches for future requests
self.exact_cache.set(messages, model, completion)
if user_query_for_semantic_cache:
self.semantic_cache.store(user_query_for_semantic_cache, completion)
self.call_records.append(record)
return completion, record
def cost_summary(self) -> dict:
if not self.call_records:
return {}
api_calls = [r for r in self.call_records if not r.cache_hit]
cache_hits = [r for r in self.call_records if r.cache_hit]
total_cost = sum(r.cost_usd for r in api_calls)
model_breakdown = {}
for record in api_calls:
model_breakdown.setdefault(record.model_used, {"calls": 0, "cost": 0.0})
model_breakdown[record.model_used]["calls"] += 1
model_breakdown[record.model_used]["cost"] += record.cost_usd
return {
"total_calls": len(self.call_records),
"api_calls": len(api_calls),
"cache_hits": len(cache_hits),
"cache_hit_rate": len(cache_hits) / len(self.call_records),
"total_cost_usd": total_cost,
"avg_cost_per_api_call": total_cost / len(api_calls) if api_calls else 0,
"model_breakdown": model_breakdown,
"estimated_savings_from_cache_usd": sum(
self._calculate_cost("gpt-4o", r.input_tokens, r.output_tokens)
for r in cache_hits
),
}
A cost optimization effort without monitoring is a one-time fix. You need ongoing visibility into four metrics:
Emit your LLMCallRecord to your observability stack (Datadog, Grafana, CloudWatch, whatever you use) as structured logs. A simple approach:
def emit_cost_metric(record: LLMCallRecord, feature_name: str) -> None:
"""
Emit structured log for ingestion by any observability system.
"""
logger.info(json.dumps({
"event": "llm_api_call",
"feature": feature_name,
"request_id": record.request_id,
"model": record.model_used,
"input_tokens": record.input_tokens,
"output_tokens": record.output_tokens,
"cost_usd": record.cost_usd,
"latency_ms": record.latency_ms,
"cache_hit": record.cache_hit,
"cache_type": record.cache_type,
}))
Set a weekly alert if your rolling 7-day cost grows more than 20% week-over-week without a corresponding growth in user activity. Cost growing faster than usage is always a signal that something has changed — a new code path, a prompt regression, or a routing misconfiguration.
Build a cost-optimized pipeline for the following scenario:
Scenario: A SaaS company has a feature that lets users ask questions about their invoice history. The system has a 2,000-token static context block (company billing policy documentation) and answers questions like "When is my next billing date?" or "Can I get a refund for the charge on March 15?"
Your tasks:
validator_fn for a waterfall that checks whether the response from the mini model actually contains a date when the user asks a date-related question.There are no trick questions here. The goal is to practice the tradeoffs — you'll find yourself arguing both for and against aggressive caching depending on the query type, which is exactly the right instinct.
Mistake 1: Caching responses that include dynamic data
If your system prompt includes Today is {date} or injects user-specific account details, exact-match caching will serve stale or wrong data. The fix: separate static from dynamic prompt elements. Cache only responses built on purely static inputs.
Mistake 2: Setting similarity thresholds too low A 0.85 cosine similarity threshold sounds high, but in practice it can match semantically related but factually distinct questions. "How do I cancel my subscription?" and "How do I pause my subscription?" might score 0.87. Serving the cancellation answer to a pause question is worse than a cache miss. Audit cache hits weekly.
Mistake 3: Over-routing to cheap models Mini models hallucinate more on multi-step reasoning, struggle with complex instruction following, and produce lower-quality structured outputs. If you route everything to the cheapest tier and then spend engineering time debugging quality regressions, the cost savings evaporate. Measure quality, not just cost.
Mistake 4: Ignoring streaming costs If you use streaming responses, your token counts are identical, but you can't cache mid-stream. Make sure your cache layer wraps the full response, not the stream. A common implementation mistake is writing to cache before the stream completes.
Mistake 5: Forgetting embedding costs in semantic caching Every cache lookup requires an embedding call (text-embedding-3-small at $0.02/M tokens). For high-volume systems, this adds up. At 100,000 lookups per day with average 50-token queries, that's $0.10/day in embedding costs — probably worth it, but account for it in your ROI calculation.
Troubleshooting: Cache hit rate below 5% Usually means either your prompts are highly variable (dynamic content mixed into static sections) or your semantic threshold is too high. Add logging to inspect cache misses and look for near-duplicate queries that should have hit.
Troubleshooting: Unexpected cost spikes Check your model routing logs first — a configuration change that accidentally routes everything to a frontier model will cause a 10x cost spike instantly. Second, look at input token counts — an upstream change that passes a full document as context instead of a summary is a common culprit.
Cost optimization for LLM APIs isn't a one-time project — it's an ongoing discipline, like database query optimization or API rate limiting. The key levers are:
max_tokensThe target is not the cheapest possible implementation — it's the cheapest implementation that meets your quality bar. Define that quality bar explicitly (through evals, user satisfaction metrics, or correctness tests), then optimize toward it. Cost and quality are in tension, but they're not opposites. A well-designed system often achieves both simultaneously by eliminating waste without sacrificing signal.
What to build next:
promptfoo is a good open-source starting point.Learning Path: Intro to AI & Prompt Engineering