Observabilidad con LangSmith
Domina la observabilidad para aplicaciones LangGraph usando trazado LangSmith, registro, depuración, evaluación de ejecuciones y monitoreo de rendimiento.
Observabilidad con LangSmith
LangSmith is LangChain's observability platform for LLM applications. It provides tracing, debugging, evaluation, and monitoring for LangGraph applications in production.
Por qué Importa la Observabilidad
LangGraph agents are complex systems with multiple LLM calls, tool executions, and conditional routing. Without proper observability:
- You can't debug why an agent made a wrong decision
- You can't measure token usage or latency
- You can't identify failing patterns
- You can't improve performance systematically
Configurando el Trazado de LangSmith
import os
# Set environment variables
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls_..."
os.environ["LANGCHAIN_PROJECT"] = "my-agent-production"Or via .env file:
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=ls_...
LANGCHAIN_PROJECT=my-agent-production
LANGCHAIN_ENDPOINT=https://api.smith.langchain.com[!NOTA] Once tracing is enabled, all LangChain and LangGraph calls are automatically traced. Each graph invocation creates a run tree showing every node, LLM call, and tool execution.
Estructura del Árbol de Ejecución
A LangGraph invocation creates a hierarchical trace:
Root Run (graph invocation)
├── Node: classify
│ └── LLM Call (classify_question)
├── Node: web_search
│ └── Tool Call (web_search)
├── Node: agent
│ └── LLM Call (generate_response)
└── [END]
Each node and internal call is a separate run with its own:
- Input/output values
- Start/end timestamps
- Token counts
- Error information
- Metadata
Agregando Metadatos Personalizados
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}Custom metadata makes filtering and searching traces much more powerful.
Etiquetado y Enmascaramiento
Add tags to runs for filtering:
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)Mask sensitive data:
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]
Be careful with sensitive data. Use hide_inputs=True for PII-heavy applications, or implement a custom masking function.
Depuración con LangSmith
Viewing Traces
LangSmith Dashboard
├── Projects
│ └── my-agent
│ ├── Runs (individual invocations)
│ │ ├── Latency distribution
│ │ ├── Token usage
│ │ ├── Error rate
│ │ └── Cost tracking
│ └── Feedback / Annotations
Finding Problematic Runs
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
Collect feedback from users or annotate runs for evaluation:
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 evaluators to automatically score runs:
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] Automated evaluation lets you track performance over time, catch regressions, and compare different agent versions.
Monitoreo de Rendimiento
Key Métricas
| Métrica | Qué mide | Objetivo |
|---|---|---|
| Latency P50 | Median response time | < 5s |
| Latency P95 | Slowest 5% of responses | < 15s |
| Token Usage | Tokens consumed per run | Varies |
| Error Rate | Percentage of failed runs | < 1% |
| Costo Per Run | API cost per invocation | Budget-dependent |
| User Rating | Average feedback score | > 4.0 / 5.0 |
Setting Up Monitors
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
# 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
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)}Preguntas Prácticas
What environment variable enables LangSmith tracing?
What is the hierarchical structure of a LangGraph trace?
How can you filter runs by user or session in LangSmith?
What is a common use of LangSmith feedback?
Which metric would you monitor to detect slow agent responses?
How do you prevent sensitive data from appearing in LangSmith traces?
What does the LangSmith evaluate() function do?
Can you compare two different versions of your agent in LangSmith?
What does a LangSmith monitor do?
Which field in a run contains token usage information?
[!ÉXITO]
Conclusiones Clave
- LangSmith provides automatic tracing of all LangGraph executions
- Run trees show hierarchical execution: graph → nodes → LLM/tool calls
- Custom metadata and tags enable powerful filtering and search
- Feedback and annotation collect user ratings for evaluation
- Automated evaluation scores outputs against test datasets
- Monitors alert on error rate, latency, and other metrics
- A/B compare experiments to validate improvements
- Mask sensitive data with hide_inputs or custom callbacks