Tracing & Debugging with LangSmith

Why Tracing Is Non-negotiable

In traditional software, a failing request leaves an error log with a stack trace. In an agent system, a failing run might have:

• 15 LLM calls
• 8 tool invocations
• 3 conditional routing decisions
• State updates at each step

When step 12 fails because step 7 put bad data in the state, you need a trace, not a log.

Setting Up LangSmith

pip install langsmith

# Set these environment variables at application startup: before any agent calls
# LangChain/LangGraph reads these at import time to initialize tracing
import os
os.environ["LANGSMITH_TRACING"] = "true"           # master switch, set to "false" to disable
os.environ["LANGSMITH_API_KEY"] = "lsv2_..."       # from app.smith.langchain.com → Settings → API Keys
os.environ["LANGSMITH_PROJECT"] = "production-agents"  # groups runs in the LangSmith dashboard
os.environ["LANGSMITH_ENDPOINT"] = "https://api.smith.langchain.com"

That’s all for basic LangGraph tracing. Every agent.invoke() and agent.astream() call now automatically creates a trace in your LangSmith project.

Adding Context to Traces

Auto-captured traces are useful; contextualized traces are essential.

from langsmith import Client, traceable, trace
from langsmith.run_helpers import get_current_run_tree

ls_client = Client()

# Approach 1: @traceable decorator: wraps the entire function as a top-level LangSmith run
# @traceable decorator auto-creates a LangSmith run, capturing inputs, outputs, and timing
@traceable(
    name="Research Agent",
    tags=["production", "v2"],          # tags appear as filter chips in the LangSmith dashboard
    metadata={"agent_version": "2.1.0"}, # static metadata attached to every run of this function
)
async def run_research_agent(query: str, user_id: str) -> str:
    # get_current_run_tree() returns the active trace: lets you add dynamic context mid-execution
    run = get_current_run_tree()
    if run:
        run.extra = {
            **(run.extra or {}),
            "user_id": user_id,         # correlate traces with your user database
            "query_length": len(query), # useful for filtering expensive long-query runs
        }
    
    result = await agent.ainvoke({"messages": [HumanMessage(content=query)]})
    return result["messages"][-1].content

# Approach 2: context manager: more explicit control, useful when you need to attach callbacks
async def run_with_trace_context(query: str, user_id: str, org_id: str) -> str:
    with trace(
        name="Agent Run",
        project_name="production-agents",
        tags=["agent", "v2"],
        metadata={
            "user_id": user_id,
            "org_id": org_id,
            "model": "gpt-5.6-sol",
            "feature_flags": {"new_rag": True},  # capture A/B variants for comparison analysis
        },
    ) as run:
        config = {
            "configurable": {"thread_id": f"{user_id}-{uuid.uuid4()}"},
            # run.get_langchain_handler() connects the LangGraph run to this LangSmith trace
            "callbacks": [run.get_langchain_handler()],
        }
        
        result = await agent.ainvoke(
            {"messages": [HumanMessage(content=query)]},
            config=config,
        )
        
        # add_metadata() attaches outcome data after the run: visible in the trace detail view
        run.add_metadata({
            "total_iterations": result.get("iterations", 0),
            "succeeded": result.get("final_answer") is not None,
        })
        
        return result["messages"][-1].content

The Run Naming Convention

Consistent run naming makes filtering 1000s of traces tractable:

def create_trace_config(
    user_id: str,
    task_type: str,
    thread_id: str,
    version: str = "v2",
) -> dict:
    return {
        "configurable": {"thread_id": thread_id},
        # run_name format "task_type:user_id:thread_prefix" makes runs scannable in the list view
        "run_name": f"{task_type}:{user_id}:{thread_id[:8]}",
        # tags support faceted filtering: find all "version:v2" runs, or all "task:research" runs
        "tags": [f"version:{version}", f"task:{task_type}", "production"],
        "metadata": {
            "user_id": user_id,
            "task_type": task_type,
            "thread_id": thread_id,    # full thread_id for LangGraph memory lookup
            "agent_version": version,
            # environment tag lets you separate dev/staging/prod traces in one project
            "environment": os.getenv("ENVIRONMENT", "development"),
        }
    }

Debugging with the Trace Viewer

When an agent fails, the LangSmith trace viewer workflow:

flowchart TD
    S1["1. Find the failing run\nFilter by: project · date · tag · error status"] --> S2["2. Open execution tree\nSee all nodes + tool calls as a trace"]
    S2 --> S3["3. Expand call_tools\nTool: fetch_page → ERROR 403 Forbidden"]
    S3 --> S4["4. Click the failing call\nSee input args · error message · timing"]
    S4 --> S5["5. Compare with a passing run\nDiff state at step 8, what changed?"]
    S5 --> FIX(["Root cause found\nFix and re-deploy"])
    style S3 fill:#fff0f0,stroke:#f87171,color:#0F172A
    style FIX fill:#f0fdf9,stroke:#0D9488,color:#0F172A

Automated Monitoring with LangSmith Rules

# Set up programmatic monitoring: use the LangSmith client to query your own run history
from langsmith import Client

ls_client = Client()

def create_cost_monitor(project_name: str, threshold_usd: float = 0.50):
    # In LangSmith UI: Monitors → Create Rule
    # Condition: run.total_cost > threshold_usd
    # Action: send email / Slack notification
    # The SDK doesn't expose rule creation yet: configure monitors in the UI
    pass

def analyze_recent_failures(project: str = "production-agents"):
    # list_runs() is a paginated iterator: it fetches runs in batches under the hood
    runs = ls_client.list_runs(
        project_name=project,
        error=True,   # filter to only runs where an exception was raised
        limit=100,
        start_time=datetime.utcnow() - timedelta(hours=24),
    )
    
    # Bucket errors by category to find systemic issues vs one-off failures
    error_categories = {}
    for run in runs:
        error = str(run.error or "unknown")
        category = categorize_error(error)
        error_categories[category] = error_categories.get(category, 0) + 1
    
    return error_categories

def compute_daily_cost(project: str = "production-agents") -> float:
    runs = ls_client.list_runs(
        project_name=project,
        start_time=datetime.utcnow() - timedelta(hours=24),
    )
    # r.total_cost is pre-computed by LangSmith from token counts: no manual price math needed
    return sum(r.total_cost or 0 for r in runs)

Building Custom Evaluators

from langsmith.evaluation import evaluate, EvaluationResult
from langsmith.schemas import Example, Run

def accuracy_evaluator(run: Run, example: Example) -> EvaluationResult:
    """Check if the agent's answer matches the expected answer."""
    # run.outputs contains what the agent actually returned for this example
    predicted = run.outputs.get("answer", "")
    # example.outputs contains the human-curated ground truth from the dataset
    expected = example.outputs.get("answer", "")
    
    # LLM-as-judge: use a model to score semantic correctness, not just string match
    eval_response = llm.invoke([
        HumanMessage(content=f"""
Score this answer 0-1 for correctness:
Question: {example.inputs.get('query')}
Expected: {expected}
Got: {predicted}

Return just a number between 0 and 1.
""")
    ])
    
    try:
        score = float(eval_response.content.strip())
    except ValueError:
        score = 0.0  # treat unparseable eval responses as failures
    
    # EvaluationResult.key becomes the metric name in LangSmith's experiment comparison view
    return EvaluationResult(
        key="accuracy",
        score=score,
        comment=f"Expected: '{expected[:50]}...', Got: '{predicted[:50]}...'"
    )

# evaluate() runs every example in the dataset through the agent, then scores each with evaluators
results = evaluate(
    lambda inputs: agent.invoke(inputs),
    data="my-agent-dataset",       # LangSmith dataset name, stores your curated test cases
    evaluators=[accuracy_evaluator],
    experiment_prefix="v2.1-test", # groups results under this name for A/B comparison
    metadata={"model": "gpt-5.6-sol", "version": "2.1"},
)

print(f"Average accuracy: {results.results.get('accuracy', {}).get('mean', 0):.2%}")

OpenTelemetry Integration

For teams that already have an observability stack (Jaeger, Zipkin, Datadog):

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

# OTLPSpanExporter ships spans to any OTLP-compatible backend: Jaeger, Zipkin, Datadog, Grafana Tempo
provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(endpoint="http://jaeger:4317")))
trace.set_tracer_provider(provider)

# get_tracer creates a named tracer: the name appears as the instrumentation library in your backend
tracer = trace.get_tracer("agent-service")

async def traced_agent_run(query: str) -> str:
    # start_as_current_span creates a span: all nested calls (including LLM calls) attach as child spans
    with tracer.start_as_current_span("agent-run") as span:
        # set_attribute adds key-value metadata: these become filterable dimensions in your dashboards
        span.set_attribute("query.length", len(query))
        span.set_attribute("agent.version", "2.1")
        
        result = await agent.ainvoke({"messages": [HumanMessage(content=query)]})
        
        # Add outcome attributes after the run: useful for SLO tracking (success rate, iteration count)
        span.set_attribute("result.iterations", result.get("iterations", 0))
        span.set_attribute("result.success", True)
        
        return result["messages"][-1].content

LangSmith for AI-specific tracing, OpenTelemetry for infrastructure-level observability, both together give you complete visibility into your agent system.