Configurando el monitoreo de IA con Langfuse

Configurando el monitoreo de IA con Langfuse

A menudo me preguntan cómo distinguir a un ingeniero de IA junior de uno experimentado. Mi respuesta es simple: revisa su stack de monitoreo. Si los prompts, logs y costos son invisibles, el equipo vuela a ciegas. En este post desglosamos por qué el monitoreo debería venir con la versión uno de tu servicio de IA y cómo desplegar Langfuse rápido.

Por qué el monitoreo va antes del fine-tuning

No llegarás al fine-tuning ni a pipelines sofisticados si no puedes ver qué hace el modelo en producción. Empieza a rastrear estas métricas desde el día uno:

• costo por solicitud (tokens de prompt + completion)
• latencia de respuesta y tasa de timeouts
• calidad de la generación y ejemplos de respuestas malas
• contexto del usuario: quién hizo la solicitud y en qué escenario

Sin estos datos no puedes explicar un pico repentino en el gasto de la API ni por qué los usuarios se quejan de un asistente "silencioso". Un error común es dejar activado el historial de conversación, lo que infla la ventana de contexto y dispara los costos exponencialmente.

Dónde ayuda Langfuse

Langfuse es una plataforma de observabilidad para aplicaciones LLM. Se distribuye tanto como capa gratuita en la nube como con opción de autoalojamiento, así que encaja fácilmente en las políticas de la empresa. Beneficios clave:

• captura trazas de solicitudes, cadenas de prompts y llamadas a herramientas anidadas
• almacena las entradas y salidas crudas del modelo para revisiones de incidentes
• cuenta tokens y gasto por usuario o funcionalidad
• soporta OpenAI, Hugging Face, Anthropic y modelos personalizados
• se integra con OpenTelemetry para que puedas combinar datos con otras métricas

Qué rastrear

Trata el monitoreo como analítica de producto:

1. Economía: a dónde va el presupuesto y cuánto cuesta ejecutar cada funcionalidad.
2. Calidad: qué prompts generan más quejas y cuánto tardan las respuestas.
3. Experimentos: etiqueta pruebas A/B de prompts o modelos para poder comparar resultados.
4. Preparación operativa: los picos de tokens y errores son señales tempranas que te permiten corregir la degradación antes de que un cliente abra un ticket.

Integrando Langfuse en un proyecto de Python

Repasemos una integración básica con OpenAI.

nodejs

// tracedOpenAi.ts

import { observeOpenAI } from '@langfuse/openai';
import { NodeSDK } from '@opentelemetry/sdk-node';
import { LangfuseSpanProcessor } from '@langfuse/otel';
import Config from '@app/config';
import { OpenAI } from 'openai/client';

const sdk = new NodeSDK({
  spanProcessors: [
    new LangfuseSpanProcessor({
      publicKey: Config.langfusePublicKey,
      secretKey: Config.langfuseSecretKey,
      baseUrl: Config.langfuseHost,
    }),
  ],
});

sdk.start();

// Wrap the OpenAI client with Langfuse tracing
export const tracedOpenAI = observeOpenAI(new OpenAI({ apiKey: Config.openAiKey }), {
  // Configure trace-level attributes for all API calls
  traceName: 'my-openai-trace', // Name for the trace
  sessionId: 'user-session-123', // Track user session
  userId: 'user-abc', // Track user identity
  tags: ['openai-integration'], // Add searchable tags
});

python

# GptClient.py

from typing import Union

from langfuse.openai import OpenAI  # type: ignore

from openai.types import ChatModel
from openai.types.chat import (
    ChatCompletionSystemMessageParam,
    ChatCompletionUserMessageParam,
)

from src.config import Config


class GptClient:
    def __init__(self, api_key: str):
        self.api_key = api_key

        self.client = OpenAI(
            api_key=Config.gpt_key(),
        )

    def exec_request(
        self,
        setup_message: str,
        request_text: str,
        temperature: float = 0.1,
        model: Union[str, ChatModel] | None = None,
    ) -> str:
        final_model = model if model else Config.gpt_model()

        messages = [
            ChatCompletionSystemMessageParam(role="system", content=setup_message),
            ChatCompletionUserMessageParam(role="user", content=request_text),
        ]

        completion = self.client.chat.completions.create(
            max_tokens=40000,
            model=final_model,
            messages=messages,
            temperature=temperature,
            stream=True,
        )

        answer_parts = []
        for chunk in completion:
            part = None
            if hasattr(chunk, "choices"):
                delta = getattr(chunk.choices[0], "delta", None)
                if delta and hasattr(delta, "content"):
                    part = delta.content
            elif isinstance(chunk, dict) and "choices" in chunk:
                part = chunk["choices"][0]["delta"].get("content", "")
            if part:
                answer_parts.append(part)

        answer = "".join(answer_parts)
        return answer

Qué inspeccionar en los dashboards

Una vez que la integración está en producción, abre Langfuse y rastrea:

• distribución de tokens por usuario y escenario
• los cinco prompts más costosos
• proporción de respuestas con error (5xx, timeouts, validación)
• ejemplos de generaciones marcadas como "malas" (añade etiquetas manuales si hace falta)

Estos hallazgos respaldan la priorización: sabrás dónde optimizar, dónde hacer fine-tuning y dónde replantear la UX.

Próximos pasos

Después de una configuración básica, explora funciones avanzadas de Langfuse: métricas personalizadas, reintentos automáticos de cadenas y recolección de feedback de usuarios directamente en la interfaz. Comparte tus hallazgos — cuanto más amplios sean nuestros kits de herramientas, más rápido construiremos juntos productos de IA confiables.