Cost Attribution por Dimensión

Descripción de la cápsula

Saber que tu sistema AI cuesta $2,400/mes no te dice nada útil. Es como saber que tu restaurante gastó $10,000 en ingredientes sin saber cuánto fue carne, cuánto vegetales, y cuánto se tiró a la basura. Para optimizar necesitas granularidad: ¿qué endpoint genera el mayor gasto? ¿Qué usuario consume más recursos? ¿Qué modelo domina la factura? ¿Los embeddings o los completions?

Cost attribution es el arte de etiquetar cada centavo con sus dimensiones de origen. No cambias cuánto gastas — cambias cuánto entiendes sobre lo que gastas. Y esa comprensión es lo que transforma "tengo que reducir costos" en "tengo que optimizar el endpoint /summarize que genera el 40% del gasto con solo el 15% del tráfico."

En esta cápsula vas a construir un sistema de cost attribution multidimensional. Cada request a tu sistema AI será etiquetada con endpoint, usuario, modelo y tipo de operación. Al terminar, podrás identificar los cost hotspots de tu sistema — ese 20% de endpoints o usuarios que genera el 80% del gasto.

¿Qué es cost attribution?

De "cuánto" a "dónde"

Imagina que recibes la factura de OpenAI: $2,400. En el Módulo 1 aprendiste a descomponerla por tipo de costo: tokens de input, tokens de output, embeddings, retries. Eso responde cuánto en cada categoría.

Cost attribution agrega una capa completamente diferente: dónde se origina cada costo. Es la diferencia entre estos dos reportes:

Reporte sin attribution:
  Total: $2,400/mes
  - Completions: $1,500
  - Embeddings: $480
  - Retries: $300
  - Storage: $120

Reporte con attribution:
  Total: $2,400/mes

  Por endpoint:
  - /chat:      $1,080 (45%)  ← cost hotspot
  - /summarize: $480  (20%)
  - /search:    $360  (15%)
  - /classify:  $240  (10%)
  - Otros:      $240  (10%)

  Por modelo:
  - GPT-4o:      $1,800 (75%)
  - GPT-4o-mini: $360   (15%)
  - Embeddings:  $240   (10%)

El primer reporte te dice "gasto mucho en completions." El segundo te dice "el endpoint /chat con GPT-4o genera casi la mitad de mi factura." Con el segundo puedes actuar.

Las cuatro dimensiones fundamentales

En un sistema AI, hay cuatro dimensiones que capturan el 95% de la información de attribution que necesitas:

Dimensión        Pregunta que responde          Ejemplo
──────────       ──────────────────────          ───────
Endpoint         ¿Qué funcionalidad gasta más?  /chat vs /search vs /summarize
Usuario          ¿Quién genera más costo?        user_847 vs promedio
Modelo           ¿Qué modelo domina la factura?  GPT-4o vs GPT-4o-mini
Tipo de op.      ¿Completions o embeddings?      completion vs embedding vs image

Puedes agregar más dimensiones según tu caso (por equipo, por tenant, por región), pero estas cuatro cubren la mayoría de escenarios.

Implementación del CostTracker

CostRecord y CostTracker

from dataclasses import dataclass, field
from datetime import datetime
from collections import defaultdict
import statistics


@dataclass
class CostRecord:
    """Un registro individual de costo con todas sus dimensiones."""
    timestamp: datetime
    endpoint: str
    user_id: str
    model: str
    operation_type: str
    input_tokens: int
    output_tokens: int
    cost: float
    metadata: dict = field(default_factory=dict)


PRICING_PER_1M = {
    "gpt-4o":      {"input": 2.50, "output": 10.00},
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},
    "gpt-4-turbo": {"input": 10.00, "output": 30.00},
    "o3-mini":     {"input": 1.10, "output": 4.40},
    "text-embedding-3-small": {"input": 0.02, "output": 0.0},
    "text-embedding-3-large": {"input": 0.13, "output": 0.0},
}


class CostTracker:
    """Tracker de costos con attribution multidimensional."""

    def __init__(self):
        self.records: list[CostRecord] = []

    def record(
        self,
        endpoint: str,
        user_id: str,
        model: str,
        operation_type: str,
        input_tokens: int,
        output_tokens: int = 0,
        metadata: dict | None = None,
    ) -> CostRecord:
        """Registra una operación con su costo calculado."""
        pricing = PRICING_PER_1M.get(model)
        if not pricing:
            raise ValueError(f"Modelo no soportado: {model}")

        cost = (
            (input_tokens / 1_000_000) * pricing["input"]
            + (output_tokens / 1_000_000) * pricing["output"]
        )

        record = CostRecord(
            timestamp=datetime.now(),
            endpoint=endpoint,
            user_id=user_id,
            model=model,
            operation_type=operation_type,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            cost=round(cost, 8),
            metadata=metadata or {},
        )
        self.records.append(record)
        return record

    def total_cost(self) -> float:
        return round(sum(r.cost for r in self.records), 6)

    def cost_by(self, dimension: str) -> dict[str, float]:
        """Agrupa costos por cualquier dimensión."""
        grouped: dict[str, float] = defaultdict(float)
        for r in self.records:
            key = getattr(r, dimension)
            grouped[key] += r.cost
        return dict(sorted(grouped.items(), key=lambda x: x[1], reverse=True))

    def cost_by_two(self, dim1: str, dim2: str) -> dict[str, dict[str, float]]:
        """Agrupa costos por dos dimensiones (cross-tabulation)."""
        grouped: dict[str, dict[str, float]] = defaultdict(lambda: defaultdict(float))
        for r in self.records:
            k1 = getattr(r, dim1)
            k2 = getattr(r, dim2)
            grouped[k1][k2] += r.cost
        return {k1: dict(v) for k1, v in grouped.items()}

tracker = CostTracker()

tracker.record("/chat", "user_847", "gpt-4o", "completion", 1500, 400)
tracker.record("/chat", "user_312", "gpt-4o", "completion", 1200, 350)
tracker.record("/summarize", "user_847", "gpt-4o", "completion", 3000, 800)
tracker.record("/search", "user_091", "gpt-4o-mini", "completion", 800, 200)
tracker.record("/search", "user_312", "text-embedding-3-small", "embedding", 500)
tracker.record("/classify", "user_558", "gpt-4o-mini", "completion", 200, 10)

print(f"Costo total: ${tracker.total_cost()}")
print(f"\nPor endpoint: {tracker.cost_by('endpoint')}")
print(f"\nPor usuario: {tracker.cost_by('user_id')}")
print(f"\nPor modelo: {tracker.cost_by('model')}")

# Output esperado:
Costo total: $0.022295

Por endpoint: {'/summarize': 0.0155, '/chat': 0.006775, '/search': 0.00024, '/classify': 1.2e-05}
Por usuario: {'user_847': 0.019275, 'user_312': 0.00301, 'user_091': 0.00012, 'user_558': 1.2e-05}
Por modelo: {'gpt-4o': 0.022275, 'gpt-4o-mini': 0.00013, 'text-embedding-3-small': 1e-05}

Ya puedes ver los patrones: /summarize es el endpoint más caro a pesar de tener menos requests, user_847 consume el 86% del gasto total, y GPT-4o domina la factura.

Cross-tabulation: cruzando dimensiones

La attribution por una dimensión es útil. Cruzar dos dimensiones es donde el análisis se vuelve poderoso.

cross = tracker.cost_by_two("endpoint", "model")
for endpoint, models in cross.items():
    print(f"\n{endpoint}:")
    for model, cost in models.items():
        print(f"  {model}: ${cost:.6f}")

# Output esperado:
/chat:
  gpt-4o: $0.006775

/summarize:
  gpt-4o: $0.015500

/search:
  gpt-4o-mini: $0.000230
  text-embedding-3-small: $0.000010

/classify:
  gpt-4o-mini: $0.000012

Ahora ves que /summarize usa exclusivamente GPT-4o. Una pregunta inmediata: ¿podría /summarize funcionar con GPT-4o-mini para ciertos tipos de documentos? Si la respuesta es sí, el ahorro es significativo.

Automatización: decorador de cost tracking

Registrar costos manualmente es útil para entender el concepto, pero en producción necesitas automatización. Un decorador intercepta cada llamada a la API y registra el costo automáticamente.

import functools
from typing import Callable


def track_cost(
    tracker: CostTracker,
    endpoint: str,
    model: str = "gpt-4o",
    operation_type: str = "completion",
):
    """Decorador que registra automáticamente el costo de una función."""
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            user_id = kwargs.get("user_id", "anonymous")
            result = func(*args, **kwargs)

            input_tokens = result.get("usage", {}).get("prompt_tokens", 0)
            output_tokens = result.get("usage", {}).get("completion_tokens", 0)

            record = tracker.record(
                endpoint=endpoint, user_id=user_id, model=model,
                operation_type=operation_type,
                input_tokens=input_tokens, output_tokens=output_tokens,
            )
            result["_cost_record"] = {"cost": record.cost, "endpoint": endpoint}
            return result
        return wrapper
    return decorator


tracked_tracker = CostTracker()


@track_cost(tracked_tracker, endpoint="/chat", model="gpt-4o")
def chat_completion(message: str, user_id: str = "anonymous") -> dict:
    return {"content": f"Respuesta para: {message}",
            "usage": {"prompt_tokens": 500, "completion_tokens": 200}}


@track_cost(tracked_tracker, endpoint="/summarize", model="gpt-4o")
def summarize_text(text: str, user_id: str = "anonymous") -> dict:
    return {"content": "Resumen del texto...",
            "usage": {"prompt_tokens": 3000, "completion_tokens": 600}}


@track_cost(tracked_tracker, endpoint="/classify", model="gpt-4o-mini")
def classify_text(text: str, user_id: str = "anonymous") -> dict:
    return {"content": "positive",
            "usage": {"prompt_tokens": 200, "completion_tokens": 5}}


chat_completion("¿Cómo funciona Python?", user_id="user_100")
chat_completion("Explica decoradores", user_id="user_200")
summarize_text("Texto largo de ejemplo...", user_id="user_100")
classify_text("Gran producto", user_id="user_300")

print(f"Total: ${tracked_tracker.total_cost()}")
for endpoint, cost in tracked_tracker.cost_by("endpoint").items():
    pct = cost / tracked_tracker.total_cost() * 100
    print(f"  {endpoint}: ${cost:.6f} ({pct:.1f}%)")

# Output esperado:
Total: $0.016533
  /summarize: $0.013500 (81.7%)
  /chat: $0.003000 (18.1%)
  /classify: $0.000033 (0.2%)

El decorador captura automáticamente los tokens reportados por la API y registra el costo con todas sus dimensiones. No necesitas modificar la lógica de negocio — solo agregas el decorador.

Identificando cost hotspots

La regla 80/20 en costos AI

En la mayoría de sistemas AI, el 20% de los endpoints (o usuarios, o modelos) generan el 80% del costo. Identificar esos hotspots es la prioridad número uno antes de optimizar.

def find_hotspots(
    tracker: CostTracker,
    dimension: str,
    threshold: float = 0.8,
) -> dict:
    """Identifica los elementos de una dimensión que acumulan el threshold% del costo."""
    costs = tracker.cost_by(dimension)
    total = sum(costs.values())

    if total == 0:
        return {"hotspots": [], "coverage": 0, "total": 0}

    hotspots = []
    cumulative = 0.0

    for key, cost in costs.items():
        cumulative += cost
        hotspots.append({
            "key": key,
            "cost": round(cost, 6),
            "percentage": round(cost / total * 100, 1),
            "cumulative_pct": round(cumulative / total * 100, 1),
        })
        if cumulative / total >= threshold:
            break

    return {
        "hotspots": hotspots,
        "concentration": f"{len(hotspots)}/{len(costs)} generan {threshold*100:.0f}% del costo",
        "total_cost": round(total, 6),
    }


result = find_hotspots(tracked_tracker, "endpoint")
print(f"Concentración: {result['concentration']}")
for h in result["hotspots"]:
    print(f"  {h['key']}: ${h['cost']} ({h['percentage']}%) — acum: {h['cumulative_pct']}%")

# Output esperado:
Concentración: 1/3 generan 80% del costo
  /summarize: $0.0135 (81.7%) — acum: 81.7%

Ahora sabes que optimizar solo /summarize cubriría más del 80% del impacto potencial.

Comparación: Flat tracking vs Multi-dimensional attribution

Cuándo flat tracking es suficiente

Si tu sistema tiene un solo endpoint, un solo modelo, y pocos usuarios, flat tracking puede ser suficiente. Pero cualquier sistema con más de 2-3 endpoints se beneficia de attribution multidimensional.

Conexión con el proyecto

Datos atribuidos = dashboards accionables

El Cost Dashboard Integration (proyecto de este módulo) necesita datos con dimensiones para ser útil. Un dashboard que solo muestra "costo total por día" es un gráfico que nadie mira. Un dashboard que muestra "costo por endpoint con tendencia semanal" genera decisiones.

El CostTracker que construiste aquí produce exactamente los datos que la cápsula 03 (Diseño de Cost Dashboards) necesita para crear visualizaciones accionables. Y el find_hotspots es la función que identifica dónde enfocar el esfuerzo de optimización.

Conexión con el baseline (cápsula 07)

El baseline formal de costos requiere attribution. "Mi sistema cuesta $2,400/mes" no es un baseline riguroso. "Mi sistema cuesta $2,400/mes, distribuido 45% en /chat, 20% en /summarize, 15% en /search" — eso sí es un baseline que permite medir el impacto de cada optimización individual.

Troubleshooting

Problema 1: "Los costos calculados no coinciden con mi factura de OpenAI"

Causa: Tu cálculo usa los precios base por token, pero la factura real incluye overhead adicional. OpenAI cobra cached input tokens al 50%, y puede haber discrepancias por redondeo en volúmenes altos.

Solución: Usa los costos calculados como referencia relativa (para comparar entre endpoints/usuarios) más que como valor absoluto. Para reconciliar con la factura, agrega un factor de ajuste de 1.05-1.10x.

Problema 2: "El user_id no siempre está disponible en la request"

Causa: Requests no autenticadas o sistemas donde el user ID llega en headers inconsistentes.

Solución: Define un user_id fallback. En el CostTracker, usa "anonymous" como default. Si "anonymous" se convierte en un hotspot, necesitas mejorar la identificación de usuarios.

Problema 3: "Tengo demasiados endpoints, el reporte es ilegible"

Causa: APIs con muchas rutas parametrizadas (/users/123/tasks/456) generan keys únicos por cada combinación.

Solución: Normaliza los paths antes de registrar:

import re

def normalize_path(path: str) -> str:
    """Reemplaza IDs numéricos y UUIDs con placeholders."""
    path = re.sub(r'/\d+', '/{id}', path)
    path = re.sub(
        r'/[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}',
        '/{uuid}', path
    )
    return path

print(normalize_path("/users/123/tasks/456"))
print(normalize_path("/docs/a1b2c3d4-e5f6-7890-abcd-ef1234567890/summary"))

# Output esperado:
/users/{id}/tasks/{id}
/docs/{uuid}/summary

Problema 4: "El CostTracker consume mucha memoria con millones de records"

Causa: Almacenar cada CostRecord en una lista en memoria no escala a producción.

Solución: Para desarrollo, la lista en memoria es suficiente. Para producción, escribe los records a una base de datos o sistema de métricas (Prometheus counters, PostgreSQL, logs). El CostTracker se convierte en una fachada que emite eventos en vez de acumular datos.

Ejercicios

Ejercicio 1: Attribution básica con datos dados (Fácil)

Tienes un sistema con estos datos de un día:

/chat:      200 requests, GPT-4o, avg 1000 input / 300 output
/search:    500 requests, GPT-4o-mini, avg 600 input / 150 output
/summarize: 50 requests, GPT-4o, avg 4000 input / 1000 output

Sin código, calcula el costo diario por endpoint y determina cuál es el hotspot.

/chat (GPT-4o: $2.50/1M input, $10.00/1M output):
  Input:  200 × 1000 = 200K tokens → 200K/1M × $2.50 = $0.50
  Output: 200 × 300 = 60K tokens → 60K/1M × $10.00 = $0.60
  Total: $1.10

/search (GPT-4o-mini: $0.15/1M input, $0.60/1M output):
  Input:  500 × 600 = 300K tokens → 300K/1M × $0.15 = $0.045
  Output: 500 × 150 = 75K tokens → 75K/1M × $0.60 = $0.045
  Total: $0.09

/summarize (GPT-4o: $2.50/1M input, $10.00/1M output):
  Input:  50 × 4000 = 200K tokens → 200K/1M × $2.50 = $0.50
  Output: 50 × 1000 = 50K tokens → 50K/1M × $10.00 = $0.50
  Total: $1.00

Hotspot: /chat y /summarize (95.9% del costo).
/search tiene 2.5x más requests pero cuesta 12x menos gracias a GPT-4o-mini.

Ejercicio 2: Construir un reporte de attribution (Fácil)

Usa el CostTracker para registrar 8 operaciones mixtas (al menos 3 endpoints, 3 usuarios, 2 modelos) y genera un reporte que muestre cost_by endpoint, user y model.

tracker = CostTracker()

tracker.record("/chat", "alice", "gpt-4o", "completion", 1000, 300)
tracker.record("/chat", "bob", "gpt-4o", "completion", 1200, 400)
tracker.record("/chat", "alice", "gpt-4o", "completion", 900, 250)
tracker.record("/summarize", "charlie", "gpt-4o", "completion", 5000, 1200)
tracker.record("/summarize", "alice", "gpt-4o", "completion", 4000, 900)
tracker.record("/search", "bob", "gpt-4o-mini", "completion", 500, 100)
tracker.record("/search", "charlie", "gpt-4o-mini", "completion", 600, 120)
tracker.record("/classify", "bob", "gpt-4o-mini", "completion", 200, 5)

total = tracker.total_cost()
print(f"=== Cost Attribution Report ===\nTotal: ${total}\n")

for dim_name, dim_field in [("Endpoint", "endpoint"), ("Usuario", "user_id"), ("Modelo", "model")]:
    print(f"--- Por {dim_name} ---")
    for key, cost in tracker.cost_by(dim_field).items():
        print(f"  {key}: ${cost:.6f} ({cost/total*100:.1f}%)")
    print()

Ejercicio 3: Cross-tabulation endpoint × modelo (Medio)

Usa cost_by_two("endpoint", "model") para generar una tabla cruzada del ejercicio anterior. Identifica la combinación más costosa y propón una optimización.

cross = tracker.cost_by_two("endpoint", "model")
total = tracker.total_cost()

print(f"{'Endpoint':<15} {'Modelo':<25} {'Costo':>12} {'%':>8}")
print("-" * 62)

rows = []
for ep, models in cross.items():
    for model, cost in models.items():
        rows.append((ep, model, cost))
rows.sort(key=lambda x: x[2], reverse=True)

for ep, model, cost in rows:
    print(f"{ep:<15} {model:<25} ${cost:>10.6f} {cost/total*100:>6.1f}%")

print(f"\n🔥 Hotspot: {rows[0][0]} + {rows[0][1]}")
print(f"Optimización: Evaluar si {rows[0][0]} puede usar GPT-4o-mini para requests simples.")

Ejercicio 4: Simular un día y detectar anomalías (Difícil)

Simula 100 requests distribuidas en 4 endpoints. Haz que un usuario (user_999) genere 30 requests caras al endpoint /summarize con GPT-4o (simula abuso). Usa find_hotspots para identificarlo.

import random

random.seed(42)
sim_tracker = CostTracker()

normal_users = [f"user_{i:03d}" for i in range(1, 21)]
endpoints = [
    ("/chat", "gpt-4o", 1000, 300),
    ("/search", "gpt-4o-mini", 500, 100),
    ("/summarize", "gpt-4o", 3000, 800),
    ("/classify", "gpt-4o-mini", 200, 5),
]

for _ in range(70):
    ep, model, inp, out = random.choice(endpoints)
    user = random.choice(normal_users)
    sim_tracker.record(ep, user, model, "completion",
                       int(inp * random.uniform(0.8, 1.2)),
                       int(out * random.uniform(0.8, 1.2)))

for _ in range(30):
    sim_tracker.record("/summarize", "user_999", "gpt-4o", "completion",
                       int(4000 * random.uniform(0.9, 1.1)),
                       int(1000 * random.uniform(0.9, 1.1)))

total = sim_tracker.total_cost()
print(f"Total: ${total:.4f} ({len(sim_tracker.records)} requests)\n")

hotspots = find_hotspots(sim_tracker, "user_id", threshold=0.5)
print(f"Concentración (50% del costo): {hotspots['concentration']}")
for h in hotspots["hotspots"]:
    print(f"  {h['key']}: ${h['cost']} ({h['percentage']}%)")

user_costs = sim_tracker.cost_by("user_id")
values = list(user_costs.values())
avg = statistics.mean(values)
std = statistics.stdev(values) if len(values) > 1 else 0

print(f"\n🚨 Anomalías (> avg + 2*std = ${avg + 2*std:.6f}):")
for user, cost in user_costs.items():
    if cost > avg + 2 * std:
        print(f"  {user}: ${cost:.6f} ({cost/avg:.1f}x el promedio)")

Resumen

• ✅ Cost attribution transforma "cuánto gasto" en "dónde gasto" — la diferencia entre un número inútil y un mapa de optimización
• ✅ Las cuatro dimensiones fundamentales (endpoint, usuario, modelo, tipo de operación) capturan el 95% de la información necesaria
• ✅ El CostTracker registra cada request con su costo calculado y todas sus dimensiones
• ✅ La cross-tabulation (cruzar dos dimensiones) revela patrones que una sola dimensión no muestra
• ✅ Decoradores automatizan el tracking sin modificar la lógica de negocio
• ✅ La regla 80/20 aplica consistentemente: pocos endpoints/usuarios generan la mayoría del costo
• ✅ find_hotspots identifica automáticamente los elementos que acumulan el X% del costo
• ✅ Normalizar paths parametrizados es esencial para attribution útil en APIs RESTful

Próxima cápsula: Diseño de Cost Dashboards — transformar estos datos atribuidos en visualizaciones que generan decisiones de optimización.

Recursos adicionales

1. OpenAI Usage API - API oficial para obtener datos de uso y costos
2. OpenAI Pricing - Precios actualizados para validar tus cálculos
3. Prometheus Python Client - Para exponer métricas de costo como counters
4. LiteLLM Cost Tracking - Cost tracking multi-proveedor
5. Pareto Principle in Software (Martin Fowler) - La regla 80/20 en software
6. FastAPI Middleware - Documentación oficial de middleware en FastAPI
7. Datadog AI Cost Tracking - Alternativa managed para cost attribution
8. OpenTelemetry Python - Estándar abierto para instrumentación con cost dimensions

Creado: Marzo 2026 / Versión: 1.0