AI Engineer Portal

Cost, latency, and failure budgets

Why this matters

AI features have an operational cost profile that is fundamentally different from traditional software. A slow database query might add 20ms to a request. A slow LLM call adds 2–10 seconds. A single GPT-4o call can cost more than 1,000 database queries. If you do not actively manage cost, latency, and failure rates, they will manage you — usually in the form of a surprising invoice or a degraded user experience that takes weeks to diagnose.

Senior AI engineers think in budgets. Every feature has a cost-per-request budget, a latency budget (what the UX can tolerate), and a failure budget (how many errors per hour is acceptable before alerting). These budgets are engineering constraints, not afterthoughts.

Core concepts

Token cost calculation. Cost is a function of model, token counts, and current provider pricing. Build a cost calculator into your application layer:

# Representative pricing (check provider docs for current rates)
COST_PER_MILLION_TOKENS = {
    "claude-haiku-4-5":   {"input": 0.80,  "output": 4.00},
    "claude-sonnet-4-5":  {"input": 3.00,  "output": 15.00},
    "gpt-4o-mini":        {"input": 0.15,  "output": 0.60},
    "gpt-4o":             {"input": 2.50,  "output": 10.00},
    "llama-3.1-70b":      {"input": 0.00,  "output": 0.00},  # self-hosted
}

def calculate_cost_usd(model: str, input_tokens: int, output_tokens: int) -> float:
    rates = COST_PER_MILLION_TOKENS.get(model)
    if not rates:
        return 0.0
    return (input_tokens * rates["input"] + output_tokens * rates["output"]) / 1_000_000

Model routing. Not every request needs your best model. A routing layer saves significant cost:

Use the cheap model (Haiku, GPT-4o mini) for: short simple queries, classification tasks, reformatting, summarization of clear text
Use the expensive model (Sonnet, GPT-4o) for: complex reasoning, code generation, multi-step analysis, tasks where quality directly impacts revenue

Route based on measurable signals: input length, task type, user tier, or even a lightweight classifier.

Latency budgets. Break down where time goes in your feature:

Stage	Typical range	Target
Context assembly	5–50ms	<50ms
Provider call (non-streaming)	1–15s	<8s
Response parsing + validation	1–10ms	<20ms
Persistence	5–30ms	<50ms

Streaming for perceived speed. For conversational UIs, streaming changes the user experience dramatically. Instead of waiting 5 seconds for a complete response, the user sees text appearing within 300ms. Implement streaming as early as possible for any user-facing text generation feature.

Failure budgets. Define error rate thresholds before you ship:

Acceptable: <0.5% of requests fail after retries
Warning: 0.5–2% failure rate — investigate
Critical: >2% failure rate — alert on-call

Track failures by error type (provider errors, timeout, validation failure) separately. A spike in validation failures means your prompt changed and broke the output format. A spike in 429 errors means you are hitting rate limits and need to throttle or upgrade your tier.

Working example

A cost tracker middleware that logs tokens and alerts on budget overruns:

import time
import threading
from dataclasses import dataclass, field
from collections import defaultdict
from typing import Callable, Optional
import logging

logger = logging.getLogger(__name__)


@dataclass
class CostRecord:
    model: str
    input_tokens: int
    output_tokens: int
    cost_usd: float
    latency_ms: int
    timestamp: float = field(default_factory=time.time)
    feature: str = "unknown"


class CostTracker:
    """Thread-safe cost tracker with budget enforcement."""

    def __init__(
        self,
        daily_budget_usd: float = 10.0,
        alert_callback: Optional[Callable[[float, float], None]] = None,
    ):
        self.daily_budget_usd = daily_budget_usd
        self.alert_callback = alert_callback or self._default_alert
        self._records: list[CostRecord] = []
        self._lock = threading.Lock()

    def _default_alert(self, spent: float, budget: float) -> None:
        logger.warning(f"Cost alert: ${spent:.4f} spent of ${budget:.2f} daily budget")

    def record(self, record: CostRecord) -> None:
        with self._lock:
            self._records.append(record)
            daily_spent = self._daily_total_locked()
            if daily_spent >= self.daily_budget_usd * 0.8:
                self.alert_callback(daily_spent, self.daily_budget_usd)

    def _daily_total_locked(self) -> float:
        cutoff = time.time() - 86400  # last 24 hours
        return sum(r.cost_usd for r in self._records if r.timestamp >= cutoff)

    def daily_total(self) -> float:
        with self._lock:
            return self._daily_total_locked()

    def report(self) -> dict:
        with self._lock:
            by_model: dict = defaultdict(lambda: {"calls": 0, "input_tokens": 0, "output_tokens": 0, "cost_usd": 0.0, "latency_ms": []})
            cutoff = time.time() - 86400
            for r in self._records:
                if r.timestamp < cutoff:
                    continue
                m = by_model[r.model]
                m["calls"] += 1
                m["input_tokens"] += r.input_tokens
                m["output_tokens"] += r.output_tokens
                m["cost_usd"] += r.cost_usd
                m["latency_ms"].append(r.latency_ms)

            return {
                model: {
                    **stats,
                    "avg_latency_ms": int(sum(stats["latency_ms"]) / len(stats["latency_ms"])) if stats["latency_ms"] else 0,
                    "p95_latency_ms": int(sorted(stats["latency_ms"])[int(len(stats["latency_ms"]) * 0.95)]) if len(stats["latency_ms"]) > 1 else 0,
                    "latency_ms": None,  # exclude raw list from report
                }
                for model, stats in by_model.items()
            }

    def is_over_budget(self) -> bool:
        return self.daily_total() >= self.daily_budget_usd


# Global tracker instance (one per application)
tracker = CostTracker(daily_budget_usd=50.0)


def tracked_llm_call(
    call_fn: Callable,
    model: str,
    feature: str = "unknown",
    **call_kwargs,
):
    """Wrapper that instruments any LLM call with cost and latency tracking."""
    if tracker.is_over_budget():
        raise RuntimeError(
            f"Daily budget of ${tracker.daily_budget_usd:.2f} exceeded. "
            "Request blocked."
        )

    start = time.monotonic()
    response = call_fn(**call_kwargs)
    elapsed_ms = int((time.monotonic() - start) * 1000)

    # Extract token usage from response (works for both OpenAI and Anthropic shapes)
    usage = getattr(response, "usage", None)
    if usage:
        input_tokens = getattr(usage, "input_tokens", 0) or getattr(usage, "prompt_tokens", 0)
        output_tokens = getattr(usage, "output_tokens", 0) or getattr(usage, "completion_tokens", 0)
    else:
        input_tokens = output_tokens = 0

    cost = calculate_cost_usd(model, input_tokens, output_tokens)
    tracker.record(CostRecord(
        model=model,
        input_tokens=input_tokens,
        output_tokens=output_tokens,
        cost_usd=cost,
        latency_ms=elapsed_ms,
        feature=feature,
    ))

    return response

The tracker is thread-safe, covers 24-hour rolling windows, alerts at 80% budget utilization, and produces a per-model report you can expose on an internal admin dashboard.

Common mistakes

No cost visibility until the invoice. By the time a monthly billing invoice shows an unexpected number, weeks of data are gone. Track cost per request in application logs from day one.
Using your most powerful model for everything. A feature that runs Sonnet on every keypress in an autocomplete widget will burn money fast. Match model capability to task complexity.
No latency budget on the user-facing path. A 12-second response is technically correct but practically broken. Set a hard timeout on the user-facing call and either stream the response or show a loading state.
Tracking only failures, not error rates. Absolute failure counts tell you nothing about whether the error rate is acceptable. Track failures as a percentage of total requests.
Ignoring output token growth. Input tokens are predictable (you control the context). Output tokens are not. If a prompt change or a new use case causes the model to produce much longer responses, your cost doubles without an obvious cause. Monitor average output token counts as a separate metric.

Try it yourself

Extend the CostTracker to support per-feature budgets. Different features in your app should have separate daily budgets — the code review assistant gets $20/day, the email drafter gets $5/day. When a feature exceeds its budget, log a warning and fall back to a cheaper model instead of blocking the request entirely.