AI Usage Tracking: Token Monitoring, Cost Attribution, and Enterprise FinOps

TL;DR — AI usage tracking provides per-user, per-team, per-feature cost attribution for LLM spend. Cloudflare: "You cannot calculate ROI on your AI spend without visibility on what you are spending. Every other line item in a business has a budget and per-team attribution and AI spend should be no different." OpenObserve: "Provider billing dashboards give monthly aggregates by API key — they tell you how much you spent, not why. Actionable cost attribution requires instrumented telemetry inside your application." LiteLLM: "Track spend for keys, users, and teams across 100+ LLMs." Behest: "Per-tenant, per-project, per-user attribution. Token budgets enforced inline. Stop overruns before the invoice lands." FinOps Foundation: "Helps predict and control costs for LLMs, which charge per token." Key components: token-level telemetry (input, output, model, cost per request), per-user attribution (JWT identity on every call), per-team chargeback (IdP groups, cost centers), real-time budget enforcement (block or downgrade), cost analytics dashboard (query by model, user, team, feature). Integrate with LLM metering, per-tenant cost tracking, LLM cost optimization, and budget caps.

OpenObserve identifies the core problem: "The root cause is structural. LLM provider billing dashboards give you monthly aggregates by API key. They tell you how much you spent. They do not tell you why you spent it. That requires LLM observability instrumented inside your own application, shipping structured telemetry into a platform that can query it at the span level."

Traceloop describes the solution: "Moving from reactive panic at monthly bills to proactive control of your LLM spend is not only possible but essential. The solution lies in shifting your perspective from the total bill to the individual request. By implementing a FinOps strategy based on granular attribution, tagging every request with user, team, project, and cost center."

AI Usage Tracking Architecture

Telemetry Signals for LLM Cost Tracking

Cost Attribution Dimensions

Implementation

import time
import json
from datetime import datetime, timezone
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class AIUsageTracker:
    """Enterprise AI usage tracking with per-user, per-team attribution."""

    # Model pricing (per 1M tokens)
    model_pricing = {
        "claude-3-haiku": {"input": 0.25, "output": 1.25},
        "claude-3-sonnet": {"input": 3.00, "output": 15.00},
        "claude-3-opus": {"input": 15.00, "output": 75.00},
        "gpt-4o": {"input": 2.50, "output": 10.00},
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
    }

    def __init__(self, db, redis, audit):
        self.db = db
        self.redis = redis
        self.audit = audit

    async def record_usage(self, tenant_id: str, user_id: str,
                           team_id: str, model: str,
                           input_tokens: int, output_tokens: int,
                           feature: str = "default",
                           project: str = "default",
                           cost_center: str = "default",
                           latency_ms: int = 0,
                           cache_hit: bool = False) -> dict:
        """Record a single LLM API call with full attribution."""

        # 1. Compute cost at instrumentation time
        pricing = self.model_pricing.get(model, {"input": 0, "output": 0})
        input_cost = (input_tokens / 1_000_000) * pricing["input"]
        output_cost = (output_tokens / 1_000_000) * pricing["output"]
        total_cost = input_cost + output_cost

        # 2. Create usage record
        record = {
            "tenant_id": tenant_id,
            "user_id": user_id,
            "team_id": team_id,
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "total_tokens": input_tokens + output_tokens,
            "input_cost": round(input_cost, 6),
            "output_cost": round(output_cost, 6),
            "total_cost": round(total_cost, 6),
            "feature": feature,
            "project": project,
            "cost_center": cost_center,
            "latency_ms": latency_ms,
            "cache_hit": cache_hit,
            "timestamp": datetime.now(timezone.utc),
        }

        # 3. Store in database
        await self.db.insert("ai_usage_logs", record)

        # 4. Update real-time budget counters in Redis
        await self._update_budget_counters(
            tenant_id, user_id, team_id, total_cost)

        # 5. Emit Prometheus metrics (via statsd or direct)
        # In production: emit to Prometheus push gateway or statsd
        # metrics.tokens_input.labels(model=model, team=team_id).inc(input_tokens)
        # metrics.tokens_output.labels(model=model, team=team_id).inc(output_tokens)
        # metrics.cost_total.labels(model=model, team=team_id).inc(total_cost)

        # 6. Audit log
        await self.audit.log(
            tenant_id=tenant_id,
            user_id=user_id,
            action="llm_api_call",
            entity_type="llm_usage",
            after={
                "model": model, "tokens": input_tokens + output_tokens,
                "cost": total_cost, "feature": feature,
            },
        )

        return record

    async def _update_budget_counters(self, tenant_id: str,
                                       user_id: str, team_id: str,
                                       cost: float):
        """Update real-time budget counters in Redis."""
        date_key = datetime.now(timezone.utc).strftime("%Y-%m-%d")
        month_key = datetime.now(timezone.utc).strftime("%Y-%m")

        # Per-user daily
        await self.redis.incrbyfloat(
            f"budget:user:{tenant_id}:{user_id}:daily:{date_key}", cost)
        await self.redis.expire(
            f"budget:user:{tenant_id}:{user_id}:daily:{date_key}", 172800)

        # Per-user monthly
        await self.redis.incrbyfloat(
            f"budget:user:{tenant_id}:{user_id}:monthly:{month_key}", cost)
        await self.redis.expire(
            f"budget:user:{tenant_id}:{user_id}:monthly:{month_key}", 2764800)

        # Per-team daily
        await self.redis.incrbyfloat(
            f"budget:team:{tenant_id}:{team_id}:daily:{date_key}", cost)

        # Per-team monthly
        await self.redis.incrbyfloat(
            f"budget:team:{tenant_id}:{team_id}:monthly:{month_key}", cost)

        # Tenant total
        await self.redis.incrbyfloat(
            f"budget:tenant:{tenant_id}:monthly:{month_key}", cost)

    async def check_budget(self, tenant_id: str, user_id: str,
                           daily_limit: float = 50.0,
                           monthly_limit: float = 1000.0) -> dict:
        """Check if user is within budget limits."""
        date_key = datetime.now(timezone.utc).strftime("%Y-%m-%d")
        month_key = datetime.now(timezone.utc).strftime("%Y-%m")

        daily_spent = float(await self.redis.get(
            f"budget:user:{tenant_id}:{user_id}:daily:{date_key}") or 0)
        monthly_spent = float(await self.redis.get(
            f"budget:user:{tenant_id}:{user_id}:monthly:{month_key}") or 0)

        return {
            "within_budget": (
                daily_spent < daily_limit and
                monthly_spent < monthly_limit),
            "daily_spent": round(daily_spent, 2),
            "daily_limit": daily_limit,
            "daily_remaining": round(daily_limit - daily_spent, 2),
            "monthly_spent": round(monthly_spent, 2),
            "monthly_limit": monthly_limit,
            "monthly_remaining": round(monthly_limit - monthly_spent, 2),
        }

    async def get_spend_report(self, tenant_id: str,
                               group_by: str = "team",
                               start_date: str = None,
                               end_date: str = None) -> list:
        """Generate spend report grouped by dimension."""
        query = f"""
            SELECT {group_by} as dimension,
                   SUM(total_cost) as total_cost,
                   SUM(input_tokens) as total_input_tokens,
                   SUM(output_tokens) as total_output_tokens,
                   COUNT(*) as request_count,
                   AVG(latency_ms) as avg_latency
            FROM ai_usage_logs
            WHERE tenant_id = $1
            {f"AND timestamp >= $2" if start_date else ""}
            {f"AND timestamp <= $3" if end_date else ""}
            GROUP BY {group_by}
            ORDER BY total_cost DESC
        """
        params = [tenant_id]
        if start_date:
            params.append(start_date)
        if end_date:
            params.append(end_date)

        results = await self.db.fetch_all(query, params)
        return [dict(r) for r in results]

    async def get_user_spend(self, tenant_id: str,
                             user_id: str) -> dict:
        """Get spend summary for a specific user."""
        query = """
            SELECT model,
                   SUM(total_cost) as cost,
                   SUM(total_tokens) as tokens,
                   COUNT(*) as requests
            FROM ai_usage_logs
            WHERE tenant_id = $1 AND user_id = $2
            AND timestamp >= NOW() - INTERVAL '30 days'
            GROUP BY model
            ORDER BY cost DESC
        """
        results = await self.db.fetch_all(query, [tenant_id, user_id])

        total_cost = sum(r["cost"] for r in results)
        total_tokens = sum(r["tokens"] for r in results)
        total_requests = sum(r["requests"] for r in results)

        return {
            "user_id": user_id,
            "total_cost_30d": round(total_cost, 2),
            "total_tokens_30d": total_tokens,
            "total_requests_30d": total_requests,
            "by_model": [dict(r) for r in results],
        }

    async def detect_anomalies(self, tenant_id: str) -> list:
        """Detect unusual spending patterns."""
        anomalies = []

        # 1. Users with sudden cost spikes (> 3x daily average)
        query = """
            SELECT user_id,
                   SUM(total_cost) as today_cost,
                   AVG(daily_avg) as avg_daily_cost
            FROM (
                SELECT user_id, total_cost,
                       AVG(total_cost) OVER (
                           PARTITION BY user_id
                           ORDER BY timestamp
                           ROWS BETWEEN 7 PRECEDING AND 1 PRECEDING
                       ) as daily_avg
                FROM ai_usage_logs
                WHERE tenant_id = $1
                AND timestamp >= NOW() - INTERVAL '1 day'
            ) sub
            GROUP BY user_id
            HAVING SUM(total_cost) > 3 * MAX(avg_daily_avg)
        """
        results = await self.db.fetch_all(query, [tenant_id])
        for r in results:
            anomalies.append({
                "type": "cost_spike",
                "user_id": r["user_id"],
                "today_cost": round(r["today_cost"], 2),
                "avg_daily_cost": round(r["avg_daily_cost"], 2),
            })

        return anomalies

AI Usage Tracking Checklist

• [ ] Instrument every LLM API call with token-level telemetry
• [ ] Capture input tokens, output tokens, model, cost per request
• [ ] Compute USD cost at instrumentation time, not from billing exports
• [ ] Authenticate users via JWT and extract identity for attribution
• [ ] Attach user_id, team_id, project, cost_center to every request
• [ ] Use IdP groups (Okta, Azure AD) for team-level attribution
• [ ] Give AI agents named identities via service tokens
• [ ] Store usage records in database (PostgreSQL) with timestamp
• [ ] Update real-time budget counters in Redis (daily and monthly)
• [ ] Emit Prometheus metrics for token counts and cost
• [ ] Emit OpenTelemetry traces with gen_ai span attributes
• [ ] Build cost analytics dashboard (per-user, per-team, per-model, per-feature)
• [ ] Track spend trends (daily, monthly, forecasts)
• [ ] Implement per-user budget limits ($/day, $/month)
• [ ] Implement per-team budget limits
• [ ] Implement per-tenant budget limits
• [ ] Start with monitoring mode (high limit) before enforcing
• [ ] Configure budget alerts at 80%, 90%, 100% thresholds
• [ ] Implement model downgrade when budget exceeded (frontier to cheap)
• [ ] Implement hard block when budget exceeded
• [ ] Track budget rejection rate as a metric
• [ ] Implement chargeback reports for finance teams
• [ ] Implement showback reports for awareness without budget transfer
• [ ] Track five chargeback dimensions: team, project, user, cost center, model
• [ ] Detect cost anomalies (sudden spikes, unusual patterns)
• [ ] Detect abuse (prompt injection, automated querying)
• [ ] Track cache hit rate for cost optimization
• [ ] Track cost per resolved query (not just cost per request)
• [ ] Track P95 latency alongside cost
• [ ] Use LiteLLM or AI gateway for automatic cost tracking
• [ ] Export usage logs to analytics platform (Grafana, Langfuse)
• [ ] Integrate with LLM usage metering for billing
• [ ] Integrate with per-tenant cost tracking
• [ ] Apply LLM cost optimization based on data
• [ ] Set AI budget caps based on usage data
• [ ] Calculate enterprise AI TCO with real data
• [ ] Avoid AI cost traps with visibility
• [ ] Consider BYOK to pass costs to customers
• [ ] Log all usage events in audit logs
• [ ] Apply RBAC to usage data access
• [ ] Apply multi-tenant usage isolation
• [ ] Set up platform monitoring for usage
• [ ] Test: cost attribution matches provider billing
• [ ] Test: budget enforcement blocks correctly
• [ ] Test: anomaly detection catches spikes
• [ ] Test: chargeback reports are accurate
• [ ] Document usage tracking architecture and metrics

FAQ

What is AI usage tracking?

AI usage tracking is the practice of continuously measuring, attributing, and controlling the financial costs generated by LLM API calls in production. OpenObserve: "LLM cost monitoring is a subdiscipline of LLM observability — the broader practice of understanding LLM system behavior through structured telemetry." Cloudflare: "You cannot calculate ROI on your AI spend without visibility on what you are spending, and you cannot protect that ROI without controls. Every other line item in a business has a budget and per-team attribution and AI spend should be no different." Key components: (1) Token-level telemetry — capture input tokens, output tokens, model, cost per request. (2) Per-user attribution — track which employee made which call. (3) Per-team attribution — group spend by IdP groups or cost centers. (4) Real-time spend controls — enforce budgets inline, not after the fact. (5) Cost analytics dashboard — query spend by model, user, team, feature.

How do you track LLM costs per user and per team?

Track LLM costs per user and per team by attaching identity metadata to every LLM request. Cloudflare: "When an employee authenticates via Cloudflare Access, we extract their identity from the JWT and attach it as metadata on the AI Gateway request. This makes per-user token consumption, team-level usage breakdowns, and cost attribution across the organization all visible in one place." LiteLLM: "Track spend for keys, users, and teams across 100+ LLMs. LiteLLM automatically tracks spend for all known models." Implementation: (1) Authenticate users via JWT or API key. (2) Attach user_id, team_id, cost_center to every request. (3) Compute cost per request based on model pricing and token counts. (4) Store in database with timestamp. (5) Aggregate by user, team, model, feature for dashboards. Behest: "Per-tenant, per-project, per-user attribution. Every Behest call carries a session ID."

What telemetry signals should you capture for LLM cost tracking?

Capture these telemetry signals on every LLM request. OpenObserve: (1) Input tokens (llm_usage_tokens_input) — primary cost driver, grows with system prompt size, RAG context, conversation history. (2) Output tokens (llm_usage_tokens_output) — often 5-10x more expensive per token than input, unbounded without max_tokens. (3) Total tokens (llm_usage_tokens_total) — context window utilization, rate-limiting surface. (4) Model identifier (gen_ai_request_model) — critical for attribution, tier differences can mean 10-100x cost difference. (5) USD cost (llm_usage_cost_total, llm_usage_cost_input, llm_usage_cost_output) — must be computed at instrumentation time, not inferred from billing exports. Solo.io: "Token metrics are exposed through Prometheus metrics and OpenTelemetry traces. Calculate costs per request, per user, or per model. Set up budget alerts and spending limits. Generate cost reports for chargeback or showback."

How do you implement AI chargeback and showback?

AI chargeback allocates LLM costs to the teams that incur them, while showback displays costs without actually transferring budget. Spheron: "Use a Grafana dashboard to join GPU-hour cost with token counts per team and compute attributed cost as: (team_tokens / total_tokens) * gpu_hour_cost. Five dimensions cover 95% of chargeback use cases: team, project, user, cost center, and model." LiteLLM: "Use /global/spend/report endpoint to get spend reports — spend per team, spend per customer, spend for specific API key, spend for internal user." Traceloop: "Moving from reactive panic at monthly bills to proactive control requires shifting from the total bill to the individual request. Tag every request with user, team, project, and cost center." Behest: "CIO chargeback per-cost-center allocation. Exact usage classification: team, project, cost center."

How do you enforce real-time AI budget limits?

Enforce real-time AI budget limits by tracking cumulative spend against dollar budgets and blocking or downgrading requests when limits are reached. Cloudflare: "AI Gateway calculates cost per request based on model pricing, and tracks cumulative spend against your limit in real time. Set per-user budgets — $500/month for individual contributors and $2,000 for senior engineers. When a user hits their limit, requests can be downgraded to a cheaper model or blocked." Behest: "Daily and monthly caps at the global, tenant, and project level. Stop overruns before the invoice lands. Token budgets enforced inline." Solo.io: "Combine cost tracking with rate limiting: set up token-based rate limiting keyed by X-User-ID. Configure daily token limit based on your budget. Monitor actual spending with metrics queries to ensure rate limits align with budget goals." Implementation: (1) Track cumulative spend per user/team in Redis. (2) Check budget before each LLM call. (3) If exceeded: block (return error), downgrade (route to cheaper model), or alert (notify admin). (4) Start with monitoring mode (high limit) before enforcing.

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