Observabilidad en IA: Cómo monitorizar LLMs, agentes de IA y sistemas RAG en producción

Has desplegado una funcionalidad con IA. Los usuarios se quejan de que a veces da respuestas incorrectas. Tus dashboards muestran que la latencia p99 es de 12 segundos. No tienes ni idea de qué parte del pipeline es lenta: ¿es el modelo de embeddings? ¿La base de datos vectorial? ¿La propia llamada al LLM? ¿La plantilla de prompt que cambiaste el martes pasado?

Bienvenido a la brecha de observabilidad en los sistemas de IA.

La observabilidad tradicional fue diseñada para software determinista: llega una petición, se ejecuta código, sale una respuesta. Los LLMs son no deterministas, con estado, costosos, y sus fallos no se parecen en nada a un error 500. Un modelo que devuelve información incorrecta con total confianza no lanza ninguna excepción. Un pipeline RAG que recupera documentos irrelevantes produce una respuesta que suena plausible pero es inútil. No vas a detectar estos fallos con checks de disponibilidad.

Este artículo cubre lo que realmente necesitas para monitorizar LLMs, agentes de IA y sistemas RAG en producción.

Por qué el APM tradicional se queda corto

Toma una traza de petición web estándar. Ves el endpoint, las consultas a base de datos, las llamadas HTTP a servicios externos, el tiempo de respuesta. Todo está estructurado, tipado y es predecible.

Ahora toma un pipeline LLM. Una única petición de usuario puede implicar:

1. Una llamada de embedding para transformar la consulta en un vector
2. Una búsqueda por similitud en una base de datos vectorial que devuelve 5 fragmentos de documentos
3. Un paso de ensamblado de prompt que inyecta esos fragmentos en una plantilla
4. Una llamada de inferencia al LLM con 1.200 tokens de entrada
5. Un paso de parseo de la respuesta
6. Una segunda llamada LLM para verificación de hechos o enrutamiento
7. Una llamada a herramienta (búsqueda, calculadora, consulta a base de datos)
8. Una llamada de síntesis final

Cada paso tiene su propio perfil de latencia, modo de fallo, coste y dimensión de calidad. El APM estándar trata todo esto como una única llamada HTTP opaca a un endpoint de API. Eso no es útil.

Las cuatro dimensiones de la observabilidad en IA

Monitorizar sistemas de IA requiere cuatro dimensiones que no existen en la observabilidad tradicional:

1. Coste

Cada llamada LLM tiene un coste en tokens. A escala, esto se convierte en un gasto operativo significativo. Necesitas rastrear:

• Tokens de entrada por petición y por funcionalidad
• Tokens de salida por petición y por funcionalidad
• Coste por modelo (GPT-4o vs GPT-4o-mini vs Claude vs Llama)
• Coste por usuario / tenant para facturación y detección de abusos
• Tendencias de coste a lo largo del tiempo (¿crecen los costes más rápido que el uso?)

2. Latencia a nivel de componente

Una respuesta de 10 segundos es mala. Una respuesta de 10 segundos donde 8 son de inferencia LLM y 2 son de recuperación es accionable. Necesitas la latencia desglosada por:

• Tiempo de generación de embeddings
• Tiempo de búsqueda vectorial
• Tiempo de ensamblado del prompt
• Latencia al primer token del LLM (TTFT)
• Latencia de completado del LLM
• Tiempo de ejecución de herramientas

3. Señales de calidad

Esta es la difícil. La calidad de un LLM no produce excepciones — produce una degradación sutil que solo aflora como insatisfacción de usuarios.

Señales que puedes medir:

• Relevancia de recuperación: ¿los fragmentos recuperados están realmente relacionados con la consulta?
• Tasa de alucinaciones: ¿la respuesta cita fuentes que no existen en el contexto?
• Completitud de la respuesta: ¿la respuesta aborda lo que preguntó el usuario?
• Tasa de rechazo: ¿con qué frecuencia rechaza el modelo responder?
• Tasa de regeneración: ¿con qué frecuencia hacen clic los usuarios en “regenerar” o “intentar de nuevo”?
• Acciones posteriores del usuario: ¿los usuarios hacen seguimiento con solicitudes de aclaración?

4. Trazado de prompt y contexto

Necesitas capturar las entradas exactas que produjeron cada salida — la plantilla de prompt, el contexto inyectado, los parámetros del modelo. Sin esto, no puedes reproducir fallos ni evaluar el efecto de los cambios en los prompts.

Instrumentando con las convenciones semánticas de OpenTelemetry para IA

La comunidad de OpenTelemetry ha definido convenciones semánticas para sistemas de IA/LLM (en desarrollo activo como gen_ai.*). Estas te proporcionan un esquema estándar para spans y atributos.

Así es como se ve un span de llamada LLM correctamente instrumentado:

from opentelemetry import trace
from opentelemetry.trace import SpanKind

tracer = trace.get_tracer("ai-pipeline")

def call_llm(prompt: str, model: str = "gpt-4o") -> str:
    with tracer.start_as_current_span(
        "gen_ai.chat",
        kind=SpanKind.CLIENT,
    ) as span:
        span.set_attribute("gen_ai.system", "openai")
        span.set_attribute("gen_ai.request.model", model)
        span.set_attribute("gen_ai.request.max_tokens", 1024)
        span.set_attribute("gen_ai.request.temperature", 0.7)
        
        # Capturar el prompt (cuidado con los datos personales)
        span.set_attribute("gen_ai.prompt", prompt[:2000])  # truncar por seguridad
        
        response = openai_client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=1024,
        )
        
        # Capturar metadatos de la respuesta
        span.set_attribute("gen_ai.response.model", response.model)
        span.set_attribute("gen_ai.usage.input_tokens", response.usage.prompt_tokens)
        span.set_attribute("gen_ai.usage.output_tokens", response.usage.completion_tokens)
        span.set_attribute("gen_ai.response.finish_reason", response.choices[0].finish_reason)
        
        # Calcular coste (tarifas de ejemplo, ajustar según precio del modelo)
        cost = (response.usage.prompt_tokens * 0.000005) + (response.usage.completion_tokens * 0.000015)
        span.set_attribute("gen_ai.usage.cost_usd", round(cost, 6))
        
        return response.choices[0].message.content

Los atributos clave del espacio de nombres gen_ai.*:

Trazando un pipeline RAG completo

Un pipeline RAG es una cadena de operaciones distintas. Cada una merece su propio span con los atributos apropiados.

from opentelemetry import trace
from opentelemetry.trace import SpanKind
import time

tracer = trace.get_tracer("rag-pipeline")

def rag_query(user_query: str) -> str:
    with tracer.start_as_current_span("rag.pipeline") as pipeline_span:
        pipeline_span.set_attribute("rag.query", user_query)
        pipeline_span.set_attribute("rag.pipeline.version", "v2.3")
        
        # Paso 1: Embeber la consulta
        with tracer.start_as_current_span("rag.embed_query", kind=SpanKind.CLIENT) as embed_span:
            embed_span.set_attribute("gen_ai.system", "openai")
            embed_span.set_attribute("gen_ai.request.model", "text-embedding-3-small")
            
            query_embedding = embed_client.embeddings.create(
                model="text-embedding-3-small",
                input=user_query
            )
            embed_span.set_attribute("gen_ai.usage.input_tokens", query_embedding.usage.total_tokens)
        
        # Paso 2: Recuperar fragmentos relevantes
        with tracer.start_as_current_span("rag.retrieve", kind=SpanKind.CLIENT) as retrieve_span:
            retrieve_span.set_attribute("db.system", "pinecone")
            retrieve_span.set_attribute("rag.retrieval.top_k", 5)
            retrieve_span.set_attribute("rag.retrieval.namespace", "product-docs")
            
            results = vector_db.query(
                vector=query_embedding.data[0].embedding,
                top_k=5,
                include_metadata=True
            )
            
            # Crítico: capturar qué se recuperó realmente
            retrieve_span.set_attribute("rag.retrieval.results_count", len(results.matches))
            retrieve_span.set_attribute("rag.retrieval.top_score", results.matches[0].score if results.matches else 0)
            retrieve_span.set_attribute("rag.retrieval.min_score", results.matches[-1].score if results.matches else 0)
            
            # Registrar IDs de documentos recuperados para depuración
            doc_ids = [m.id for m in results.matches]
            retrieve_span.set_attribute("rag.retrieval.document_ids", str(doc_ids))
        
        # Paso 3: Ensamblar el prompt
        with tracer.start_as_current_span("rag.assemble_prompt") as assemble_span:
            context_chunks = [m.metadata["text"] for m in results.matches]
            context_text = "\n\n---\n\n".join(context_chunks)
            
            prompt = f"""Usa el siguiente contexto para responder la pregunta. 
Si la respuesta no está en el contexto, dilo explícitamente.

Contexto:
{context_text}

Pregunta: {user_query}

Respuesta:"""
            
            assemble_span.set_attribute("rag.prompt.template_version", "v1.2")
            assemble_span.set_attribute("rag.prompt.context_chunks", len(context_chunks))
            assemble_span.set_attribute("rag.prompt.total_chars", len(prompt))
        
        # Paso 4: Generar la respuesta
        answer = call_llm(prompt, model="gpt-4o")
        
        pipeline_span.set_attribute("rag.answer_length", len(answer))
        
        return answer

Esta traza te da una imagen completa de lo que ocurrió en cada petición de usuario: qué documentos se recuperaron y con qué puntuaciones de relevancia, cuántos tokens se consumieron en cada paso, y exactamente qué plantilla de prompt produjo la salida.

Monitorizando agentes de IA

Los agentes son más difíciles de trazar porque son dinámicos — el número de pasos no se conoce en el momento de la petición. Un agente puede llamar a una herramienta, o puede ejecutar 12 llamadas en un bucle antes de llegar a una respuesta (o alcanzar el límite de detección de bucles).

El patrón clave para la observabilidad de agentes es una estructura de spans jerárquica:

def run_agent(user_goal: str) -> str:
    with tracer.start_as_current_span("agent.run") as agent_span:
        agent_span.set_attribute("agent.goal", user_goal)
        agent_span.set_attribute("agent.model", "gpt-4o")
        
        messages = [{"role": "user", "content": user_goal}]
        step = 0
        max_steps = 10
        
        while step < max_steps:
            step += 1
            
            with tracer.start_as_current_span(f"agent.step") as step_span:
                step_span.set_attribute("agent.step.number", step)
                
                # El LLM decide qué hacer a continuación
                response = call_llm_with_tools(messages, tools=AVAILABLE_TOOLS)
                
                if response.finish_reason == "stop":
                    # El agente decidió responder directamente
                    step_span.set_attribute("agent.step.type", "final_answer")
                    agent_span.set_attribute("agent.total_steps", step)
                    return response.content
                
                elif response.finish_reason == "tool_calls":
                    # El agente decidió usar una herramienta
                    for tool_call in response.tool_calls:
                        with tracer.start_as_current_span("agent.tool_call") as tool_span:
                            tool_span.set_attribute("agent.tool.name", tool_call.function.name)
                            tool_span.set_attribute("agent.tool.arguments", tool_call.function.arguments)
                            
                            # Ejecutar la herramienta
                            t0 = time.time()
                            result = execute_tool(tool_call.function.name, tool_call.function.arguments)
                            tool_span.set_attribute("agent.tool.execution_ms", int((time.time() - t0) * 1000))
                            tool_span.set_attribute("agent.tool.result_length", len(str(result)))
                            
                            messages.append({"role": "tool", "content": str(result)})
        
        # El agente alcanzó el límite de pasos
        agent_span.set_attribute("agent.hit_step_limit", True)
        agent_span.set_attribute("agent.total_steps", step)
        raise AgentStepLimitError(f"El agente superó {max_steps} pasos sin llegar a una respuesta final")

Métricas críticas a derivar de las trazas de agentes:

• Pasos por completado — ¿cuántas llamadas LLM necesita para responder? Un número alto de pasos indica problemas de diseño del prompt o las herramientas.
• Distribución de llamadas a herramientas — ¿qué herramientas se usan más? ¿Cuáles son más lentas?
• Tasa de límite de pasos alcanzado — si los agentes frecuentemente alcanzan tu límite máximo de pasos, el sistema falla silenciosamente con regularidad.
• Tasa de éxito del agente — ¿con qué frecuencia el agente produce una respuesta final frente a terminar en error?

Métricas a rastrear en producción

Más allá de las trazas, necesitas métricas agregadas para dashboards y alertas:

from opentelemetry import metrics

meter = metrics.get_meter("ai-pipeline")

# Seguimiento de costes
llm_cost_counter = meter.create_counter(
    "gen_ai.cost.usd",
    description="Coste total del LLM en USD",
    unit="USD",
)

# Uso de tokens
input_token_counter = meter.create_counter(
    "gen_ai.usage.input_tokens",
    description="Total de tokens de entrada consumidos",
)
output_token_counter = meter.create_counter(
    "gen_ai.usage.output_tokens",
    description="Total de tokens de salida generados",
)

# Histogramas de latencia
llm_latency = meter.create_histogram(
    "gen_ai.request.duration",
    description="Duración de la petición al LLM",
    unit="s",
)

# Señales de calidad
retrieval_score = meter.create_histogram(
    "rag.retrieval.top_score",
    description="Puntuación de similitud más alta de la búsqueda vectorial",
)

refusal_counter = meter.create_counter(
    "gen_ai.response.refusals",
    description="Número de veces que el modelo rechazó responder",
)

# Registrar estas métricas en el código de instrumentación
def record_llm_call(model: str, feature: str, input_tokens: int, output_tokens: int, duration_s: float, cost: float):
    labels = {"gen_ai.request.model": model, "feature": feature}
    
    input_token_counter.add(input_tokens, labels)
    output_token_counter.add(output_tokens, labels)
    llm_cost_counter.add(cost, labels)
    llm_latency.record(duration_s, labels)

Dashboards clave a construir:

Dashboard de costes:

• Coste por hora por modelo
• Coste por funcionalidad (chat, búsqueda, resumen)
• Coste por usuario (para sistemas multi-tenant)
• Tendencia de coste vs volumen de peticiones (¿eres más eficiente con el tiempo?)

Dashboard de rendimiento:

• Latencia LLM p50/p90/p99 por modelo
• TTFT (tiempo al primer token) para respuestas en streaming
• Latencia de recuperación por base de datos vectorial
• Latencia end-to-end del pipeline

Dashboard de calidad:

• Tasa de rechazo a lo largo del tiempo
• Distribución de puntuación de recuperación (¿están bajando las puntuaciones? puede haber deriva en los datos)
• Tasa de regeneración del usuario (si la registras)
• Distribución de razones de finalización (¿con qué frecuencia se detiene la generación por length vs stop?)

Manejo de datos sensibles en trazas de IA

Los prompts y las respuestas frecuentemente contienen datos sensibles de usuarios. Antes de capturarlos en trazas, establece políticas claras:

Qué capturar:

• Conteos de tokens (siempre seguros)
• Nombre y parámetros del modelo (siempre seguros)
• Latencia y coste (siempre seguros)
• IDs de documentos recuperados, no el contenido del documento (generalmente seguros)
• Prompts truncados con datos personales eliminados

Qué evitar:

• Consultas de usuario en bruto si pueden contener datos de salud, financieros o personales
• Texto completo del prompt si contiene contenido de usuario inyectado
• Respuestas del LLM si pueden reflejar de vuelta información sensible

Un enfoque práctico es hashear los campos de identificación de usuarios y capturar solo metadatos sobre el contenido del prompt:

import hashlib

def safe_span_attributes(span, user_query: str, response: str):
    # Seguro: identificador de usuario hasheado para correlación sin exponer contenido
    span.set_attribute("user.query.hash", hashlib.sha256(user_query.encode()).hexdigest()[:16])
    
    # Seguro: metadatos estructurales sobre el prompt
    span.set_attribute("prompt.word_count", len(user_query.split()))
    span.set_attribute("prompt.has_code", "```" in user_query)
    
    # Seguro: metadatos estructurales sobre la respuesta
    span.set_attribute("response.word_count", len(response.split()))
    span.set_attribute("response.has_citations", "Fuente:" in response)
    
    # Condicional: capturar contenido solo en desarrollo/staging
    import os
    if os.getenv("CAPTURE_PROMPT_CONTENT", "false") == "true":
        span.set_attribute("prompt.content", user_query[:500])

Alertando sobre modos de fallo específicos de IA

Las alertas estándar de tasa de errores no funcionan para sistemas de IA. Aquí están las alertas que realmente necesitas:

Usando OpenLLMetry para instrumentación automática

Si no quieres escribir toda esta instrumentación manualmente, OpenLLMetry (de Traceloop) proporciona auto-instrumentación basada en OpenTelemetry para la mayoría de SDKs de LLM:

pip install opentelemetry-sdk traceloop-sdk

from traceloop.sdk import Traceloop

Traceloop.init(
    app_name="mi-servicio-rag",
    api_endpoint="http://localhost:4318",  # Tu endpoint del colector OTLP
    disable_batch=False,
)

# A partir de aquí, todas las llamadas a openai, anthropic, langchain, llama-index,
# pinecone, chromadb, etc. se trazan automáticamente con atributos gen_ai.*

OpenLLMetry instrumenta: OpenAI, Anthropic, Azure OpenAI, Cohere, Bedrock, LangChain, LlamaIndex, Pinecone, Chroma, Qdrant, Weaviate y más. Para la mayoría de equipos, este es el punto de partida correcto antes de construir instrumentación personalizada.

La checklist de preparación para producción

Antes de lanzar una funcionalidad de IA a producción, verifica:

• Cada llamada LLM está trazada con atributos gen_ai.* (modelo, conteo de tokens, latencia, coste)
• Los spans de recuperación RAG capturan IDs de documentos y puntuaciones de similitud
• Las trazas de agentes incluyen conteo de pasos y detalles de llamadas a herramientas
• El coste se rastrea por funcionalidad y por modelo, no solo de forma global
• Los prompts y respuestas se capturan con protecciones de datos personales
• Existen alertas para picos de coste, tasa de rechazo y alcance del límite de pasos
• Puedes reproducir cualquier fallo reproduciendo el prompt y el contexto trazados
• Los dashboards muestran señales de calidad, no solo disponibilidad y latencia
• La correlación log-traza está en su lugar (cada línea de log lleva el trace ID)
• Tienes una línea base para el uso “normal” de tokens para detectar anomalías

La observabilidad para sistemas de IA no es un lujo. Cuando tu pipeline LLM empieza a devolver respuestas incorrectas en producción, o cuando una anomalía de costes impacta tu factura a fin de mes, tus trazas son el único camino hacia la causa raíz.

Construye la instrumentación antes de necesitarla.