Observability with LangSmith

Advanced⏱35 minutes📖3 min readLesson 2 of 10

Observability with LangSmith

Master observability for LangGraph applications using LangSmith tracing, logging, debugging, run evaluation, and performance monitoring.

LangSmith is LangChain's observability platform for LLM applications. It provides tracing, debugging, evaluation, and monitoring for LangGraph applications in production.

Why Observability Matters

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

Setting Up LangSmith Tracing

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"

Or via .env file:

bash

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

ℹ️Note

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.

Run Tree Structure

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

Adding Custom Metadata

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}

Custom metadata makes filtering and searching traces much more powerful.

Tagging and Masking

Add tags to runs for filtering:

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)

Mask sensitive data:

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
)

⚠️Warning

Be careful with sensitive data. Use hide_inputs=True for PII-heavy applications, or implement a custom masking function.

Debugging with LangSmith

Viewing Traces

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

Finding Problematic Runs

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 and Annotation

Collect feedback from users or annotate runs for evaluation:

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"
)

Evaluation

Define evaluators to automatically score runs:

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"
)

✅Success

Automated evaluation lets you track performance over time, catch regressions, and compare different agent versions.

Performance Monitoring

Key Metrics

Metric	What It Measures	Target
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%
Cost Per Run	API cost per invocation	Budget-dependent
User Rating	Average feedback score	> 4.0 / 5.0

Setting Up Monitors

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
)

Comparing Runs

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")

Custom Logging in Nodes

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)}