Module 5: Structured Logging for AI Systems

4. Token and Cost Tracking

Next →

Description

For an AI Engineer, not knowing how much each request costs is like a backend engineer who doesn't monitor their database. The OpenAI bill can surprise you at the end of the month if you don't track in real time. This capsule implements a complete tracking system: how to get usage from the API, how to calculate cost by model, how to log and accumulate costs per request, and how to detect outliers — those 5% of requests that consume 60% of the budget.

Why cost tracking is the "wow" feature

Real scenario without cost tracking:
→ You launch your AI app
→ Week 1: $5 on OpenAI, normal
→ Week 2: $8, expected
→ Week 3: $47, what happened?

Without cost logs, debugging is:
"Let me review the code... did someone change the model?
is there a loop that calls the LLM multiple times?
did someone forget to set max_tokens?"

→ Hours of debugging

With cost tracking in logs:
jq -s 'sort_by(-.cost_usd) | .[0:5]' logs.json
→ The 5 most expensive requests are all from the /summarize endpoint
→ Each one uses ~15,000 input tokens
→ A user is sending 100-page documents
→ Fix: truncate the input in the sanitizer

→ 5 minutes of debugging

Getting usage from the OpenAI API

# The OpenAI API returns usage in every response
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Analyze this text..."}]
)

# Structure of the usage object:
usage = response.usage
print(usage.prompt_tokens)      # Tokens in the input (including the system prompt)
print(usage.completion_tokens)  # Tokens in the generated output
print(usage.total_tokens)       # prompt_tokens + completion_tokens

# Example:
# usage.prompt_tokens = 245
# usage.completion_tokens = 87
# usage.total_tokens = 332

# When is it NOT available?
# - Some models or configurations may not return usage
# - stream=True can behave differently
# Solution: use tiktoken to count if usage isn't available

Calculate cost precisely

# src/cost_calculator.py
from dataclasses import dataclass
from typing import Optional

@dataclass
class CostCalculation:
    model: str
    input_tokens: int
    output_tokens: int
    total_tokens: int
    input_cost_usd: float
    output_cost_usd: float
    total_cost_usd: float

# Prices per 1M tokens (update periodically from openai.com/pricing)
# Do NOT hardcode in business code — centralize here for easy updates
MODEL_PRICING = {
    # GPT-4o-mini: the high-efficiency model
    "gpt-4o-mini": {
        "input": 0.150,   # $0.15 per 1M input tokens
        "output": 0.600,  # $0.60 per 1M output tokens
    },
    # GPT-4o: the balanced model
    "gpt-4o": {
        "input": 2.50,
        "output": 10.00,
    },
    # GPT-4o-2024-11-20 (alias for GPT-4o latest)
    "gpt-4o-2024-11-20": {
        "input": 2.50,
        "output": 10.00,
    },
    # GPT-4 Turbo
    "gpt-4-turbo": {
        "input": 10.00,
        "output": 30.00,
    },
    # GPT-4 (base, legacy)
    "gpt-4": {
        "input": 30.00,
        "output": 60.00,
    },
    # GPT-3.5 Turbo (legacy, very cheap)
    "gpt-3.5-turbo": {
        "input": 0.50,
        "output": 1.50,
    },
}

# Fallback for unlisted models (conservative best estimate)
DEFAULT_PRICING = {"input": 0.150, "output": 0.600}

def calculate_cost(
    model: str,
    input_tokens: int,
    output_tokens: int
) -> CostCalculation:
    """
    Calculates the exact cost of an LLM call.
    
    Formula: (tokens / 1_000_000) * price_per_million
    """
    # Normalize the model name (it may have version suffixes)
    pricing = _get_pricing(model)
    
    input_cost = (input_tokens / 1_000_000) * pricing["input"]
    output_cost = (output_tokens / 1_000_000) * pricing["output"]
    
    return CostCalculation(
        model=model,
        input_tokens=input_tokens,
        output_tokens=output_tokens,
        total_tokens=input_tokens + output_tokens,
        input_cost_usd=round(input_cost, 8),
        output_cost_usd=round(output_cost, 8),
        total_cost_usd=round(input_cost + output_cost, 8),
    )

def _get_pricing(model: str) -> dict:
    """
    Gets the pricing for a model, with intelligent fallback.
    Handles models with version suffixes like "gpt-4o-mini-2024-07-18".
    """
    if model in MODEL_PRICING:
        return MODEL_PRICING[model]
    
    # Try to match by prefix (for timestamped versions)
    for key in MODEL_PRICING:
        if model.startswith(key):
            return MODEL_PRICING[key]
    
    # Fall back to the default
    return DEFAULT_PRICING

Log the cost in every request

# src/llm_wrapper.py (updated with complete cost tracking)
import time
import structlog
from src.cost_calculator import calculate_cost, CostCalculation
from src.tracing import get_request_id

log = structlog.get_logger()

# Thresholds for alerts
HIGH_COST_THRESHOLD_USD = 0.05    # $0.05 per request is high for most apps
VERY_HIGH_COST_THRESHOLD_USD = 0.20

HIGH_TOKENS_THRESHOLD = 10_000    # 10K input tokens is suspicious
HIGH_LATENCY_THRESHOLD_MS = 10_000

def call_llm_with_cost_tracking(
    client,
    model: str,
    messages: list,
    **kwargs
) -> tuple:
    """
    Makes an LLM call and tracks tokens and cost.
    Returns (response, cost_calculation).
    """
    start_time = time.time()
    request_id = get_request_id() or "unknown"
    
    # Warn if the input looks excessively long
    total_input_chars = sum(len(m.get("content", "")) for m in messages)
    if total_input_chars > 40_000:  # Approx 10K tokens
        log.warning(
            "large_input_detected",
            input_chars=total_input_chars,
            estimated_tokens=total_input_chars // 4,  # Rough estimate
            model=model
        )
    
    try:
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            **kwargs
        )
        
        duration_ms = (time.time() - start_time) * 1000
        
        # Calculate cost
        cost = calculate_cost(
            model=model,
            input_tokens=response.usage.prompt_tokens,
            output_tokens=response.usage.completion_tokens
        )
        
        # INFO log: cost and tokens (always)
        log.info(
            "llm_request_completed",
            model=model,
            input_tokens=cost.input_tokens,
            output_tokens=cost.output_tokens,
            total_tokens=cost.total_tokens,
            input_cost_usd=cost.input_cost_usd,
            output_cost_usd=cost.output_cost_usd,
            total_cost_usd=cost.total_cost_usd,
            duration_ms=round(duration_ms, 1)
        )
        
        # Alerts for anomalous cost
        if cost.total_cost_usd > VERY_HIGH_COST_THRESHOLD_USD:
            log.warning(
                "very_high_cost_request",
                total_cost_usd=cost.total_cost_usd,
                total_tokens=cost.total_tokens,
                model=model,
                note="Investigate whether the input is being truncated correctly"
            )
        elif cost.total_cost_usd > HIGH_COST_THRESHOLD_USD:
            log.warning(
                "high_cost_request",
                total_cost_usd=cost.total_cost_usd,
                model=model
            )
        
        # Alert for excessively long input
        if cost.input_tokens > HIGH_TOKENS_THRESHOLD:
            log.warning(
                "excessive_input_tokens",
                input_tokens=cost.input_tokens,
                model=model,
                note="Is the sanitizer truncating correctly?"
            )
        
        return response, cost
    
    except Exception as e:
        duration_ms = (time.time() - start_time) * 1000
        log.error(
            "llm_request_failed",
            error_type=type(e).__name__,
            duration_ms=round(duration_ms, 1),
            model=model
        )
        raise

Accumulate cost in requests with multiple LLM calls

# Some pipelines make multiple LLM calls per request
# (e.g., guardrail judge + main analysis)
# It's important to track the request's total cost, not just each call's

from dataclasses import dataclass, field
from typing import List
from src.cost_calculator import CostCalculation

@dataclass
class RequestCostAccumulator:
    """Accumulates the cost of multiple LLM calls in a single request."""
    request_id: str
    calls: List[CostCalculation] = field(default_factory=list)
    
    def add(self, cost: CostCalculation):
        self.calls.append(cost)
    
    @property
    def total_cost_usd(self) -> float:
        return round(sum(c.total_cost_usd for c in self.calls), 8)
    
    @property
    def total_tokens(self) -> int:
        return sum(c.total_tokens for c in self.calls)
    
    @property
    def num_calls(self) -> int:
        return len(self.calls)
    
    def to_log_dict(self) -> dict:
        return {
            "total_cost_usd": self.total_cost_usd,
            "total_tokens": self.total_tokens,
            "llm_calls_count": self.num_calls,
            "cost_breakdown": [
                {"model": c.model, "tokens": c.total_tokens, "cost": c.total_cost_usd}
                for c in self.calls
            ]
        }

# Usage in the endpoint:
# accumulator = RequestCostAccumulator(request_id=request_id)
# 
# # Call to the guardrail LLM judge
# _, cost1 = call_llm_with_cost_tracking(client, model, guard_messages)
# accumulator.add(cost1)
# 
# # Main call
# _, cost2 = call_llm_with_cost_tracking(client, model, main_messages)
# accumulator.add(cost2)
# 
# log.info("request_total_cost", **accumulator.to_log_dict())

Estimate tokens WITHOUT calling the API: tiktoken

# Useful for:
# 1. Validating that the input doesn't exceed the limit before calling
# 2. Estimating cost before the call
# 3. Counting tokens when the API doesn't return usage

import tiktoken

def count_tokens(text: str, model: str = "gpt-4o-mini") -> int:
    """
    Counts tokens exactly as the model does.
    Requires: pip install tiktoken
    """
    try:
        enc = tiktoken.encoding_for_model(model)
    except KeyError:
        # Fallback for unknown models
        enc = tiktoken.get_encoding("cl100k_base")
    
    return len(enc.encode(text))

def count_messages_tokens(messages: list, model: str = "gpt-4o-mini") -> int:
    """
    Counts tokens for a list of messages (as the API sends them).
    Includes the overhead tokens from the message structure.
    """
    try:
        enc = tiktoken.encoding_for_model(model)
    except KeyError:
        enc = tiktoken.get_encoding("cl100k_base")
    
    # Overhead per message: 3 tokens (role, content, separators)
    overhead_per_message = 3
    total = 0
    
    for message in messages:
        total += overhead_per_message
        for key, value in message.items():
            total += len(enc.encode(str(value)))
    
    total += 3  # Overhead from the response format
    return total

def estimate_cost(
    messages: list,
    model: str = "gpt-4o-mini",
    expected_output_tokens: int = 500
) -> dict:
    """
    Estimates the cost of a call BEFORE making it.
    Useful for validating that the input isn't excessively expensive.
    """
    from src.cost_calculator import calculate_cost
    
    input_tokens = count_messages_tokens(messages, model)
    cost = calculate_cost(model, input_tokens, expected_output_tokens)
    
    return {
        "estimated_input_tokens": input_tokens,
        "estimated_output_tokens": expected_output_tokens,
        "estimated_cost_usd": cost.total_cost_usd,
        "model": model
    }

# Example of usage to reject very expensive requests:
def pre_validate_cost(messages, model, max_cost_usd=0.20):
    estimate = estimate_cost(messages, model)
    if estimate["estimated_cost_usd"] > max_cost_usd:
        raise ValueError(
            f"Request estimated too expensive: ${estimate['estimated_cost_usd']:.4f} "
            f"(maximum: ${max_cost_usd}). "
            f"Input tokens: {estimate['estimated_input_tokens']}"
        )

Cost analysis from the logs

# scripts/analyze_costs.py
# Script to analyze costs from the JSON logs

import json
import sys
from collections import defaultdict
from datetime import datetime, timezone
from pathlib import Path

def load_logs(log_file: str) -> list:
    """Loads logs from a JSON Lines file."""
    logs = []
    with open(log_file) as f:
        for line in f:
            line = line.strip()
            if line:
                try:
                    logs.append(json.loads(line))
                except json.JSONDecodeError:
                    pass
    return logs

def analyze_costs(logs: list, date_filter: str = None) -> dict:
    """Complete cost analysis from the logs."""
    
    # Filter by date if specified
    if date_filter:
        logs = [l for l in logs if l.get("timestamp", "").startswith(date_filter)]
    
    # Only completed-request logs with cost
    cost_logs = [
        l for l in logs
        if l.get("event") == "llm_request_completed" and "total_cost_usd" in l
    ]
    
    if not cost_logs:
        return {"error": "No cost logs found"}
    
    total_cost = sum(l["total_cost_usd"] for l in cost_logs)
    total_tokens = sum(l.get("total_tokens", 0) for l in cost_logs)
    
    # By model
    by_model = defaultdict(lambda: {"cost": 0, "tokens": 0, "count": 0})
    for l in cost_logs:
        model = l.get("model", "unknown")
        by_model[model]["cost"] += l["total_cost_usd"]
        by_model[model]["tokens"] += l.get("total_tokens", 0)
        by_model[model]["count"] += 1
    
    # Top 10 most expensive requests
    top_expensive = sorted(cost_logs, key=lambda l: l["total_cost_usd"], reverse=True)[:10]
    
    # Cost distribution
    buckets = {"<$0.001": 0, "$0.001-$0.01": 0, "$0.01-$0.05": 0, "$0.05-$0.20": 0, ">$0.20": 0}
    for l in cost_logs:
        cost = l["total_cost_usd"]
        if cost < 0.001: buckets["<$0.001"] += 1
        elif cost < 0.01: buckets["$0.001-$0.01"] += 1
        elif cost < 0.05: buckets["$0.01-$0.05"] += 1
        elif cost < 0.20: buckets["$0.05-$0.20"] += 1
        else: buckets[">$0.20"] += 1
    
    return {
        "summary": {
            "total_cost_usd": round(total_cost, 6),
            "total_tokens": total_tokens,
            "total_requests": len(cost_logs),
            "avg_cost_per_request": round(total_cost / len(cost_logs), 8),
            "avg_tokens_per_request": total_tokens // len(cost_logs),
        },
        "by_model": {
            model: {
                "cost_usd": round(data["cost"], 6),
                "tokens": data["tokens"],
                "requests": data["count"],
                "pct_of_total": round(data["cost"] / total_cost * 100, 1)
            }
            for model, data in by_model.items()
        },
        "top_expensive_requests": [
            {
                "request_id": l.get("request_id"),
                "cost_usd": l["total_cost_usd"],
                "tokens": l.get("total_tokens"),
                "model": l.get("model")
            }
            for l in top_expensive
        ],
        "cost_distribution": buckets
    }

if __name__ == "__main__":
    log_file = sys.argv[1] if len(sys.argv) > 1 else "logs/app.json"
    date_filter = sys.argv[2] if len(sys.argv) > 2 else None
    
    logs = load_logs(log_file)
    analysis = analyze_costs(logs, date_filter)
    print(json.dumps(analysis, indent=2))

Cost comparison between models

# Helps make decisions about which model to use:

COST_COMPARISON = {
    "Sentiment analysis (300 input, 100 output tokens)": {
        "gpt-4o-mini": calculate_cost("gpt-4o-mini", 300, 100).total_cost_usd,
        "gpt-4o": calculate_cost("gpt-4o", 300, 100).total_cost_usd,
        "gpt-4-turbo": calculate_cost("gpt-4-turbo", 300, 100).total_cost_usd,
    }
}

# gpt-4o-mini:  $0.0001050 per request
# gpt-4o:       $0.0017500 per request  (16.7x more expensive)
# gpt-4-turbo:  $0.0060000 per request  (57.1x more expensive)

# For 10,000 requests/month:
# gpt-4o-mini:  $1.05/month
# gpt-4o:       $17.50/month
# gpt-4-turbo:  $60.00/month

# Conclusion: for simple sentiment analysis, gpt-4o-mini is the obvious choice
# Only use gpt-4o for cases that genuinely require more capacity

Cost tracker tests

# tests/unit/test_cost_calculator.py
import pytest
from src.cost_calculator import calculate_cost

def test_gpt4o_mini_cost():
    """Verifies cost calculation for gpt-4o-mini."""
    cost = calculate_cost("gpt-4o-mini", input_tokens=500, output_tokens=200)
    
    # Input: 500 / 1,000,000 * 0.15 = 0.000075
    # Output: 200 / 1,000,000 * 0.60 = 0.000120
    # Total: 0.000195
    assert abs(cost.total_cost_usd - 0.000195) < 0.000001

def test_unknown_model_uses_default():
    """Unknown models use the default pricing without crashing."""
    cost = calculate_cost("gpt-unknown-model-2099", 100, 100)
    assert cost.total_cost_usd > 0  # Doesn't crash, returns something reasonable

@pytest.mark.parametrize("model,inp,out,expected_total", [
    ("gpt-4o-mini", 1_000_000, 0, 0.15),   # 1M input tokens = $0.15
    ("gpt-4o-mini", 0, 1_000_000, 0.60),   # 1M output tokens = $0.60
    ("gpt-4o", 1_000_000, 0, 2.50),         # 1M input GPT-4o = $2.50
])
def test_pricing_table(model, inp, out, expected_total):
    """Verifies the pricing table for the most important models."""
    cost = calculate_cost(model, inp, out)
    assert abs(cost.total_cost_usd - expected_total) < 0.01

Exercises

Exercise 1: Calculate costs manually

For each scenario, calculate the cost before running the code:

  1. Analyzing a tweet (50 input tokens, 30 output tokens) with gpt-4o-mini
  2. Summarizing an article (2,000 input tokens, 500 output tokens) with gpt-4o-mini
  3. The same summary with gpt-4o
See solution

Scenario 1: Tweet with gpt-4o-mini

  • Input: 50 / 1,000,000 × $0.15 = $0.0000075
  • Output: 30 / 1,000,000 × $0.60 = $0.000018
  • Total: $0.0000255 (~$0.000026)

Scenario 2: Article with gpt-4o-mini

  • Input: 2,000 / 1,000,000 × $0.15 = $0.0003
  • Output: 500 / 1,000,000 × $0.60 = $0.0003
  • Total: $0.0006

Scenario 3: Article with gpt-4o

  • Input: 2,000 / 1,000,000 × $2.50 = $0.005
  • Output: 500 / 1,000,000 × $10.00 = $0.005
  • Total: $0.010 (16.7x more expensive than gpt-4o-mini for this case)

Exercise 2: Detect the outlier request

You have these cost logs. Which one is the outlier and what could be the cause?

{"cost_usd": 0.00019, "input_tokens": 300, "output_tokens": 87}
{"cost_usd": 0.00021, "input_tokens": 320, "output_tokens": 95}
{"cost_usd": 0.00018, "input_tokens": 280, "output_tokens": 80}
{"cost_usd": 0.04250, "input_tokens": 65000, "output_tokens": 1200}
{"cost_usd": 0.00020, "input_tokens": 310, "output_tokens": 88}
See solution

The 4th log has 65,000 input tokens vs an average of ~300 for the others. That's an outlier of more than 200x.

Possible causes:

  1. The sanitizer didn't truncate the input correctly
  2. The user uploaded an entire document instead of a fragment
  3. There's a bug in the prompt construction (accumulated context, loop without reset)

Cost of the outlier: ~225× the average, which means a single request of this type costs the same as 225 normal requests.

Exercise 3: Alert policy

Define alert thresholds for an app with these characteristics:

  • Average cost per request: $0.0002
  • Average requests/hour: 100
  • Monthly budget: $100
See guide

Expected budget: $0.0002 × 100 × 24 × 30 = $14.40/month (much less than $100)

Reasonable thresholds:

  • high_cost_per_request: > $0.02 (100× the average) → WARNING
  • very_high_cost_per_request: > $0.10 (500× the average) → ERROR
  • hourly_cost_alert: > $5/hour (vs $0.02/hour normal) → immediate ERROR
  • daily_cost_alert: > $20/day (vs $0.48/day normal) → WARNING
  • monthly_budget_alert: > $80 (80% of the budget) → ERROR, notify the team

Summary

  • response.usage is the source of truth for tokens — use it whenever it's available
  • Centralize the prices in an updatable table, don't hardcode them in business code
  • Log the cost in every request with input_cost_usd, output_cost_usd, total_cost_usd
  • Cost alerts let you detect outliers before they impact the bill
  • tiktoken lets you estimate tokens before calling the API — useful for validating inputs
  • Analysis scripts over the JSON logs reveal cost patterns that are invisible without logging

Additional resources

  1. OpenAI Pricing — Updated prices (check periodically)
  2. tiktoken GitHub — Counting tokens before API calls
  3. OpenAI Usage Dashboard — Official usage dashboard
  4. LangSmith — AI observability platform with built-in cost tracking
  5. Helicone — Proxy that adds observability (including cost tracking) without changing code
← Previous3. Request Tracing with Correlation IDsNext →5. JSON Logs with structlog