Cómo monitorear y alertar tus costos de OpenAI en tiempo real con Python y Grafana

Datos y precios verificados al 29 de mayo de 2026. Los precios de la API de OpenAI son susceptibles de cambiar; consulta la página oficial en openai.com/api/pricing antes de tomar decisiones de presupuesto.

Si alguna vez abriste la factura mensual de OpenAI y el número fue mayor al esperado, no eres el único. Monitorear los costos de OpenAI en tiempo real ha pasado de ser un nice-to-have a una necesidad operativa real: según datos de Zylo publicados en abril de 2026, el gasto promedio anual de una organización en la API de OpenAI alcanza los $384,500 USD, y el gasto en aplicaciones nativas de IA creció un 108% durante 2025. Para rematar, el 78% de los líderes de TI reportan haber recibido cargos inesperados ligados a consumo de IA, y el 60% admite no tener visibilidad completa sobre el uso de IA generativa en su organización.

El problema no es solo el costo. Es que el comportamiento en producción es impredecible. Un endpoint que en staging consumía 400 tokens por llamada puede dispararse a 2,000 en producción por culpa de prompts dinámicos mal controlados, usuarios con conversaciones largas, o un bug que repite llamadas al modelo. Sin un sistema de alertas activo, te enteras del problema cuando llega la factura, no cuando ocurre.

En este artículo vas a construir, paso a paso, un pipeline de observabilidad completo: un exporter en Python que consume la Usage API de OpenAI y expone métricas en formato Prometheus, conectado a un dashboard en Grafana con alertas configuradas en Alertmanager. Sin magia negra, sin herramientas de pago. Solo código que puedes entender y mantener.

Por qué los dashboards nativos de OpenAI no son suficientes

La trampa del billing por sorpresa

OpenAI ofrece un panel de uso en platform.openai.com/account/usage que muestra consumo por día, por modelo y por clave API. También puedes configurar un límite de gasto y recibir una notificación por email cuando lo alcanzas. Suena bien. El problema es que ese umbral es reactivo y binario: o no lo has cruzado, o ya lo cruzaste. No hay granularidad de "llevas el 60% del presupuesto diario y el ritmo sugiere que lo excederás a las 3pm".

Además, el dashboard nativo no tiene integración con tus sistemas de alertas (PagerDuty, Slack, OpsGenie), no permite agregar contexto de negocio (¿qué feature del producto genera más tokens?), y tiene una latencia de actualización que puede ser de horas. Para operaciones serias, eso es inaceptable.

Qué métricas realmente importan

Antes de construir cualquier cosa, define qué quieres observar. Las métricas de costo no son solo "cuánto gasté hoy". Las más útiles en producción son:

Costo por modelo y por hora: saber si un spike de costos lo genera GPT-4o o GPT-4o-mini cambia completamente la respuesta.
Tokens de entrada vs. tokens de salida por endpoint: los tokens de salida cuestan más. Si tu ratio output/input sube, algo cambió en tus prompts o en los patrones de uso.
Costo acumulado del ciclo de facturación vs. presupuesto: la única métrica que le importa a finanzas.
Velocidad de gasto (burn rate): cuánto estás gastando por hora en este momento. Con esto puedes proyectar si terminarás el mes dentro del presupuesto.
Errores de tasa (429 rate limit): un indicador indirecto de que algo está generando demasiadas llamadas.

Arquitectura del sistema de monitoreo de costos de OpenAI en tiempo real

Componentes del stack

El stack que vamos a construir tiene tres capas, todas open source y sin costo de licencia:

Python Exporter: un script que corre cada 60 segundos, consulta los endpoints /v1/organization/costs y /v1/organization/usage/completions de OpenAI, y expone las métricas en formato Prometheus a través de un servidor HTTP local en el puerto 8000.
Prometheus: hace scraping del exporter cada minuto, almacena las series de tiempo y evalúa las reglas de alerta.
Grafana: se conecta a Prometheus como datasource, renderiza los dashboards y envía notificaciones vía Alertmanager a Slack, email o PagerDuty.

El flujo de datos


OUTPUT
OpenAI Usage API
      │
      ▼ (cada 60 seg)
Python Exporter  ──► :8000/metrics (formato Prometheus)
      │
      ▼ (scraping cada 60 seg)
Prometheus  ──► Evaluación de reglas de alerta
      │                    │
      ▼                    ▼
Grafana Dashboard    Alertmanager ──► Slack / Email

Este diseño es deliberadamente simple. No hay Kafka, no hay base de datos adicional, no hay infraestructura pesada. Si ya tienes Prometheus y Grafana corriendo en tu organización (como es común en equipos con Kubernetes), solo necesitas agregar el exporter.

Implementación: el exporter de Python que hace el trabajo real

Paso 1 – Preparar el entorno

Necesitas Python 3.10+ y tres dependencias:


bash
pip install openai prometheus-client requests python-dotenv

Crea un archivo .env en la raíz del proyecto:


bash
OPENAI_API_KEY=sk-org-...      # Tu API key de organización (no la personal)
BUDGET_DAILY_USD=50.0          # Presupuesto diario máximo en USD
BUDGET_MONTHLY_USD=1200.0      # Presupuesto mensual máximo en USD
EXPORTER_PORT=8000             # Puerto donde escucha el exporter

Importante: usa una API key de organización (sk-org-...), no una personal. El endpoint /v1/organization/costs requiere permisos de organización. Puedes generarla desde Organization Settings > API Keys.

Paso 2 – El exporter completo


python
# openai_cost_exporter.py
# Exporter de métricas de costo de OpenAI para Prometheus
# Requiere: openai, prometheus-client, requests, python-dotenv

import os
import time
import requests
from datetime import datetime, timezone
from dotenv import load_dotenv
from prometheus_client import (
    start_http_server,   # Servidor HTTP para exponer métricas
    Gauge,               # Tipo de métrica para valores que suben y bajan
    Counter,             # Tipo de métrica para valores que solo crecen
    REGISTRY,
)

# ─── Carga de configuración ───────────────────────────────────────────────────
load_dotenv()  # Lee variables desde el archivo .env

OPENAI_API_KEY    = os.getenv("OPENAI_API_KEY")
BUDGET_DAILY      = float(os.getenv("BUDGET_DAILY_USD", "50.0"))
BUDGET_MONTHLY    = float(os.getenv("BUDGET_MONTHLY_USD", "1200.0"))
EXPORTER_PORT     = int(os.getenv("EXPORTER_PORT", "8000"))
SCRAPE_INTERVAL   = 60  # Segundos entre cada consulta a la API de OpenAI

# Encabezados HTTP reutilizados en todas las llamadas a la API
HEADERS = {
    "Authorization": f"Bearer {OPENAI_API_KEY}",
    "Content-Type": "application/json",
}

# ─── Definición de métricas Prometheus ───────────────────────────────────────
# Gauge: el costo actual del día en curso (puede bajar si se aplican créditos)
costo_hoy = Gauge(
    "openai_costo_diario_usd",
    "Costo acumulado del día actual en USD",
)

# Gauge: costo del ciclo de facturación mensual en curso
costo_mensual = Gauge(
    "openai_costo_mensual_usd",
    "Costo acumulado del ciclo de facturación mensual en USD",
)

# Gauge: porcentaje del presupuesto diario consumido (0–100+)
pct_presupuesto_diario = Gauge(
    "openai_pct_presupuesto_diario",
    "Porcentaje del presupuesto diario consumido",
)

# Gauge con etiquetas: tokens consumidos, desglosados por modelo y tipo
tokens_por_modelo = Gauge(
    "openai_tokens_total",
    "Tokens consumidos por modelo",
    ["modelo", "tipo"],  # etiquetas: gpt-4o, gpt-4o-mini, etc. | input / output
)

# Gauge: burn rate estimado (USD por hora, promedio últimas 3 horas)
burn_rate_usd_hora = Gauge(
    "openai_burn_rate_usd_hora",
    "Velocidad de gasto estimada en USD por hora",
)

# ─── Función para obtener costos desde la API de OpenAI ──────────────────────
def obtener_costos_diarios() -> dict:
    """
    Consulta el endpoint /v1/organization/costs de OpenAI.
    Devuelve los costos en USD del día actual y los últimos 30 días.
    """
    ahora = int(time.time())
    # Inicio del día actual en UTC (medianoche)
    inicio_dia = int(
        datetime.now(timezone.utc).replace(
            hour=0, minute=0, second=0, microsecond=0
        ).timestamp()
    )
    # Inicio del mes para calcular el acumulado mensual
    inicio_mes = int(
        datetime.now(timezone.utc).replace(
            day=1, hour=0, minute=0, second=0, microsecond=0
        ).timestamp()
    )

    url = "https://api.openai.com/v1/organization/costs"

    # Consulta los costos desde el inicio del mes hasta ahora
    resp = requests.get(
        url,
        headers=HEADERS,
        params={
            "start_time": inicio_mes,     # Unix timestamp: inicio del mes
            "end_time": ahora,            # Unix timestamp: ahora mismo
            "bucket_width": "1d",         # Granularidad: un bucket por día
            "limit": 31,                  # Máximo 31 días (un mes)
        },
        timeout=10,
    )
    resp.raise_for_status()  # Lanza excepción si la respuesta no es 2xx
    datos = resp.json()

    total_mes = 0.0
    total_hoy = 0.0
    historial_horas = []  # Para calcular el burn rate

    # Itera sobre cada bucket diario devuelto por la API
    for bucket in datos.get("data", []):
        # El campo 'amount' contiene el objeto con valor y divisa
        valor = bucket.get("amount", {}).get("value", 0.0)
        ts    = bucket.get("start_time", 0)   # Unix timestamp del inicio del bucket

        total_mes += valor  # Suma todos los días para obtener el total mensual

        if ts >= inicio_dia:
            total_hoy = valor  # Solo el bucket del día actual
        
        historial_horas.append(valor)

    return {
        "total_hoy": total_hoy,
        "total_mes": total_mes,
        "historial": historial_horas,
    }

# ─── Función para obtener tokens por modelo ──────────────────────────────────
def obtener_tokens_por_modelo():
    """
    Consulta /v1/organization/usage/completions agrupado por modelo.
    Devuelve una lista de tuplas (modelo, tipo, cantidad_tokens).
    """
    ahora = int(time.time())
    inicio_dia = int(
        datetime.now(timezone.utc).replace(
            hour=0, minute=0, second=0, microsecond=0
        ).timestamp()
    )

    resp = requests.get(
        "https://api.openai.com/v1/organization/usage/completions",
        headers=HEADERS,
        params={
            "start_time": inicio_dia,
            "bucket_width": "1h",      # Granularidad horaria para el burn rate
            "group_by": ["model"],     # Agrupar resultados por modelo
            "limit": 24,               # Últimas 24 horas
        },
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json().get("data", [])

# ─── Función principal de recolección y actualización de métricas ─────────────
def recolectar_metricas():
    """
    Orquesta la recolección de datos y actualiza todos los Gauge de Prometheus.
    Esta función se llama en cada ciclo del loop principal.
    """
    try:
        # 1. Costos diarios y mensuales
        costos = obtener_costos_diarios()
        costo_hoy.set(costos["total_hoy"])
        costo_mensual.set(costos["total_mes"])
        
        # Porcentaje del presupuesto diario consumido
        pct = (costos["total_hoy"] / BUDGET_DAILY * 100) if BUDGET_DAILY > 0 else 0
        pct_presupuesto_diario.set(pct)

        # Burn rate: promedio de los últimos 3 días dividido entre 24 horas
        historial = costos["historial"]
        if len(historial) >= 3:
            promedio_diario = sum(historial[-3:]) / 3
            burn_rate_usd_hora.set(promedio_diario / 24)

        # 2. Tokens por modelo
        uso = obtener_tokens_por_modelo()
        for bucket in uso:
            for resultado in bucket.get("results", []):
                modelo = resultado.get("model_id", "desconocido")
                input_tokens  = resultado.get("input_tokens", 0)
                output_tokens = resultado.get("output_tokens", 0)
                # Actualiza el Gauge con la etiqueta correspondiente
                tokens_por_modelo.labels(modelo=modelo, tipo="input").set(input_tokens)
                tokens_por_modelo.labels(modelo=modelo, tipo="output").set(output_tokens)

        print(f"[OK] Métricas actualizadas | Hoy: ${costos['total_hoy']:.4f} | Mes: ${costos['total_mes']:.2f}")

    except requests.exceptions.HTTPError as e:
        # Error de autenticación o límite de tasa de la API de OpenAI
        print(f"[ERROR HTTP] {e.response.status_code}: {e.response.text}")
    except Exception as e:
        print(f"[ERROR] {type(e).__name__}: {e}")

# ─── Punto de entrada ─────────────────────────────────────────────────────────
if __name__ == "__main__":
    print(f"Iniciando exporter en puerto {EXPORTER_PORT}...")
    start_http_server(EXPORTER_PORT)  # Expone /metrics en localhost:8000
    print("Servidor de métricas activo. Iniciando ciclo de recolección.")

    while True:
        recolectar_metricas()          # Actualiza todos los Gauge
        time.sleep(SCRAPE_INTERVAL)    # Espera 60 segundos antes del siguiente ciclo

El script hace exactamente una cosa por ciclo: habla con la API de OpenAI, transforma los datos y actualiza los Gauge de Prometheus. Los Gauge quedan disponibles en http://localhost:8000/metrics en el formato que Prometheus espera. No hay base de datos propia, no hay estado complejo: Prometheus es quien almacena y consulta la serie temporal.

Configurando Prometheus, Grafana y Alertmanager para costos de OpenAI

Prometheus: scraping y reglas de alerta

Agrega el job del exporter a tu prometheus.yml:


yaml
# prometheus.yml – añade este bloque al array scrape_configs
scrape_configs:
  - job_name: 'openai-cost-exporter'
    scrape_interval: 60s        # Igual que el SCRAPE_INTERVAL del exporter
    static_configs:
      - targets: ['localhost:8000']  # O la IP del host donde corre el exporter

Para las alertas, crea un archivo openai_alerts.yml y referencíalo desde prometheus.yml bajo rule_files:


yaml
# openai_alerts.yml
groups:
  - name: openai_costos
    rules:
      # Alerta si el gasto del día supera el 80% del presupuesto
      - alert: OpenAICostoAlto
        expr: openai_pct_presupuesto_diario > 80
        for: 5m       # Debe mantenerse por 5 minutos para evitar falsos positivos
        labels:
          severity: warning
        annotations:
          summary: "Gasto diario de OpenAI al {{ $value | printf \"%.0f\" }}%"
          description: "El gasto de hoy ha superado el 80% del presupuesto diario definido."

      # Alerta crítica al 100% del presupuesto diario
      - alert: OpenAIPresupuestoExcedido
        expr: openai_pct_presupuesto_diario >= 100
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "¡Presupuesto diario de OpenAI EXCEDIDO!"
          description: "El gasto actual es ${{ $value | printf \"%.2f\" }} USD."

      # Alerta si el burn rate proyecta exceder el presupuesto mensual
      - alert: OpenAIBurnRatePeligroso
        expr: (openai_burn_rate_usd_hora * 24 * 30) > 1200
        for: 15m
        labels:
          severity: warning
        annotations:
          summary: "Burn rate de OpenAI proyecta exceder el presupuesto mensual"

Dashboard en Grafana

Grafana tiene una integración oficial mantenida por el equipo de Grafana Labs que incluye dashboards preconfigurados y 5 alertas base para OpenAI. Puedes instalarla desde Connections > OpenAI en tu instancia de Grafana Cloud o autohospedada. Si prefieres conectar tu propio Prometheus, agrega el datasource y crea paneles con estas queries PromQL:


promql
# Costo acumulado del día en USD
openai_costo_diario_usd

# Burn rate proyectado al mes (en USD)
openai_burn_rate_usd_hora * 24 * 30

# Tokens de salida en el último minuto, por modelo
rate(openai_tokens_total{tipo="output"}[1m])

# Porcentaje del presupuesto diario consumido
openai_pct_presupuesto_diario

El panel más útil en la práctica es el de burn rate proyectado: toma el ritmo de gasto de las últimas horas y lo extrapola al mes. Si a las 10am ya ves que el mes va a costar el doble del presupuesto, puedes intervenir horas antes de que el problema escale.

Notificaciones a Slack con Alertmanager


yaml
# alertmanager.yml
global:
  slack_api_url: 'https://hooks.slack.com/services/TU_WEBHOOK_URL'

route:
  receiver: 'slack-openai'
  group_by: ['alertname']
  group_wait: 30s
  group_interval: 5m
  repeat_interval: 4h        # No spamear: repite alerta cada 4 horas

receivers:
  - name: 'slack-openai'
    slack_configs:
      - channel: '#alertas-infra'
        title: '{{ .CommonAnnotations.summary }}'
        text: '{{ .CommonAnnotations.description }}'
        send_resolved: true  # Notifica también cuando la alerta se resuelve

Con esto, cuando el exporter detecte que el gasto supera el 80% del presupuesto diario, Prometheus lo captura, Alertmanager lo enruta y tu canal de Slack recibe una notificación con el contexto necesario para actuar.

Referencia rápida de precios por modelo (mayo 2026)

Para configurar bien tus presupuestos, aquí tienes los precios actuales de los modelos más usados en producción:

Modelo	Input / 1M tokens	Output / 1M tokens	Contexto
GPT-5.5 (flagship)	$5.00	$30.00	—
GPT-4.1	$2.00	$8.00	1M tokens
GPT-4o	$2.50	$10.00	128K tokens
GPT-4o-mini	$0.15	$0.60	128K tokens
GPT-5.4 Nano	$0.20	$1.25	—

El Batch API ofrece un 50% de descuento en todos los modelos a cambio de procesamiento asíncrono en 24 horas, ideal para pipelines de embeddings o clasificación masiva que no requieren respuesta inmediata.

Conclusión: observabilidad de costos de IA no es opcional en 2026

El gasto en APIs de IA ya no es una línea pequeña en el presupuesto de infraestructura: es, en muchos equipos, el segundo o tercer mayor costo de operación. Construir un sistema de monitoreo como el que describimos aquí, con un exporter Python de menos de 150 líneas más Prometheus y Grafana, te da visibilidad en tiempo real, alertas accionables y el contexto suficiente para intervenir antes de que un problema de costos se convierta en un problema de negocio.

Los siguientes pasos concretos para ti:

Despliega el exporter en el mismo host o pod donde corren tus servicios que consumen la API.
Define presupuestos reales basados en tu consumo histórico más un 20% de margen.
Conecta Alertmanager a tu canal de Slack o PagerDuty con una regla al 80% y otra al 100%.
Revisa el dashboard semanalmente para detectar modelos o features que consumen desproporcionadamente. Si tu equipo ya tiene Prometheus y Grafana para infraestructura, el costo de implementar esto es menos de 2 horas de trabajo. El ROI de evitar un solo spike de costos no planeado lo justifica de sobra.

¿Quieres ir más lejos? Añade una dimensión de costo por usuario o por feature interceptando las llamadas en un middleware que inyecte metadatos en los logs de Loki, y tendrás un sistema de FinOps de IA de nivel enterprise sin pagar por herramientas SaaS adicionales.