Monitoring de LLMs en Producción: Observability, Costos y Performance (Guía 2025)
Tienes tu LLM en producción, pero ¿sabes cuánto te está costando cada respuesta? ¿Detectas cuándo alucina? ¿Mides la latencia real que experimentan tus usuarios? Sin monitoring adecuado, tu aplicación LLM es una caja negra cara y potencialmente peligrosa. En esta guía aprenderás a implementar observability profesional con las mejores herramientas open source y comerciales del 2025.
¿Por Qué el Monitoring de LLMs es Diferente?
El monitoring tradicional de software no es suficiente para LLMs. Los modelos de lenguaje presentan desafíos únicos:
Desafíos Específicos de LLMs
No-determinismo: A diferencia del código tradicional, el mismo input puede generar outputs diferentes. Esto dificulta la detección de regresiones y errores.
Costos variables: Cada request consume tokens diferentes según el contexto y la respuesta. Un bug en tu prompt puede multiplicar tus costos por 10x sin que te des cuenta.
Calidad subjetiva: ¿Cómo mides si una respuesta es “buena”? No hay un assert simple. Necesitas métricas como groundedness, relevancia, y toxicidad.
Latencia impredecible: El tiempo de respuesta varía según la longitud del output, la carga del modelo, y factores que no controlas si usas APIs externas.
Alucinaciones: El modelo puede generar información falsa con total confianza. Sin monitoring, no sabes cuándo está pasando.
Métricas Críticas en Producción
Según las mejores prácticas de 2025, estas son las métricas que debes trackear:
| Métrica | Descripción | Por Qué Importa |
|---|---|---|
| Latency (p50, p95, p99) | Tiempo de respuesta al usuario | UX y SLAs |
| Token Usage | Tokens de input/output por request | Control de costos |
| Cost per Request | Costo real en $ por llamada | ROI y presupuesto |
| Groundedness | ¿La respuesta se basa en el contexto? | Prevenir alucinaciones |
| Relevance | ¿La respuesta responde la pregunta? | Calidad del servicio |
| Toxicity/Bias | Contenido inapropiado o sesgado | Compliance y reputación |
| Error Rate | Fallas de API, timeouts, rate limits | Reliability |
| TTFT (Time to First Token) | Tiempo hasta el primer token streaming | UX percibida |
Herramientas de Observability para LLMs: Comparativa 2025
LangSmith: El Estándar de LangChain
LangSmith es la plataforma de observability oficial de LangChain, diseñada específicamente para aplicaciones LLM.
Características clave:
- Tracing end-to-end de chains, agents y RAG pipelines
- No añade latencia (usa callbacks asíncronos)
- Soporte para OpenTelemetry desde 2025
- Self-hosting disponible en plan Enterprise
- Evaluaciones automatizadas con LLM-as-judge
Cuándo usarlo:
- Si ya usas LangChain o LangGraph
- Necesitas debugging profundo de agents complejos
- Quieres tracing distribuido con OTel
import os
from langsmith import Client
from langchain.callbacks import LangChainTracer
# Configuración (no añade latencia)
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "tu-api-key"
# Automáticamente trackeará todas las llamadas
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
llm = ChatOpenAI(model="gpt-4")
prompt = ChatPromptTemplate.from_messages([
("system", "Eres un asistente experto en {domain}"),
("user", "{question}")
])
chain = prompt | llm
# Esta llamada se trackeará automáticamente
response = chain.invoke({
"domain": "machine learning",
"question": "¿Qué es overfitting?"
})
Pricing:
- Developer: Gratis hasta 50K traces/mes
- Plus: $39/mes (500K traces)
- Enterprise: Self-hosted, custom pricing
Langfuse: El Open Source Más Popular
Langfuse es la herramienta open source de observability más usada en 2025, con integración nativa de OpenTelemetry.
Ventajas sobre LangSmith:
- 100% open source y self-hostable
- Soporte para múltiples frameworks (no solo LangChain)
- Interfaz más moderna y customizable
- Evaluaciones con modelos propios o externos
Implementación básica:
from langfuse import Langfuse
from langfuse.decorators import observe
# Inicializar cliente
langfuse = Langfuse(
public_key="pk-...",
secret_key="sk-...",
host="https://cloud.langfuse.com" # o tu instancia self-hosted
)
# Trackear con decoradores (lo más simple)
@observe()
def retrieve_context(query: str) -> list[str]:
# Tu lógica de retrieval
docs = vector_db.similarity_search(query, k=5)
return [doc.page_content for doc in docs]
@observe()
def generate_answer(query: str, context: list[str]) -> str:
prompt = f"Context: {context}\n\nQuestion: {query}"
response = openai.ChatCompletion.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
# RAG pipeline completo con tracing automático
@observe()
def rag_pipeline(user_query: str) -> str:
context = retrieve_context(user_query)
answer = generate_answer(user_query, context)
return answer
# Uso normal - el tracing es transparente
result = rag_pipeline("¿Cómo funciona un transformer?")
Self-hosting con Docker:
# Clonar repo
git clone https://github.com/langfuse/langfuse.git
cd langfuse
# Levantar con Docker Compose
docker compose up -d
# Ya está corriendo en http://localhost:3000
Phoenix (Arize AI): Detección de Alucinaciones Built-in
Phoenix destaca por su herramienta integrada de detección de alucinaciones y su enfoque en debugging visual.
Features únicas:
- Hallucination detector integrado
- Visualización de embeddings para debugging
- Análisis de drift en producción
- 100% open source
Ejemplo con detección de alucinaciones:
import phoenix as px
from phoenix.trace import using_project
from openinference.instrumentation.openai import OpenAIInstrumentor
# Iniciar Phoenix localmente
px.launch_app()
# Auto-instrumentar OpenAI
OpenAIInstrumentor().instrument()
# Tu código normal
import openai
with using_project("production-rag"):
response = openai.ChatCompletion.create(
model="gpt-4",
messages=[
{"role": "system", "content": "Responde basándote SOLO en el contexto."},
{"role": "user", "content": f"Context: {context}\n\nQuestion: {query}"}
]
)
# Phoenix automáticamente:
# - Detecta si la respuesta alucina (no está groundeada en el context)
# - Calcula métricas de relevancia
# - Trackea latencia y costos
Evaluación de groundedness:
from phoenix.evals import (
HallucinationEvaluator,
RelevanceEvaluator,
run_evals
)
# Evaluar si las respuestas están groundeadas
hallucination_eval = HallucinationEvaluator(model="gpt-4")
results = run_evals(
dataframe=production_traces_df,
evaluators=[hallucination_eval],
provide_explanation=True
)
# Ver qué % de respuestas alucinan
print(f"Hallucination rate: {results['hallucination'].mean():.2%}")
Helicone: Cost Tracking Extremadamente Simple
Helicone es perfecto si tu prioridad #1 es controlar costos. Integración en 1 línea de código.
Setup ultra-rápido:
# ANTES
import openai
openai.api_base = "https://api.openai.com/v1"
# DESPUÉS (solo cambiar 2 líneas)
import openai
openai.api_base = "https://oai.helicone.ai/v1"
openai.default_headers = {
"Helicone-Auth": "Bearer sk-helicone-..."
}
# Todo tu código sigue igual
response = openai.ChatCompletion.create(...)
# Helicone ya está trackeando costos, latencia, y requests
Features de cost tracking:
# Tracking por usuario
openai.default_headers["Helicone-User-Id"] = "user-123"
# Tracking por proyecto
openai.default_headers["Helicone-Property-Project"] = "chatbot-v2"
# Caching para reducir costos (beta 2025)
openai.default_headers["Helicone-Cache-Enabled"] = "true"
response = openai.ChatCompletion.create(
model="gpt-4",
messages=[{"role": "user", "content": "¿Qué es Python?"}]
)
# Si ya se hizo esta pregunta, usa la respuesta cacheada (costo $0)
Pricing:
- Free: 100K requests/mes
- Pro: $20/mes (unlimited requests)
- Enterprise: Custom (self-hosted disponible)
Weights & Biases: Para Equipos ML Completos
W&B es ideal si ya lo usas para training y quieres unificar observability.
Ventajas:
- Unifica training + serving en una plataforma
- Prompt versioning y A/B testing
- Integraciones con todos los frameworks
import wandb
from wandb.integration.langchain import WandbTracer
# Inicializar proyecto
wandb.init(project="llm-production", name="rag-v2")
# Trackear prompts como Artifacts
prompt_template = ChatPromptTemplate.from_messages([...])
wandb.log_artifact(prompt_template, name="rag-prompt", type="prompt")
# Usar con LangChain
chain = prompt | llm
tracer = WandbTracer()
response = chain.invoke(
{"query": "..."},
config={"callbacks": [tracer]}
)
# W&B trackea automáticamente:
# - Latencia end-to-end
# - Token usage y costos
# - Input/output completos
# - Versión del prompt usado
Comparativa: ¿Qué Herramienta Elegir?
| Herramienta | Mejor Para | Open Source | Self-Hosting | Precio (Free Tier) |
|---|---|---|---|---|
| LangSmith | Apps con LangChain/LangGraph | ❌ | ✅ Enterprise | 50K traces/mes |
| Langfuse | Máxima flexibilidad y control | ✅ | ✅ Completo | Ilimitado (self-hosted) |
| Phoenix | Detección de alucinaciones | ✅ | ✅ Completo | Ilimitado (self-hosted) |
| Helicone | Cost tracking simple y rápido | ✅ (parcial) | ✅ Enterprise | 100K requests/mes |
| W&B | Equipos ML con training+serving | ❌ | ❌ | Limitado |
Recomendación por caso de uso:
- Startup con LangChain: Empieza con LangSmith (free tier generoso)
- Empresa con compliance estricto: Langfuse self-hosted (control total de datos)
- Prioridad anti-alucinaciones: Phoenix (detector built-in)
- Necesitas controlar costos YA: Helicone (setup en 5 minutos)
- Equipo ML maduro: W&B (si ya lo usas para training)
OpenTelemetry: El Futuro del LLM Monitoring
OpenTelemetry (OTel) se está convirtiendo en el estándar para observability de LLMs en 2025. Evita vendor lock-in y permite unificar traces de múltiples herramientas.
¿Por Qué OTel para LLMs?
- Vendor-neutral: Cambia de backend sin reescribir código
- Estandarización: Semantic conventions para LLM traces
- Integración: Unifica traces de LLM + microservices tradicionales
- Futuro-proof: Adoptado por LangSmith, Langfuse, Phoenix
Implementación con OpenLLMetry
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
# Setup OTel
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)
# Exportar a tu backend (Langfuse, Jaeger, etc.)
otlp_exporter = OTLPSpanExporter(endpoint="http://localhost:4317")
trace.get_tracer_provider().add_span_processor(
BatchSpanProcessor(otlp_exporter)
)
# Auto-instrumentar con OpenLLMetry
from traceloop.sdk import Traceloop
Traceloop.init(app_name="production-rag")
# Tu código normal - se instrumenta automáticamente
import openai
response = openai.ChatCompletion.create(
model="gpt-4",
messages=[{"role": "user", "content": "Hola"}]
)
# El trace se envía a tu backend OTel automáticamente
Semantic Conventions para LLMs (2025)
OpenTelemetry define convenciones semánticas estándar:
# Atributos estándar en spans de LLM
span.set_attribute("gen_ai.system", "openai")
span.set_attribute("gen_ai.request.model", "gpt-4")
span.set_attribute("gen_ai.request.temperature", 0.7)
span.set_attribute("gen_ai.request.max_tokens", 500)
span.set_attribute("gen_ai.usage.input_tokens", 150)
span.set_attribute("gen_ai.usage.output_tokens", 320)
span.set_attribute("gen_ai.usage.total_tokens", 470)
# Para RAG
span.set_attribute("gen_ai.retrieval.documents_count", 5)
span.set_attribute("gen_ai.retrieval.source", "pinecone")
Métricas Clave: Cómo Medirlas en Producción
1. Groundedness (Prevenir Alucinaciones)
Groundedness mide si la respuesta está soportada por el contexto proporcionado. Es crítico en RAG.
from phoenix.evals import HallucinationEvaluator
# Evaluar groundedness con LLM-as-judge
evaluator = HallucinationEvaluator(model="gpt-4")
def evaluate_groundedness(query: str, context: str, answer: str) -> float:
result = evaluator.evaluate(
input=query,
context=context,
output=answer
)
return result.score # 0.0 (alucinación) a 1.0 (100% groundeado)
# En producción
query = "¿Cuál es el horario de la tienda?"
context = "La tienda abre de lunes a viernes de 9am a 6pm."
answer = llm.generate(query, context)
score = evaluate_groundedness(query, context, answer)
# Alertar si groundedness < 0.7
if score < 0.7:
logging.warning(f"Possible hallucination detected! Score: {score}")
send_alert_to_slack(query, answer, score)
2. Latency: p50, p95, p99
No uses promedios para latencia. Usa percentiles.
import time
from collections import deque
from typing import List
class LatencyTracker:
def __init__(self, window_size: int = 1000):
self.latencies = deque(maxlen=window_size)
def record(self, latency_ms: float):
self.latencies.append(latency_ms)
def get_percentiles(self) -> dict:
sorted_latencies = sorted(self.latencies)
n = len(sorted_latencies)
return {
"p50": sorted_latencies[int(n * 0.50)] if n > 0 else 0,
"p95": sorted_latencies[int(n * 0.95)] if n > 0 else 0,
"p99": sorted_latencies[int(n * 0.99)] if n > 0 else 0,
}
# Uso
tracker = LatencyTracker()
def call_llm(prompt: str) -> str:
start = time.time()
response = openai.ChatCompletion.create(...)
latency_ms = (time.time() - start) * 1000
tracker.record(latency_ms)
# Logear cada request
logger.info(f"Latency: {latency_ms:.2f}ms", extra={
"trace_id": get_trace_id(),
"user_id": get_user_id(),
"model": "gpt-4"
})
return response.choices[0].message.content
# Cada minuto, reportar percentiles
import sched
scheduler = sched.scheduler(time.time, time.sleep)
def report_metrics():
percentiles = tracker.get_percentiles()
print(f"Latency - p50: {percentiles['p50']:.0f}ms, "
f"p95: {percentiles['p95']:.0f}ms, "
f"p99: {percentiles['p99']:.0f}ms")
# Alertar si p95 > 3 segundos
if percentiles['p95'] > 3000:
send_alert("High latency detected!", percentiles)
scheduler.enter(60, 1, report_metrics)
scheduler.enter(60, 1, report_metrics)
scheduler.run()
3. Cost per Request
# Precios OpenAI (2025, verificar en docs oficiales)
PRICING = {
"gpt-4": {"input": 0.03 / 1000, "output": 0.06 / 1000},
"gpt-4-turbo": {"input": 0.01 / 1000, "output": 0.03 / 1000},
"gpt-3.5-turbo": {"input": 0.0005 / 1000, "output": 0.0015 / 1000},
}
def calculate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
if model not in PRICING:
raise ValueError(f"Unknown model: {model}")
cost = (
input_tokens * PRICING[model]["input"] +
output_tokens * PRICING[model]["output"]
)
return cost
# Wrapper con cost tracking
def call_llm_with_cost_tracking(messages: list, model: str = "gpt-4") -> dict:
response = openai.ChatCompletion.create(
model=model,
messages=messages
)
usage = response.usage
cost = calculate_cost(
model=model,
input_tokens=usage.prompt_tokens,
output_tokens=usage.completion_tokens
)
# Logear a tu sistema de monitoring
logger.info(f"LLM call completed", extra={
"model": model,
"input_tokens": usage.prompt_tokens,
"output_tokens": usage.completion_tokens,
"total_tokens": usage.total_tokens,
"cost_usd": cost,
"user_id": get_current_user_id()
})
# Si el costo es anormalmente alto, alertar
if cost > 0.10: # $0.10 por request es mucho
send_alert(f"High cost request: ${cost:.4f}")
return {
"response": response.choices[0].message.content,
"cost": cost,
"tokens": usage.total_tokens
}
Dashboard de Producción: Qué Monitorear
Un buen dashboard de LLM en producción debe mostrar:
Métricas de Negocio (Top)
- Requests/min (tráfico)
- Success rate (reliability)
- p95 latency (UX)
- Cost/hour (burn rate)
Métricas de Calidad (Middle)
- Groundedness score (promedio últimas 1h)
- Relevance score
- User feedback (thumbs up/down)
- Error rate por tipo (rate limit, timeout, etc.)
Métricas Técnicas (Bottom)
- Token usage (input vs output ratio)
- Model distribution (% gpt-4 vs gpt-3.5)
- TTFT (Time to First Token) para streaming
- Cache hit rate (si usas caching)
Alertas Críticas
Configura alertas para:
# Ejemplo con Prometheus + Alertmanager
alerts = [
{
"name": "HighLatency",
"condition": "latency_p95 > 5000", # 5 segundos
"severity": "warning",
"message": "p95 latency exceeds 5s"
},
{
"name": "HighCosts",
"condition": "cost_per_hour > 100", # $100/hora
"severity": "critical",
"message": "Spending $100/hour - check for issues!"
},
{
"name": "LowGroundedness",
"condition": "groundedness_avg_1h < 0.6",
"severity": "warning",
"message": "Many hallucinations detected"
},
{
"name": "HighErrorRate",
"condition": "error_rate_5m > 0.05", # >5% errores
"severity": "critical",
"message": "More than 5% of requests failing"
}
]
Case Study: Optimización Real de Costos
Caso real: Una startup tenía costos de $15K/mes en OpenAI. Tras implementar monitoring, redujeron a $2K/mes (87% de ahorro).
Problema Detectado
# ANTES (sin monitoring)
def generate_summary(document: str) -> str:
return openai.ChatCompletion.create(
model="gpt-4",
messages=[{
"role": "user",
"content": f"Summarize this document:\n\n{document}"
}]
).choices[0].message.content
# Problema: documentos de 50K tokens -> $1.50 por request
# Con 10K requests/día = $15K/mes 🔥
Solución con Monitoring
- Detectaron el problema con Helicone cost tracking
- Implementaron chunking para documentos largos
- Usaron gpt-3.5-turbo para summaries (10x más barato)
- Agregaron caching para documentos repetidos
# DESPUÉS (con optimizaciones)
import hashlib
from functools import lru_cache
@lru_cache(maxsize=1000)
def generate_summary_cached(document_hash: str, document: str) -> str:
# Chunk si es muy largo
if len(document) > 15000: # ~4K tokens
chunks = chunk_document(document, max_length=10000)
chunk_summaries = [
summarize_chunk(chunk) for chunk in chunks
]
# Combinar summaries
combined = "\n".join(chunk_summaries)
return summarize_chunk(combined) # Summary de summaries
# Usar modelo más barato para summaries
return openai.ChatCompletion.create(
model="gpt-3.5-turbo", # 20x más barato que gpt-4
messages=[{
"role": "user",
"content": f"Summarize concisely:\n\n{document[:15000]}"
}],
max_tokens=300 # Limitar output
).choices[0].message.content
def generate_summary(document: str) -> str:
doc_hash = hashlib.md5(document.encode()).hexdigest()
return generate_summary_cached(doc_hash, document)
# Resultado:
# - Chunking evitó inputs gigantes
# - gpt-3.5 redujo costo 20x
# - Cache evitó duplicados
# - Total: $15K/mes -> $2K/mes ✅
Best Practices de Monitoring LLM
1. Usa Sampling Inteligente
No traces el 100% del tráfico en producción (es caro y ruidoso).
import random
def should_trace_request(user_tier: str) -> bool:
sampling_rates = {
"free": 0.01, # 1% de usuarios free
"pro": 0.10, # 10% de usuarios pro
"enterprise": 1.0 # 100% de enterprise
}
return random.random() < sampling_rates.get(user_tier, 0.01)
# En producción
if should_trace_request(user.tier):
with tracer.start_span("llm_request"):
response = call_llm(prompt)
else:
# Solo trackear métricas básicas (latencia, costo)
response = call_llm(prompt)
2. Anonimiza Datos Sensibles
import re
def anonymize_pii(text: str) -> str:
# Emails
text = re.sub(r'\b[\w.-]+@[\w.-]+\.\w+\b', '[EMAIL]', text)
# Teléfonos (formato internacional)
text = re.sub(r'\+?\d{1,3}[-.\s]?\(?\d{1,4}\)?[-.\s]?\d{1,4}[-.\s]?\d{1,9}', '[PHONE]', text)
# Tarjetas de crédito
text = re.sub(r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b', '[CARD]', text)
return text
# Antes de enviar a monitoring
def trace_request_safely(input: str, output: str):
langfuse.trace(
name="llm_request",
input=anonymize_pii(input),
output=anonymize_pii(output)
)
3. Correlaciona con User Feedback
# Capturar feedback del usuario
def submit_feedback(trace_id: str, rating: int, comment: str = ""):
langfuse.score(
trace_id=trace_id,
name="user_rating",
value=rating, # 1-5 estrellas
comment=comment
)
# En tu app
trace_id = langfuse.get_current_trace_id()
response = call_llm(prompt)
# Mostrar UI para feedback
show_rating_widget(trace_id)
# Ahora puedes filtrar en dashboard:
# "Mostrar traces con rating < 3" -> debuggear los malos
4. Automatiza Evaluaciones
from apscheduler.schedulers.background import BackgroundScheduler
def evaluate_production_sample():
# Tomar 100 requests random de las últimas 24h
traces = langfuse.get_traces(
limit=100,
from_timestamp=datetime.now() - timedelta(hours=24)
)
# Evaluar groundedness
evaluator = HallucinationEvaluator()
results = evaluator.evaluate_batch(traces)
# Calcular métricas
hallucination_rate = (results < 0.7).mean()
if hallucination_rate > 0.15: # >15% alucinaciones
send_slack_alert(
f"⚠️ High hallucination rate: {hallucination_rate:.1%}\n"
f"Check dashboard: https://langfuse.com/traces"
)
# Ejecutar cada 6 horas
scheduler = BackgroundScheduler()
scheduler.add_job(evaluate_production_sample, 'interval', hours=6)
scheduler.start()
Preguntas Frecuentes (FAQs)
1. ¿El monitoring añade latencia a mis requests?
No, si usas herramientas modernas. LangSmith, Langfuse y Phoenix usan callbacks asíncronos que envían datos en background sin bloquear el request principal. La latencia agregada es <5ms, imperceptible para usuarios.
2. ¿Cuánto cuesta el monitoring en producción?
Depende de la herramienta:
- Open source self-hosted (Langfuse, Phoenix): Solo infraestructura (~$50-200/mes en cloud)
- SaaS free tiers: LangSmith (50K traces/mes), Helicone (100K requests/mes) son generosos
- SaaS pagado: ~$20-100/mes para startups, más para enterprise
El ROI es positivo si detectas 1 solo bug costoso o optimizas costos en 10-20%.
3. ¿Puedo usar múltiples herramientas a la vez?
Sí, gracias a OpenTelemetry. Puedes enviar traces a LangSmith para debugging Y a Prometheus para alertas.
# Exportar a múltiples backends
from opentelemetry.sdk.trace.export import BatchSpanProcessor
trace_provider.add_span_processor(BatchSpanProcessor(langsmith_exporter))
trace_provider.add_span_processor(BatchSpanProcessor(prometheus_exporter))
4. ¿Cómo mido la calidad sin ground truth?
Usa LLM-as-judge: un modelo evalúa outputs de otro.
# GPT-4 evalúa respuestas de GPT-3.5
judge_prompt = """
Evalúa esta respuesta del 1-10:
Question: {question}
Answer: {answer}
Criteria:
- Correctitud
- Relevancia
- Claridad
Score (1-10):
"""
score = gpt4.generate(judge_prompt.format(...))
Alternativamente: user feedback (thumbs up/down) es ground truth real.
5. ¿Qué hago si detecto una alucinación en producción?
- Inmediato: Logear el caso completo (input, context, output, trace_id)
- Corto plazo: Revisar el prompt – ¿es lo suficientemente específico?
- Mediano plazo: Aumentar groundedness threshold, agregar ejemplos al prompt
- Largo plazo: Fine-tuning del modelo con casos correctos
if groundedness_score < 0.6:
# Logear para análisis posterior
log_hallucination_case(trace_id, input, context, output, score)
# Respuesta de fallback
return "Lo siento, no tengo suficiente información para responder con confianza."
6. ¿Debo monitorear en desarrollo o solo en producción?
Ambos, pero de forma diferente:
- Desarrollo: 100% tracing detallado para debugging
- Staging: 100% tracing + evaluaciones automatizadas
- Producción: Sampling (1-10%) + métricas agregadas + alertas
7. ¿Cómo detecto model drift?
Compara distribuciones de métricas a lo largo del tiempo:
# Calcular groundedness promedio por semana
weekly_groundedness = {
"week_1": 0.85,
"week_2": 0.83,
"week_3": 0.79, # ⚠️ Drift detectado
"week_4": 0.76
}
# Alertar si baja >10% en un mes
if weekly_groundedness["week_4"] < weekly_groundedness["week_1"] * 0.90:
alert("Model quality degrading! Possible drift.")
Phoenix tiene visualizaciones específicas para drift.
8. ¿Qué métricas son más importantes para RAG?
Para RAG, prioriza:
- Groundedness (anti-alucinaciones)
- Retrieval precision (¿recuperaste docs relevantes?)
- Context utilization (¿el LLM usó el contexto?)
- Latency breakdown (retrieval vs generation)
# Trackear retrieval por separado
@observe()
def retrieve(query: str):
start = time.time()
docs = vector_db.search(query, k=5)
retrieval_latency = time.time() - start
langfuse.log({
"retrieval_latency_ms": retrieval_latency * 1000,
"docs_retrieved": len(docs),
"top_score": docs[0].score if docs else 0
})
return docs
9. ¿Cómo justifico el ROI del monitoring a management?
Datos reales de equipos que implementaron monitoring:
- Reducción de costos: 30-80% en promedio (optimizando prompts y modelos)
- Mejora de calidad: 25-50% menos quejas de usuarios (detectando alucinaciones)
- Time to resolution: 5x más rápido debuggear issues (con tracing)
- Prevención de incidents: Detectar rate limits antes de afectar a todos los usuarios
Ejemplo: Si gastas $10K/mes en LLMs, un 30% de ahorro = $3K/mes = $36K/año. El monitoring cuesta ~$1K/año. ROI = 36x.
10. ¿Necesito un equipo de MLOps para esto?
No para empezar. Herramientas como Helicone o LangSmith se configuran en <30 minutos. Para self-hosting (Langfuse, Phoenix) necesitas conocimientos básicos de Docker.
Para producción enterprise sí recomiendo alguien dedicado a observability.
11. ¿Cómo monitoreo agents complejos con múltiples tools?
Usa tracing jerárquico:
@observe() # Span padre
def agent_loop(task: str):
plan = planner_llm.generate(task)
for step in plan.steps:
execute_step(step) # Span hijo
return synthesize_result()
@observe() # Span hijo
def execute_step(step: str):
if "search" in step:
return web_search(step) # Span nieto
elif "calculate" in step:
return calculator(step) # Span nieto
El dashboard mostrará un flamegraph con toda la jerarquía.
12. ¿Puedo monitorear modelos open source self-hosted?
Sí, funciona igual. Ejemplo con vLLM:
from openai import OpenAI # vLLM expone API compatible con OpenAI
# Cliente vLLM local
client = OpenAI(
base_url="http://localhost:8000/v1",
api_key="dummy"
)
# Instrumentar con Langfuse
from langfuse.openai import OpenAI as LangfuseOpenAI
instrumented_client = LangfuseOpenAI(
base_url="http://localhost:8000/v1",
api_key="dummy"
)
# Ahora todas las llamadas a vLLM se trackean
response = instrumented_client.chat.completions.create(
model="meta-llama/Llama-2-70b-chat-hf",
messages=[...]
)
Para costos, calcula basándote en tu infraestructura (GPU hours * costo/hora).
13. ¿Qué pasa con la privacidad de los datos de usuarios?
Opciones:
- Self-hosting (Langfuse, Phoenix): Datos nunca salen de tu infra
- Anonimización (regex para PII antes de enviar a SaaS)
- Políticas de retención: Auto-borrar traces después de 30 días
- Modo solo-métricas: Enviar solo stats agregadas, no inputs/outputs completos
# Configurar Langfuse para NO enviar prompts/outputs
langfuse = Langfuse(
enabled=True,
sample_rate=0.1,
mask_inputs=True, # Solo envía métricas, no contenido
mask_outputs=True
)
14. ¿Cómo integro monitoring con mi CI/CD?
Evalúa en cada PR:
# .github/workflows/llm-eval.yml
name: LLM Quality Check
on: pull_request
jobs:
evaluate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Run evaluations
run: |
python -m pytest tests/llm_evals.py
- name: Check groundedness
run: |
python scripts/eval_groundedness.py --threshold 0.75
- name: Cost estimate
run: |
python scripts/estimate_cost_change.py
Si las métricas empeoran, el PR no pasa.
15. ¿Qué herramienta recomiendas para empezar HOY?
Para empezar en <1 hora:
- Si usas LangChain: LangSmith (2 variables de entorno)
- Si no: Helicone (cambiar base_url)
- Si quieres self-hosted: Langfuse (docker compose up)
Todas tienen free tiers generosos para validar el ROI.
Conclusión: Monitoring es Obligatorio, No Opcional
En 2025, no puedes tener LLMs en producción sin monitoring. Es como conducir con los ojos cerrados.
Empieza simple:
- Elige una herramienta (recomiendo Langfuse o Helicone)
- Implementa tracing básico (10-30 minutos)
- Configura 3 alertas críticas (latency, cost, error rate)
- Itera mejorando prompts basándote en datos reales
Recuerda:
- El monitoring se paga solo si detectas 1 bug costoso
- Las alucinaciones solo se detectan si las mides
- Los costos solo se optimizan si los trackeas
¿Siguiente paso? Elige tu herramienta y configúrala HOY. Tu yo del futuro (y tu CFO) te lo agradecerán.
Relacionado:
