Monitoreo, Cost Management y FinOps para IA
El Costo de la IA en Producción
Las aplicaciones de IA generativa tienen costos que escalan rápidamente si no se monitorean:
Costo por query típico en producción:
├── Embedding query: $0.0001 (text-embedding-3-small)
├── Vector search: $0.0002 (OpenSearch/Pinecone)
├── Embedding retrieve: $0.0001 (re-embed chunks)
├── LLM generation: $0.005 (GPT-4o, ~500 tokens out)
└── Total: ~$0.006 por query
Con 100K queries/día = $600/día = $18,000/mes
Con agentes (3x LLM calls): ~$54,000/mes
CloudWatch para Monitoreo de IA
Métricas Custom
import boto3
cloudwatch = boto3.client("cloudwatch")
def publish_ai_metrics(query_id: str, metrics: dict):
cloudwatch.put_metric_data(
Namespace="AIService/RAG",
MetricData=[
{
"MetricName": "QueryLatency",
"Value": metrics["latency_ms"],
"Unit": "Milliseconds",
"Dimensions": [
{"Name": "Model", "Value": metrics["model"]},
{"Name": "Environment", "Value": "production"},
],
},
{
"MetricName": "TokensUsed",
"Value": metrics["total_tokens"],
"Unit": "Count",
"Dimensions": [
{"Name": "Model", "Value": metrics["model"]},
{"Name": "TokenType", "Value": "total"},
],
},
{
"MetricName": "QueryCost",
"Value": metrics["cost_usd"],
"Unit": "None",
"Dimensions": [
{"Name": "Model", "Value": metrics["model"]},
],
},
{
"MetricName": "RetrievalScore",
"Value": metrics["avg_retrieval_score"],
"Unit": "None",
"Dimensions": [
{"Name": "Pipeline", "Value": "rag-v2"},
],
},
],
)
Alarmas
cloudwatch.put_metric_alarm(
AlarmName="HighAICost",
MetricName="QueryCost",
Namespace="AIService/RAG",
Statistic="Sum",
Period=3600, # 1 hora
EvaluationPeriods=1,
Threshold=100.0, # $100/hora
ComparisonOperator="GreaterThanThreshold",
AlarmActions=["arn:aws:sns:us-east-1:xxx:ai-alerts"],
AlarmDescription="Costo de IA supera $100/hora",
)
cloudwatch.put_metric_alarm(
AlarmName="HighLatency",
MetricName="QueryLatency",
Namespace="AIService/RAG",
Statistic="p99",
Period=300, # 5 minutos
EvaluationPeriods=3, # 3 períodos consecutivos
Threshold=30000, # 30 segundos
ComparisonOperator="GreaterThanThreshold",
AlarmActions=["arn:aws:sns:us-east-1:xxx:ai-alerts"],
)
Estrategias de Optimización de Costos
1. Model Routing (Cascading)
class ModelRouter:
"""Enruta queries al modelo más económico que pueda responderlas."""
MODELS = {
"simple": {"model": "gpt-4o-mini", "cost_1k_tokens": 0.00015},
"standard": {"model": "gpt-4o", "cost_1k_tokens": 0.005},
"complex": {"model": "claude-sonnet", "cost_1k_tokens": 0.003},
}
def __init__(self, classifier_llm):
self.classifier = classifier_llm
def route(self, query: str) -> str:
complexity = self.classifier.invoke(
f"Clasifica la complejidad de esta query como 'simple', 'standard' o 'complex':\n{query}"
).content.strip().lower()
return self.MODELS.get(complexity, self.MODELS["standard"])["model"]
2. Caching de Respuestas
import hashlib
import redis
class ResponseCache:
def __init__(self, redis_client: redis.Redis, ttl: int = 3600):
self.cache = redis_client
self.ttl = ttl
def get_or_generate(self, query: str, generate_fn) -> dict:
cache_key = f"rag:{hashlib.sha256(query.encode()).hexdigest()}"
cached = self.cache.get(cache_key)
if cached:
return {"answer": cached.decode(), "cached": True, "cost": 0}
result = generate_fn(query)
self.cache.setex(cache_key, self.ttl, result["answer"])
return {**result, "cached": False}
3. Prompt Compression
from llmlingua import PromptCompressor
compressor = PromptCompressor(model_name="microsoft/llmlingua-2-bert-base-multilingual-cased-meetingbank")
# Comprimir contexto recuperado antes de enviarlo al LLM
compressed = compressor.compress_prompt(
context_texts,
instruction="Responde la pregunta basándote en el contexto.",
question=query,
target_token=500, # Reducir a 500 tokens máx
)
# Reduce tokens de contexto ~4x con pérdida mínima de calidad
4. Batching de Requests
import asyncio
from collections import defaultdict
class EmbeddingBatcher:
"""Agrupa requests de embedding para reducir API calls.
Problema: 100 requests individuales = 100 API calls ($$$, rate limits).
Solución: acumular requests y enviar 1 batch call con los 100 textos.
Flujo completo:
embed() → encola (text, future) → espera → process_batches agrupa →
1 API call con N textos → resuelve Futures → cada caller recibe su vector
"""
def __init__(self, embeddings, batch_size=100, max_wait_ms=50):
self.embeddings = embeddings
self.batch_size = batch_size
self.max_wait = max_wait_ms / 1000 # Convertir ms → segundos
self.queue = asyncio.Queue()
self.results = {}
async def embed(self, text: str) -> list[float]:
# asyncio.Future actúa como mecanismo de sincronización:
# permite que embed() espere (await) hasta que process_batches()
# resuelva el future con set_result() — coordinación entre coroutines
future = asyncio.Future()
await self.queue.put((text, future))
return await future # Se bloquea aquí hasta que el batch se procese
async def process_batches(self):
while True:
batch = []
try:
while len(batch) < self.batch_size:
# wait_for con timeout implementa micro-batching:
# espera max_wait ms por el siguiente item; si no llega,
# TimeoutError fuerza el procesamiento del batch parcial
# Esto balancea latencia (no esperar demasiado) vs eficiencia (agrupar más)
item = await asyncio.wait_for(self.queue.get(), timeout=self.max_wait)
batch.append(item)
except asyncio.TimeoutError:
pass # Timeout = procesar lo que tengamos hasta ahora
if batch:
texts = [t for t, _ in batch]
vectors = self.embeddings.embed_documents(texts) # 1 API call para N textos
for (_, future), vector in zip(batch, vectors):
future.set_result(vector) # Resuelve cada Future → desbloquea cada caller
Dashboard de FinOps para IA
┌────────────────────────────────────────────────────────┐
│ AI FinOps Dashboard │
├────────────────────────────────────────────────────────┤
│ │
│ Costo Diario Costo Mensual (est.) Budget │
│ [$523] [$15,690] [$20,000] │
│ ▲ 12% vs ayer ▲ 8% vs mes ant. 78% used │
│ │
│ ┌─────────────────────────────────────┐ │
│ │ Costo por Modelo (últimos 7 días) │ │
│ │ ████████████████ GPT-4o $2,100 │ │
│ │ ████████ Claude $980 │ │
│ │ ███ Embeddings $340 │ │
│ │ █ Bedrock $140 │ │
│ └─────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────┐ │
│ │ Costo por Feature │ │
│ │ RAG Chat: $1,800/sem │ │
│ │ Agent Analysis: $1,200/sem │ │
│ │ Document Index: $360/sem │ │
│ └─────────────────────────────────────┘ │
│ │
│ Métricas de Eficiencia: │
│ Cache Hit Rate: 34% (target: 50%) │
│ Avg tokens/query: 2,450 (target: 2,000) │
│ Queries routed to mini: 45% (target: 60%) │
└────────────────────────────────────────────────────────┘
AWS Cost Explorer API
ce = boto3.client("ce")
# Obtener costos de servicios de IA
response = ce.get_cost_and_usage(
TimePeriod={"Start": "2026-03-01", "End": "2026-03-17"},
Granularity="DAILY",
Metrics=["BlendedCost"],
Filter={
"Dimensions": {
"Key": "SERVICE",
"Values": ["Amazon Bedrock", "Amazon OpenSearch Service", "Amazon SageMaker"],
}
},
GroupBy=[{"Type": "DIMENSION", "Key": "SERVICE"}],
)
Resumen
FinOps para IA requiere:
- Monitoreo granular — costo por query, por modelo, por feature
- Model routing — usar el modelo más barato que resuelva la tarea
- Caching agresivo — respuestas, embeddings, resultados de retrieval
- Compression de prompts — reducir tokens enviados al LLM
- Alertas proactivas — detectar anomalías de costo antes de que escalen
🧠 Preguntas de Repaso
1. ¿Cuál es el costo aproximado por query en un pipeline RAG típico en producción con GPT-4o?
- A) $0.0001 por query
- B) ~$0.006 por query (embedding $0.0001 + vector search $0.0002 + LLM ~$0.005)
- C) $0.50 por query
- D) $1.00 por query
Respuesta: B) — Un query RAG típico cuesta
$0.006: embedding ($0.0001) + vector search ($0.0002) + generación LLM con GPT-4o ($0.005 por ~500 tokens). A 100K queries/día esto suma ~$18,000/mes.
2. ¿Qué es "Model Routing" y cómo reduce costos?
- A) Enviar todos los requests al modelo más barato
- B) Clasificar queries en simple/standard/complex y rutear cada tipo al modelo más barato que pueda resolverlo correctamente
- C) Rotar entre modelos para distribuir la carga
- D) Usar un solo modelo pero con diferentes temperatures
Respuesta: B) — Model Routing clasifica los queries por complejidad y los envía al modelo más económico capaz de resolverlos. Ej: queries simples → gpt-4o-mini ($0.00015/1k tokens), complejos → claude-sonnet ($0.003/1k). Esto puede reducir costos un 60%+ si el 45% de queries son simples.
3. ¿Qué técnica reduce ~4x los tokens enviados al LLM con pérdida mínima de calidad?
- A) Caching de respuestas en Redis
- B) Prompt Compression con modelos como LLMLingua-2
- C) Batching de requests
- D) Reducción de dimensiones de embeddings
Respuesta: B) — Prompt Compression usa modelos como LLMLingua-2 para comprimir el contexto del prompt, reduciendo ~4x los tokens enviados al LLM con pérdida mínima de información relevante.
4. En el dashboard FinOps del ejemplo, el Cache Hit Rate es del 34% con un target del 50%. ¿Por qué es importante mejorar esta métrica?
- A) Un cache hit rate más alto mejora la calidad de las respuestas
- B) Cada cache hit evita una llamada completa al LLM, reduciendo costos a $0 por esas queries y la latencia a <1ms
- C) El cache hit rate afecta la seguridad del sistema
- D) Un cache hit rate más alto aumenta la capacidad de almacenamiento
Respuesta: B) — Cada cache hit significa una respuesta servida sin llamar al LLM, con costo $0 y latencia <1ms. Subir del 34% al 50% target significaría que la mitad de las queries no generarían costo de LLM, un ahorro significativo en el presupuesto mensual.