LLM APIs bill by the token. A single call costs cents, but when an agent task runs dozens of rounds or a product racks up a million calls a day, cost becomes a metric you have to watch like a hawk. Alibaba's Qwen3.8-Max gives a concrete number: explicit cache-hit input at ¥1/M tokens versus cache-miss at ¥12, a 12x gap (subject to official pricing). The same million-token input costs ¥1 when cached and ¥12 when not. For long-horizon tasks to be viable, each call has to be cheap enough that you can afford to let the model try, fail, and try again. This SOP breaks cost optimization into six steps: baseline measurement, prefix caching, model routing, batch API, context optimization, and monitoring. Each step ships with copy-paste Python code, followed by pitfalls and FAQ. The four core levers: make expensive models do less, make cheap models do more, let repeated inputs hit cache, and push non-urgent tasks into batch queues. Unlike our site's LLM Fine-Tuning SOP, which covers how to train, this one covers how to save, but both share a common prerequisite: measure first, optimize second.
1. Quantify Your Current Cost Baseline
Measure before you optimize. If you don't know how much you spend daily, where it goes, or which model eats the bulk, any optimization is blind. Step one is not cutting cost, it is getting the ledger right.
Every API response's usage field carries token statistics (prompt_tokens, completion_tokens, and for some providers cached_tokens). Log these numbers, aggregate by model, by endpoint, by time window, and you can pinpoint where the money flows.
import time
from collections import defaultdict
class CostTracker:
"""Lightweight call-cost tracker: logs token usage and spend per call"""
def __init__(self):
self.logs = []
self.cost_by_model = defaultdict(float)
def log(self, model, input_tokens, output_tokens,
cached_tokens=0, cost=0.0):
self.logs.append({
"ts": time.time(),
"model": model,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cached_tokens": cached_tokens,
"cost": cost,
})
self.cost_by_model[model] += cost
def summary(self):
total_cost = sum(l["cost"] for l in self.logs)
total_input = sum(l["input_tokens"] for l in self.logs)
total_cached = sum(l["cached_tokens"] for l in self.logs)
hit_rate = total_cached / total_input if total_input else 0
return {
"total_cost": round(total_cost, 4),
"total_input_tokens": total_input,
"cache_hit_rate": f"{hit_rate:.1%}",
"cost_by_model": dict(self.cost_by_model),
}Run this for a week and read the summary() output: total spend, per-model breakdown, cache hit rate. If 80% of your calls are just translation or classification but they all hit the expensive model, routing has huge room. If your cache hit rate is under 5%, prefix caching is your first priority. Without this data layer, every optimization that follows is guesswork.
2. Prefix Caching (Prompt Caching)
The core principle: if multiple requests share an identical prefix (system prompt + long document + few-shot examples), the provider stores that prefix's KV-cache and charges subsequent hits at a "cache read" rate far below the regular input price. Qwen3.8-Max's cache-hit ¥1/M versus cache-miss ¥12/M, a 12x gap, is a textbook example.
Providers implement this differently:
| Provider | Mechanism | How to Enable |
|---|---|---|
| Anthropic | Explicit cache, requires cache_control tag | Add cache_control to system/content blocks |
| OpenAI | Automatic (prompts >1024 tokens auto-qualify) | No code change needed, put static content first |
| DeepSeek | Prefix caching, automatic hit | Put stable content at the prompt start |
| Qwen (Alibaba) | Explicit / automatic caching | Refer to official docs |
Anthropic's API is the most explicit, so we use it as the example:
import anthropic
client = anthropic.Anthropic() # env var ANTHROPIC_API_KEY
# Stable long doc / rules / examples go in system, tagged cache_control
LONG_DOC = "(thousands of tokens of document content, rules, few-shot examples)"
response = client.messages.create(
model="claude-sonnet-4-5-20250929", # check official docs
max_tokens=1024,
system=[
{
"type": "text",
"text": f"You are a doc assistant. Reference doc:\n{LONG_DOC}",
"cache_control": {"type": "ephemeral"},
}
],
messages=[
{"role": "user", "content": "Summarize the key points."}
],
)
# Response usage reports cache hit status
print(f"Cache read tokens: {response.usage.cache_read_input_tokens}")
print(f"Cache write tokens: {response.usage.cache_creation_input_tokens}")
print(f"Regular input tokens: {response.usage.input_tokens}")The critical constraint: the prefix must match token-for-token. Change one character in the system prompt, add or remove a tool, reorder few-shot examples, and the cache is invalidated entirely. Adopting prefix caching means freezing your prompt structure: stable segments go first (system + long doc + rules), variable segments go last (the user's current input).
Pitfall reminders:
- Cache has a TTL (Anthropic defaults to 5 minutes, extendable to 1 hour via beta header; OpenAI/DeepSeek each have their own windows, check official docs). Idle too long and it expires.
- The first request is always a cache write (some providers charge slightly above the regular input price for writes). Hits start from the second request. High-frequency callers see the most benefit.
- Injecting timestamps, random IDs, or user names at the prompt start will punch through the cache. Any dynamic content must go at the very end of the prompt.
3. Model Routing (Tiered Dispatch)
Not every task needs the strongest model. Translation, classification, summarization, and format conversion are tasks where cheap models and expensive models produce nearly identical output, but the price differs by several-fold. Model routing means dispatching by task complexity: the expensive model only gets the hard problems, the cheap model handles the easy ones.
def route_model(query: str) -> str:
"""Route to different models by task complexity (model names per official docs)"""
q = query.lower()
# Complex reasoning: code, math, long-form analysis, multi-step -> strong model
hard_keywords = [
"code", "debug", "math", "reason", "analyze", "essay",
"code", "math", "reasoning", "analysis",
]
if any(kw in q for kw in hard_keywords):
return "qwen-max" # strong model
# Simple tasks: translation, classification, summary, format -> light model
return "qwen-turbo" # cheap model
def call_with_fallback(query: str) -> str:
"""Routing call with fallback"""
from openai import OpenAI
client = OpenAI() # OpenAI-compatible endpoint (Qwen/DeepSeek both support it)
primary = route_model(query)
fallback = "qwen-plus" # degrade here if primary is unavailable
for model in [primary, fallback]:
try:
resp = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": query}],
max_tokens=512,
)
return resp.choices[0].message.content
except Exception as e:
print(f"Model {model} failed: {e}, trying fallback")
raise RuntimeError("All models unavailable")Routing strategies can evolve from simple to sophisticated:
- Keyword matching (above): simplest, covers high-frequency cases, ships in dozens of lines.
- Classifier routing: train a lightweight classifier (or use embedding similarity) to judge difficulty, more accurate than keywords.
- Cascade routing: let the cheap model try first, escalate to the expensive model only if confidence is low or quality is subpar. Most cost-effective but doubles latency.
Whatever you choose, always leave a fallback: if the primary model rate-limits or times out, automatically degrade to the backup. A downgrade is always better than a 500 error.
4. Batch API
Large volumes of non-urgent tasks (data labeling, batch translation, content generation, evaluation scoring) do not need real-time responses. They can go through the batch API. OpenAI and Anthropic both offer batch endpoints: you submit a JSONL file of requests, get results back within hours, and pay a significant discount (commonly around 50% off, check official pricing).
Using the OpenAI Batch API as the example:
import json
from openai import OpenAI
client = OpenAI() # env var OPENAI_API_KEY
# 1. Write requests to a JSONL file
texts = ["hello", "thank you", "goodbye", "excuse me", "no problem"]
requests = [
{
"custom_id": f"req-{i}",
"method": "POST",
"url": "/v1/chat/completions",
"body": {
"model": "gpt-4o-mini",
"messages": [{"role": "user",
"content": f"Translate to Chinese: {text}"}],
"max_tokens": 200,
},
}
for i, text in enumerate(texts)
]
with open("batch_input.jsonl", "w", encoding="utf-8") as f:
for r in requests:
f.write(json.dumps(r, ensure_ascii=False) + "\n")
# 2. Upload file + create batch job
batch_file = client.files.create(
file=open("batch_input.jsonl", "rb"),
purpose="batch",
)
batch = client.batches.create(
input_file_id=batch_file.id,
endpoint="/v1/chat/completions",
completion_window="24h",
)
print(f"Batch ID: {batch.id}, status: {batch.status}")
# 3. Poll for completion, then download results
# batch = client.batches.retrieve(batch.id)
# if batch.status == "completed":
# result = client.files.content(batch.output_file_id)Key limitations:
- Latency for price: batch jobs complete within a 24-hour window, unsuitable for real-time scenarios.
- Per-file limits: request count and file size have caps (check official docs), split into multiple files if exceeded.
- No streaming: batch returns complete results only, no streaming.
5. Context and Repeat-Request Optimization
In long conversations and multi-round agent tasks, context length is a cost killer. A 20-round dialogue, if untrimmed, resends the entire history with each call. Token count grows linearly, and so does cost. Context optimization does two things: trim history and deduplicate repeat requests.
def trim_context(messages: list, max_messages: int = 20) -> list:
"""Keep system + most recent N messages, trim the middle history"""
if len(messages) <= max_messages:
return messages
system_msgs = [m for m in messages if m["role"] == "system"]
dialog = [m for m in messages if m["role"] != "system"]
keep = max_messages - len(system_msgs)
recent = dialog[-keep:] if keep > 0 else []
return system_msgs + recent
def compress_with_summary(messages: list, llm_call) -> list:
"""Long conversations: compress early messages into a summary, keep recent raw"""
if len(messages) <= 20:
return messages
system_msgs = [m for m in messages if m["role"] == "system"]
dialog = [m for m in messages if m["role"] != "system"]
to_summarize = dialog[:-10]
recent = dialog[-10:]
# Use a cheap model to compress early dialogue into a summary
summary_input = "\n".join(
f"{m['role']}: {m['content']}" for m in to_summarize)
summary = llm_call(f"Compress the following dialogue into a 200-word summary:\n{summary_input}")
summary_msg = {
"role": "system",
"content": f"Early dialogue summary: {summary}",
}
return system_msgs + [summary_msg] + recentBeyond trimming, watch for repeat-request deduplication. The same prompt called multiple times in quick succession (user refreshes, retries) can hit a local cache layer:
import hashlib
import time
class ResponseCache:
"""Simple response cache (only for idempotent requests)"""
def __init__(self, ttl_seconds=300):
self.cache = {}
self.ttl = ttl_seconds
def _key(self, model, messages):
raw = f"{model}:{str(messages)}"
return hashlib.md5(raw.encode()).hexdigest()
def get(self, model, messages):
key = self._key(model, messages)
if key in self.cache:
ts, result = self.cache[key]
if time.time() - ts < self.ttl:
return result
return None
def set(self, model, messages, result):
key = self._key(model, messages)
self.cache[key] = (time.time(), result)Note: only cache idempotent requests (like "translate this text"). Do not cache requests that need real-time data (like "what's the weather today"), or you will serve stale results.
6. Monitoring and Iteration
Cost optimization is not a one-time project. After going live, continuously monitor these metrics:
| Metric | Meaning | Alert Threshold (reference) |
|---|---|---|
| Daily total spend | API cost per day | Day-over-day increase > 50% |
| Cache hit rate | cached_tokens / input_tokens | < 30% (room to optimize) |
| Model distribution | Per-model call share | Expensive model > 60% (check routing) |
| Cost per request | Total spend / request count | Week-over-week increase > 20% |
| Batch share | Batch requests / total requests | < 20% (room to grow) |
Wire the Step 1 CostTracker into production, run summary() daily, and watch trends over a week. If cache hit rate drops from 60% to 20%, someone likely changed the prompt structure and caused prefix drift. If the expensive model's share suddenly spikes, check whether routing rules have a leak letting simple tasks through to the wrong tier.
7. Pitfalls
Pitfall 1: Cache hit requires an exact prefix match. Changing one punctuation mark in the system prompt, reordering few-shot examples, or adding/removing a tool invalidates the entire cache. Strictly separate stable and variable segments, stable goes first.
Pitfall 2: Batch API has latency, not for real-time. Batch is a "results within hours" model. Real-time chat and online inference cannot use it. It is only for offline bulk tasks.
Pitfall 3: Routing without fallback will break. If the primary model rate-limits or times out with no fallback, the user gets a 500. Always configure a backup model. A downgrade beats an error.
Pitfall 4: Dynamic content at the prompt start kills cache. Timestamps, random IDs, or user names at the beginning of the prompt change every call, so the cache never hits. Put these at the very end (the user message portion).
Pitfall 5: Over-trimming context loses critical info. Setting max_messages too small causes the agent to lose task context, leading to repeated questions or off-target answers. Rule of thumb: keep 10-20 messages for chat, use summary compression rather than hard truncation for agents.
Pitfall 6: Watching unit price, not total bill. Cutting per-call cost by 50% but doubling call frequency means total spend goes up. Optimize the total bill, not the per-call cost.
FAQ
Q1: How much can prefix caching save?
It depends on your prompt structure and call frequency. If the system prompt + long document accounts for 80% of input and you call frequently, that 80% is billed at the cache rate. With Qwen3.8-Max, the cache-hit price is 1/12 of the miss price (check official pricing), so high-hit-rate scenarios can cut input cost to a tenth or better. Low-frequency callers (a few calls a day) see limited benefit because the cache expires between calls.
Q2: How long does batch API take to return results?
OpenAI's Batch API completion_window is set to 24h, and it usually completes within a few hours, but real-time is not guaranteed. Anthropic's Message Batches works similarly. Batch is suited for data labeling, batch translation, and offline evaluation, not real-time scenarios.
Q3: Which step should a small team start with?
Start with measuring the baseline (Step 1), then adopt prefix caching (Step 2). These two have the smallest investment and most direct payoff: CostTracker is dozens of lines of code, and prefix caching for Anthropic is one line of cache_control. Routing and batch come next, as they require changing call logic. Context trimming and monitoring are the third phase, an ongoing iteration.
Q4: Does model routing hurt quality?
Not for simple tasks. Translation, classification, and summarization show minimal quality gap between cheap and expensive models, but the price differs severalfold. The key is routing accuracy: if a hard problem is misclassified as easy and sent to the cheap model, quality drops. Start conservative (only route obviously simple tasks) and expand gradually. Cascade mode (cheap model tries first, escalates if quality is insufficient) is safer but doubles latency.
Q5: How do I improve cache hit rate?
Three things: (1) freeze your prompt structure, stable segments (system + doc + rules) first, variable segments (user input) last; (2) avoid injecting dynamic content (timestamps, random numbers, user IDs) into the prefix, put them in the user message portion; (3) call frequently enough that a subsequent request arrives within the TTL. If call frequency is low, consider longer-TTL cache options (such as Anthropic's 1-hour cache, check official docs).
Perspective
The essence of cost optimization is not "saving money" but "spending it where it counts." When an agent task needs 50 rounds to converge, if each call costs too much, you would never dare let it run. You either abandon midway or artificially cut iterations, sacrificing quality. Prefix caching compresses per-call cost to a tenth or less, meaning the same budget buys ten times the iterations, or the same iterations cost a tenth as much. That is the real value of cost optimization: it changes the decision threshold of "dare I let the model try fully." Qwen3.8-Max setting a 12x gap between cache hits and misses is essentially telling developers: long-horizon tasks are viable now, as long as you use prefix caching well.
References