LLM Cost Optimization: Eight Production Techniques for 50-90% Savings

TL;DR — LLM cost optimization reduces inference spend 50-90% without quality degradation. Prodinit: "Eight techniques ranked by cost impact — semantic caching + model routing alone cut spend 47-80%." AIStackInsights: "An LLM Gateway is the single highest-ROI change you can make. It does five things: cache, classify, route, validate, govern." TrueFoundry: "Combined, they routinely deliver 50-80% cost reduction on production LLM workloads." HLD Handbook: "Five levers: prompt caching (50-90% discount), semantic caching (bypass model entirely), model routing (cheap models for easy queries), cascading (up to 98% savings), context compression (2-20x reduction)." CalibreOS: "The four-layer cost architecture: caching → routing → compression → batching." Zylos: "Teams applying the full stack consistently report 60-80% reductions in token spend." Implementation sequence: prompt caching (free, 1 day) → batch API (50% discount, 1 day) → semantic caching (20-40%, 2-4 weeks) → model routing (40-70%, 2-4 weeks) → prompt compression (2-20x, 1 week) → cascade (up to 98%, 4-8 weeks) → distillation/quantization (4-8 weeks). Integrate with AI cost trap, enterprise TCO, semantic caching, and LLM metering.

CalibreOS frames the problem: "A consumer LLM product at moderate scale, say 100K daily active users, 5 queries each, burns through 500K queries times ~3K tokens times $5/M tokens = ~$7,500/day = ~$2.7M/year on a frontier model. At 1M DAU that's $27M/year. These numbers are not theoretical: companies with public LLM products routinely report monthly inference bills in the $1M-$10M range."

AIStackInsights identifies the architectural solution: "This is the most common production failure in AI engineering today, and it has nothing to do with prompts, models, or fine-tuning. It is an architecture problem. You are missing the LLM Gateway — a routing and caching layer that sits between your application and your model providers."

LLM Gateway Architecture

Eight Techniques Ranked by Impact

Gateway Metrics Dashboard

Source: AIStackInsights

Implementation

import hashlib
import json
import time
from dataclasses import dataclass, field
from typing import Optional, Literal

@dataclass
class LLMGateway:
    """AI Gateway: routing, caching, compression, governance."""

    # Model tiers (cheapest to most expensive)
    model_tiers = [
        {"name": "claude-3-haiku", "input": 0.25, "output": 1.25,
         "min_prefix": 1024},
        {"name": "claude-3-sonnet", "input": 3.00, "output": 15.00,
         "min_prefix": 1024},
        {"name": "claude-3-opus", "input": 15.00, "output": 75.00,
         "min_prefix": 4096},
    ]

    # Cache config
    cache_threshold: float = 0.10
    cache_ttl: int = 86400

    # Budget limits (per tenant)
    default_daily_budget: float = 100.0  # $100/day per tenant

    def __init__(self, redis, embedding_model, llm_clients, audit):
        self.redis = redis
        self.embeddings = embedding_model
        self.llm = llm_clients  # dict of model_name → client
        self.audit = audit
        self.metrics = {
            "cache_hits": 0, "cache_misses": 0,
            "tier_distribution": {"tier1": 0, "tier2": 0, "tier3": 0},
            "escalations": 0, "total_cost": 0.0,
            "budget_rejections": 0,
        }

    async def complete(self, tenant_id: str, user_id: str,
                       query: str, system_prompt: str = "",
                       max_tokens: int = 1000,
                       batch: bool = False) -> dict:
        """Gateway-managed LLM call with all optimizations."""

        # 1. Budget check
        if not await self._check_budget(tenant_id, user_id):
            self.metrics["budget_rejections"] += 1
            return {"error": "budget_exceeded",
                    "daily_budget": self.default_daily_budget}

        # 2. Semantic cache check
        cached = await self._check_cache(tenant_id, query)
        if cached:
            self.metrics["cache_hits"] += 1
            return {"response": cached, "model": "cache",
                    "cost": 0, "cache_hit": True}

        self.metrics["cache_misses"] += 1

        # 3. Classify query complexity
        complexity = self._classify_complexity(query, system_prompt)

        # 4. Route to appropriate tier (cascade)
        for tier_idx in range(complexity, len(self.model_tiers)):
            model = self.model_tiers[tier_idx]

            # 5. Optimize prompt
            optimized = await self._optimize_prompt(
                system_prompt, query, model)

            # 6. Call LLM
            response = await self.llm[model["name"]].complete(
                system=optimized["system"],
                prompt=optimized["user"],
                max_tokens=max_tokens,
                cache_prefix=True,
                batch=batch,
            )

            # 7. Validate response quality
            if self._validate_response(response["text"], complexity):
                # Success — store in cache and return
                await self._store_cache(tenant_id, query, response["text"])

                cost = self._calculate_cost(
                    model, response["input_tokens"],
                    response["output_tokens"], optimized["cached_tokens"])

                await self._record_spend(tenant_id, user_id, cost)

                tier_key = f"tier{tier_idx + 1}"
                self.metrics["tier_distribution"][tier_key] += 1
                self.metrics["total_cost"] += cost
                if tier_idx > complexity:
                    self.metrics["escalations"] += 1

                return {
                    "response": response["text"],
                    "model": model["name"],
                    "cost": cost,
                    "cache_hit": False,
                    "tier": tier_idx + 1,
                    "escalated": tier_idx > complexity,
                }

            # Quality check failed — escalate to next tier

        # All tiers exhausted
        return {"error": "all_tiers_exhausted"}

    def _classify_complexity(self, query: str,
                             system_prompt: str) -> int:
        """Classify query complexity → starting tier index."""
        query_lower = query.lower()

        # Tier 0 (cheap): short, simple tasks
        if (len(query) < 100 and
            any(w in query_lower for w in [
                "classify", "categorize", "extract", "yes", "no",
                "summarize", "label", "tag"])):
            return 0

        # Tier 0: simple system prompt
        if len(system_prompt) < 500:
            return 0

        # Tier 1 (mid): moderate complexity
        if len(query) < 500 and len(system_prompt) < 2000:
            return 1

        # Tier 2 (frontier): complex reasoning
        return 2

    def _validate_response(self, response: str,
                           expected_tier: int) -> bool:
        """Validate response quality (simplified)."""
        if not response or len(response.strip()) < 10:
            return False
        # In production: use eval harness, LLM-as-judge, or
        # deterministic checks specific to your use case
        return True

    async def _optimize_prompt(self, system_prompt: str,
                               query: str, model: dict) -> dict:
        """Apply prompt compression and identify cacheable prefix."""
        # Compress: remove redundancy
        compressed_system = self._compress(system_prompt)
        compressed_query = self._compress(query)

        # Identify cacheable prefix (stable system prompt)
        cached_tokens = 0
        if len(compressed_system) > model["min_prefix"]:
            cached_tokens = len(compressed_system.split())

        return {
            "system": compressed_system,
            "user": compressed_query,
            "cached_tokens": cached_tokens,
        }

    def _compress(self, text: str) -> str:
        """Compress text by removing redundancy (simplified LLMLingua)."""
        lines = text.split("\n")
        seen = set()
        unique = []
        for line in lines:
            stripped = line.strip().lower()
            if stripped and stripped not in seen:
                seen.add(stripped)
                unique.append(line)
        return "\n".join(unique)

    def _calculate_cost(self, model: dict, input_tokens: int,
                        output_tokens: int,
                        cached_tokens: int) -> float:
        """Calculate cost with prompt caching discount."""
        # Cached tokens: 90% discount (0.1x)
        cached_cost = (cached_tokens / 1_000_000) * model["input"] * 0.1
        # Uncached input tokens
        uncached_input = input_tokens - cached_tokens
        input_cost = (uncached_input / 1_000_000) * model["input"]
        # Output tokens (no caching)
        output_cost = (output_tokens / 1_000_000) * model["output"]

        total = cached_cost + input_cost + output_cost

        # Batch discount
        # (applied at API level, but track for metrics)

        return round(total, 6)

    async def _check_cache(self, tenant_id: str,
                           query: str) -> Optional[str]:
        """Check semantic cache for similar queries."""
        embedding = await self.embeddings.embed(query)
        results = await self.redis.vector_search(
            key=f"llm_cache:{tenant_id}",
            embedding=embedding, count=1)

        if results and results[0]["score"] >= self.cache_threshold:
            cached = await self.redis.get(
                f"llm_cache:{tenant_id}:{results[0]['id']}")
            if cached:
                return json.loads(cached)["response"]
        return None

    async def _store_cache(self, tenant_id: str, query: str,
                           response: str):
        """Store query-response in semantic cache."""
        embedding = await self.embeddings.embed(query)
        cache_id = hashlib.sha256(query.encode()).hexdigest()[:16]

        await self.redis.vector_add(
            key=f"llm_cache:{tenant_id}",
            id=cache_id, embedding=embedding)
        await self.redis.setex(
            f"llm_cache:{tenant_id}:{cache_id}",
            self.cache_ttl,
            json.dumps({"response": response, "query": query}))

    async def _check_budget(self, tenant_id: str,
                            user_id: str) -> bool:
        """Check if user is within daily budget."""
        key = f"budget:{tenant_id}:{user_id}:{time.strftime('%Y-%m-%d')}"
        spent = float(await self.redis.get(key) or 0)
        return spent < self.default_daily_budget

    async def _record_spend(self, tenant_id: str, user_id: str,
                            cost: float):
        """Record spend for budget tracking."""
        key = f"budget:{tenant_id}:{user_id}:{time.strftime('%Y-%m-%d')}"
        await self.redis.incrbyfloat(key, cost)
        await self.redis.expire(key, 172800)  # 48 hour TTL

    def get_metrics(self) -> dict:
        """Get gateway metrics for observability."""
        total = self.metrics["cache_hits"] + self.metrics["cache_misses"]
        hit_rate = (self.metrics["cache_hits"] / total * 100
                    if total > 0 else 0)
        return {
            "cache_hit_rate": round(hit_rate, 1),
            "tier_distribution": self.metrics["tier_distribution"],
            "escalation_rate": round(
                self.metrics["escalations"] / max(total, 1) * 100, 1),
            "total_cost": round(self.metrics["total_cost"], 2),
            "budget_rejections": self.metrics["budget_rejections"],
        }

LLM Cost Optimization Checklist

• [ ] Instrument current spend — track cost per call, per team, per feature
• [ ] Implement per-tenant or per-team cost attribution
• [ ] Deploy AI gateway as single control layer between apps and providers
• [ ] Enable provider-native prompt caching (OpenAI automatic, Anthropic explicit)
• [ ] Structure prompts with stable prefix at top for cache eligibility
• [ ] Set max_tokens to limit output length
• [ ] Use batch API for offline workloads (50% discount, 24h SLA)
• [ ] Implement semantic caching with embedding similarity
• [ ] Choose cache threshold based on use case (0.05 strict to 0.25 loose)
• [ ] Set 24-hour TTL for cache entries
• [ ] Use GPTCache or Redis vector search for semantic caching
• [ ] Implement model routing with complexity classifier
• [ ] Route 60-80% of queries to small models (Haiku, Mini, 8B)
• [ ] Implement cascade: cheapest first, escalate on quality failure
• [ ] Log escalations as training data for small model fine-tuning
• [ ] Implement prompt compression with LLMLingua or structured formats
• [ ] Convert prose to YAML/JSON for lossless compression
• [ ] Remove filler words and redundant instructions
• [ ] Implement response validation (eval harness, LLM-as-judge, deterministic)
• [ ] Set per-user daily budget limits
• [ ] Track budget rejection rate
• [ ] Implement failover chain (primary → secondary → local model)
• [ ] Use token-based rate limiting (not just request count)
• [ ] Apply dual-layer limiting: request RPM + token TPM
• [ ] Monitor gateway metrics: cache hit rate, tier distribution, escalation rate
• [ ] Track cost per request, p95 latency, budget rejection rate
• [ ] A/B test each optimization layer against previous baseline
• [ ] Consider knowledge distillation for stable high-volume tasks
• [ ] Consider quantization for self-hosted open-source models
• [ ] Consider token-budget pool routing for self-hosted vLLM fleets
• [ ] Use continuous batching for self-hosted inference (vLLM)
• [ ] Right-size GPU instances and implement autoscaling
• [ ] Calculate enterprise AI TCO with optimizations
• [ ] Avoid AI cost traps from day one
• [ ] Use semantic answer caching for RAG
• [ ] Implement LLM usage metering for billing
• [ ] Track per-tenant cost for multi-tenant
• [ ] Set AI budget caps to prevent runaway costs
• [ ] Track AI usage for cost attribution
• [ ] Consider BYOK to pass API costs to customers
• [ ] Log all gateway decisions in audit logs
• [ ] Test: each optimization layer preserves output quality
• [ ] Test: cascade escalation produces correct results
• [ ] Test: budget limits prevent overspend
• [ ] Test: failover chain maintains availability
• [ ] Document gateway architecture and optimization stack

FAQ

What is LLM cost optimization?

LLM cost optimization is the practice of reducing AI inference expenditure without degrading output quality. LeanLM: "LLM costs can be reduced 50-90% using five techniques — model routing, semantic caching, knowledge distillation, prompt compression, and quantization. The goal is to pay for the intelligence each task actually requires, not the maximum available." Prodinit: "The three highest-impact techniques — model routing, semantic caching, and prompt prefix caching — deliver 40-90% savings on the token categories they address. Applied together, they produce the 47-80% total cost reduction most teams are targeting." CalibreOS: "Cost is the constraint that separates we built a demo from we shipped a product." The four-layer cost architecture: caching to routing to compression to batching.

What is an AI gateway and why do you need one?

An AI gateway is a routing and caching layer that sits between your application and model providers, intercepting every LLM request. AIStackInsights: "It is the single highest-ROI change you can make to a production AI system. Your application code never picks a model directly. It calls gateway.complete(query) and the gateway makes the decision." TrueFoundry: "Without something sitting between your applications and your model providers, every optimization lives in application code, duplicated across every team, with no governance and no observability." Five functions: (1) Cache — return stored response for semantically equivalent queries. (2) Classify — assess query complexity. (3) Route — pick cheapest capable model. (4) Validate — check response, escalate on low confidence. (5) Govern — enforce per-user budgets, redact PII, log traces. TrueFoundry: "Combined, they routinely deliver 50-80% cost reduction on production LLM workloads."

How does cascade model routing work?

Cascade model routing tries the cheapest model first and escalates to more expensive models only when the cheaper model's output fails a quality check. HLD Handbook: "Cascading: try small first, escalate on low confidence for up to 98% savings at matched quality." Zylos: "A well-implemented cascade system that routes 90% of queries to cheaper models while reserving the expensive tier for genuinely complex tasks can achieve 87% cost reduction." Implementation: (1) Semantic cache check — return cached response (100% savings). (2) Complexity classification — route simple tasks to lightweight models. (3) Escalation on failure — if cheaper model output fails quality check, retry with next tier. AIStackInsights: "Every escalation from Tier 1 to Tier 2 is logged as training data. Periodic fine-tunes of the small model close the gap; tier mix shifts down on its own."

How do you implement prompt caching and compression together?

Prompt caching and compression are complementary layers. HLD Handbook: "Prompt caching: 50 to 90% input discount, zero quality risk. Context compression: 2 to 20x token reduction." Prompt caching: Anthropic offers 90% discount on cached reads (0.1x cost), writes cost 1.25x, 5-minute TTL. Minimum cacheable prefix: 1,024 tokens for Sonnet, 4,096 for Opus. Break-even: two cache hits pay for one write. OpenAI offers 50% discount automatically. Prompt compression: (1) Lossless — convert prose to YAML/JSON, remove redundancy. (2) Lossy — LLMLingua uses small model to identify and remove low-information tokens by perplexity scoring. SitePoint: "Microsoft Research LLMLingua evaluates token-level perplexity and prunes low-information tokens while preserving semantic integrity." Apply order: compress first (fewer tokens), then cache the compressed prefix (90% discount on compressed tokens).

What is the optimal sequence for implementing LLM cost optimization?

The optimal sequence starts with zero-risk, low-effort techniques and progresses to higher-investment ones. Prodinit: "Start with low-complexity wins — prompt caching, batch inference, and output length control deliver significant savings with minimal architectural change. Model routing and semantic caching require more infrastructure but produce the largest absolute cost reductions." HLD Handbook: "Start with zero-risk levers (prompt caching, max_tokens, batch API for offline), then add semantic caching with conservative thresholds, then model routing with eval gates, then cascading with deterministic verifiers." Recommended sequence: (1) Prompt prefix caching — free, provider-native, 1 day. (2) Output length control (max_tokens) — 1 day. (3) Batch API for offline workloads — 50% discount, 1 day. (4) Semantic caching — 20-40% savings, 2-4 weeks. (5) Model routing — 40-70% savings, 2-4 weeks. (6) Prompt compression — 2-20x, 1 week. (7) Cascade with escalation — up to 98% savings, 4-8 weeks. (8) Knowledge distillation / quantization — for stable high-volume tasks, 4-8 weeks. Each layer requires an A/B test against the previous baseline.

Want a self-hosted AI company brain that does all of this out of the box?
Book a demo →