FromZero2AI
CoursesPlaygroundBlog
Support on Ko-fi
AI Observability & Production Monitoring • Module A: Tracing & DebuggingLesson 2: Token Cost Tracking & Budget Enforcement
PreviousNext

Lesson 2: Token Cost Tracking & Budget Enforcement

Track token spend per user, per feature, and per model. Build a cost dashboard, set per-request budgets, and implement automatic cost alerts before monthly bills spike.

Built with AI for beginners. Free forever.

Support on Ko-fi•About•Blog•Privacy Policy•Terms of Service

A naive LLM application can spend $50K/month on API calls that cost $5K with the right architecture. Cost tracking turns your token usage data into a map of where money is being spent — and optimization patterns to cut it.

Computing Cost from Token Usage

import anthropic
from dataclasses import dataclass

client = anthropic.Anthropic()

# Pricing per 1K tokens (verify at anthropic.com/pricing — changes)
COST_PER_1K = {
    "claude-haiku-4-5-20251001": {"input": 0.00025, "output": 0.00125},
    "claude-sonnet-4-6":         {"input": 0.003,   "output": 0.015},
    "claude-opus-4-8":           {"input": 0.015,   "output": 0.075},
}

@dataclass
class CostRecord:
    model: str
    input_tokens: int
    output_tokens: int
    operation: str   # e.g. "chat", "rag-retrieval", "summarization"
    user_id: str | None = None
    tenant_id: str | None = None

    @property
    def cost_usd(self) -> float:
        p = COST_PER_1K.get(self.model, {"input": 0, "output": 0})
        return (
            self.input_tokens / 1000 * p["input"]
            + self.output_tokens / 1000 * p["output"]
        )

def tracked_llm_call(prompt: str, model: str, operation: str) -> tuple[str, CostRecord]:
    response = client.messages.create(
        model=model,
        max_tokens=500,
        messages=[{"role": "user", "content": prompt}],
    )
    record = CostRecord(
        model=model,
        input_tokens=response.usage.input_tokens,
        output_tokens=response.usage.output_tokens,
        operation=operation,
    )
    # Write record to your cost DB / data warehouse
    write_cost_record(record)
    return response.content[0].text, record

Cost Aggregation Query (PostgreSQL)

-- Daily cost breakdown by model and operation
SELECT
    DATE_TRUNC('day', created_at) AS day,
    model,
    operation,
    COUNT(*) AS api_calls,
    SUM(input_tokens) AS total_input_tokens,
    SUM(output_tokens) AS total_output_tokens,
    SUM(cost_usd) AS total_cost_usd
FROM llm_cost_records
WHERE created_at >= NOW() - INTERVAL '30 days'
GROUP BY 1, 2, 3
ORDER BY total_cost_usd DESC;

-- Top cost operations (where to focus optimization)
SELECT
    operation,
    ROUND(SUM(cost_usd)::numeric, 4) AS total_cost,
    ROUND(AVG(cost_usd)::numeric, 6) AS avg_cost_per_call,
    COUNT(*) AS total_calls,
    ROUND(AVG(input_tokens)) AS avg_input_tokens,
    ROUND(AVG(output_tokens)) AS avg_output_tokens
FROM llm_cost_records
GROUP BY operation
ORDER BY total_cost DESC;

-- Cost anomaly detection: alert if daily cost > 2x trailing 7-day avg
WITH daily AS (
    SELECT DATE_TRUNC('day', created_at) AS day, SUM(cost_usd) AS daily_cost
    FROM llm_cost_records
    GROUP BY 1
)
SELECT day, daily_cost,
    AVG(daily_cost) OVER (ORDER BY day ROWS BETWEEN 7 PRECEDING AND 1 PRECEDING) AS trailing_7d_avg
FROM daily
HAVING daily_cost > trailing_7d_avg * 2;

Cost Reduction Technique Explorer

Select a technique to see estimated savings, when to apply it, trade-offs, and the relevant implementation code.

Prompt Caching

~90% on repeated system prompts

Best scenarios

Large stable system prompts or documents sent on every request — documentation Q&A, code review bots, any app where a big context prefix is constant.

Trade-off / caveat

Cache TTL is ~5 minutes (Anthropic ephemeral). Cold start (first call after expiry) pays full price. Only supported on specific models.

Implementation

import anthropic

client = anthropic.Anthropic()

# Large stable system prompt + documents = cache this prefix
# Only the user's question changes per request
response = client.beta.prompt_caching.messages.create(
    model="claude-haiku-4-5-20251001",
    max_tokens=500,
    system=[
        {
            "type": "text",
            "text": "You are an expert assistant. Here is the full product documentation:",
        },
        {
            "type": "text",
            "text": LARGE_PRODUCT_DOCS,   # 50K tokens — cached after first call
            "cache_control": {"type": "ephemeral"},  # Mark for caching
        },
    ],
    messages=[{"role": "user", "content": user_question}],
)
# First call: 50K tokens @ $0.003/1K = $0.15
# Subsequent calls: 50K @ $0.0003/1K (90% discount) + small variable portion
# Breakeven: 2 calls — every call after that saves ~$0.13

Build a cost budget alert. Set a daily cost threshold (e.g., $500) and alert on Slack/PagerDuty if exceeded. Cost spikes are often the first signal of an abuse pattern, runaway retry loop, or accidental infinite agent recursion — catching it at $500 prevents it becoming $50,000.

Attribute costs to features, not just models.Tag every API call with the feature/operation that triggered it (operation="rag-chat", operation="doc-summary"). When you need to cut costs, you'll know which feature to optimize — not just which model is most expensive.

Lesson 3 covers structured logging for LLM systems: what to log, what to never log (PII), and how to build dashboards that answer production questions in seconds.