Observabilidade com LangSmith
Domine observabilidade para aplicações LangGraph usando rastreamento LangSmith, registro em log, depuração, avaliação de execuções e monitoramento de desempenho.
Observabilidade com LangSmith
LangSmith is LangChain's observability platform for LLM applications. It provides tracing, debugging, evaluation, and monitoring for LangGraph applications in production.
Por que a Observabilidade é Importante
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 o Rastreamento 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.
Estrutura da Árvore de Execução
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
Adicionando Metadados 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.
Marcação e Mascaramento
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
)[!AVISO]
Be careful with sensitive data. Use hide_inputs=True for PII-heavy applications, or implement a custom masking function.
Depuração com 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 e Anotação
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"
)Avaliação
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"
)[!SUCESSO] Automated evaluation lets you track performance over time, catch regressions, and compare different agent versions.
Monitoramento de Desempenho
Key Métricas
| Métrica | O que mede | Alvo |
|---|---|---|
| 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% |
| Custo 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 Execuções
# 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 em Nós
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)}Perguntas Práticas
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?
[!SUCESSO]
Principais Conclusões
- 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