Module 12: LangSmith and Production

Tracing and Observability

Capsule overview

Tracing records every operation your agent performs: model calls, tool executions, state changes, routing decisions. In LangSmith, all of it shows up on a visual timeline you can walk through step by step.

But setting up tracing is one line of code. The real skill is reading traces. Knowing how to spot which operation was slowest, see exactly which prompt was sent to the model, catch redundant tool calls that waste tokens, and find the exact point where the agent made a wrong decision. That's what this capsule teaches you.


Setup: one line, full tracing

Setting up LangSmith tracing takes exactly one environment variable:

from dotenv import load_dotenv
load_dotenv()

# In your .env:
# LANGSMITH_TRACING=true
# LANGSMITH_API_KEY=lsv2_pt_...
# LANGSMITH_PROJECT=research-assistant

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")
response = model.invoke("What is tracing in AI?")
print(response.content[:100])
# Expected output:
# Tracing in AI is the process of recording and monitoring every operation...

That's it. With LANGSMITH_TRACING=true, every LangChain and LangGraph operation gets sent to LangSmith automatically. You don't need decorators, wrappers, or code changes. Your existing agents — all of them, from Module 1 through Module 11 — get traced automatically.

Confirm tracing is on

from dotenv import load_dotenv
load_dotenv()

import os

tracing = os.getenv("LANGSMITH_TRACING", "false")
api_key = os.getenv("LANGSMITH_API_KEY", "")
project = os.getenv("LANGSMITH_PROJECT", "default")

print(f"Tracing on: {tracing}")
print(f"API key set: {'✅' if api_key else '❌'}")
print(f"Project: {project}")
# Expected output:
# Tracing on: true
# API key set: ✅
# Project: research-assistant

What a trace contains

A trace is the complete record of one execution. Each trace contains a hierarchy of runs (individual operations):

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

@tool
def search_web(query: str) -> str:
    """Search the web for information."""
    return f"Results for '{query}': AI safety is an active research field with 500+ papers in 2025."

@tool
def calculate(expression: str) -> str:
    """Evaluate a math expression."""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"Error: {e}"

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

result = agent.invoke({"messages": [{"role": "user", "content": "How many AI safety papers were published in 2025? Multiply that by 12."}]})
print(result["messages"][-1].content)
# Expected output:
# According to the search, over 500 AI safety papers were published in 2025. 500 × 12 = 6000.

In LangSmith, this trace shows:

Trace: "How many AI safety papers..."
│
├─ [LLM] gpt-4.1-mini — decides to use search_web
│   Input: system prompt + user message
│   Output: tool_call(search_web, "AI safety papers 2025")
│   Tokens: 180 input, 45 output
│   Latency: 0.8s
│
├─ [Tool] search_web("AI safety papers 2025")
│   Output: "Results for 'AI safety papers 2025': ..."
│   Latency: 0.01s
│
├─ [LLM] gpt-4.1-mini — decides to use calculate
│   Input: history + tool result
│   Output: tool_call(calculate, "500 * 12")
│   Tokens: 290 input, 30 output
│   Latency: 0.6s
│
├─ [Tool] calculate("500 * 12")
│   Output: "6000"
│   Latency: 0.001s
│
└─ [LLM] gpt-4.1-mini — produces the final answer
    Input: full history
    Output: "According to the search... 500 × 12 = 6000."
    Tokens: 350 input, 40 output
    Latency: 0.5s

Total: 3 LLM calls, 2 tool calls, 820 input tokens, 115 output tokens, 1.9s

Every run has: the exact input, the exact output, tokens used, latency, and its position in the hierarchy.


Reading traces: the real skill

Setting up tracing takes 30 seconds. Reading traces is the skill that separates a developer from a production AI engineer.

Skill 1: Find the bottleneck

The slowest operation dominates total latency. In a trace with 5 operations, if one takes 3 seconds and the others take 0.5 each, optimizing the four fast ones has no impact. You have to find and optimize the slow one.

from dotenv import load_dotenv
load_dotenv()

import time
from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    query: str
    search_results: str
    analysis: str
    report: str

def search(state: State) -> dict:
    time.sleep(2.0)
    return {"search_results": f"5 sources found for '{state['query']}'"}

def analyze(state: State) -> dict:
    time.sleep(0.3)
    return {"analysis": f"Analysis of: {state['search_results'][:40]}"}

def write_report(state: State) -> dict:
    time.sleep(0.5)
    return {"report": f"Report: {state['analysis']}"}

builder = StateGraph(State)
builder.add_node("search", search)
builder.add_node("analyze", analyze)
builder.add_node("report", write_report)
builder.add_edge(START, "search")
builder.add_edge("search", "analyze")
builder.add_edge("analyze", "report")
builder.add_edge("report", END)

graph = builder.compile()

start = time.time()
result = graph.invoke({"query": "AI in healthcare", "search_results": "", "analysis": "", "report": ""})
elapsed = time.time() - start

print(f"Result: {result['report']}")
print(f"Total time: {elapsed:.1f}s")
print(f"\nIn LangSmith you'll see:")
print(f"  search:  2.0s (71% of the total) ← BOTTLENECK")
print(f"  analyze: 0.3s (11%)")
print(f"  report:  0.5s (18%)")
print(f"\n→ Optimizing 'search' has 4x more impact than optimizing 'analyze'.")
# Expected output:
# Result: Report: Analysis of: 5 sources found for 'AI in healt
# Total time: 2.8s
#
# In LangSmith you'll see:
#   search:  2.0s (71% of the total) ← BOTTLENECK
#   analyze: 0.3s (11%)
#   report:  0.5s (18%)
#
# → Optimizing 'search' has 4x more impact than optimizing 'analyze'.

In the LangSmith dashboard, operations show up as bars on a timeline. The longest bar is your bottleneck. You don't guess — you see it.

Skill 2: See the exact prompt

When the model produces an unexpected answer, the first thing you need to know is: what prompt did it actually receive? Not the prompt you wrote — the complete prompt that was sent to the model, including the system prompt, the message history, tool results, and metadata.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage

model = init_chat_model("openai:gpt-4.1-mini")

messages = [
    SystemMessage(content="You are a financial analyst. Answer ONLY with numbers and percentages. Do not use narrative text."),
    HumanMessage(content="What is the impact of AI on the banking sector?"),
]

response = model.invoke(messages)
print(f"Answer: {response.content[:200]}")
print(f"\nIn LangSmith you'll see the EXACT input:")
print(f"  system: 'You are a financial analyst. Answer ONLY with numbers...'")
print(f"  human: 'What is the impact of AI on the banking sector?'")
print(f"\n→ If the answer is narrative, the prompt is clear but the model is ignoring it.")
print(f"→ If the answer is correct, the prompt is doing its job.")
# Expected output:
# Answer: • Operating cost reduction: 20-30%...
#
# In LangSmith you'll see the EXACT input:
#   system: 'You are a financial analyst. Answer ONLY with numbers...'
#   human: 'What is the impact of AI on the banking sector?'

In LangSmith, you click on the model's run and you see: input (the exact messages), output (the exact answer), model parameters (temperature, model name), and token counts. Prompt debugging becomes trivial.

Skill 3: Catch redundant tool calls

An agent can call the same tool twice with the same query, or call a tool it doesn't need. Every redundant tool call costs time and tokens.

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

call_log = []

@tool
def search_database(query: str) -> str:
    """Search the research database."""
    call_log.append(query)
    return f"3 results found for '{query}'"

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

result = agent.invoke({
    "messages": [{"role": "user", "content": "Search for papers on transformers and also search for papers on attention mechanisms"}]
})
print(f"Answer: {result['messages'][-1].content[:150]}")
print(f"\nTool calls made: {len(call_log)}")
for i, q in enumerate(call_log):
    print(f"  Call {i+1}: search_database('{q}')")
print(f"\n→ In LangSmith, you'll see each tool call on the timeline.")
print(f"→ If 'transformers' and 'attention mechanisms' return the same results,")
print(f"   one of those two calls is redundant.")
# Expected output:
# Answer: I found results on both topics...
#
# Tool calls made: 2
#   Call 1: search_database('transformers')
#   Call 2: search_database('attention mechanisms')
#
# → In LangSmith, you'll see each tool call on the timeline.
# → If 'transformers' and 'attention mechanisms' return the same results,
#    one of those two calls is redundant.

Skill 4: Find wrong decisions

In an agent with conditional routing, the model decides which path to take. If it takes the wrong one, the trace shows you exactly what input it got and what decision it made:

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

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

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

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

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

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

builder = StateGraph(State)
builder.add_node("router", router)
builder.add_node("technical", technical_handler)
builder.add_node("business", business_handler)

builder.add_edge(START, "router")
builder.add_conditional_edges("router", route_decision, {
    "technical": "technical",
    "business": "business",
})
builder.add_edge("technical", END)
builder.add_edge("business", END)

graph = builder.compile()

result = graph.invoke({"query": "How much does it cost to run RAG in production?", "route": "", "result": ""})
print(f"Route taken: {result['route']}")
print(f"Result: {result['result']}")
print(f"\nIn LangSmith you'll see:")
print(f"  1. The exact prompt the router received")
print(f"  2. The model's answer ('{result['route']}')")
print(f"  3. Which node ran as a result")
print(f"\n→ If the route is wrong, you see exactly why.")
# Expected output:
# Route taken: business
# Result: [BUSINESS] Business analysis of: How much does it cost to run RAG in production?
#
# In LangSmith you'll see:
#   1. The exact prompt the router received
#   2. The model's answer ('business')
#   3. Which node ran as a result

Trace hierarchy: run → child runs

A trace is hierarchical. The root run (the graph invocation) contains child runs (the nodes), which in turn contain child runs (model calls, tool calls):

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

@tool
def get_weather(city: str) -> str:
    """Get the weather for a city."""
    return f"In {city}: 22°C, partly cloudy"

@tool
def get_population(city: str) -> str:
    """Get the population of a city."""
    return f"{city}: 8.3 million people"

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

result = agent.invoke({
    "messages": [{"role": "user", "content": "Give me the weather and the population of Mexico City"}]
})
print(result["messages"][-1].content[:200])
print(f"\nTrace hierarchy in LangSmith:")
print(f"  Root run: agent.invoke()")
print(f"    ├─ LLM call: decides to use tools")
print(f"    ├─ Tool: get_weather('Mexico City')")
print(f"    ├─ Tool: get_population('Mexico City')")
print(f"    └─ LLM call: produces the final answer with both results")
# Expected output:
# In Mexico City the weather is 22°C, partly cloudy. The population is 8.3 million...
#
# Trace hierarchy in LangSmith:
#   Root run: agent.invoke()
#     ├─ LLM call: decides to use tools
#     ├─ Tool: get_weather('Mexico City')
#     ├─ Tool: get_population('Mexico City')
#     └─ LLM call: produces the final answer with both results

In the dashboard, you can expand and collapse levels of the hierarchy. For quick debugging, you look at the top level. For deep debugging, you expand until you see the input/output of each individual operation.


Custom metadata: filtering and organizing traces

In production, you generate thousands of traces. Without metadata, finding a specific trace is a needle-in-a-haystack problem. LangSmith lets you attach custom metadata so you can filter:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig

model = init_chat_model("openai:gpt-4.1-mini")

config = RunnableConfig(
    tags=["research", "production", "v7"],
    metadata={
        "user_id": "user_abc123",
        "session_id": "session_789",
        "agent_version": "7.0.1",
        "environment": "production",
    },
    run_name="Research Query - AI Safety",
)

response = model.invoke(
    "What are the main risks of running AI in production?",
    config=config,
)
print(f"Answer: {response.content[:120]}...")
print(f"\nIn LangSmith you can filter by:")
print(f"  - Tags: 'research', 'production', 'v7'")
print(f"  - Metadata: user_id='user_abc123'")
print(f"  - Run name: 'Research Query - AI Safety'")
# Expected output:
# Answer: The main risks of running AI in production include: hallucinations, unpredictable cost...
#
# In LangSmith you can filter by:
#   - Tags: 'research', 'production', 'v7'
#   - Metadata: user_id='user_abc123'
#   - Run name: 'Research Query - AI Safety'

Metadata worth having in production

MetadataWhat it's forExample
user_idFilter traces by userDebugging a reported issue
session_idGroup the traces of one sessionSeeing the full flow of a conversation
agent_versionCompare versions"Is v7.0.1 slower than v7.0.0?"
environmentSeparate prod/staging/devOnly looking at production traces
query_typeCategorize queries"Are research queries slower than Q&A ones?"

Tracing in LangGraph: seeing the graph structure

When you trace a LangGraph, LangSmith shows you not just the individual operations but the structure of the graph: which nodes ran, in what order, and which edges were taken:

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

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: 3 articles about {state['topic']}"]}

def search_papers(state: ResearchState) -> dict:
    return {"sources": [f"Papers: 2 papers about {state['topic']}"]}

def analyze(state: ResearchState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Analyze this briefly (1 sentence): {', '.join(state['sources'])}"
    )
    return {"analysis": response.content}

def write_report(state: ResearchState) -> dict:
    return {"report": f"Report on '{state['topic']}': {state['analysis']}"}

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

result = graph.invoke({"topic": "LLM observability", "sources": [], "analysis": "", "report": ""})
print(f"Report: {result['report'][:150]}")
print(f"Sources: {len(result['sources'])}")
print(f"\nIn LangSmith you'll see:")
print(f"  START → [search_web + search_papers] (parallel) → analyze → report → END")
print(f"  Parallel nodes show up as simultaneous bars on the timeline")
# Expected output:
# Report: Report on 'LLM observability': The sources indicate that...
# Sources: 2
#
# In LangSmith you'll see:
#   START → [search_web + search_papers] (parallel) → analyze → report → END
#   Parallel nodes show up as simultaneous bars on the timeline

A LangGraph trace shows the execution as a hierarchical tree where the root run contains one child run per node. If there are parallel nodes (like search_web and search_papers), LangSmith shows them simultaneously on the timeline — you can see that they ran at the same time.


Filtering traces in the dashboard

In production, you generate hundreds of traces a day. LangSmith gives you filters to find what you need:

By project

Each project in LangSmith groups traces. Organize by environment or by application:

# .env for development
LANGSMITH_PROJECT=research-assistant-dev

# .env for production
LANGSMITH_PROJECT=research-assistant-prod

By status (success/error)

Filter for traces that ended in an error, for debugging:

from dotenv import load_dotenv
load_dotenv()

from langsmith import Client

client = Client()

runs = list(client.list_runs(
    project_name="research-assistant-prod",
    is_root=True,
    filter='eq(status, "error")',
    limit=5,
))

print(f"Last {len(runs)} traces with an error:")
for run in runs:
    print(f"  - {run.name}: {run.error[:80] if run.error else 'No message'}")
    print(f"    ID: {run.id}")
    print(f"    Date: {run.start_time}")
# Expected output:
# Last 3 traces with an error:
#   - Research Query: RateLimitError: Rate limit exceeded...
#     ID: abc123-def456-...
#     Date: 2025-12-15 14:32:00
#   ...

By tags

Filter by the tags you added in the metadata:

from dotenv import load_dotenv
load_dotenv()

from langsmith import Client

client = Client()

runs = list(client.list_runs(
    project_name="research-assistant-prod",
    is_root=True,
    filter='has(tags, "research")',
    limit=5,
))

print(f"Last {len(runs)} traces tagged 'research':")
for run in runs:
    latency = (run.end_time - run.start_time).total_seconds() if run.end_time else 0
    print(f"  - {run.name}: {latency:.1f}s, {run.total_tokens or 0} tokens")
# Expected output:
# Last 5 traces tagged 'research':
#   - Research Query - AI Safety: 3.2s, 1450 tokens
#   - Research Query - RAG: 2.8s, 1200 tokens
#   ...

By date

Filter by date range to investigate incidents:

from dotenv import load_dotenv
load_dotenv()

from langsmith import Client
from datetime import datetime, timedelta

client = Client()

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

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

print(f"Traces in the last 24 hours: {len(runs)}")
if runs:
    latencies = [(r.end_time - r.start_time).total_seconds() for r in runs if r.end_time]
    if latencies:
        print(f"  Average latency: {sum(latencies)/len(latencies):.1f}s")
        print(f"  Max latency: {max(latencies):.1f}s")
        print(f"  Min latency: {min(latencies):.1f}s")
# Expected output:
# Traces in the last 24 hours: 10
#   Average latency: 3.4s
#   Max latency: 8.1s
#   Min latency: 1.2s

Naming conventions for traces

A good trace name makes debugging easy. A bad one makes it impossible:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig

model = init_chat_model("openai:gpt-4.1-mini")

queries = [
    ("What is RAG?", "simple-qa"),
    ("Research AI in finance, analyze trends, produce an executive report", "full-research"),
    ("Summarize this article in 3 bullets", "summarization"),
]

for query, query_type in queries:
    config = RunnableConfig(
        run_name=f"[{query_type}] {query[:50]}",
        metadata={"query_type": query_type},
    )
    response = model.invoke(query, config=config)
    print(f"[{query_type}] → {response.content[:60]}...")
# Expected output:
# [simple-qa] → RAG (Retrieval-Augmented Generation) is a pattern that...
# [full-research] → The impact of artificial intelligence on the finance...
# [summarization] → Without an article provided, I can't summarize...

In the dashboard, those traces show up as:

[simple-qa] What is RAG?
[full-research] Research AI in finance, analyze trends...
[summarization] Summarize this article in 3 bullets

Far more useful than three traces all named "ChatOpenAI".


Programmatic tracing: reaching runs from code

You can reach your traces programmatically with the LangSmith SDK:

from dotenv import load_dotenv
load_dotenv()

from langsmith import Client
from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig
import time

client = Client()
model = init_chat_model("openai:gpt-4.1-mini")

config = RunnableConfig(
    run_name="Programmatic Trace Demo",
    metadata={"demo": "true"},
    tags=["demo"],
)

response = model.invoke("What is LangSmith? Answer in one sentence.", config=config)
print(f"Answer: {response.content}")

time.sleep(3)

runs = list(client.list_runs(
    project_name="research-assistant",
    filter='has(tags, "demo")',
    limit=1,
))

if runs:
    run = runs[0]
    print(f"\nRun found:")
    print(f"  Name: {run.name}")
    print(f"  Status: {run.status}")
    print(f"  Input tokens: {run.prompt_tokens}")
    print(f"  Output tokens: {run.completion_tokens}")
    print(f"  Total tokens: {run.total_tokens}")
    latency = (run.end_time - run.start_time).total_seconds() if run.end_time else 0
    print(f"  Latency: {latency:.2f}s")
# Expected output:
# Answer: LangSmith is an observability platform for...
#
# Run found:
#   Name: Programmatic Trace Demo
#   Status: success
#   Input tokens: 28
#   Output tokens: 25
#   Total tokens: 53
#   Latency: 0.85s

Programmatic access is useful for: monitoring scripts, automated cost reports, custom alerts, and integration with external dashboards.


Troubleshooting

Problem 1: "My traces don't show up in LangSmith"

Symptom: You run LangChain code but you see no traces in the dashboard.

Cause: LANGSMITH_TRACING isn't set, or the API key is wrong.

Fix:

import os
print(f"LANGSMITH_TRACING: {os.getenv('LANGSMITH_TRACING')}")
print(f"LANGSMITH_API_KEY: {'set' if os.getenv('LANGSMITH_API_KEY') else 'NOT SET'}")
print(f"LANGSMITH_PROJECT: {os.getenv('LANGSMITH_PROJECT', 'default')}")

Check that LANGSMITH_TRACING=true (lowercase), that the API key is valid, and that load_dotenv() ran before you imported LangChain.

Problem 2: "Traces take a while to appear"

Symptom: The trace doesn't show up in the dashboard immediately.

Cause: LangSmith processes traces asynchronously. There's a 1-5 second delay.

Fix: Wait a few seconds and reload the dashboard. For programmatic access, add time.sleep(3) before you look for the trace.

Problem 3: "My traces have no token counts"

Symptom: The trace shows up but total_tokens is None.

Cause: The provider doesn't return usage info, or the model doesn't support token counting.

Fix: Check that you're using a model that reports tokens (OpenAI, Anthropic). Some local models or custom wrappers don't include that information.

Problem 4: "I have too many traces and can't find the one I need"

Symptom: The dashboard shows thousands of traces.

Cause: You're not using metadata or tags to organize them.

Fix: Add tags and metadata in RunnableConfig and use the dashboard filters. Organize by project (LANGSMITH_PROJECT), keeping environments separate.

Problem 5: "Tracing is making my agent slower"

Symptom: You notice extra latency when tracing is on.

Cause: Shipping traces to LangSmith adds minimal overhead (the send is asynchronous), but on slow networks it can be noticeable.

Fix: The send is asynchronous and shouldn't affect perceived latency. If it does, check your network connection. For performance tests, you can turn tracing off temporarily with LANGSMITH_TRACING=false.


Exercises

Exercise 1: First trace with metadata (Easy)

Write a script that invokes a model with a RunnableConfig that includes: a descriptive run_name, an "exercise" tag, and metadata with your name and the date. Run it and confirm the trace shows up in LangSmith with the right metadata.

See solution
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig
from datetime import date

model = init_chat_model("openai:gpt-4.1-mini")

config = RunnableConfig(
    run_name="Exercise 1 - First Trace",
    tags=["exercise", "module-12"],
    metadata={
        "author": "student",
        "date": str(date.today()),
        "exercise": "01-first-trace",
    },
)

response = model.invoke(
    "What is observability in AI systems? Answer in 2 sentences.",
    config=config,
)
print(f"Answer: {response.content}")
print(f"\n→ Open LangSmith, filter by the 'exercise' tag, and check:")
print(f"  - Run name: 'Exercise 1 - First Trace'")
print(f"  - Metadata: author='student', date='{date.today()}'")
# Expected output:
# Answer: Observability in AI systems is the ability to monitor...
#
# → Open LangSmith, filter by the 'exercise' tag, and check:
#   - Run name: 'Exercise 1 - First Trace'
#   - Metadata: author='student', date='2025-12-15'

Exercise 2: Trace a multi-node graph (Easy)

Build a StateGraph with 3 sequential nodes (fetch → process → format). Run it with tracing on. In LangSmith, identify: how many child runs the trace has, which node was slowest, and how many tokens were used in total.

See solution
from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    input: str
    fetched: str
    processed: str
    formatted: str

def fetch(state: State) -> dict:
    time.sleep(0.5)
    return {"fetched": f"Data about '{state['input']}'"}

def process(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Analyze in 1 sentence: {state['fetched']}")
    return {"processed": response.content}

def format_output(state: State) -> dict:
    return {"formatted": f"📊 RESULT: {state['processed']}"}

builder = StateGraph(State)
builder.add_node("fetch", fetch)
builder.add_node("process", process)
builder.add_node("format", format_output)
builder.add_edge(START, "fetch")
builder.add_edge("fetch", "process")
builder.add_edge("process", "format")
builder.add_edge("format", END)

graph = builder.compile()

start = time.time()
result = graph.invoke({"input": "AI observability", "fetched": "", "processed": "", "formatted": ""})
elapsed = time.time() - start

print(f"Result: {result['formatted']}")
print(f"Total time: {elapsed:.1f}s")
print(f"\n→ In LangSmith you'll see 3 child runs (fetch, process, format)")
print(f"→ 'process' should be the slowest (it includes the LLM call)")
print(f"→ Only 'process' burns tokens (fetch and format are local)")
# Expected output:
# Result: 📊 RESULT: AI observability involves monitoring and...
# Total time: 1.5s

Exercise 3: Find the bottleneck with timings (Medium)

Build a graph with 4 nodes where each one simulates a different latency (0.1s, 2.0s, 0.3s, 0.5s). Run it, then use the LangSmith SDK to fetch the trace and compute programmatically which node ate the largest share of the total time.

See solution
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_core.runnables import RunnableConfig

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

def fast_node(state: State) -> dict:
    time.sleep(0.1)
    return {"data": "fast", "log": ["fast:0.1s"]}

def slow_node(state: State) -> dict:
    time.sleep(2.0)
    return {"data": "slow", "log": ["slow:2.0s"]}

def medium_node(state: State) -> dict:
    time.sleep(0.3)
    return {"data": "medium", "log": ["medium:0.3s"]}

def final_node(state: State) -> dict:
    time.sleep(0.5)
    return {"data": "final", "log": ["final:0.5s"]}

builder = StateGraph(State)
builder.add_node("fast", fast_node)
builder.add_node("slow", slow_node)
builder.add_node("medium", medium_node)
builder.add_node("final", final_node)

builder.add_edge(START, "fast")
builder.add_edge("fast", "slow")
builder.add_edge("slow", "medium")
builder.add_edge("medium", "final")
builder.add_edge("final", END)

graph = builder.compile()

config = RunnableConfig(
    run_name="Bottleneck Analysis",
    tags=["bottleneck-exercise"],
)

start = time.time()
result = graph.invoke({"data": "", "log": []}, config=config)
total_elapsed = time.time() - start

print(f"Log: {result['log']}")
print(f"Total time: {total_elapsed:.1f}s")

expected_times = {"fast": 0.1, "slow": 2.0, "medium": 0.3, "final": 0.5}
total_expected = sum(expected_times.values())

print(f"\nBottleneck analysis:")
for name, t in sorted(expected_times.items(), key=lambda x: -x[1]):
    pct = (t / total_expected) * 100
    bar = "█" * int(pct / 2)
    print(f"  {name:>8}: {t:.1f}s ({pct:.0f}%) {bar}")

print(f"\n→ 'slow' eats {expected_times['slow']/total_expected*100:.0f}% of the total time")
print(f"→ Optimizing 'slow' has the biggest impact")
# Expected output:
# Log: ['fast:0.1s', 'slow:2.0s', 'medium:0.3s', 'final:0.5s']
# Total time: 2.9s
#
# Bottleneck analysis:
#      slow: 2.0s (69%) ██████████████████████████████████
#     final: 0.5s (17%) ████████
#    medium: 0.3s (10%) █████
#      fast: 0.1s (3%) █
#
# → 'slow' eats 69% of the total time
# → Optimizing 'slow' has the biggest impact

Exercise 4: Catch redundant tool calls (Medium)

Build an agent with a search tool that logs every call. Give it a prompt that could cause redundant calls. After running it, analyze the log to identify whether there were duplicate or very similar queries. Print a report.

See solution
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

call_log = []

@tool
def search(query: str) -> str:
    """Search for information on a topic."""
    call_log.append(query)
    return f"Results for '{query}': relevant information found."

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

call_log = []
result = agent.invoke({
    "messages": [{"role": "user", "content": (
        "I need information about machine learning. "
        "Search for machine learning applications. "
        "Also search for ML use cases."
    )}]
})

print(f"Answer: {result['messages'][-1].content[:120]}...")
print(f"\n=== TOOL CALL REPORT ===")
print(f"Total calls: {len(call_log)}")
for i, q in enumerate(call_log):
    print(f"  {i+1}. search('{q}')")

if len(call_log) > 1:
    print(f"\n=== REDUNDANCY ANALYSIS ===")
    from difflib import SequenceMatcher
    for i in range(len(call_log)):
        for j in range(i+1, len(call_log)):
            similarity = SequenceMatcher(None, call_log[i].lower(), call_log[j].lower()).ratio()
            status = "⚠️ POSSIBLY REDUNDANT" if similarity > 0.5 else "✅ Different"
            print(f"  '{call_log[i]}' vs '{call_log[j]}': {similarity:.0%} similar → {status}")
# Expected output:
# Answer: Here's the information about machine learning...
#
# === TOOL CALL REPORT ===
# Total calls: 3
#   1. search('machine learning')
#   2. search('machine learning applications')
#   3. search('ML use cases')
#
# === REDUNDANCY ANALYSIS ===
#   'machine learning' vs 'machine learning applications': 72% similar → ⚠️ POSSIBLY REDUNDANT
#   'machine learning' vs 'ML use cases': 25% similar → ✅ Different
#   'machine learning applications' vs 'ML use cases': 28% similar → ✅ Different

Exercise 5: Latency monitoring with the SDK (Medium)

Run 5 model invocations with inputs of different lengths. Then use the LangSmith SDK to fetch the traces and compute: average latency, max latency, and the correlation between input length and latency.

See solution
from dotenv import load_dotenv
load_dotenv()

import time
from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig
from langsmith import Client

model = init_chat_model("openai:gpt-4.1-mini")
client = Client()

inputs = [
    "What is AI?",
    "Explain the concept of machine learning and its main types in 2 sentences.",
    "Describe in detail how the attention mechanism works in transformers, including query, key, value, and multi-head attention. Be concise.",
    "Compare and contrast 5 language model architectures: GPT, BERT, T5, LLaMA, and Mixtral. For each one, mention: release year, size, and main use case.",
    "Write a complete analysis of the state of the art in Retrieval-Augmented Generation (RAG) in 2025, covering: chunking strategies, embedding models, vector databases, hybrid search, reranking, and evaluation metrics. Include the pros and cons of each approach.",
]

latencies = []
for i, input_text in enumerate(inputs):
    config = RunnableConfig(
        run_name=f"Latency Test {i+1}",
        tags=["latency-test"],
        metadata={"input_length": len(input_text), "test_index": i},
    )
    start = time.time()
    model.invoke(input_text, config=config)
    elapsed = time.time() - start
    latencies.append({"index": i+1, "input_len": len(input_text), "latency": elapsed})
    print(f"Test {i+1}: {len(input_text):>3} chars → {elapsed:.2f}s")

print(f"\n=== LATENCY ANALYSIS ===")
lats = [l["latency"] for l in latencies]
print(f"  Average: {sum(lats)/len(lats):.2f}s")
print(f"  Max:     {max(lats):.2f}s")
print(f"  Min:     {min(lats):.2f}s")

print(f"\n→ In LangSmith, filter by the 'latency-test' tag to see the 5 traces")
print(f"→ Compare each one's token count with its latency")
# Expected output:
# Test 1:  12 chars → 0.65s
# Test 2:  80 chars → 0.82s
# Test 3: 170 chars → 1.10s
# Test 4: 220 chars → 1.45s
# Test 5: 350 chars → 2.30s
#
# === LATENCY ANALYSIS ===
#   Average: 1.26s
#   Max:     2.30s
#   Min:     0.65s

Exercise 6: A trace dashboard with the SDK (Advanced)

Write a script that uses the LangSmith SDK to produce a text report of the last 10 traces in your project. For each trace, show: name, status, latency, total tokens, and whether it errored. At the end, compute aggregate statistics.

See solution
from dotenv import load_dotenv
load_dotenv()

from langsmith import Client
from datetime import datetime, timedelta

client = Client()

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

print(f"{'='*70}")
print(f" TRACE REPORT — {project_name}")
print(f" Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}")
print(f"{'='*70}\n")

total_tokens = 0
total_latency = 0
errors = 0

for i, run in enumerate(runs):
    latency = (run.end_time - run.start_time).total_seconds() if run.end_time else 0
    tokens = run.total_tokens or 0
    status = "✅" if run.status == "success" else "❌"

    print(f"{i+1}. {status} {run.name or 'Unnamed'}")
    print(f"   Status: {run.status} | Latency: {latency:.1f}s | Tokens: {tokens}")
    if run.error:
        print(f"   Error: {run.error[:80]}")
        errors += 1

    total_tokens += tokens
    total_latency += latency

print(f"\n{'='*70}")
print(f" AGGREGATE STATISTICS")
print(f"{'='*70}")
if runs:
    print(f"  Total traces: {len(runs)}")
    print(f"  Successful: {len(runs) - errors} | Errors: {errors}")
    print(f"  Average latency: {total_latency/len(runs):.1f}s")
    print(f"  Total tokens: {total_tokens:,}")
    print(f"  Average tokens: {total_tokens//len(runs) if runs else 0:,}")
    print(f"  Error rate: {errors/len(runs)*100:.0f}%")
# Expected output:
# ======================================================================
#  TRACE REPORT — research-assistant
#  Generated: 2025-12-15 15:30
# ======================================================================
#
# 1. ✅ Research Query - AI Safety
#    Status: success | Latency: 3.2s | Tokens: 1450
# 2. ✅ Exercise 1 - First Trace
#    Status: success | Latency: 0.9s | Tokens: 53
# ...

Summary

In this capsule you learned:

  • Setting up tracing is one line: LANGSMITH_TRACING=true. Every LangChain/LangGraph operation gets traced automatically without touching your code
  • The real skill is reading traces, not configuring them. Four skills: finding bottlenecks (the slowest operation dominates latency), seeing the exact prompt (prompt debugging), catching redundant tool calls (cost optimization), and finding wrong decisions (routing debugging)
  • A trace is hierarchical: root run → child runs → nested operations. In LangGraph, the graph's nodes show up as child runs with their own sub-runs (model calls, tool calls)
  • Custom metadata is essential in production: tags, metadata, and run_name in RunnableConfig let you filter and organize thousands of traces. Without metadata, finding a specific trace is impossible
  • Tracing in LangGraph shows the graph structure: which nodes ran, in what order, which ones were parallel, and which edges were taken
  • The LangSmith SDK gives you programmatic access to traces: listing runs, filtering by status/tags/date, and computing statistics — the foundation for monitoring scripts and alerts

Next capsule: Visual Agent Debugging — how to use LangSmith to go from "add a print and pray" to "click the trace and SEE what happened," with a systematic debugging workflow.


Additional resources

  1. LangSmith — Tracing Concepts — Core concepts: runs, traces, projects
  2. LangSmith — Annotate traces with metadata — How to add tags, metadata, and run names
  3. LangSmith — Filter traces in the application — Advanced filters in the dashboard
  4. LangSmith SDK Reference — Python SDK reference for programmatic access
  5. LangGraph Tracing — How tracing works specifically with LangGraph
  6. OpenTelemetry for LLMs — LangChain Blog — Context on tracing standards in LLM applications

Module 12 — LangChain & LangGraph: From Chains to Agents