Complete Guide to LLM Cost Optimization

Introduction

LLM costs scale with usage — every token costs money. In production, unoptimized LLM applications can rack up thousands of dollars per day. Cost optimization is not optional; it is a production requirement. This guide walks through model routing, prompt compression, caching strategies, batch API usage, token management, and budget control patterns that reduce LLM costs by 50-90% without sacrificing quality.

Cost Breakdown

LLM Cost Components:
1. Input tokens: Tokens sent to the model (prompt + context)
2. Output tokens: Tokens generated by the model
3. Embedding tokens: Tokens embedded for vector search
4. Fine-tuning: Training compute + storage
5. Image/audio: Per-image or per-minute costs

Pricing per 1M tokens (approximate):
  gpt-4o: $2.50 input / $10.00 output
  gpt-4o-mini: $0.15 input / $0.60 output
  o1: $15.00 input / $60.00 output
  o1-mini: $3.00 input / $12.00 output
  text-embedding-3-small: $0.02 per 1M
  text-embedding-3-large: $0.13 per 1M

Cost ratio: input vs output
  gpt-4o: output is 4x more expensive than input
  gpt-4o-mini: output is 4x more expensive than input
  → Reducing output tokens saves more than reducing input tokens

Model Routing

Smart Model Selection

from openai import OpenAI
import re

client = OpenAI()

class ModelRouter:
    def __init__(self):
        self.routing_rules = [
            {"pattern": r"^(yes|no|true|false|\d+)$", "model": "gpt-4o-mini"},
            {"pattern": r"^(summarize|classify|extract)", "model": "gpt-4o-mini"},
            {"pattern": r"^(write|create|generate|code|debug)", "model": "gpt-4o"},
            {"pattern": r"^(analyze|reason|compare|evaluate)", "model": "gpt-4o"},
        ]
        self.default_model = "gpt-4o-mini"
    
    def select_model(self, prompt: str) -> str:
        prompt_lower = prompt.lower().strip()
        
        for rule in self.routing_rules:
            if re.search(rule["pattern"], prompt_lower):
                return rule["model"]
        
        # Token-based routing: short prompts use mini
        estimated_tokens = len(prompt) // 4
        if estimated_tokens < 100:
            return "gpt-4o-mini"
        
        return self.default_model
    
    def complete(self, prompt: str, **kwargs) -> str:
        model = self.select_model(prompt)
        
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            **kwargs
        )
        
        return {
            "content": response.choices[0].message.content,
            "model": model,
            "input_tokens": response.usage.prompt_tokens,
            "output_tokens": response.usage.completion_tokens,
            "cost": self._calculate_cost(model, response.usage.prompt_tokens, response.usage.completion_tokens)
        }
    
    def _calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        pricing = {
            "gpt-4o": {"input": 2.50, "output": 10.00},
            "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        }
        if model not in pricing:
            return 0.0
        
        p = pricing[model]
        return (input_tokens / 1_000_000 * p["input"]) + (output_tokens / 1_000_000 * p["output"])

router = ModelRouter()

# Simple task → uses gpt-4o-mini
result = router.complete("Classify this review as positive or negative: Great product!")
print(f"Model: {result['model']}, Cost: ${result['cost']:.6f}")

# Complex task → uses gpt-4o
result = router.complete("Write a Python function to implement quicksort with detailed comments")
print(f"Model: {result['model']}, Cost: ${result['cost']:.6f}")

Cascading Model Fallback

class CascadingRouter:
    """Try cheap model first, fall back to expensive if quality is insufficient."""
    
    def __init__(self, confidence_threshold: float = 0.7):
        self.confidence_threshold = confidence_threshold
        self.client = OpenAI()
    
    def complete(self, prompt: str) -> dict:
        # Try gpt-4o-mini first
        response = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "Answer the question. If you are not confident, say 'UNCERTAIN'."},
                {"role": "user", "content": prompt}
            ],
            temperature=0.0
        )
        
        answer = response.choices[0].message.content
        
        if "UNCERTAIN" not in answer:
            return {
                "answer": answer,
                "model": "gpt-4o-mini",
                "cost": self._cost("gpt-4o-mini", response.usage)
            }
        
        # Fall back to gpt-4o
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.0
        )
        
        return {
            "answer": response.choices[0].message.content,
            "model": "gpt-4o",
            "cost": self._cost("gpt-4o", response.usage)
        }
    
    def _cost(self, model, usage):
        pricing = {"gpt-4o": (2.50, 10.00), "gpt-4o-mini": (0.15, 0.60)}
        p = pricing.get(model, (0, 0))
        return (usage.prompt_tokens / 1e6 * p[0]) + (usage.completion_tokens / 1e6 * p[1])

Prompt Compression

Token Reduction Techniques

class PromptOptimizer:
    @staticmethod
    def compress_system_prompt(prompt: str) -> str:
        """Remove redundancy and verbosity from system prompts."""
        # Remove filler phrases
        fillers = [
            "You are a helpful assistant.",
            "Please note that",
            "It is important to",
            "You should always",
            "Make sure to",
        ]
        for filler in fillers:
            prompt = prompt.replace(filler, "")
        
        # Collapse whitespace
        import re
        prompt = re.sub(r'\s+', ' ', prompt).strip()
        
        return prompt
    
    @staticmethod
    def truncate_context(context: str, max_tokens: int = 2000) -> str:
        """Truncate context to fit within token budget."""
        # Rough estimate: 1 token ≈ 4 chars
        max_chars = max_tokens * 4
        
        if len(context) <= max_chars:
            return context
        
        # Keep beginning and end, truncate middle
        keep_start = max_chars // 2
        keep_end = max_chars // 2
        return context[:keep_start] + "\n[...truncated...]\n" + context[-keep_end:]
    
    @staticmethod
    def remove_examples(prompt: str) -> str:
        """Remove few-shot examples for cost reduction."""
        import re
        # Remove example blocks
        prompt = re.sub(r'Example \d+:.*?(?=Example \d+:|$)', '', prompt, flags=re.DOTALL)
        return prompt.strip()

# Before: 500-token system prompt
long_prompt = """
You are a helpful assistant. You should always be polite and professional.
Please note that it is important to provide accurate information.
Make sure to format your responses clearly. You should always cite sources
when possible. It is important to be concise but thorough. Please note that
you should avoid making assumptions.
"""

# After: 50-token system prompt
optimized = PromptOptimizer.compress_system_prompt(long_prompt)
print(f"Before: {len(long_prompt)} chars")
print(f"After: {len(optimized)} chars")

Caching Strategies

Exact Match Cache

import hashlib
import json
from datetime import datetime, timedelta

class LLMCache:
    def __init__(self, ttl_hours: int = 24):
        self.cache: dict[str, dict] = {}
        self.ttl = timedelta(hours=ttl_hours)
    
    def _key(self, model: str, messages: list, **kwargs) -> str:
        data = json.dumps({"model": model, "messages": messages, **kwargs}, sort_keys=True)
        return hashlib.sha256(data.encode()).hexdigest()
    
    def get(self, model: str, messages: list, **kwargs) -> str | None:
        key = self._key(model, messages, **kwargs)
        if key in self.cache:
            entry = self.cache[key]
            if datetime.now() - entry["timestamp"] < self.ttl:
                return entry["response"]
            del self.cache[key]
        return None
    
    def set(self, model: str, messages: list, response: str, **kwargs):
        key = self._key(model, messages, **kwargs)
        self.cache[key] = {
            "response": response,
            "timestamp": datetime.now()
        }

cache = LLMCache(ttl_hours=24)

def cached_complete(prompt: str, model: str = "gpt-4o") -> str:
    messages = [{"role": "user", "content": prompt}]
    
    # Check cache
    cached = cache.get(model, messages)
    if cached:
        return cached
    
    # Make API call
    response = client.chat.completions.create(model=model, messages=messages)
    result = response.choices[0].message.content
    
    # Cache result
    cache.set(model, messages, result)
    
    return result

Semantic Cache

import numpy as np
from openai import OpenAI

client = OpenAI()

class SemanticCache:
    def __init__(self, similarity_threshold: float = 0.95, max_entries: int = 1000):
        self.entries: list[dict] = []
        self.threshold = similarity_threshold
        self.max_entries = max_entries
    
    def _embed(self, text: str) -> np.ndarray:
        response = client.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        return np.array(response.data[0].embedding)
    
    def get(self, query: str) -> str | None:
        query_embedding = self._embed(query)
        
        for entry in self.entries:
            similarity = np.dot(query_embedding, entry["embedding"]) / (
                np.linalg.norm(query_embedding) * np.linalg.norm(entry["embedding"])
            )
            if similarity >= self.threshold:
                return entry["response"]
        
        return None
    
    def set(self, query: str, response: str):
        embedding = self._embed(query)
        
        self.entries.append({
            "query": query,
            "response": response,
            "embedding": embedding
        })
        
        # Evict oldest if over capacity
        if len(self.entries) > self.max_entries:
            self.entries.pop(0)

semantic_cache = SemanticCache(similarity_threshold=0.95)

def semantically_cached_complete(prompt: str, model: str = "gpt-4o") -> str:
    # Check semantic cache
    cached = semantic_cache.get(prompt)
    if cached:
        return cached  # Cache hit — no API cost
    
    # Make API call
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}]
    )
    result = response.choices[0].message.content
    
    # Cache result
    semantic_cache.set(prompt, result)
    
    return result

Batch API for Cost Reduction

import json

class BatchProcessor:
    """Use OpenAI Batch API for 50% cost reduction on non-time-sensitive workloads."""
    
    def __init__(self, client):
        self.client = client
    
    def prepare_batch(self, prompts: list[str], model: str = "gpt-4o-mini") -> str:
        """Prepare batch file and upload."""
        requests = []
        for i, prompt in enumerate(prompts):
            requests.append({
                "custom_id": f"task-{i}",
                "method": "POST",
                "url": "/v1/chat/completions",
                "body": {
                    "model": model,
                    "messages": [{"role": "user", "content": prompt}]
                }
            })
        
        # Write to JSONL
        filename = "batch_requests.jsonl"
        with open(filename, "w") as f:
            for req in requests:
                f.write(json.dumps(req) + "\n")
        
        # Upload file
        file = self.client.files.create(
            file=open(filename, "rb"),
            purpose="batch"
        )
        
        return file.id
    
    def submit_batch(self, file_id: str) -> str:
        """Submit batch for processing."""
        batch = self.client.batches.create(
            input_file_id=file_id,
            endpoint="/v1/chat/completions",
            completion_window="24h"
        )
        return batch.id
    
    def get_results(self, batch_id: str) -> list[dict]:
        """Retrieve batch results."""
        batch = self.client.batches.retrieve(batch_id)
        
        if batch.status != "completed":
            return []
        
        results = []
        content = self.client.files.content(batch.output_file_id)
        
        for line in content.text.strip().split("\n"):
            result = json.loads(line)
            if not result.get("error"):
                content = result["response"]["body"]["choices"][0]["message"]["content"]
                results.append({
                    "custom_id": result["custom_id"],
                    "content": content
                })
        
        return results

# Usage: Process 1000 sentiment classifications at 50% cost
processor = BatchProcessor(client)

prompts = [f"Classify sentiment (positive/negative/neutral): {review}" for review in reviews]
file_id = processor.prepare_batch(prompts, model="gpt-4o-mini")
batch_id = processor.submit_batch(file_id)

# Results available within 24 hours
# results = processor.get_results(batch_id)

Token Management

Token Counting

import tiktoken

class TokenManager:
    def __init__(self, model: str = "gpt-4o"):
        self.encoder = tiktoken.encoding_for_model(model)
    
    def count_tokens(self, text: str) -> int:
        return len(self.encoder.encode(text))
    
    def count_messages(self, messages: list[dict]) -> int:
        """Count tokens in a list of messages."""
        total = 0
        for msg in messages:
            total += 4  # Message overhead tokens
            for key, value in msg.items():
                total += len(self.encoder.encode(str(value)))
                if key == "name":
                    total -= 1  # Name tokens are cheaper
        total += 2  # Conversation overhead
        return total
    
    def truncate_to_budget(self, text: str, max_tokens: int) -> str:
        """Truncate text to fit within token budget."""
        tokens = self.encoder.encode(text)
        if len(tokens) <= max_tokens:
            return text
        
        truncated = tokens[:max_tokens]
        return self.encoder.decode(truncated) + "..."
    
    def estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        pricing = {
            "gpt-4o": {"input": 2.50, "output": 10.00},
            "gpt-4o-mini": {"input": 0.15, "output": 0.60},
            "o1": {"input": 15.00, "output": 60.00},
            "o1-mini": {"input": 3.00, "output": 12.00},
        }
        p = pricing.get(model, {"input": 0, "output": 0})
        return (input_tokens / 1e6 * p["input"]) + (output_tokens / 1e6 * p["output"])

tm = TokenManager()

# Count tokens before sending
prompt = "Explain Python decorators with examples"
token_count = tm.count_tokens(prompt)
print(f"Prompt tokens: {token_count}")
print(f"Estimated cost (gpt-4o): ${tm.estimate_cost('gpt-4o', token_count, 500):.6f}")
print(f"Estimated cost (gpt-4o-mini): ${tm.estimate_cost('gpt-4o-mini', token_count, 500):.6f}")

Budget Control

Per-Request Budget

class BudgetController:
    def __init__(self, daily_budget: float = 10.0):
        self.daily_budget = daily_budget
        self.spent: dict[str, float] = {}  # date -> amount
        self.pricing = {
            "gpt-4o": {"input": 2.50, "output": 10.00},
            "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        }
    
    def _today(self) -> str:
        from datetime import date
        return date.today().isoformat()
    
    def can_spend(self, estimated_cost: float) -> bool:
        today = self._today()
        spent_today = self.spent.get(today, 0)
        return spent_today + estimated_cost <= self.daily_budget
    
    def record(self, model: str, input_tokens: int, output_tokens: int):
        p = self.pricing.get(model, {"input": 0, "output": 0})
        cost = (input_tokens / 1e6 * p["input"]) + (output_tokens / 1e6 * p["output"])
        
        today = self._today()
        self.spent[today] = self.spent.get(today, 0) + cost
    
    def remaining_budget(self) -> float:
        today = self._today()
        return max(0, self.daily_budget - self.spent.get(today, 0))
    
    def report(self) -> dict:
        today = self._today()
        return {
            "daily_budget": self.daily_budget,
            "spent_today": self.spent.get(today, 0),
            "remaining": self.remaining_budget(),
            "utilization": (self.spent.get(today, 0) / self.daily_budget) * 100
        }

budget = BudgetController(daily_budget=10.0)

# Before making a call, check budget
estimated_cost = 0.02  # Estimate based on token count
if budget.can_spend(estimated_cost):
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Explain quantum computing"}]
    )
    budget.record("gpt-4o", response.usage.prompt_tokens, response.usage.completion_tokens)
    print(budget.report())
else:
    print("Daily budget exceeded. Using cheaper model or cache.")

Cost Monitoring Dashboard

from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class CostMetric:
    timestamp: float
    model: str
    input_tokens: int
    output_tokens: int
    cost: float
    endpoint: str
    user_id: str = ""

class CostMonitor:
    def __init__(self):
        self.metrics: list[CostMetric] = []
    
    def record(self, model: str, input_tokens: int, output_tokens: int, endpoint: str = "chat", user_id: str = ""):
        pricing = {
            "gpt-4o": (2.50, 10.00),
            "gpt-4o-mini": (0.15, 0.60),
            "o1": (15.00, 60.00),
        }
        p = pricing.get(model, (0, 0))
        cost = (input_tokens / 1e6 * p[0]) + (output_tokens / 1e6 * p[1])
        
        self.metrics.append(CostMetric(
            timestamp=time.time(),
            model=model,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            cost=cost,
            endpoint=endpoint,
            user_id=user_id
        ))
    
    def summary(self, hours: int = 24) -> dict:
        cutoff = time.time() - hours * 3600
        recent = [m for m in self.metrics if m.timestamp >= cutoff]
        
        if not recent:
            return {"total_cost": 0, "total_calls": 0}
        
        by_model = {}
        by_user = {}
        for m in recent:
            by_model[m.model] = by_model.get(m.model, 0) + m.cost
            if m.user_id:
                by_user[m.user_id] = by_user.get(m.user_id, 0) + m.cost
        
        return {
            "period_hours": hours,
            "total_cost": sum(m.cost for m in recent),
            "total_calls": len(recent),
            "total_input_tokens": sum(m.input_tokens for m in recent),
            "total_output_tokens": sum(m.output_tokens for m in recent),
            "avg_cost_per_call": sum(m.cost for m in recent) / len(recent),
            "by_model": by_model,
            "by_user": by_user,
        }
    
    def alert_if_over(self, threshold: float) -> bool:
        summary = self.summary(hours=1)
        return summary["total_cost"] > threshold

import time
monitor = CostMonitor()

# Record API calls
monitor.record("gpt-4o", 1500, 800, "chat", "user_123")
monitor.record("gpt-4o-mini", 200, 100, "classify", "user_456")

print(json.dumps(monitor.summary(24), indent=2))

FAQ

How much can I save with model routing?

Routing 50% of requests to gpt-4o-mini instead of gpt-4o saves approximately 90% on those requests. If 50% of your traffic is simple tasks (classification, extraction, short answers), you can reduce overall costs by 40-50%. Measure your task distribution to estimate savings.

Is semantic caching worth the embedding cost?

Yes, for high-traffic applications. Embedding a query costs ~$0.00000002 (text-embedding-3-small). A cached gpt-4o response saves ~$0.01-0.05. You need a cache hit rate of only 0.1% to break even. For applications with repeated queries (FAQ bots, documentation search), cache hit rates of 10-30% are common, yielding massive savings.

Should I use the Batch API?

Use the Batch API for any workload that does not need real-time responses. It costs 50% less and handles up to 50,000 requests per batch. Good use cases: bulk classification, data enrichment, content generation, report creation. Not suitable for interactive chat, real-time recommendations, or user-facing responses.

How do I set max_tokens to control costs?

Set max_tokens based on your expected output length. For classification: 10 tokens. For summaries: 200 tokens. For code generation: 1000 tokens. For detailed explanations: 500 tokens. Setting max_tokens too low truncates responses. Setting it too high wastes money if the model generates unnecessary content.

What is the cheapest way to do RAG?

Use gpt-4o-mini for the generation step (not gpt-4o). Use text-embedding-3-small for embeddings (not large). Cache embedding results. Pre-compute embeddings for your document corpus instead of re-embedding on each query. Use pgvector instead of managed vector databases for small-to-medium datasets. Limit retrieved chunks to 3-5 to reduce context tokens.

How do I handle budget overruns?

Implement a BudgetController that tracks daily spending. Check budget before each API call. When budget is low, route to cheaper models or use cached responses. Set up alerts at 50%, 80%, and 100% of budget. Consider per-user budgets for multi-tenant applications. Implement circuit breakers that stop API calls when budget is exhausted.

See Also

• Complete Guide to LLM Application Architecture
• Complete Guide to LLM Evaluation
• Complete Guide to OpenAI API Mastery
• Complete Guide to GraphQL Caching
• Complete Guide to Application-Level Caching