Batching, Model Routing, and Cost Dashboards
Route simple requests to cheaper models like GPT-4o-mini and complex ones to GPT-4o, batch non-urgent requests, and build a cost dashboard tracking spending by feature.
Batching, Model Routing, and Cost Dashboards is a free AI Engineering Academy lesson on CoddyKit — lesson 4 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the AI Engineering Academy learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.
Three More Levers for Cost Optimization
After caching, three additional strategies dramatically reduce LLM operating costs: batching (defer non-urgent requests and submit them in bulk at a lower API rate), model routing (route simple queries to cheap models and complex ones to powerful models), and cost dashboards (track spending per feature to identify where optimizations have the highest ROI). Together these can reduce costs by another 40-60 percent beyond caching.
OpenAI Batch API: 50% Off for Async Workloads
OpenAI's Batch API accepts a JSONL file containing up to 50,000 requests and processes them asynchronously within 24 hours at 50 percent of the standard price. This is ideal for non-interactive workloads: embedding large document corpora, generating product descriptions, running nightly evaluations, or preprocessing training data. The trade-off is latency — results are available hours later, not immediately.
import json
from openai import OpenAI
client = OpenAI()
# Prepare batch file
requests = [
{
'custom_id': f'req_{i}',
'method': 'POST',
'url': '/v1/chat/completions',
'body': {
'model': 'gpt-4o-mini',
'messages': [
{'role': 'user', 'content': f'Summarize: {document}'}
],
'max_tokens': 150,
}
}
for i, document in enumerate(documents_to_process)
]
# Write JSONL batch file
with open('/tmp/batch_requests.jsonl', 'w') as f:
for req in requests:
f.write(json.dumps(req) + '\n')
# Upload and submit batch
with open('/tmp/batch_requests.jsonl', 'rb') as f:
batch_file = client.files.create(file=f, purpose='batch')
batch = client.batches.create(
input_file_id=batch_file.id,
endpoint='/v1/chat/completions',
completion_window='24h',
)
print(f'Batch {batch.id} submitted, status: {batch.status}')Polling Batch Results
After submitting a batch, poll its status until it completes (status changes from in_progress to completed). Once complete, download the output file containing results for all requests. Each output line is a JSON object with the custom_id from the request and either a response or an error field — always handle both since individual requests within a batch can fail independently.
import time
def wait_for_batch(batch_id: str, poll_interval: int = 60) -> str:
while True:
batch = client.batches.retrieve(batch_id)
print(f'Status: {batch.status}, completed: {batch.request_counts.completed}')
if batch.status == 'completed':
return batch.output_file_id
elif batch.status == 'failed':
raise RuntimeError(f'Batch failed: {batch.errors}')
time.sleep(poll_interval)
def download_batch_results(output_file_id: str) -> list[dict]:
content = client.files.content(output_file_id)
results = []
for line in content.text.strip().split('\n'):
results.append(json.loads(line))
return results
output_file_id = wait_for_batch(batch.id)
results = download_batch_results(output_file_id)
for result in results[:3]:
print(result['custom_id'], result.get('response', {}).get('body', {}).get('choices', [{}])[0])Model Routing: Match Complexity to Model Size
Model routing assigns each request to the cheapest model capable of handling it well. GPT-4o-mini costs about 30x less than GPT-4o but handles simple classification, extraction, and short Q&A tasks equally well. Route simple, structured tasks to small cheap models and complex reasoning, long-context synthesis, and nuanced generation to large powerful models. Even routing 60 percent of traffic to a cheap model saves significant costs.
CHEAP_MODEL = 'gpt-4o-mini'
POWERFUL_MODEL = 'gpt-4o'
def classify_query_complexity(query: str) -> str:
# Heuristic-based routing (replace with ML classifier for production)
words = query.split()
has_code = any(c in query for c in ['```', 'def ', 'class ', 'SELECT ', 'function '])
is_multi_step = any(w in query.lower() for w in ['compare', 'analyze', 'explain why', 'evaluate'])
is_long = len(words) > 50
if has_code or is_multi_step or is_long:
return POWERFUL_MODEL
return CHEAP_MODEL
def routed_completion(messages: list[dict]) -> str:
user_query = messages[-1].get('content', '')
model = classify_query_complexity(user_query)
print(f'Routing to: {model}')
response = client.chat.completions.create(model=model, messages=messages)
return response.choices[0].message.contentLLM-Based Routing for Higher Accuracy
Heuristic routing is fast but brittle. A more accurate approach uses a small, cheap classification model to decide which model to route to. Fine-tune a small model on examples of simple versus complex queries in your domain, or use few-shot prompting with GPT-4o-mini itself. The classifier call costs a few hundred input tokens — far less than incorrectly routing a complex query to a cheap model that produces a wrong answer.
CLASSIFIER_SYSTEM = '''You are a query complexity classifier.
Classify the user query as SIMPLE or COMPLEX.
SIMPLE: factual lookup, extraction, classification with clear answer.
COMPLEX: multi-step reasoning, synthesis, comparison, code generation, long-form writing.
Reply with just SIMPLE or COMPLEX.'''
def llm_classify_complexity(query: str) -> str:
response = client.chat.completions.create(
model='gpt-4o-mini', # use cheap model for routing
messages=[
{'role': 'system', 'content': CLASSIFIER_SYSTEM},
{'role': 'user', 'content': query},
],
max_tokens=10,
temperature=0,
)
label = response.choices[0].message.content.strip()
return POWERFUL_MODEL if label == 'COMPLEX' else CHEAP_MODELTracking Cost Per Feature
To know where to focus optimization effort, you need to track cost per application feature, not just total spend. Wrap every LLM call with a feature tag and accumulate token costs per tag. 'search_summarization' might consume 40 percent of your budget while serving only 5 percent of traffic, making it a high-priority optimization target. 'user_onboarding' might be expensive but serves a high-value flow you do not want to degrade.
from collections import defaultdict
cost_tracker = defaultdict(lambda: {'prompt_tokens': 0, 'completion_tokens': 0, 'cost_usd': 0.0})
MODEL_PRICING = {
'gpt-4o-mini': {'input': 0.15 / 1e6, 'output': 0.60 / 1e6},
'gpt-4o': {'input': 2.50 / 1e6, 'output': 10.00 / 1e6},
}
def tracked_completion(feature: str, messages: list[dict], model: str = 'gpt-4o-mini') -> str:
response = client.chat.completions.create(model=model, messages=messages)
usage = response.usage
pricing = MODEL_PRICING.get(model, {'input': 0, 'output': 0})
cost = usage.prompt_tokens * pricing['input'] + usage.completion_tokens * pricing['output']
cost_tracker[feature]['prompt_tokens'] += usage.prompt_tokens
cost_tracker[feature]['completion_tokens'] += usage.completion_tokens
cost_tracker[feature]['cost_usd'] += cost
return response.choices[0].message.content
def print_cost_report():
print(f'{"Feature":<30} {"Prompt":<10} {"Completion":<12} {"Cost USD":<12}')
for feature, stats in sorted(cost_tracker.items(), key=lambda x: -x[1]['cost_usd']):
print(f'{feature:<30} {stats["prompt_tokens"]:<10} {stats["completion_tokens"]:<12} ${stats["cost_usd"]:.4f}')Building a Simple Cost Dashboard
A practical cost dashboard aggregates feature-level spend data and exposes it via a simple HTTP endpoint. Store cumulative costs in Redis with daily rollup keys so you can trend spend over time. Add this dashboard to your internal developer tools so the team can see the cost impact of feature releases in near real-time and catch runaway spending before it becomes a large bill.
from fastapi import FastAPI
import datetime
app = FastAPI()
async def record_cost(feature: str, model: str, prompt_tokens: int, completion_tokens: int):
pricing = MODEL_PRICING.get(model, {'input': 0, 'output': 0})
cost = prompt_tokens * pricing['input'] + completion_tokens * pricing['output']
today = datetime.date.today().isoformat()
key = f'cost:{today}:{feature}:{model}'
await async_r.incrbyfloat(key, cost)
await async_r.expire(key, 86400 * 30) # keep 30 days
@app.get('/dashboard/costs')
async def cost_dashboard():
today = datetime.date.today().isoformat()
pattern = f'cost:{today}:*'
costs = {}
async for key in async_r.scan_iter(match=pattern):
value = await async_r.get(key)
parts = key.split(':')
feature_model = ':'.join(parts[2:])
costs[feature_model] = float(value or 0)
return {'date': today, 'costs': costs, 'total': sum(costs.values())}Monthly Budget Alerts
Set monthly budget alerts to catch unexpected cost spikes before they become large bills. Compute a rolling daily spend from your cost tracker, project it to month-end, and fire a Slack alert when the projection exceeds your budget threshold. A simple projection — daily_spend * days_remaining — catches runaway requests early even if actual patterns are nonlinear.
import datetime
import httpx
SLACK_WEBHOOK = 'https://hooks.slack.com/services/YOUR/WEBHOOK'
MONTHLY_BUDGET_USD = 500.0
async def check_budget_alert():
today = datetime.date.today()
days_in_month = 30
day_of_month = today.day
days_remaining = days_in_month - day_of_month
# Sum today's costs
today_total = sum(cost_tracker[f]['cost_usd'] for f in cost_tracker)
avg_daily = today_total # simplified: just today's spend
projected_month = avg_daily * days_in_month
if projected_month > MONTHLY_BUDGET_USD:
message = (
f'LLM Budget Alert: Projected monthly spend ${projected_month:.2f} '
f'exceeds budget ${MONTHLY_BUDGET_USD:.2f}. '
f'Today spend: ${today_total:.2f}'
)
async with httpx.AsyncClient() as client:
await client.post(SLACK_WEBHOOK, json={'text': message})Request Queue for Rate Limit Management
When traffic spikes, requests hit OpenAI's rate limits and fail with 429 Too Many Requests. A request queue buffers incoming requests and submits them at a controlled rate, smoothing traffic peaks. Use an async queue backed by Redis or a message broker like RabbitMQ for production, and implement exponential backoff retry logic for temporary 429 errors.
import asyncio
from asyncio import Queue
class RateLimitedLLMClient:
def __init__(self, requests_per_minute: int = 500):
self.rpm = requests_per_minute
self.queue: Queue = Queue(maxsize=1000)
self.interval = 60.0 / requests_per_minute
async def start(self):
asyncio.create_task(self._worker())
async def _worker(self):
while True:
request_fn, future = await self.queue.get()
try:
result = await request_fn()
future.set_result(result)
except Exception as e:
future.set_exception(e)
await asyncio.sleep(self.interval)
async def submit(self, request_fn) -> str:
loop = asyncio.get_event_loop()
future = loop.create_future()
await self.queue.put((request_fn, future))
return await futurePutting It All Together: Cost Optimization Stack
A complete LLM cost optimization stack operates in layers: exact cache eliminates calls for repeated identical queries, semantic cache eliminates calls for similar queries, prefix caching reduces input cost for all remaining calls, model routing uses cheap models for simple queries, batching defers non-urgent work for 50 percent off, and dashboards and alerts keep costs visible and bounded. Implement them incrementally in order of impact for your specific application.
# Decision framework for cost optimization priority:
#
# 1. Enable prefix caching (free, zero effort, automatic)
# 2. Add exact caching (high hit rate for FAQ/support bots)
# 3. Add model routing (simple heuristics first, ML classifier later)
# 4. Add semantic caching (complex, high ROI for paraphrase-heavy use cases)
# 5. Enable batch API (only for non-real-time pipelines)
# 6. Build cost dashboard (essential for ongoing monitoring)
#
# Typical combined result in a customer support bot:
# Before: $1,000/month
# After step 1-2: $400/month (-60%)
# After step 3-4: $200/month (-50% of remaining)
# After step 5-6: $150/month and visibleCascading Fallback on Cheap Model Failure
When routing to a cheap model, you must handle cases where it produces an unsatisfactory answer. Implement a quality check on the cheap model's output — check response length, presence of required fields, or run a fast LLM-as-judge score — and automatically fall back to the powerful model if quality is insufficient. This safety net lets you be aggressive in routing to cheap models without risking degraded user experience.
async def routing_with_fallback(messages: list[dict], min_length: int = 50) -> str:
# Try cheap model first
cheap_response = await async_client.chat.completions.create(
model=CHEAP_MODEL, messages=messages, temperature=0.0
)
answer = cheap_response.choices[0].message.content
# Quality check: response too short indicates poor answer
if len(answer.strip()) < min_length:
print(f'Cheap model answer too short ({len(answer)} chars), escalating...')
powerful_response = await async_client.chat.completions.create(
model=POWERFUL_MODEL, messages=messages, temperature=0.0
)
return powerful_response.choices[0].message.content
return answerQuick Check
Test your understanding of batching, model routing, and cost dashboards from this lesson.
Lesson Recap
In this lesson you learned: OpenAI Batch API provides 50 percent off for async non-real-time workloads, model routing uses cheap models like GPT-4o-mini for simple tasks and expensive models for complex ones, and per-feature cost tracking reveals which parts of your application consume the most budget so you can prioritize optimizations effectively. Combined with caching strategies from previous lessons, these techniques can reduce LLM infrastructure costs by 60-80 percent. You have now completed the LLM caching and cost optimization course.
Frequently asked questions
Is the “Batching, Model Routing, and Cost Dashboards” lesson free?
Yes — the full text of “Batching, Model Routing, and Cost Dashboards” is free to read here on the web, and the AI Engineering Academy course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the AI Engineering Academy course, upgrade to CoddyKit PRO.
What will I learn in “Batching, Model Routing, and Cost Dashboards”?
Route simple requests to cheaper models like GPT-4o-mini and complex ones to GPT-4o, batch non-urgent requests, and build a cost dashboard tracking spending by feature. You practise AI Engineering Academy with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.
Do I need any experience to start AI Engineering Academy?
No prior experience is required. AI Engineering Academy on CoddyKit is structured for beginners through advanced learners; this is — lesson 4 of 4, so you can start here or from the beginning and move at your own pace.
How long does the “Batching, Model Routing, and Cost Dashboards” lesson take?
Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.
Can I write and run code in this AI Engineering Academy lesson?
Yes. Every AI Engineering Academy lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.
All lessons in this course
- Exact Caching with Redis
- Semantic Caching with Embeddings
- OpenAI Prompt Prefix Caching
- Batching, Model Routing, and Cost Dashboards