Visual Agent Debugging

Capsule overview

LangSmith turns agent debugging from "add print statements and pray" into "click the trace and SEE what happened." When your agent produces an unexpected result, you don't need to re-run it, you don't need to guess, you don't need to spend money on extra API calls. You open the trace, walk the timeline, and find the exact step where things went wrong.

This isn't a nice-to-have. It's the difference between fixing a bug in 2 minutes and spending an hour guessing. And in production, where bugs cost real money and affect real users, that difference is critical.

In the previous capsule you learned how to set up tracing and read individual traces. In this one, you learn the full debugging workflow: from "the output is wrong" to "I found the root cause and fixed it."

The debugging workflow with LangSmith

Every time your agent produces an unexpected result, follow these 7 steps:

1. The agent produces unexpected output
     ↓
2. Open LangSmith → find the trace for that run
     ↓
3. Look at the timeline: how many steps? Any errors?
     ↓
4. Walk it step by step: click each run to see input/output
     ↓
5. Spot the step where things went off the rails
     ↓
6. Examine that step's input and output in detail
     ↓
7. Root cause → fix → verify

Let's walk each step with a concrete example.

A debugging scenario: the mixed-up report

Your Research Assistant researched "AI in education" but the report mixes education data with finance data. What happened?

Step 1: Reproduce and trace

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig

class ResearchState(TypedDict):
    topic: str
    sources: Annotated[list[str], operator.add]
    analysis: str
    report: str

def search_web(state: ResearchState) -> dict:
    return {"sources": [
        f"Web: AI is transforming education with personalized tutors",
        f"Web: Banks use AI for fraud detection",
    ]}

def search_papers(state: ResearchState) -> dict:
    return {"sources": [
        f"Paper: 'Adaptive Learning with LLMs' (2025)",
    ]}

def analyze(state: ResearchState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Analyze these sources about '{state['topic']}'. "
        f"Identify the 2 main trends.\n\n"
        f"Sources:\n" + "\n".join(f"- {s}" for s in state["sources"])
    )
    return {"analysis": response.content}

def write_report(state: ResearchState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Write a short report (3 sentences) about '{state['topic']}' "
        f"based on this analysis:\n\n{state['analysis']}"
    )
    return {"report": response.content}

builder = StateGraph(ResearchState)
builder.add_node("search_web", search_web)
builder.add_node("search_papers", search_papers)
builder.add_node("analyze", analyze)
builder.add_node("report", write_report)

builder.add_edge(START, "search_web")
builder.add_edge(START, "search_papers")
builder.add_edge("search_web", "analyze")
builder.add_edge("search_papers", "analyze")
builder.add_edge("analyze", "report")
builder.add_edge("report", END)

graph = builder.compile()

config = RunnableConfig(
    run_name="Debug: Mixed-Up Report",
    tags=["debug-exercise"],
    metadata={"scenario": "mixed-report"},
)

result = graph.invoke(
    {"topic": "AI in education", "sources": [], "analysis": "", "report": ""},
    config=config,
)

print(f"FINAL REPORT:\n{result['report']}")
print(f"\n⚠️  Does the report mention finance/banks? If so, there's a bug.")
print(f"\nTo debug, follow the steps in LangSmith:")
print(f"  1. Open the trace 'Debug: Mixed-Up Report'")
print(f"  2. Look at the child run 'search_web' → its output includes finance data")
print(f"  3. Root cause: search_web returned an irrelevant source about banks")
print(f"  4. Fix: improve the search so it filters by relevance to the topic")
# Expected output:
# FINAL REPORT:
# AI is transforming education with personalized tutors...
# (possibly mentions banking fraud detection)
#
# ⚠️  Does the report mention finance/banks? If so, there's a bug.

Steps 2-4: What you see in LangSmith

In the LangSmith dashboard, the trace "Debug: Mixed-Up Report" shows:

Trace: "Debug: Mixed-Up Report"           Total: 2.3s
│
├─ search_web (0.01s)
│   Output: ["Web: AI is transforming education...",
│            "Web: Banks use AI for fraud detection"]
│                                          ← ⚠️ IRRELEVANT DATA
│
├─ search_papers (0.01s)
│   Output: ["Paper: 'Adaptive Learning with LLMs' (2025)"]
│
├─ analyze (1.2s) — gpt-4.1-mini
│   Input: "Analyze these sources about 'AI in education'...
│           - Web: AI is transforming education...
│           - Web: Banks use AI for fraud detection  ← THERE IT IS
│           - Paper: 'Adaptive Learning with LLMs'"
│   Output: "Trend 1: Personalized tutors...
│            Trend 2: Fraud detection..."  ← BUG PROPAGATED
│
└─ report (1.1s) — gpt-4.1-mini
    Input: analysis contaminated with finance data
    Output: a report that mixes education and finance

Steps 5-7: Root cause found

The bug is in search_web: it returns a source about banks that isn't relevant to "AI in education." The analysis includes it because it received the sources unfiltered. The report reflects it because it was built on the contaminated analysis.

The root cause isn't the model — it's the search. The fix is to add relevance filtering after the search.

Visualizing graph execution

LangSmith shows the graph's structure in the trace. For every run, you can see exactly which nodes were visited and in what order:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig

class State(TypedDict):
    query: str
    category: str
    result: str
    path: str

def classify(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Classify this query as 'factual' or 'opinion'. "
        f"Answer with ONE word only.\n\n{state['query']}"
    )
    category = response.content.strip().lower()
    return {"category": category}

def handle_factual(state: State) -> dict:
    return {
        "result": f"[FACTUAL] Data-based answer for: {state['query']}",
        "path": "classify → factual → end",
    }

def handle_opinion(state: State) -> dict:
    return {
        "result": f"[OPINION] Analytical perspective on: {state['query']}",
        "path": "classify → opinion → end",
    }

def route(state: State) -> str:
    return "factual" if "factual" in state["category"] else "opinion"

builder = StateGraph(State)
builder.add_node("classify", classify)
builder.add_node("factual", handle_factual)
builder.add_node("opinion", handle_opinion)

builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route, {
    "factual": "factual",
    "opinion": "opinion",
})
builder.add_edge("factual", END)
builder.add_edge("opinion", END)

graph = builder.compile()

queries = [
    ("How many parameters does GPT-4 have?", "factual-query"),
    ("Is it better to use RAG or fine-tuning?", "opinion-query"),
]

for query, tag in queries:
    config = RunnableConfig(
        run_name=f"Routing: {query[:40]}",
        tags=[tag, "routing-debug"],
    )
    result = graph.invoke({"query": query, "category": "", "result": "", "path": ""}, config=config)
    print(f"Query: {query}")
    print(f"  Category: {result['category']}")
    print(f"  Path: {result['path']}")
    print(f"  Result: {result['result'][:60]}...")
    print()
# Expected output:
# Query: How many parameters does GPT-4 have?
#   Category: factual
#   Path: classify → factual → end
#   Result: [FACTUAL] Data-based answer for: How many parameters d...
#
# Query: Is it better to use RAG or fine-tuning?
#   Category: opinion
#   Path: classify → opinion → end
#   Result: [OPINION] Analytical perspective on: Is it better to us...

In LangSmith, the two traces show different paths through the graph. If the classification is wrong (the factual query got routed to "opinion"), you see exactly: what prompt the classifier received, what it answered, and which node the query was sent to.

Finding bottlenecks: the slowest node

When your agent's latency is unacceptable, you need to know where the time goes. LangSmith shows the duration of each operation as bars on a timeline — the bottleneck is visual and immediate:

from dotenv import load_dotenv
load_dotenv()

import time
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig

class State(TypedDict):
    topic: str
    web_results: str
    paper_results: str
    analysis: str
    report: str
    timings: Annotated[list[str], operator.add]

def search_web(state: State) -> dict:
    start = time.time()
    time.sleep(0.3)
    elapsed = time.time() - start
    return {
        "web_results": f"3 articles about {state['topic']}",
        "timings": [f"search_web: {elapsed:.2f}s"],
    }

def search_papers(state: State) -> dict:
    start = time.time()
    time.sleep(0.2)
    elapsed = time.time() - start
    return {
        "paper_results": f"2 papers about {state['topic']}",
        "timings": [f"search_papers: {elapsed:.2f}s"],
    }

def analyze(state: State) -> dict:
    start = time.time()
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Analyze this briefly (1 sentence): {state['web_results']}, {state['paper_results']}"
    )
    elapsed = time.time() - start
    return {
        "analysis": response.content,
        "timings": [f"analyze: {elapsed:.2f}s"],
    }

def write_report(state: State) -> dict:
    start = time.time()
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Write a 1-sentence summary: {state['analysis']}"
    )
    elapsed = time.time() - start
    return {
        "report": response.content,
        "timings": [f"write_report: {elapsed:.2f}s"],
    }

builder = StateGraph(State)
builder.add_node("search_web", search_web)
builder.add_node("search_papers", search_papers)
builder.add_node("analyze", analyze)
builder.add_node("report", write_report)

builder.add_edge(START, "search_web")
builder.add_edge(START, "search_papers")
builder.add_edge("search_web", "analyze")
builder.add_edge("search_papers", "analyze")
builder.add_edge("analyze", "report")
builder.add_edge("report", END)

graph = builder.compile()

config = RunnableConfig(
    run_name="Bottleneck Analysis",
    tags=["performance"],
)

total_start = time.time()
result = graph.invoke(
    {"topic": "AI observability", "web_results": "", "paper_results": "",
     "analysis": "", "report": "", "timings": []},
    config=config,
)
total_elapsed = time.time() - total_start

print(f"Report: {result['report'][:80]}...")
print(f"\n=== PERFORMANCE ANALYSIS ===")
print(f"Total: {total_elapsed:.2f}s\n")

for timing in result["timings"]:
    name, duration_str = timing.split(": ")
    duration = float(duration_str.replace("s", ""))
    pct = (duration / total_elapsed) * 100
    bar = "█" * int(pct / 2)
    print(f"  {name:<15} {duration_str:>6} ({pct:4.0f}%) {bar}")

print(f"\n→ In LangSmith, the timeline bars are proportional to time")
print(f"→ The longest node is your bottleneck — focus your optimization there")
# Expected output:
# Report: The sources analyzed show that AI observability is...
#
# === PERFORMANCE ANALYSIS ===
# Total: 2.50s
#
#   search_web       0.30s ( 12%) ██████
#   search_papers    0.20s (  8%) ████
#   analyze          1.20s ( 48%) ████████████████████████
#   write_report     0.80s ( 32%) ████████████████

Debugging tool calls: arguments and responses

When an agent uses tools, each tool call shows up in the trace with the exact arguments it sent and the response it got back:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from langchain_core.runnables import RunnableConfig

@tool
def search_database(query: str, limit: int = 5) -> str:
    """Search the research database."""
    if "error" in query.lower():
        raise ValueError(f"Invalid query: '{query}'")
    return f"[DB] {limit} results for '{query}': relevant data found."

@tool
def analyze_data(data: str, method: str = "statistical") -> str:
    """Analyze data with a specific method."""
    return f"[ANALYSIS] Method {method}: {data[:50]}... → 3 insights found."

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_react_agent(model, [search_database, analyze_data])

config = RunnableConfig(
    run_name="Debug Tool Calls",
    tags=["tool-debug"],
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "Search for papers on RAG and analyze them statistically"}]},
    config=config,
)

print(f"Answer: {result['messages'][-1].content[:150]}...")
print(f"\nIn LangSmith you'll see each tool call:")
print(f"  1. search_database(query='RAG', limit=5) → output")
print(f"  2. analyze_data(data=..., method='statistical') → output")
print(f"\n→ If the agent passed the wrong arguments, you see it right away")
print(f"→ If the tool returned unexpected data, you see it right away")
# Expected output:
# Answer: I found 5 results about RAG in the database...
#
# In LangSmith you'll see each tool call:
#   1. search_database(query='RAG', limit=5) → output
#   2. analyze_data(data=..., method='statistical') → output

Comparing traces: a good run vs a bad one

One of the most powerful uses of LangSmith is comparing two runs: one that produced a good result and one that failed. The difference between them shows you exactly what changed:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig

class State(TypedDict):
    query: str
    context: str
    response: str

def get_context(state: State) -> dict:
    contexts = {
        "good": "RAG combines retrieval with generation. It's implemented with vector stores. "
                "The key metrics are precision@k and recall@k.",
        "bad": "",
    }
    variant = "good" if "RAG" in state["query"] else "bad"
    return {"context": contexts.get(variant, "")}

def generate(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    prompt = f"Answer based ONLY on this context:\n\n"
    prompt += f"Context: {state['context'] or 'No context available.'}\n\n"
    prompt += f"Question: {state['query']}"
    response = model.invoke(prompt)
    return {"response": response.content}

builder = StateGraph(State)
builder.add_node("context", get_context)
builder.add_node("generate", generate)
builder.add_edge(START, "context")
builder.add_edge("context", "generate")
builder.add_edge("generate", END)

graph = builder.compile()

config_good = RunnableConfig(
    run_name="[GOOD] RAG query",
    tags=["comparison", "good"],
    metadata={"expected_quality": "high"},
)
good = graph.invoke(
    {"query": "How does RAG work?", "context": "", "response": ""},
    config=config_good,
)

config_bad = RunnableConfig(
    run_name="[BAD] Empty context query",
    tags=["comparison", "bad"],
    metadata={"expected_quality": "low"},
)
bad = graph.invoke(
    {"query": "How does quantum computing work?", "context": "", "response": ""},
    config=config_bad,
)

print(f"GOOD run:")
print(f"  Context: {good.get('context', '')[:60]}...")
print(f"  Response: {good['response'][:80]}...")
print(f"\nBAD run:")
print(f"  Context: '{bad.get('context', '')[:60] or 'EMPTY'}'")
print(f"  Response: {bad['response'][:80]}...")
print(f"\n=== COMPARISON ===")
print(f"Difference: the good run has relevant context.")
print(f"The bad one has empty context → the model improvises → low-quality answer.")
print(f"\nIn LangSmith, filter by the 'comparison' tag and compare both traces:")
print(f"  [GOOD] → context has data → generate produces an informed answer")
print(f"  [BAD]  → context is empty → generate produces a generic answer")
# Expected output:
# GOOD run:
#   Context: RAG combines retrieval with generation. It's implemented wi...
#   Response: RAG works by combining a retrieval system with generation...
#
# BAD run:
#   Context: 'EMPTY'
#   Response: Without specific context available, I can't provide...

In LangSmith, you open both traces side by side. In the "GOOD" trace, the context node produces useful data. In the "BAD" trace, it produces an empty string. The root cause is obvious: the context node found no data for "quantum computing."

Error traces: when something blows up

When a node throws an exception, LangSmith captures the full error including the traceback, the agent's state at that moment, and which nodes had already run:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.runnables import RunnableConfig

class State(TypedDict):
    data: str
    processed: str
    log: Annotated[list[str], operator.add]

def fetch_data(state: State) -> dict:
    return {"data": "raw data from API", "log": ["fetch: OK"]}

def process_data(state: State) -> dict:
    if not state["data"]:
        raise ValueError("There is no data to process")
    result = state["data"].upper()
    return {"processed": result, "log": ["process: OK"]}

def validate(state: State) -> dict:
    if len(state["processed"]) < 5:
        raise ValueError(f"Processed data is too short: '{state['processed']}'")
    return {"log": ["validate: OK"]}

builder = StateGraph(State)
builder.add_node("fetch", fetch_data)
builder.add_node("process", process_data)
builder.add_node("validate", validate)
builder.add_edge(START, "fetch")
builder.add_edge("fetch", "process")
builder.add_edge("process", "validate")
builder.add_edge("validate", END)

graph = builder.compile()

config = RunnableConfig(
    run_name="Error Trace Demo",
    tags=["error-debug"],
)

try:
    result = graph.invoke({"data": "", "processed": "", "log": []}, config=config)
    print(f"Result: {result}")
except Exception as e:
    print(f"Error caught: {type(e).__name__}: {e}")
    print(f"\nIn LangSmith you'll see:")
    print(f"  ✅ fetch: done (returned 'raw data from API')")
    print(f"  ✅ process: done (returned 'RAW DATA FROM API')")
    print(f"  ✅ validate: done (data > 5 characters)")
    print(f"\nIf fetch had returned empty data, you'd see:")
    print(f"  ✅ fetch: done (returned '')")
    print(f"  ❌ process: ERROR — 'There is no data to process'")
    print(f"     State at the moment of the error: data=''")
    print(f"  ⏭️  validate: never ran")
# Expected output:
# Result: {'data': 'raw data from API', 'processed': 'RAW DATA FROM API', 'log': ['fetch: OK', 'process: OK', 'validate: OK']}

In the LangSmith trace, an error shows up in red with the full traceback. You can see exactly what state the agent was in when it failed, which nodes had already run successfully, and which nodes never ran at all.

Connection with time-travel debugging (M8)

In Module 8 you learned time-travel debugging with checkpoints: walking your agent's state history locally. LangSmith complements that with a visual, remote layer:

Use time-travel when you need to manipulate state (fork, replay, update_state). Use LangSmith when you need the full picture of a run and want to share the debugging with your team.

Debugging patterns in production

Pattern 1: Reactive debugging — "a user reported a problem"

from dotenv import load_dotenv
load_dotenv()

from langsmith import Client

client = Client()

user_id = "user_abc123"
runs = list(client.list_runs(
    project_name="research-assistant-prod",
    filter=f'has(metadata, {{"user_id": "{user_id}"}})',
    limit=5,
))

print(f"Latest runs for {user_id}:")
for run in runs:
    status = "✅" if run.status == "success" else "❌"
    latency = (run.end_time - run.start_time).total_seconds() if run.end_time else 0
    print(f"  {status} {run.name} — {latency:.1f}s — {run.total_tokens or 0} tokens")
    if run.error:
        print(f"     Error: {run.error[:100]}")
# Expected output:
# Latest runs for user_abc123:
#   ✅ Research Query - AI Safety — 3.2s — 1450 tokens
#   ❌ Research Query - Quantum — 0.5s — 0 tokens
#      Error: RateLimitError: Rate limit exceeded...
#   ✅ Research Query - RAG — 2.8s — 1200 tokens

Pattern 2: Proactive debugging — "is anything weird today?"

from dotenv import load_dotenv
load_dotenv()

from langsmith import Client
from datetime import datetime, timedelta

client = Client()

since = datetime.now() - timedelta(hours=6)

runs = list(client.list_runs(
    project_name="research-assistant-prod",
    is_root=True,
    start_time=since,
    limit=50,
))

if runs:
    errors = [r for r in runs if r.status == "error"]
    latencies = [(r.end_time - r.start_time).total_seconds() for r in runs if r.end_time]
    slow_runs = [l for l in latencies if l > 5.0]

    print(f"=== HEALTH CHECK (last 6 hours) ===")
    print(f"  Total runs: {len(runs)}")
    print(f"  Errors: {len(errors)} ({len(errors)/len(runs)*100:.0f}%)")
    if latencies:
        print(f"  Average latency: {sum(latencies)/len(latencies):.1f}s")
        print(f"  Slow runs (>5s): {len(slow_runs)}")

    if len(errors) / len(runs) > 0.1:
        print(f"\n  ⚠️  ALERT: error rate > 10%")
    if slow_runs and len(slow_runs) / len(runs) > 0.2:
        print(f"  ⚠️  ALERT: > 20% of runs are slow")
# Expected output:
# === HEALTH CHECK (last 6 hours) ===
#   Total runs: 45
#   Errors: 3 (7%)
#   Average latency: 3.2s
#   Slow runs (>5s): 4

Pattern 3: Comparative debugging — "is the new version better?"

When you change an agent's prompt, you use version tags to compare:

from dotenv import load_dotenv
load_dotenv()

from langsmith import Client

client = Client()

def get_version_stats(project: str, version_tag: str) -> dict:
    runs = list(client.list_runs(
        project_name=project,
        is_root=True,
        filter=f'has(tags, "{version_tag}")',
        limit=20,
    ))
    if not runs:
        return {"count": 0, "avg_latency": 0, "avg_tokens": 0, "error_rate": 0}

    latencies = [(r.end_time - r.start_time).total_seconds() for r in runs if r.end_time]
    tokens = [r.total_tokens for r in runs if r.total_tokens]
    errors = sum(1 for r in runs if r.status == "error")

    return {
        "count": len(runs),
        "avg_latency": sum(latencies) / len(latencies) if latencies else 0,
        "avg_tokens": sum(tokens) / len(tokens) if tokens else 0,
        "error_rate": errors / len(runs) if runs else 0,
    }

v1 = get_version_stats("research-assistant-prod", "v6")
v2 = get_version_stats("research-assistant-prod", "v7")

print(f"{'Metric':<20} {'v6':>10} {'v7':>10} {'Change':>10}")
print(f"{'-'*50}")
print(f"{'Runs':<20} {v1['count']:>10} {v2['count']:>10}")
print(f"{'Latency (s)':<20} {v1['avg_latency']:>10.1f} {v2['avg_latency']:>10.1f}")
print(f"{'Average tokens':<20} {v1['avg_tokens']:>10.0f} {v2['avg_tokens']:>10.0f}")
print(f"{'Error rate':<20} {v1['error_rate']:>9.0%} {v2['error_rate']:>9.0%}")
# Expected output:
# Metric                     v6         v7     Change
# --------------------------------------------------
# Runs                       15         20
# Latency (s)               3.5        2.8
# Average tokens           1400       1100
# Error rate                 7%         5%

Troubleshooting

Problem 1: "I don't see the graph structure in the trace"

Symptom: The trace shows flat operations instead of the graph hierarchy.

Cause: Tracing captures the hierarchy automatically, but if you call model.invoke() outside a graph, there's no graph structure to show.

Fix: Check that you're invoking the compiled graph (graph.invoke()), not individual nodes. The hierarchy appears automatically when LangGraph manages the execution.

Problem 2: "The trace doesn't show the tool call's arguments"

Symptom: You can see the tool ran, but not which arguments it received.

Cause: The tool isn't decorated properly, or tracing didn't capture the details.

Fix: Make sure you're using @tool from langchain_core.tools. Native LangChain tools report their arguments automatically.

Problem 3: "I want to compare two traces but I can't find them"

Symptom: You need to compare a good run with a bad one.

Cause: Without consistent naming/tagging, finding specific traces is hard.

Fix: Use descriptive run_name values and tags to categorize. For example: tags=["research", "v7", "user_abc"] lets you filter by any combination.

Problem 4: "The error trace doesn't give me enough context"

Symptom: You can see a node failed but you don't understand why.

Cause: The agent's state at the moment of the error may not be obvious from the traceback alone.

Fix: Combine LangSmith with time-travel debugging (M8). The trace tells you where it failed; get_state_history() gives you the complete state at that point.

Exercises

Exercise 1: Debugging a wrong route (Easy)

Build a graph with a router that classifies queries as "technical" or "business." Send it an ambiguous query ("How much does it cost to implement RAG?") and use the trace in LangSmith to check which classification the model chose and why.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig

class State(TypedDict):
    query: str
    classification: str
    result: str

def classify(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Classify as 'technical' or 'business'. Answer with ONE word only.\n\n"
        f"Query: {state['query']}"
    )
    return {"classification": response.content.strip().lower()}

def technical(state: State) -> dict:
    return {"result": f"[TECH] Technical analysis: {state['query']}"}

def business(state: State) -> dict:
    return {"result": f"[BIZ] Business analysis: {state['query']}"}

def route(state: State) -> str:
    return "technical" if "technical" in state["classification"] else "business"

builder = StateGraph(State)
builder.add_node("classify", classify)
builder.add_node("technical", technical)
builder.add_node("business", business)

builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route, {"technical": "technical", "business": "business"})
builder.add_edge("technical", END)
builder.add_edge("business", END)

graph = builder.compile()

config = RunnableConfig(
    run_name="Route Debug: Ambiguous Query",
    tags=["routing-debug"],
)

result = graph.invoke(
    {"query": "How much does it cost to implement RAG?", "classification": "", "result": ""},
    config=config,
)

print(f"Query: {result['query']}")
print(f"Classification: {result['classification']}")
print(f"Result: {result['result']}")
print(f"\n→ Open the trace in LangSmith")
print(f"→ Click the 'classify' run → see the prompt and the model's answer")
print(f"→ Is the classification right? The query is ambiguous (it has a technical AND a business side)")
# Expected output:
# Query: How much does it cost to implement RAG?
# Classification: business
# Result: [BIZ] Business analysis: How much does it cost to implement RAG?

Exercise 2: Find the problem node (Medium)

Build a 4-node graph where the third node has an intentional bug (it adds junk data to the result). Run it with tracing on. Then use the LangSmith SDK to fetch the trace's child runs and find programmatically which node introduced the junk.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.runnables import RunnableConfig

class State(TypedDict):
    data: Annotated[list[str], operator.add]

def collect_a(state: State) -> dict:
    return {"data": ["Valid data A: transformers"]}

def collect_b(state: State) -> dict:
    return {"data": ["Valid data B: attention"]}

def collect_c(state: State) -> dict:
    return {"data": ["JUNK: buy now with a discount!!!"]}

def summarize(state: State) -> dict:
    return {"data": [f"Summary of {len(state['data'])} items"]}

builder = StateGraph(State)
builder.add_node("a", collect_a)
builder.add_node("b", collect_b)
builder.add_node("c_buggy", collect_c)
builder.add_node("summarize", summarize)

builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("b", "c_buggy")
builder.add_edge("c_buggy", "summarize")
builder.add_edge("summarize", END)

graph = builder.compile()

config = RunnableConfig(
    run_name="Find the Buggy Node",
    tags=["bug-hunt"],
)

result = graph.invoke({"data": []}, config=config)

print(f"Final data: {result['data']}")
print(f"\n=== AUTOMATIC ANALYSIS ===")

suspect_keywords = ["buy", "discount", "free", "click", "spam"]
for i, item in enumerate(result["data"]):
    is_suspect = any(kw in item.lower() for kw in suspect_keywords)
    status = "⚠️ SUSPICIOUS" if is_suspect else "✅ OK"
    print(f"  [{status}] {item}")

print(f"\n→ In LangSmith, click 'c_buggy' → you'll see its output is junk")
print(f"→ Nodes 'a' and 'b' produced valid data")
# Expected output:
# Final data: ['Valid data A: transformers', 'Valid data B: attention', 'JUNK: buy now with a discount!!!', 'Summary of 3 items']
#
# === AUTOMATIC ANALYSIS ===
#   [✅ OK] Valid data A: transformers
#   [✅ OK] Valid data B: attention
#   [⚠️ SUSPICIOUS] JUNK: buy now with a discount!!!
#   [✅ OK] Summary of 3 items

Exercise 3: Compare a good run vs a bad one (Medium)

Build an agent that generates summaries. Run it twice: once with rich context and once with empty context. Tag them "good" and "bad." Then use the SDK to fetch both traces and compare: latency, tokens, and status.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig
import time

class State(TypedDict):
    context: str
    summary: str

def summarize(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    prompt = f"Summarize in 1 sentence:\n\n{state['context'] or 'No context.'}"
    response = model.invoke(prompt)
    return {"summary": response.content}

builder = StateGraph(State)
builder.add_node("summarize", summarize)
builder.add_edge(START, "summarize")
builder.add_edge("summarize", END)
graph = builder.compile()

good_config = RunnableConfig(
    run_name="[GOOD] Rich Context",
    tags=["comparison-ex", "good"],
)
good_result = graph.invoke(
    {"context": "RAG combines retrieval and generation. It uses vector stores like Chroma or Pinecone. "
                "The key metrics are precision@k and recall@k. In 2025, hybrid search beats dense retrieval.",
     "summary": ""},
    config=good_config,
)

bad_config = RunnableConfig(
    run_name="[BAD] Empty Context",
    tags=["comparison-ex", "bad"],
)
bad_result = graph.invoke(
    {"context": "", "summary": ""},
    config=bad_config,
)

print(f"GOOD: {good_result['summary'][:80]}...")
print(f"BAD:  {bad_result['summary'][:80]}...")

time.sleep(3)

from langsmith import Client
client = Client()

for tag_filter, label in [("good", "GOOD"), ("bad", "BAD")]:
    runs = list(client.list_runs(
        project_name="research-assistant",
        filter=f'and(has(tags, "comparison-ex"), has(tags, "{tag_filter}"))',
        limit=1,
    ))
    if runs:
        run = runs[0]
        latency = (run.end_time - run.start_time).total_seconds() if run.end_time else 0
        print(f"\n[{label}] {run.name}")
        print(f"  Status: {run.status}")
        print(f"  Latency: {latency:.2f}s")
        print(f"  Tokens: {run.total_tokens or 'N/A'}")
# Expected output:
# GOOD: RAG is a technique that combines retrieval with generation using...
# BAD:  There is no context available to summarize.
#
# [GOOD] [GOOD] Rich Context
#   Status: success
#   Latency: 0.95s
#   Tokens: 120
#
# [BAD] [BAD] Empty Context
#   Status: success
#   Latency: 0.62s
#   Tokens: 35

Exercise 4: Automated health check (Advanced)

Write a script that works as a health check: it fetches the last 20 production traces, computes error rate, p50/p95 latency, average tokens, and produces a report with alerts if any metric crosses a threshold.

from dotenv import load_dotenv
load_dotenv()

from langsmith import Client
from datetime import datetime, timedelta

client = Client()

runs = list(client.list_runs(
    project_name="research-assistant",
    is_root=True,
    limit=20,
))

print(f"{'='*60}")
print(f" HEALTH CHECK — {datetime.now().strftime('%Y-%m-%d %H:%M')}")
print(f" Last {len(runs)} runs")
print(f"{'='*60}\n")

if not runs:
    print("  ⚠️  No traces available.")
else:
    errors = [r for r in runs if r.status == "error"]
    latencies = sorted(
        [(r.end_time - r.start_time).total_seconds() for r in runs if r.end_time]
    )
    tokens = [r.total_tokens for r in runs if r.total_tokens]

    error_rate = len(errors) / len(runs)
    p50 = latencies[len(latencies) // 2] if latencies else 0
    p95 = latencies[int(len(latencies) * 0.95)] if latencies else 0
    avg_tokens = sum(tokens) / len(tokens) if tokens else 0

    THRESHOLDS = {
        "error_rate": 0.10,
        "p95_latency": 5.0,
        "avg_tokens": 3000,
    }

    print(f"  Metric              Value        Threshold    Status")
    print(f"  {'-'*55}")

    er_status = "❌ ALERT" if error_rate > THRESHOLDS["error_rate"] else "✅ OK"
    print(f"  Error rate          {error_rate:>6.0%}       <{THRESHOLDS['error_rate']:.0%}         {er_status}")

    p95_status = "❌ ALERT" if p95 > THRESHOLDS["p95_latency"] else "✅ OK"
    print(f"  Latency p50         {p50:>6.1f}s      —            —")
    print(f"  Latency p95         {p95:>6.1f}s      <{THRESHOLDS['p95_latency']:.0f}s          {p95_status}")

    tok_status = "❌ ALERT" if avg_tokens > THRESHOLDS["avg_tokens"] else "✅ OK"
    print(f"  Average tokens      {avg_tokens:>6.0f}       <{THRESHOLDS['avg_tokens']}        {tok_status}")

    alerts = []
    if error_rate > THRESHOLDS["error_rate"]:
        alerts.append(f"High error rate: {error_rate:.0%}")
    if p95 > THRESHOLDS["p95_latency"]:
        alerts.append(f"High p95 latency: {p95:.1f}s")
    if avg_tokens > THRESHOLDS["avg_tokens"]:
        alerts.append(f"High token usage: {avg_tokens:.0f}")

    if alerts:
        print(f"\n  ⚠️  ALERTS:")
        for alert in alerts:
            print(f"    - {alert}")
    else:
        print(f"\n  ✅ Everything within normal parameters")
# Expected output:
# ============================================================
#  HEALTH CHECK — 2025-12-15 15:45
#  Last 20 runs
# ============================================================
#
#   Metric              Value        Threshold    Status
#   -------------------------------------------------------
#   Error rate             5%       <10%         ✅ OK
#   Latency p50           2.3s      —            —
#   Latency p95           4.8s      <5s          ✅ OK
#   Average tokens        1200       <3000        ✅ OK
#
#   ✅ Everything within normal parameters

Summary

In this capsule you learned:

• The debugging workflow with LangSmith is systematic: unexpected output → find the trace → walk the timeline → spot the problem step → examine input/output → root cause → fix. No more "add a print and re-run"
• Visualizing graph execution shows which nodes were visited, in what order, and which path the routing took. If the agent went down the wrong path, you see it immediately
• Finding bottlenecks is visual: the timeline bars are proportional to time. The longest bar is your bottleneck — focus your optimization there
• Debugging tool calls shows the exact arguments and responses. If the agent sent the wrong arguments to a tool, you don't have to guess — you see it in the trace
• Comparing traces (good vs bad) reveals exactly what changed between a successful run and a failed one. It's the most efficient debugging there is
• Error traces capture the agent's complete state at the moment of failure: traceback, state, nodes that ran, nodes that never did
• Three debugging patterns in production: reactive (a user reported something), proactive (a periodic health check), and comparative (is the new version better?)

Next capsule: Evaluation: Datasets and Evaluators — how to go from "seems to work" to systematic quality measurement against specific criteria.

Additional resources

1. LangSmith — Tracing FAQ — Frequently asked questions about tracing and debugging
2. LangSmith — Filter runs — How to filter and search traces in the dashboard
3. LangGraph — Debugging — Debugging graphs with built-in tracing
4. LangSmith SDK — list_runs — SDK reference for programmatic access to traces
5. LangSmith — Comparison View — How to compare runs in the dashboard

Module 12 — LangChain & LangGraph: From Chains to Agents