Optimizing Costs at Scale
On this page
Cost as an Engineering Constraint
A poorly optimized prompt costing $0.002 per call is negligible during development. At one million calls, it costs $2,000—and it will keep costing you that every month until someone rewrites it. This shift in scale turns cost into an engineering constraint: you don’t solve it by negotiating a contract, you solve it by changing code.
Visibility is the prerequisite for any optimization. A monthly bill tells you how much you spent, not where. That’s why this lesson starts with instrumentation before covering reduction levers.
Knowing Where the Money Goes
Four elements determine the cost of a call, and they don’t behave the same way. Input tokens, which aggregate the system prompt, conversation history, and user message, grow with session duration. Output tokens—the generated response—are billed significantly higher than input. Cached input tokens receive roughly a 50% discount off the input rate, making their tracking essential to understanding any bill. Finally, the model itself: GPT-5.6 Luna is far cheaper than GPT-5.6 Sol, and this factor dominates all others.
Tracking means converting every usage object returned by the API into a dollar amount, applying the correct rate, and isolating the cached portion.
import openai
from dataclasses import dataclass, field
client = openai.OpenAI()
# Rates per million tokens (July 2026) — cached_input is indicative (~50% of input),
# adjust according to the official pricing grid
RATES = {
"gpt-5.6-luna": {"input": 0.20, "output": 1.20, "cached_input": 0.02},
"gpt-5.6-terra": {"input": 2.00, "output": 12.00, "cached_input": 0.20},
"gpt-5.6-sol": {"input": 5.00, "output": 30.00, "cached_input": 0.50},
}
@dataclass
class CostTracker:
"""Tracks costs in real time."""
calls: list[dict] = field(default_factory=list)
def record(self, model: str, usage: dict) -> float:
input_tokens = usage.input_tokens
cached_tokens = getattr(
usage.input_tokens_details, "cached_tokens", 0
)
output_tokens = usage.output_tokens
uncached_input_tokens = input_tokens - cached_tokens
rate = RATES.get(model, RATES["gpt-5.6-terra"])
cost = (
(uncached_input_tokens / 1_000_000) * rate["input"]
+ (cached_tokens / 1_000_000) * rate["cached_input"]
+ (output_tokens / 1_000_000) * rate["output"]
)
self.calls.append({
"model": model,
"input_tokens": input_tokens,
"cached_tokens": cached_tokens,
"output_tokens": output_tokens,
"cost": cost,
})
return cost
@property
def total_cost(self) -> float:
return sum(a["cost"] for a in self.calls)
def report(self) -> str:
if not self.calls:
return "No calls recorded."
return (
f"Calls: {len(self.calls)} | "
f"Total cost: ${self.total_cost:.4f} | "
f"Average cost: ${self.total_cost / len(self.calls):.6f}"
)
tracker = CostTracker()The RATES table requires constant attention: it becomes outdated as soon as the official pricing grid changes. Treat it as configuration data to verify, not as a program constant, and make average cost per call a metric you review weekly.
Four Levers for Reduction
The first lever, and by far the most profitable, is avoiding paying for a high-end model when the task doesn’t require it. Explicit routing based on complexity prevents the default drift where all traffic ends up on the most expensive model simply because it appeared in the first copied example.
async def economic_router(prompt: str, complexity: str) -> str:
"""Routes to the cheapest capable model."""
if complexity == "simple":
model = "gpt-5.6-luna"
elif complexity == "medium":
model = "gpt-5.6-terra"
else:
model = "gpt-5.6-sol"
response = client.responses.create(model=model, input=prompt)
tracker.record(model, response.usage)
return response.output_textThe second lever goes further when complexity isn’t known in advance. Instead of classifying the request, the cascade tries the cheapest model first and escalates only if the response fails a quality check. The check used here is deliberately rudimentary—a minimum length and the absence of an ignorance admission—and this is the point you’ll need to refine: a cascade that escalates too often becomes more expensive than a direct call to the right model.
async def model_cascade(prompt: str) -> str:
"""Tries the cheapest model first."""
models = ["gpt-5.6-luna", "gpt-5.6-terra", "gpt-5.6-sol"]
for model in models:
response = client.responses.create(
model=model,
input=prompt,
max_output_tokens=500,
)
tracker.record(model, response.usage)
# Check quality (simple heuristic)
text = response.output_text
if len(text) > 50 and "i don't know" not in text.lower():
return text
return text # Last result even if imperfectThe third lever targets input tokens, which grow silently in any conversational application: by the twentieth turn, you’re repaying for the previous nineteen with every message. Compressing history by keeping the first message (which holds context) and the last (which holds the question), then filling the remaining budget with the most recent exchanges, preserves essential coherence at a fraction of the cost.
def compress_history(
messages: list[dict],
max_tokens: int = 4000,
) -> list[dict]:
"""Keeps the most recent messages within budget."""
# Always keep the first (context) and last (question)
if len(messages) <= 2:
return messages
first = messages[0]
last = messages[-1]
middle = messages[1:-1]
# Estimate tokens (approximation: 1 token ≈ 4 characters)
fixed_tokens = (len(first["content"]) + len(last["content"])) // 4
middle_budget = max_tokens - fixed_tokens
# Keep the most recent messages
result = []
tokens_used = 0
for msg in reversed(middle):
msg_tokens = len(msg["content"]) // 4
if tokens_used + msg_tokens > middle_budget:
break
result.insert(0, msg)
tokens_used += msg_tokens
return [first] + result + [last]The fourth lever requires no algorithmic finesse, only discipline: the Batch API offers a 50% reduction, and you just need to identify what can wait. Let the task itself carry urgency from its creation, rather than deducing it afterward—this is the only way to prevent everything from defaulting to the real-time queue.
def classify_urgency(tasks: list[dict]) -> dict:
"""Separates tasks into real-time vs batch."""
real_time = []
batch = []
for task in tasks:
if task.get("urgency") == "immediate":
real_time.append(task)
else:
batch.append(task)
return {
"real_time": real_time,
"batch": batch,
"estimated_savings": len(batch) * 0.001, # Example
}Getting Warned Before the Bill
All these optimizations can be undone by a single unlucky deployment: an unbounded retry loop, a tripled system prompt, a model changed by mistake. A cost alert is the safety net that warns you within hours rather than at month-end. Note the intermediate threshold at 80%: it turns a brutal alarm into an early signal, giving you time to act before exceeding the limit.
class CostAlert:
"""Triggers alerts when costs exceed a threshold."""
def __init__(self, daily_threshold: float = 50.0):
self.threshold = daily_threshold
self.tracker = CostTracker()
def check(self) -> str | None:
if self.tracker.total_cost > self.threshold:
return (
f"ALERT: daily cost ${self.tracker.total_cost:.2f} "
f"exceeds threshold of ${self.threshold:.2f}"
)
if self.tracker.total_cost > self.threshold * 0.8:
return (
f"WARNING: cost at {self.tracker.total_cost / self.threshold:.0%} "
f"of daily threshold"
)
return NoneKey Takeaways
- Track your costs in real time using
response.usage - Route to the cheapest model capable of the task
- Use the Batch API (-50%) for anything not requiring real-time processing
- Reduce input tokens by compressing history
- Set up cost alerts to avoid surprises