Observabilidad con LangSmith

Avanzado⏱35 min📖3 min de lecturaLección 2 de 10

Observabilidad con LangSmith

Domina la observabilidad para aplicaciones LangGraph usando trazado LangSmith, registro, depuración, evaluación de ejecuciones y monitoreo de rendimiento.

LangSmith es la plataforma de observabilidad de LangChain para aplicaciones con LLM. Ofrece trazas, depuración, evaluación y monitoreo para aplicaciones LangGraph en producción.

Por qué Importa la Observabilidad

Los agentes LangGraph son sistemas complejos con múltiples llamadas a LLM, ejecuciones de herramientas y enrutamiento condicional. Sin observabilidad adecuada:

No puedes depurar por qué un agente tomó una decisión incorrecta
No puedes medir el consumo de tokens ni la latencia
No puedes identificar patrones que fallan
No puedes mejorar el rendimiento de forma sistemática

Configurando el Trazado de LangSmith

python

import os

# Set environment variables
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls_..."
os.environ["LANGCHAIN_PROJECT"] = "my-agent-production"

O a través del archivo .env:

bash

LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=ls_...
LANGCHAIN_PROJECT=my-agent-production
LANGCHAIN_ENDPOINT=https://api.smith.langchain.com

ℹ️Nota

Una vez activado el rastreo, todas las llamadas de LangChain y LangGraph se rastrean automáticamente. Cada invocación de grafo crea un árbol de ejecución (run tree) que muestra cada nodo, cada llamada a LLM y cada ejecución de herramienta.

Estructura del Árbol de Ejecución

Una invocación de LangGraph crea una traza jerárquica:

Root Run (graph invocation)
├── Node: classify
│   └── LLM Call (classify_question)
├── Node: web_search
│   └── Tool Call (web_search)
├── Node: agent
│   └── LLM Call (generate_response)
└── [END]

Cada nodo y cada llamada interna son una ejecución (run) separada, con su propio:

Valores de entrada/salida
Marcas de tiempo de inicio/fin
Token counts
Error information
Metadata

Agregando Metadatos Personalizados

python

def node_with_metadata(state: State) -> dict:
    # Add metadata to the LLM call
    response = llm.invoke(
        state["messages"],
        metadata={
            "user_id": state.get("user_id"),
            "session_type": "premium",
            "node_name": "classify",
            "version": "2.1.0"
        }
    )
    return {"response": response.content}

Los metadatos personalizados hacen que filtrar y buscar en las trazas sea mucho más potente.

Etiquetado y Enmascaramiento

Añade etiquetas a las ejecuciones para filtrar:

python

from langchain_core.tracers.context import tracing_v2_enabled

with tracing_v2_enabled(
    project="my-agent",
    tags=["production", "user-facing"],
    metadata={"deployment": "us-east-1"}
):
    result = app.invoke(input_data)

Enmascara datos sensibles:

python

from langchain_core.tracers import LangChainTracer

tracer = LangChainTracer(
    project="my-agent",
    # Tags to redact from traces
    hide_inputs=True,   # Don't log raw inputs
    hide_outputs=False  # Log outputs for debugging
)

⚠️Advertencia

Ten cuidado con los datos sensibles. Usa hide_inputs=True en aplicaciones con muchos PII, o implementa una función de enmascaramiento personalizada.

Depuración con LangSmith

Visualizando trazas

LangSmith Dashboard
├── Projects
│   └── my-agent
│       ├── Runs (individual invocations)
│       │   ├── Latency distribution
│       │   ├── Token usage
│       │   ├── Error rate
│       │   └── Cost tracking
│       └── Feedback / Annotations

Encontrando ejecuciones problemáticas

python

from langsmith import Client

client = Client()

# Find runs with errors
runs = client.list_runs(
    project_name="my-agent",
    error=True,
    start_time="2024-01-01"
)

for run in runs:
    print(f"Run {run.id}: {run.name} failed with {run.error}")

Feedback y Anotación

Recopila comentarios de los usuarios o anota ejecuciones para evaluar:

python

from langsmith import Client

client = Client()

# Add user feedback
client.create_feedback(
    run_id=run_id,
    key="user_rating",
    score=4,  # 1-5 scale
    comment="The agent understood my question correctly"
)

# Add evaluator annotation
client.create_feedback(
    run_id=run_id,
    key="correctness",
    score=1.0,  # 0.0 - 1.0
    comment="Correctly answered the query"
)

Evaluación

Define evaluadores para puntuar ejecuciones automáticamente:

python

from langsmith.evaluation import evaluate, StringEvaluator

def correctness_evaluator(example: dict, prediction: dict) -> dict:
    """Evaluate if the agent's answer matches the expected answer."""
    expected = example["output"]["answer"]
    actual = prediction["answer"]

    score = 1.0 if expected in actual else 0.0
    return {"key": "correctness", "score": score}

# Run evaluation
results = evaluate(
    app.invoke,
    data="test-dataset",  # Dataset in LangSmith
    evaluators=[correctness_evaluator],
    experiment_prefix="my-agent-v2"
)

✅Éxito

La evaluación automatizada te permite seguir el rendimiento a lo largo del tiempo, detectar regresiones y comparar distintas versiones del agente.

Monitoreo de Rendimiento

Métricas clave

Métrica	Qué mide	Objetivo
Latencia P50	Tiempo medio de respuesta	< 5s
Latencia P95	5% más lentas de las respuestas	< 15s
Consumo de tokens	Tokens consumidos por ejecución	Varía
Tasa de error	Porcentaje de ejecuciones fallidas	< 1%
Coste por ejecución	Coste de la API por invocación	Dependiente del presupuesto
Calificación del usuario	Puntuación media de comentarios	> 4.0 / 5.0

Configurando Monitores

python

from langsmith import Client

client = Client()

# Create a monitor for error rate
client.create_monitor(
    project_name="my-agent",
    metric="error_rate",
    threshold=0.05,  # Alert if > 5%
    interval_minutes=60
)

# Create a monitor for latency
client.create_monitor(
    project_name="my-agent",
    metric="latency_p95",
    threshold=15.0,  # Alert if > 15 seconds
    interval_minutes=60
)

Comparando Ejecuciones

python

# Compare two experiments
results = client.compare_experiments(
    experiment_ids=["exp_1", "exp_2"],
    metrics=["correctness", "latency", "cost"]
)

for result in results:
    print(f"Experiment {result.experiment_id}:")
    print(f"  Correctness: {result.metrics['correctness']}")
    print(f"  Avg Latency: {result.metrics['latency']}s")

Registro Personalizado en Nodos

python

import logging

logger = logging.getLogger("langgraph.agent")

def monitored_node(state: State, config: dict) -> dict:
    thread_id = config["configurable"]["thread_id"]
    logger.info(f"[{thread_id}] Starting node execution")
    start = time.time()

    try:
        result = process(state)
        elapsed = time.time() - start
        logger.info(f"[{thread_id}] Node completed in {elapsed:.2f}s")
        return result
    except Exception as e:
        logger.error(f"[{thread_id}] Node failed: {e}", exc_info=True)
        return {"error": str(e)}