Why this matters

LLM API costs scale with usage in a way that traditional compute costs do not. A popular feature that calls GPT-4o for every request can produce a $50,000 monthly bill before the team realizes what happened. The engineering response is not to avoid capable models — it is to use the right model for each request type, cache responses that can be reused, and enforce budget boundaries that trigger before costs spiral.

Core concepts

Token budgeting

A token budget is a per-request or per-session constraint that caps how much of the context window you use. This matters for cost because LLM pricing is linear with tokens: a request with 8,000 prompt tokens costs 8x a request with 1,000 prompt tokens.

from dataclasses import dataclass


@dataclass
class TokenBudget:
    max_prompt_tokens: int
    max_completion_tokens: int
    reserved_system_tokens: int = 200

    @property
    def available_context_tokens(self) -> int:
        return self.max_prompt_tokens - self.reserved_system_tokens


def fit_context_to_budget(chunks: list[str], budget: TokenBudget) -> list[str]:
    '''Greedily select chunks that fit within the token budget.'''
    # Approximate: 1 token ≈ 4 characters
    char_budget = budget.available_context_tokens * 4
    selected = []
    used = 0
    for chunk in chunks:
        chunk_chars = len(chunk)
        if used + chunk_chars > char_budget:
            break
        selected.append(chunk)
        used += chunk_chars
    return selected

The reserved_system_tokens prevents the system prompt from being squeezed out by context chunks.

Exact-match caching

The cheapest LLM call is one that never happens. For requests where the input is identical — the same user asking the same question again, a documentation lookup, a template rendering — an exact-match cache returns the stored response instantly.

import hashlib
import json
import time
from dataclasses import dataclass
import redis


@dataclass
class CachedResponse:
    content: str
    model: str
    cached_at: float
    hit_count: int


class ExactMatchCache:
    def __init__(self, redis_client: redis.Redis, ttl_seconds: int = 3600):
        self.redis = redis_client
        self.ttl = ttl_seconds

    def _key(self, messages: list[dict], model: str) -> str:
        payload = json.dumps({"messages": messages, "model": model}, sort_keys=True)
        return f"llm:exact:{hashlib.sha256(payload.encode()).hexdigest()}"

    def get(self, messages: list[dict], model: str) -> CachedResponse | None:
        key = self._key(messages, model)
        raw = self.redis.get(key)
        if raw is None:
            return None
        data = json.loads(raw)
        data["hit_count"] += 1
        self.redis.setex(key, self.ttl, json.dumps(data))
        return CachedResponse(**data)

    def set(self, messages: list[dict], model: str, content: str) -> None:
        key = self._key(messages, model)
        data = {
            "content": content,
            "model": model,
            "cached_at": time.time(),
            "hit_count": 0,
        }
        self.redis.setex(key, self.ttl, json.dumps(data))

Exact-match caching works well for FAQ-style features, help documentation, and any use case where user inputs cluster around a small set of common questions.

Semantic caching

Semantic caching uses embedding similarity to return cached responses for semantically equivalent questions, even if they are phrased differently. "What is your return policy?" and "How do I return something?" might be different strings but should return the same cached answer.

import numpy as np
from dataclasses import dataclass, field


@dataclass
class SemanticCacheEntry:
    question: str
    answer: str
    embedding: list[float]
    model_used: str


class SemanticCache:
    def __init__(self, similarity_threshold: float = 0.92):
        self.threshold = similarity_threshold
        self._entries: list[SemanticCacheEntry] = []

    def _cosine_similarity(self, a: list[float], b: list[float]) -> float:
        va, vb = np.array(a), np.array(b)
        denom = np.linalg.norm(va) * np.linalg.norm(vb)
        return float(np.dot(va, vb) / denom) if denom > 0 else 0.0

    def get(self, query_embedding: list[float]) -> SemanticCacheEntry | None:
        best_score = 0.0
        best_entry = None
        for entry in self._entries:
            score = self._cosine_similarity(query_embedding, entry.embedding)
            if score > best_score:
                best_score = score
                best_entry = entry
        if best_score >= self.threshold:
            return best_entry
        return None

    def set(self, question: str, answer: str, embedding: list[float], model: str) -> None:
        self._entries.append(SemanticCacheEntry(
            question=question,
            answer=answer,
            embedding=embedding,
            model_used=model,
        ))

The threshold controls the tradeoff between cache hit rate and answer relevance. 0.92 is a reasonable starting point; calibrate against your specific use case.

Model routing for cost optimization

Not every request needs your most expensive model. A model router inspects the incoming request and routes it to the appropriate tier:

from enum import Enum


class TaskComplexity(Enum):
    simple = "simple"      # factual lookup, short answer, classification
    moderate = "moderate"  # summarization, extraction, structured output
    complex = "complex"    # reasoning, code generation, long-form writing


MODEL_TIERS = {
    TaskComplexity.simple: "gpt-4o-mini",      # ~$0.00015/1k tokens
    TaskComplexity.moderate: "gpt-4o-mini",    # same — handle with prompt
    TaskComplexity.complex: "gpt-4o",          # ~$0.0025/1k tokens
}


def classify_complexity(prompt: str, context_tokens: int) -> TaskComplexity:
    '''Heuristic classifier — replace with a real classifier in production.'''
    if context_tokens > 6000:
        return TaskComplexity.complex
    if len(prompt) < 120 and "?" in prompt:
        return TaskComplexity.simple
    return TaskComplexity.moderate


def route_request(prompt: str, context_tokens: int) -> str:
    complexity = classify_complexity(prompt, context_tokens)
    return MODEL_TIERS[complexity]

A more sophisticated version uses a small, cheap classifier model to make the routing decision, costing fractions of a cent to avoid spending dollars on an expensive model.

Monitoring spend

Track token usage and cost per request, per feature, and per user. This data is essential for spotting runaway costs before the monthly bill arrives.

from dataclasses import dataclass, field
from collections import defaultdict


# Cost per 1000 tokens (input/output) — update as pricing changes
MODEL_COSTS = {
    "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
    "gpt-4o": {"input": 0.0025, "output": 0.01},
    "claude-3-5-haiku-20241022": {"input": 0.0008, "output": 0.004},
    "claude-sonnet-4-5": {"input": 0.003, "output": 0.015},
}


@dataclass
class CostEvent:
    feature: str
    model: str
    input_tokens: int
    output_tokens: int

    @property
    def cost_usd(self) -> float:
        if self.model not in MODEL_COSTS:
            return 0.0
        rates = MODEL_COSTS[self.model]
        return (
            self.input_tokens / 1000 * rates["input"] +
            self.output_tokens / 1000 * rates["output"]
        )


class CostTracker:
    def __init__(self):
        self._events: list[CostEvent] = []
        self._by_feature: dict[str, float] = defaultdict(float)

    def record(self, event: CostEvent) -> None:
        self._events.append(event)
        self._by_feature[event.feature] += event.cost_usd

    def report(self) -> dict:
        return {
            "total_usd": sum(e.cost_usd for e in self._events),
            "by_feature": dict(self._by_feature),
            "request_count": len(self._events),
        }

Working example

A request handler that applies caching, model routing, and cost tracking together:

async def handle_request(
    query: str,
    feature_name: str,
    cache: ExactMatchCache,
    tracker: CostTracker,
) -> str:
    messages = [{"role": "user", "content": query}]

    # Try exact-match cache first
    cached = cache.get(messages, model="any")
    if cached:
        return cached.content

    # Route to appropriate model tier
    model = route_request(query, context_tokens=len(query) // 4)

    # Call the model
    response = await call_provider(messages, model=model)

    # Cache and track
    cache.set(messages, model, response.content)
    tracker.record(CostEvent(
        feature=feature_name,
        model=model,
        input_tokens=response.input_tokens,
        output_tokens=response.output_tokens,
    ))

    return response.content

Common mistakes

No token budget enforcement. A user who pastes a 50,000-token document into a chat box can generate a $1 request. Set hard limits on prompt construction and truncate before calling the API.

Caching without TTL. A semantic cache entry from three months ago may reference outdated information. Set expiration based on your content freshness requirements.

Model routing without evaluation. Routing "simple" requests to a cheaper model only saves money if quality is maintained. Run your eval suite against the cheaper model before deploying routing.

Tracking tokens without tracking cost. Token counts alone do not tell you your spending rate. Different models have dramatically different prices; track cost in dollars alongside token counts.

No spend alerts. Set a budget alert in your cloud or provider billing console. A missing alert means the first signal of a cost problem is the monthly invoice.

Try it yourself

1. Implement the ExactMatchCache above backed by a Python dict instead of Redis (for local testing). Add a hit_rate() method that returns the fraction of get() calls that returned a cached result.
2. Build a cost tracker that groups by hour and returns hourly cost totals for the last 24 hours.
3. Write a token budget function that, given a list of retrieved chunks and a budget, returns the maximum set of chunks that fits. Assert that the result never exceeds the budget.