Agent Cost Monitoring

Why This Matters

LLM API costs scale linearly with usage—a single GPT-4 call costs 0.06, but at 100K calls/day that becomes 6,000/day (180K/month). Without cost monitoring, a production agent can burn through a budget in hours. Token tracking, budget alerts, and cost optimization aren't optional features—they're survival mechanisms. The difference between a profitable AI product and a money pit is often just visibility into where every dollar goes.

Real-World Analogy

Cost monitoring for AI agents is like monitoring fuel consumption in a fleet of delivery trucks. Each truck (LLM call) consumes fuel (tokens) at different rates depending on load (prompt complexity). Without dashboards showing miles-per-gallon, fuel costs per route, and driver efficiency, you'd never know which trucks are gas guzzlers. A fleet manager who tracks fuel per delivery can optimize routes, switch to efficient trucks for short hauls, and catch fuel theft—just like a cost-conscious agent architect.

Token Tracking System

import time
import logging
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum
from collections import defaultdict

logger = logging.getLogger(__name__)


class ModelTier(Enum):
    PREMIUM = "premium"
    STANDARD = "standard"
    ECONOMY = "economy"


@dataclass(frozen=True)
class ModelPricing:
    model_name: str
    input_cost_per_1k: float
    output_cost_per_1k: float
    tier: ModelTier
    rate_limit_rpm: int = 60


@dataclass
class TokenUsage:
    input_tokens: int
    output_tokens: int
    total_tokens: int
    model: str
    timestamp: float = field(default_factory=time.time)
    request_id: Optional[str] = None
    user_id: Optional[str] = None
    session_id: Optional[str] = None


class TokenTracker:
    def __init__(self) -> None:
        self._pricing: dict[str, ModelPricing] = {}
        self._history: list[TokenUsage] = []
        self._user_costs: dict[str, float] = defaultdict(float)
        self._session_costs: dict[str, float] = defaultdict(float)
        self._daily_costs: dict[str, float] = defaultdict(float)

    def register_model(self, pricing: ModelPricing) -> None:
        self._pricing[pricing.model_name] = pricing

    def track(
        self, model: str, input_tokens: int, output_tokens: int,
        request_id: Optional[str] = None, user_id: Optional[str] = None,
        session_id: Optional[str] = None,
    ) -> float:
        pricing = self._pricing.get(model)
        if not pricing:
            raise ValueError(f"No pricing defined for model: {model}")

        cost = (
            (input_tokens / 1000) * pricing.input_cost_per_1k
            + (output_tokens / 1000) * pricing.output_cost_per_1k
        )
        usage = TokenUsage(
            input_tokens=input_tokens, output_tokens=output_tokens,
            total_tokens=input_tokens + output_tokens, model=model,
            request_id=request_id, user_id=user_id, session_id=session_id,
        )
        self._history.append(usage)

        if user_id:
            self._user_costs[user_id] += cost
        if session_id:
            self._session_costs[session_id] += cost
        today = time.strftime("%Y-%m-%d")
        self._daily_costs[today] += cost
        return cost

    def stats(self, hours: int = 24) -> dict[str, Any]:
        cutoff = time.time() - (hours * 3600)
        recent = [u for u in self._history if u.timestamp > cutoff]
        if not recent:
            return {"total_requests": 0, "total_tokens": 0, "total_cost": 0.0}

        total_cost = 0.0
        cost_by_model: dict[str, float] = defaultdict(float)
        for u in recent:
            p = self._pricing.get(u.model)
            if p:
                cost = (u.input_tokens / 1000) * p.input_cost_per_1k + (u.output_tokens / 1000) * p.output_cost_per_1k
                total_cost += cost
                cost_by_model[u.model] += cost

        return {
            "total_requests": len(recent),
            "total_tokens": sum(u.total_tokens for u in recent),
            "total_cost": total_cost,
            "avg_tokens_per_request": sum(u.total_tokens for u in recent) / len(recent),
            "cost_by_model": dict(cost_by_model),
        }

    def user_cost(self, user_id: str) -> float:
        return self._user_costs.get(user_id, 0.0)

    def daily_cost(self, date: Optional[str] = None) -> float:
        return self._daily_costs.get(date or time.strftime("%Y-%m-%d"), 0.0)

    def predict_monthly(self) -> float:
        return self.daily_cost() * 30

Budget Manager

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

logger = logging.getLogger(__name__)


class AlertLevel(Enum):
    INFO = "info"
    WARNING = "warning"
    CRITICAL = "critical"
    EMERGENCY = "emergency"


@dataclass(frozen=True)
class BudgetConfig:
    daily_limit: float = 100.0
    monthly_limit: float = 3000.0
    per_user_limit: float = 10.0
    alert_thresholds: tuple[float, ...] = (0.5, 0.75, 0.9, 1.0)
    hard_limit_multiplier: float = 1.5


@dataclass(frozen=True)
class Alert:
    level: AlertLevel
    message: str
    current_cost: float
    limit: float
    threshold: float
    timestamp: float = field(default_factory=time.time)


class BudgetManager:
    def __init__(self, config: Optional[BudgetConfig] = None) -> None:
        self._config = config or BudgetConfig()
        self._daily: dict[str, float] = defaultdict(float)
        self._monthly: dict[str, float] = defaultdict(float)
        self._user: dict[str, float] = defaultdict(float)
        self._alerts: list[Alert] = []
        self._callbacks: list[Callable] = []
        self._blocked = False
        self._blocked_until = 0.0

    def on_alert(self, callback: Callable) -> None:
        self._callbacks.append(callback)

    def check(self, cost: float, user_id: Optional[str] = None) -> tuple[bool, list[Alert]]:
        alerts: list[Alert] = []
        today = time.strftime("%Y-%m-%d")
        month = time.strftime("%Y-%m")
        self._daily[today] += cost
        self._monthly[month] += cost
        if user_id:
            self._user[user_id] += cost

        if self._blocked:
            if time.time() < self._blocked_until:
                return False, [Alert(AlertLevel.EMERGENCY, "Budget exceeded, blocked",
                                     self._daily[today], self._config.daily_limit, 1.0)]
            self._blocked = False

        daily = self._daily[today]
        monthly = self._monthly[month]
        for threshold in self._config.alert_thresholds:
            if daily >= self._config.daily_limit * threshold:
                level = AlertLevel.CRITICAL if threshold >= 1.0 else AlertLevel.WARNING
                alerts.append(Alert(level, f"Daily budget {threshold*100:.0f}% reached",
                                    daily, self._config.daily_limit, threshold))
            if monthly >= self._config.monthly_limit * threshold:
                level = AlertLevel.CRITICAL if threshold >= 1.0 else AlertLevel.WARNING
                alerts.append(Alert(level, f"Monthly budget {threshold*100:.0f}% reached",
                                    monthly, self._config.monthly_limit, threshold))

        if user_id and self._user[user_id] >= self._config.per_user_limit:
            alerts.append(Alert(AlertLevel.WARNING, f"User {user_id} per-user limit reached",
                                self._user[user_id], self._config.per_user_limit, 1.0))

        if daily >= self._config.daily_limit * self._config.hard_limit_multiplier:
            self._blocked = True
            self._blocked_until = time.time() + 3600
            alerts.append(Alert(AlertLevel.EMERGENCY, "Hard limit exceeded, blocking 1 hour",
                                daily, self._config.daily_limit, self._config.hard_limit_multiplier))

        for alert in alerts:
            self._alerts.append(alert)
            for cb in self._callbacks:
                try:
                    cb(alert)
                except Exception:
                    pass

        has_critical = any(a.level in (AlertLevel.CRITICAL, AlertLevel.EMERGENCY) for a in alerts)
        return not has_critical, alerts

    def status(self) -> dict[str, Any]:
        today = time.strftime("%Y-%m-%d")
        month = time.strftime("%Y-%m")
        daily = self._daily[today]
        monthly = self._monthly[month]
        return {
            "daily": {"spent": daily, "limit": self._config.daily_limit,
                       "remaining": max(0, self._config.daily_limit - daily),
                       "pct": daily / self._config.daily_limit * 100},
            "monthly": {"spent": monthly, "limit": self._config.monthly_limit,
                         "remaining": max(0, self._config.monthly_limit - monthly),
                         "pct": monthly / self._config.monthly_limit * 100},
            "is_blocked": self._blocked,
        }

Cost Optimization Engine

import hashlib
import time
import logging
from dataclasses import dataclass, field
from typing import Any, Callable, Optional
from enum import Enum

logger = logging.getLogger(__name__)


class OptimizationStrategy(Enum):
    MODEL_ROUTING = "model_routing"
    CACHING = "caching"
    PROMPT_OPTIMIZATION = "prompt_optimization"
    BATCH_PROCESSING = "batch_processing"


@dataclass(frozen=True)
class OptimizationRule:
    strategy: OptimizationStrategy
    condition: Callable
    action: Callable
    priority: int = 1


class CostOptimizer:
    def __init__(self) -> None:
        self._rules: list[OptimizationRule] = []
        self._savings: list[dict] = []
        self._cache: dict[str, dict] = {}
        self._cache_ttl: float = 3600.0

    def add_rule(self, rule: OptimizationRule) -> None:
        self._rules.append(rule)
        self._rules.sort(key=lambda r: r.priority, reverse=True)

    async def optimize(self, query: str, context: dict, model: str) -> tuple[str, dict]:
        applied: list[str] = []
        total_savings = 0.0
        for rule in self._rules:
            try:
                if await rule.condition(query, context, model):
                    query, model, savings = await rule.action(query, context, model)
                    applied.append(rule.strategy.value)
                    total_savings += savings
            except Exception:
                continue
        return query, {
            "optimizations": applied, "savings": total_savings, "model": model,
        }

    def cache_key(self, query: str, model: str) -> str:
        return hashlib.md5(f"{query}:{model}".encode()).hexdigest()

    async def get_cached(self, query: str, model: str) -> Optional[Any]:
        key = self.cache_key(query, model)
        entry = self._cache.get(key)
        if entry and time.time() - entry["ts"] < self._cache_ttl:
            return entry["response"]
        return None

    async def set_cached(self, query: str, model: str, response: Any) -> None:
        self._cache[self.cache_key(query, model)] = {"response": response, "ts": time.time()}

    def stats(self) -> dict[str, Any]:
        if not self._savings:
            return {"total_savings": 0, "count": 0}
        total = sum(s["savings"] for s in self._savings)
        strategies: dict[str, int] = {}
        for s in self._savings:
            for opt in s["optimizations"]:
                strategies[opt] = strategies.get(opt, 0) + 1
        return {"total_savings": total, "count": len(self._savings),
                "strategies": strategies, "avg_per_request": total / len(self._savings)}

Mathematical Foundations

Total Cost of Ownership:

Cost Efficiency Score:

Token Utilization Rate:

Cache Hit Savings:

ROI of Optimization:

Cost per Task:

Performance Considerations

Component	Latency	Overhead	Accuracy
Token counting	<1ms	Negligible	Exact
Cost calculation	<1ms	Negligible	Exact
Budget check	<1ms	Negligible	Exact
Model routing decision	1-5ms	Low	Depends on classifier
Cache lookup	1-10ms	Low	Depends on strategy
Optimization pipeline	5-50ms	Medium	Varies by rules

Security Considerations

Encrypt cost data at rest since it may reveal business metrics and usage patterns.
Access control on cost dashboards—not all users should see organizational spending.
Rate limit cost API endpoints to prevent abuse or information leakage.
Audit budget override events to detect unauthorized limit changes.
Protect API keys used for LLM access—rotate regularly and use secret managers.

Common Pitfalls

Pitfall	Consequence	Solution
No cost tracking from day one	Blind to spending	Implement tracking before launch
Ignoring small costs	Death by a thousand cuts	Aggregate and monitor all expenses
Over-optimization	Degraded quality	Balance cost with quality metrics
No alerting	Surprise bills	Set up multi-level alerts
Ignoring cache opportunities	Unnecessary API calls	Analyze query patterns
Single model usage	Missing cheaper alternatives	Implement model routing
No budget reviews	Unchecked spending	Regular analysis and adjustment
Manual cost reporting	Slow response to issues	Automate with dashboards

Interview Q&A

1. How do you implement token tracking in AI agents?

Token tracking: (1) register model pricing for each LLM provider (input/output per 1K tokens), (2) capture token counts from API responses—most providers return usage.input_tokens and usage.output_tokens, (3) calculate costs using pricing models, (4) track by user, session, and time period using metadata tags, (5) store in a time-series database (InfluxDB, TimescaleDB) for analysis. Implement as middleware that wraps every LLM call to ensure zero漏 tracking.

2. What are effective cost optimization strategies?

Key strategies: (1) model routing—classify query complexity and route simple queries to economy models (GPT-3.5, Haiku) and complex queries to premium (GPT-4, Opus), typically saves 30-50%, (2) caching—store responses for repeated or semantically similar queries, (3) prompt optimization—reduce token count while maintaining quality (shorter system prompts, few-shot examples), (4) batch processing—combine multiple small requests, (5) compression—use structured output formats instead of verbose text. Measure savings and adjust strategies quarterly.

3. How do you set appropriate budget limits?

Budget setting: (1) analyze historical usage patterns for 30+ days, (2) define daily limits based on expected growth (current + 50% buffer), (3) set per-user limits based on access tier (free: 5/day, enterprise: custom), (4) configure alert thresholds (50% = info, 75% = warning, 90% = critical, 100% = emergency), (5) set hard limits with temporary blocking (1 hour after 150% daily), (6) review and adjust monthly. Factor in seasonal patterns and planned feature launches.

4. What metrics should you monitor for cost optimization?

Key metrics: (1) cost per request (trend over time), (2) tokens per request (prompt efficiency), (3) cache hit rate (higher = lower costs), (4) model distribution (% using each model), (5) cost by user/tenant (identifies heavy users), (6) cost by use case (support vs. analysis vs. writing), (7) optimization savings (actual $ saved), (8) ROI of optimizations (savings - implementation cost). Use dashboards (Grafana, Datadog) with real-time alerts for anomalies.

5. How do you handle cost spikes in production?

Spike handling: (1) immediate alerting when thresholds exceeded (PagerDuty, Slack), (2) automatic rate limiting for high-cost users or endpoints, (3) model downgrade to cheaper alternatives during spikes, (4) request queuing during peak times with priority-based processing, (5) emergency budget increase with approval workflow, (6) post-spike analysis to identify root cause (bot traffic? new feature?). Implement circuit breakers that temporarily block requests when costs exceed hard limits.

6. What is model routing and how does it reduce costs?

Model routing selects the appropriate LLM based on task complexity: (1) classify incoming queries using a lightweight classifier (rules or small ML model), (2) simple queries (FAQ, greetings) → economy models at 15/1M tokens, (4) creative tasks → models optimized for creativity. Implementation: maintain a model capabilities matrix, route based on requirements. Typically reduces costs 30-50% with <5% quality impact when done well.

7. How do you measure the ROI of cost optimizations?

ROI measurement: (1) establish baseline costs over 30 days before optimization, (2) implement optimization (model routing, caching, prompt engineering), (3) measure actual savings over 30 days, (4) calculate implementation cost (engineering time + infrastructure), (5) compute ROI = (Savings - Cost) / Cost. Consider: direct token savings, indirect benefits (faster response → better UX → higher retention), and ongoing maintenance costs. Track ROI quarterly and adjust strategies.

8. How do you handle cost allocation across teams or tenants?

Cost allocation: (1) tag all requests with metadata (user_id, team, project, use_case), (2) calculate costs per tag in the token tracker, (3) generate chargeback reports (weekly/monthly), (4) set budgets per team/tenant in the budget manager, (5) provide self-service cost visibility dashboards per team, (6) automate billing integration for SaaS (Stripe, Billing). Use cost centers for internal allocation and tenant-level tracking for multi-tenant products.

KnowledgeCheck

What is the primary purpose of token tracking?
- a) Improve response latency
- b) Monitor and optimize costs
- c) Increase model accuracy
- d) Reduce memory usage
What is model routing?
- a) Selecting models randomly
- b) Routing queries to appropriate models based on complexity
- c) Loading models dynamically
- d) Managing model versions
What is a common cost optimization strategy?
- a) Using only premium models
- b) Caching responses for repeated queries
- c) Increasing token limits
- d) Removing all constraints
What alert threshold should trigger immediate action?
- a) 50% of budget
- b) 75% of budget
- c) 90% of budget
- d) 100% of budget
How is Cost Efficiency Score calculated?
- a) Total cost / Value delivered
- b) Value delivered / Total cost
- c) Tokens used / Cost
- d) Requests / Cost
What is the benefit of per-user cost tracking?
- a) Reduces latency
- b) Enables fair allocation and billing
- c) Increases accuracy
- d) Simplifies code

Answers: 1-b, 2-b, 3-b, 4-c, 5-b, 6-b

Agent Cost Monitoring: Token Tracking, Budget Alerts & Optimization