Módulo 5: Structured Logging para AI Systems

4. Token y Cost Tracking

Siguiente →

Descripción

Para un AI Engineer, no saber cuánto cuesta cada request es como un ingeniero de backend que no monitorea su base de datos. La factura de OpenAI puede sorprenderte al final del mes si no trackeas en tiempo real. Esta cápsula implementa un sistema completo de tracking: cómo obtener usage de la API, cómo calcular el costo por modelo, cómo loguear y acumular costos por request, y cómo detectar outliers — esos 5% de requests que consumen el 60% del budget.

Por qué el cost tracking es la feature "wow"

Escenario real sin cost tracking:
→ Lanzas tu app AI
→ Semana 1: $5 en OpenAI, normal
→ Semana 2: $8, esperado
→ Semana 3: $47, ¿qué pasó?

Sin logs de costo, el debugging es:
"Déjame revisar el código... ¿alguien cambió el modelo?
¿hay un loop que llama al LLM múltiples veces?
¿alguien olvidó poner max_tokens?"

→ Horas de debugging

Con cost tracking en logs:
jq -s 'sort_by(-.cost_usd) | .[0:5]' logs.json
→ Los 5 requests más caros son todos del endpoint /summarize
→ Cada uno usa ~15,000 tokens de input
→ Un usuario está enviando documentos de 100 páginas
→ Fix: truncar input en el sanitizador

→ 5 minutos de debugging

Obtener usage de la API de OpenAI

# La API de OpenAI devuelve el usage en cada respuesta
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Analyze this text..."}]
)

# Estructura del objeto usage:
usage = response.usage
print(usage.prompt_tokens)      # Tokens en el input (incluyendo system prompt)
print(usage.completion_tokens)  # Tokens en el output generado
print(usage.total_tokens)       # prompt_tokens + completion_tokens

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

# ¿Cuándo NO está disponible?
# - Algunos modelos o configuraciones pueden no retornar usage
# - stream=True puede tener comportamiento diferente
# Solución: usar tiktoken para contar si usage no está disponible

Calcular costo con precisión

# 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

# Precios por 1M tokens (actualizar periódicamente desde openai.com/pricing)
# NO hardcodear en código de negocio — centralizar aquí para fácil actualización
MODEL_PRICING = {
    # GPT-4o-mini: el modelo de alta eficiencia
    "gpt-4o-mini": {
        "input": 0.150,   # $0.15 por 1M tokens input
        "output": 0.600,  # $0.60 por 1M tokens output
    },
    # GPT-4o: el modelo equilibrado
    "gpt-4o": {
        "input": 2.50,
        "output": 10.00,
    },
    # GPT-4o-2024-11-20 (alias para 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, muy barato)
    "gpt-3.5-turbo": {
        "input": 0.50,
        "output": 1.50,
    },
}

# Fallback para modelos no listados (mejor estimación conservadora)
DEFAULT_PRICING = {"input": 0.150, "output": 0.600}

def calculate_cost(
    model: str,
    input_tokens: int,
    output_tokens: int
) -> CostCalculation:
    """
    Calcula el costo exacto de una llamada al LLM.
    
    Fórmula: (tokens / 1_000_000) * price_per_million
    """
    # Normalizar el nombre del modelo (puede tener sufijos de versión)
    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:
    """
    Obtiene el pricing para un modelo, con fallback inteligente.
    Maneja modelos con sufijos de versión como "gpt-4o-mini-2024-07-18".
    """
    if model in MODEL_PRICING:
        return MODEL_PRICING[model]
    
    # Intentar match por prefijo (para versiones con timestamp)
    for key in MODEL_PRICING:
        if model.startswith(key):
            return MODEL_PRICING[key]
    
    # Fallback al default
    return DEFAULT_PRICING

Loguear costo en cada request

# src/llm_wrapper.py (actualizado con cost tracking completo)
import time
import structlog
from src.cost_calculator import calculate_cost, CostCalculation
from src.tracing import get_request_id

log = structlog.get_logger()

# Umbrales para alertas
HIGH_COST_THRESHOLD_USD = 0.05    # $0.05 por request es alto para la mayoría de apps
VERY_HIGH_COST_THRESHOLD_USD = 0.20

HIGH_TOKENS_THRESHOLD = 10_000    # 10K tokens de input es sospechoso
HIGH_LATENCY_THRESHOLD_MS = 10_000

def call_llm_with_cost_tracking(
    client,
    model: str,
    messages: list,
    **kwargs
) -> tuple:
    """
    Realiza una llamada al LLM y trackea tokens y costo.
    Retorna (response, cost_calculation).
    """
    start_time = time.time()
    request_id = get_request_id() or "unknown"
    
    # Advertir si el input parece excesivamente largo
    total_input_chars = sum(len(m.get("content", "")) for m in messages)
    if total_input_chars > 40_000:  # Aprox 10K tokens
        log.warning(
            "large_input_detected",
            input_chars=total_input_chars,
            estimated_tokens=total_input_chars // 4,  # Estimación rough
            model=model
        )
    
    try:
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            **kwargs
        )
        
        duration_ms = (time.time() - start_time) * 1000
        
        # Calcular costo
        cost = calculate_cost(
            model=model,
            input_tokens=response.usage.prompt_tokens,
            output_tokens=response.usage.completion_tokens
        )
        
        # Log INFO: costo y tokens (siempre)
        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)
        )
        
        # Alertas por costo anómalo
        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="Investigar si el input está siendo truncado correctamente"
            )
        elif cost.total_cost_usd > HIGH_COST_THRESHOLD_USD:
            log.warning(
                "high_cost_request",
                total_cost_usd=cost.total_cost_usd,
                model=model
            )
        
        # Alerta por input excesivamente largo
        if cost.input_tokens > HIGH_TOKENS_THRESHOLD:
            log.warning(
                "excessive_input_tokens",
                input_tokens=cost.input_tokens,
                model=model,
                note="¿El sanitizador está truncando correctamente?"
            )
        
        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

Acumular costo en requests con múltiples llamadas al LLM

# Algunos pipelines hacen múltiples llamadas al LLM por request
# (e.g., guardrail judge + análisis principal)
# Es importante trackear el costo total del request, no solo de cada llamada

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

@dataclass
class RequestCostAccumulator:
    """Acumula el costo de múltiples llamadas al LLM en un solo 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
            ]
        }

# Uso en el endpoint:
# accumulator = RequestCostAccumulator(request_id=request_id)
# 
# # Llamada al guardrail LLM judge
# _, cost1 = call_llm_with_cost_tracking(client, model, guard_messages)
# accumulator.add(cost1)
# 
# # Llamada principal
# _, cost2 = call_llm_with_cost_tracking(client, model, main_messages)
# accumulator.add(cost2)
# 
# log.info("request_total_cost", **accumulator.to_log_dict())

Estimar tokens SIN llamar al API: tiktoken

# Útil para:
# 1. Validar que el input no excede el límite antes de llamar
# 2. Estimar costo antes de la llamada
# 3. Contar tokens cuando el API no retorna usage

import tiktoken

def count_tokens(text: str, model: str = "gpt-4o-mini") -> int:
    """
    Cuenta tokens exactamente como lo hace el modelo.
    Requiere: pip install tiktoken
    """
    try:
        enc = tiktoken.encoding_for_model(model)
    except KeyError:
        # Fallback para modelos no conocidos
        enc = tiktoken.get_encoding("cl100k_base")
    
    return len(enc.encode(text))

def count_messages_tokens(messages: list, model: str = "gpt-4o-mini") -> int:
    """
    Cuenta tokens de una lista de mensajes (como los envía el API).
    Incluye los tokens de overhead por la estructura del mensaje.
    """
    try:
        enc = tiktoken.encoding_for_model(model)
    except KeyError:
        enc = tiktoken.get_encoding("cl100k_base")
    
    # Overhead por mensaje: 3 tokens (role, content, separadores)
    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 del formato del response
    return total

def estimate_cost(
    messages: list,
    model: str = "gpt-4o-mini",
    expected_output_tokens: int = 500
) -> dict:
    """
    Estima el costo de una llamada ANTES de hacerla.
    Útil para validar que el input no es excesivamente caro.
    """
    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
    }

# Ejemplo de uso para rechazar requests muy caros:
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 demasiado caro estimado: ${estimate['estimated_cost_usd']:.4f} "
            f"(máximo: ${max_cost_usd}). "
            f"Input tokens: {estimate['estimated_input_tokens']}"
        )

Análisis de costos desde los logs

# scripts/analyze_costs.py
# Script para analizar costos desde los 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:
    """Carga logs desde un archivo JSON Lines."""
    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:
    """Análisis completo de costos desde los logs."""
    
    # Filtrar por fecha si se especifica
    if date_filter:
        logs = [l for l in logs if l.get("timestamp", "").startswith(date_filter)]
    
    # Solo logs de requests completados con costo
    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)
    
    # Por modelo
    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 requests más caros
    top_expensive = sorted(cost_logs, key=lambda l: l["total_cost_usd"], reverse=True)[:10]
    
    # Distribución de costos
    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))

Comparación de costos entre modelos

# Ayuda a tomar decisiones de qué modelo usar:

COST_COMPARISON = {
    "Análisis de sentimiento (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 por request
# gpt-4o:       $0.0008750 por request  (8.3x más caro)
# gpt-4-turbo:  $0.0040000 por request  (38x más caro)

# Para 10,000 requests/mes:
# gpt-4o-mini:  $1.05/mes
# gpt-4o:       $8.75/mes
# gpt-4-turbo:  $40.00/mes

# Conclusión: para análisis de sentimiento simple, gpt-4o-mini es la elección obvia
# Solo usar gpt-4o para casos que genuinamente requieren más capacidad

Tests del cost tracker

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

def test_gpt4o_mini_cost():
    """Verifica cálculo de costo para 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():
    """Modelos desconocidos usan el pricing por defecto sin crash."""
    cost = calculate_cost("gpt-unknown-model-2099", 100, 100)
    assert cost.total_cost_usd > 0  # No crashea, retorna algo razonable

@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):
    """Verifica la tabla de precios para los modelos más importantes."""
    cost = calculate_cost(model, inp, out)
    assert abs(cost.total_cost_usd - expected_total) < 0.01

Ejercicios

Ejercicio 1: Calcular costos manualmente

Para cada escenario, calcula el costo antes de ejecutar el código:

  1. Análisis de un tweet (50 input tokens, 30 output tokens) con gpt-4o-mini
  2. Resumen de un artículo (2,000 input tokens, 500 output tokens) con gpt-4o-mini
  3. El mismo resumen con gpt-4o
Ver solución

Escenario 1: Tweet con 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)

Escenario 2: Artículo con 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

Escenario 3: Artículo con 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 más caro que gpt-4o-mini para este caso)

Ejercicio 2: Detectar el request outlier

Tienes estos logs de costo. ¿Cuál es el outlier y cuál podría ser la causa?

{"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}
Ver solución

El 4to log tiene 65,000 input tokens vs un promedio de ~300 para los demás. Eso es un outlier de más de 200x.

Causas posibles:

  1. El sanitizador no truncó el input correctamente
  2. El usuario subió un documento entero en lugar de un fragmento
  3. Hay un bug en la construcción del prompt (context acumulado, loop sin reset)

Costo del outlier: ~225× el promedio, lo que significa que un solo request de este tipo cuesta lo mismo que 225 requests normales.

Ejercicio 3: Política de alertas

Define umbrales de alerta para una app con estas características:

  • Costo promedio por request: $0.0002
  • Requests/hora promedio: 100
  • Budget mensual: $100
Ver guía

Budget esperado: $0.0002 × 100 × 24 × 30 = $14.40/mes (mucho menos que $100)

Umbrales razonables:

  • high_cost_per_request: > $0.02 (100× el promedio) → WARNING
  • very_high_cost_per_request: > $0.10 (500× el promedio) → ERROR
  • hourly_cost_alert: > $5/hora (vs $0.02/hora normal) → ERROR inmediato
  • daily_cost_alert: > $20/día (vs $0.48/día normal) → WARNING
  • monthly_budget_alert: > $80 (80% del budget) → ERROR, notificar equipo

Resumen

  • response.usage es la fuente de verdad para tokens — úsalo siempre que esté disponible
  • Centraliza los precios en una tabla actualizable, no los hardcodees en el código de negocio
  • Logguea costo en cada request con input_cost_usd, output_cost_usd, total_cost_usd
  • Alertas de costo permiten detectar outliers antes de que impacten la factura
  • tiktoken permite estimar tokens antes de llamar al API — útil para validar inputs
  • Scripts de análisis sobre los JSON logs revelan patterns de costo que son invisibles sin logging

Recursos adicionales

  1. OpenAI Pricing — Precios actualizados (revisar periódicamente)
  2. tiktoken GitHub — Counting tokens before API calls
  3. OpenAI Usage Dashboard — Dashboard oficial de uso
  4. LangSmith — Plataforma de observabilidad AI con cost tracking incorporado
  5. Helicone — Proxy que añade observabilidad (incluyendo cost tracking) sin cambiar código
← Anterior3. Request Tracing con Correlation IDsSiguiente →5. JSON Logs con structlog