Observability and Tracing with LangGraph: Complete Guide

Observability and Tracing with LangGraph: Complete Guide

Building agentic workflows with LangGraph is powerful, but as your graphs grow in complexity—with multiple nodes, conditional edges, tool calls, and LLM invocations—understanding what happens inside them becomes critical. Observability and tracing give you the visibility needed to debug, optimize, and monitor your agents in production. This guide walks through everything you need to know to make your LangGraph applications transparent and debuggable.

What Is Observability in LangGraph?

Observability is the ability to inspect the internal state and behavior of a system from the outside. In the context of LangGraph, it means tracking every step your graph takes: which nodes executed, what data flowed between them, how long each LLM call took, what tools were invoked, and where errors occurred.

Tracing is the specific technique of recording a hierarchical timeline of execution. A trace captures the full call tree—parent operations and their nested children—so you can see exactly how a single user request propagated through your graph.

LangGraph is built on top of LangChain's instrumentation layer, which means every LLM call, tool execution, and graph node is automatically traceable when you configure a tracing backend.

Why Observability Matters for Agentic Workflows

Agentic systems behave differently from traditional applications. They are non-deterministic, multi-step, and often involve reasoning that is hard to predict. Without observability, you are flying blind. Here is why it matters:

• Debugging loops and dead ends: Agents can get stuck in cycles or take unexpected paths through conditional edges. Traces reveal exactly which branches were taken and why.
• Cost tracking: Each LLM call consumes tokens. Tracing lets you attribute token usage to specific nodes so you can identify expensive operations.
• Latency analysis: You can pinpoint which node or tool call is the bottleneck in your graph's execution.
• Quality evaluation: By reviewing traces, you can assess whether your agent's reasoning and outputs meet quality standards.
• Production monitoring: In production, observability tools alert you to failures, regressions, and performance degradation over time.

Setting Up LangSmith for Tracing

LangSmith is LangChain's official observability platform and the most straightforward way to trace LangGraph applications. It provides a visual interface for inspecting traces, comparing runs, and building evaluation datasets.

To get started, sign up for a LangSmith account and obtain your API key. Then configure your environment variables:

import os

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls__your_api_key_here"
os.environ["LANGCHAIN_PROJECT"] = "my-langgraph-project"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"

Once these environment variables are set, LangGraph automatically emits traces to LangSmith for every run. You do not need to modify your graph code at all. Let's build a simple graph and see this in action:

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]

llm = ChatOpenAI(model="gpt-4o", temperature=0)

def call_model(state: AgentState) -> AgentState:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("tools", "agent")

app = workflow.compile()

# Run it — this will automatically be traced in LangSmith
result = app.invoke({
    "messages": [HumanMessage(content="What is 25 multiplied by 13?")]
})
print(result["messages"][-1].content)

After running this code, navigate to your LangSmith dashboard. You will see a trace showing the graph execution, the LLM call within the agent node, token usage, latency, and the full message history. This visibility comes for free just by setting environment variables.

Using LangGraph's Built-in Time Travel and State Inspection

Beyond external tracing tools, LangGraph itself provides powerful introspection capabilities. Every compiled graph maintains a checkpointable state, and you can inspect or replay from any point in execution.

To enable checkpointing, use a checkpointer when compiling your graph:

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)

# Run with a thread_id to enable state tracking
config = {"configurable": {"thread_id": "thread-1"}}
result = app.invoke(
    {"messages": [HumanMessage(content="Hello, who are you?")]},
    config=config
)

# Inspect the full state history
for state in app.get_state_history(config):
    print(f"Step: {state.next}, Messages: {len(state.values.get('messages', []))}")
    print(f"  Created at: {state.created_at}")
    print(f"  Config: {state.config}")

This state history is invaluable for debugging. You can see every intermediate state your graph passed through. You can even replay from a specific checkpoint or modify state before resuming:

# Get the state history and pick an earlier checkpoint
states = list(app.get_state_history(config))
earlier_state = states[-2]  # Pick an earlier point

# Resume execution from that checkpoint
for event in app.stream(
    None,
    earlier_state.config,
    stream_mode="values"
):
    if "messages" in event:
        last_msg = event["messages"][-1]
        print(f"Message: {last_msg.content[:100]}...")

Custom Callbacks for Fine-Grained Tracing

Sometimes you need more control over what gets traced. LangGraph and LangChain support a callback system that lets you hook into every event. This is useful when you want to log to a custom backend, add business-specific metadata, or filter which events get recorded.

from langchain_core.callbacks import BaseCallbackHandler
from typing import Any, Dict, List, Optional
from uuid import UUID
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("langgraph_tracer")

class CustomTracingHandler(BaseCallbackHandler):
    def __init__(self):
        self.start_times = {}

    def on_llm_start(
        self,
        serialized: Dict[str, Any],
        prompts: List[str],
        *,
        run_id: UUID,
        **kwargs: Any
    ) -> None:
        self.start_times[run_id] = time.time()
        model_name = serialized.get("name", "unknown")
        logger.info(f"[LLM START] Model: {model_name}, Run ID: {run_id}")

    def on_llm_end(
        self,
        response,
        *,
        run_id: UUID,
        **kwargs: Any
    ) -> None:
        duration = time.time() - self.start_times.get(run_id, time.time())
        token_usage = response.llm_output.get("token_usage", {}) if response.llm_output else {}
        logger.info(
            f"[LLM END] Run ID: {run_id}, Duration: {duration:.2f}s, "
            f"Tokens: {token_usage}"
        )

    def on_tool_start(
        self,
        serialized: Dict[str, Any],
        input_str: str,
        *,
        run_id: UUID,
        **kwargs: Any
    ) -> None:
        tool_name = serialized.get("name", "unknown")
        logger.info(f"[TOOL START] Tool: {tool_name}, Input: {input_str[:100]}")

    def on_tool_end(
        self,
        output: str,
        *,
        run_id: UUID,
        **kwargs: Any
    ) -> None:
        logger.info(f"[TOOL END] Output: {str(output)[:100]}")

    def on_chain_error(
        self,
        error: BaseException,
        *,
        run_id: UUID,
        **kwargs: Any
    ) -> None:
        logger.error(f"[CHAIN ERROR] Run ID: {run_id}, Error: {error}")

# Attach the callback to your LLM
llm_with_callbacks = ChatOpenAI(
    model="gpt-4o",
    temperature=0,
    callbacks=[CustomTracingHandler()]
)

You can also attach callbacks at the graph level by passing them in the config during invocation:

result = app.invoke(
    {"messages": [HumanMessage(content="What is the weather in Tokyo?")]},
    config={
        "callbacks": [CustomTracingHandler()],
        "configurable": {"thread_id": "thread-2"}
    }
)

Streaming for Real-Time Observability

Tracing after the fact is useful, but sometimes you need real-time visibility. LangGraph supports multiple streaming modes that let you observe execution as it happens:

# Stream mode options:
# - "values": streams the full state after each step
# - "updates": streams only the updates from each step
# - "messages": streams LLM tokens as they are generated
# - "debug": streams detailed debug events including task execution

config = {"configurable": {"thread_id": "thread-3"}}

print("=== Streaming Updates ===")
for event in app.stream(
    {"messages": [HumanMessage(content="Explain quantum computing in 2 sentences.")]},
    config=config,
    stream_mode="updates"
):
    for node_name, node_output in event.items():
        print(f"Node '{node_name}' produced:")
        if "messages" in node_output:
            for msg in node_output["messages"]:
                print(f"  {msg.type}: {msg.content[:150]}...")

print("\n=== Streaming Debug Info ===")
for event in app.stream(
    {"messages": [HumanMessage(content="What is 5 + 3?")]},
    config={"configurable": {"thread_id": "thread-4"}},
    stream_mode="debug"
):
    if event.get("type") == "task":
        print(f"Task: {event.get('name', 'unknown')}")
    elif event.get("type") == "task_result":
        print(f"Result received for: {event.get('name', 'unknown')}")

Integrating with OpenTelemetry

If your organization uses OpenTelemetry for distributed tracing, you can integrate LangGraph traces into your existing observability stack. The langfuse package and OpenTelemetry-compatible backends can capture LangChain/LangGraph events:

# Install: pip install langfuse opentelemetry-sdk opentelemetry-exporter-otlp

from langfuse.callback import CallbackHandler
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

# Set up OpenTelemetry
provider = TracerProvider()
processor = BatchSpanProcessor(
    OTLPSpanExporter(endpoint="http://localhost:4317")
)
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

# Set up Langfuse handler for LangGraph
langfuse_handler = CallbackHandler(
    public_key="pk-lf-your-key",
    secret_key="sk-lf-your-secret",
    host="https://cloud.langfuse.com"
)

# Use it in your graph invocation
result = app.invoke(
    {"messages": [HumanMessage(content="Summarize the French Revolution.")]},
    config={
        "callbacks": [langfuse_handler],
        "configurable": {"thread_id": "thread-5"}
    }
)

Adding Custom Metadata to Traces

To make traces more useful in production, attach metadata that helps you filter and search. This is especially valuable when you have many users or different graph configurations:

result = app.invoke(
    {"messages": [HumanMessage(content="What are the symptoms of flu?")]},
    config={
        "configurable": {"thread_id": "user-123-thread-1"},
        "metadata": {
            "user_id": "user-123",
            "session_id": "session-abc",
            "environment": "production",
            "graph_version": "v2.1.0",
            "feature_flags": {"extended_reasoning": True}
        },
        "tags": ["health-query", "production"]
    }
)

In LangSmith, you can then filter traces by these metadata fields, tags, or user IDs to find specific runs or analyze patterns across a subset of traffic.

Building a Custom Logging Wrapper for Nodes

For maximum control, you can wrap your node functions with custom logging logic. This approach works well when you want structured logs alongside traces:

import functools
import logging
import json
from datetime import datetime

logger = logging.getLogger("langgraph_nodes")
logger.setLevel(logging.DEBUG)

def trace_node(func):
    """Decorator that adds logging and tracing to graph nodes."""
    @functools.wraps(func)
    def wrapper(state):
        node_name = func.__name__
        start = datetime.now()
        logger.info(f"Entering node '{node_name}'")

        # Log incoming state (truncate large values)
        try:
            state_summary = {
                k: str(v)[:200] if not isinstance(v, list) else f"[{len(v)} items]"
                for k, v in state.items()
            }
            logger.debug(f"Node '{node_name}' input state: {json.dumps(state_summary, default=str)}")
        except Exception as e:
            logger.warning(f"Could not serialize state for node '{node_name}': {e}")

        try:
            result = func(state)
            duration = (datetime.now() - start).total_seconds()
            logger.info(f"Node '{node_name}' completed in {duration:.3f}s")

            # Log output summary
            if isinstance(result, dict):
                output_keys = list(result.keys())
                logger.debug(f"Node '{node_name}' output keys: {output_keys}")

            return result
        except Exception as e:
            duration = (datetime.now() - start).total_seconds()
            logger.error(
                f"Node '{node_name}' failed after {duration:.3f}s: {e}",
                exc_info=True
            )
            raise

    return wrapper

# Apply the decorator to your nodes
@trace_node
def call_model(state: AgentState) -> AgentState:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

@trace_node
def call_tools(state: AgentState) -> AgentState:
    # Tool execution logic here
    return {"messages": []}

Best Practices for LangGraph Observability

• Always enable tracing in development: Set up LangSmith or another tracing backend from day one. Waiting until production makes debugging exponentially harder.
• Use meaningful node names: When you call workflow.add_node("retrieve_documents", retrieve_node), the name you choose appears in traces. Descriptive names make traces readable.
• Attach metadata consistently: Standardize on metadata fields like user_id, session_id, and environment across all invocations so you can query traces effectively.
• Use thread IDs for conversation tracking: Always pass a unique thread_id in your config. This groups related invocations and enables state inspection across turns.
• Monitor token usage proactively: Set up alerts on token consumption. Agentic loops can spiral, and without monitoring, costs can escalate quickly.
• Log tool inputs and outputs: Tool calls are often where things go wrong. Make sure your tracing captures what was sent to each tool and what came back.
• Sample in production: If you have high traffic, consider tracing a sample of requests rather than every single one to manage costs and storage.
• Version your graphs: Include a version identifier in your metadata so you can compare performance across graph versions and detect regressions.
• Combine traces with evaluation: Use traced runs as datasets for evaluation. LangSmith lets you convert traces into test cases, closing the loop between observability and quality improvement.
• Handle PII carefully: Traces may contain sensitive user data. Configure redaction or avoid tracing certain fields in compliance with your data policies.

Putting It All Together: A Production-Ready Setup

Here is a complete example that combines checkpointing, LangSmith tracing, custom callbacks, and metadata into a production-ready configuration:

import os
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from typing import TypedDict, Annotated
import operator
import logging

# Configure tracing
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls__your_api_key"
os.environ["LANGCHAIN_PROJECT"] = "production-agent"

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("agent")

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    iteration_count: int

llm = ChatOpenAI(model="gpt-4o", temperature=0)

def agent_node(state: AgentState) -> AgentState:
    logger.info(f"Agent iteration: {state.get('iteration_count', 0)}")
    system_prompt = SystemMessage(content=(
        "You are a helpful assistant. Answer concisely. "
        "If you need to use a tool, do so. Otherwise, provide a direct answer."
    ))
    messages = [system_prompt] + state["messages"]
    response = llm.invoke(messages)
    return {
        "messages": [response],
        "iteration_count": state.get("iteration_count", 0) + 1
    }

def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        if state.get("iteration_count", 0) >= 10:
            logger.warning("Max iterations reached, stopping.")
            return END
        return "tools"
    return END

def tools_node(state: AgentState) -> AgentState:
    # Placeholder for actual tool execution
    return {"iteration_count": state.get("iteration_count", 0) + 1}

# Build and compile with checkpointing
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tools_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("tools", "agent")

checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)

# Production invocation with full observability
def run_agent(user_input: str, user_id: str, session_id: str):
    config = {
        "configurable": {"thread_id": f"{user_id}-{session_id}"},
        "metadata": {
            "user_id": user_id,
            "session_id": session_id,
            "environment": os.getenv("ENV", "development"),
            "graph_version": "1.0.0",
        },
        "tags": ["user-facing", "conversational"]
    }

    result = app.invoke(
        {"messages": [HumanMessage(content=user_input)], "iteration_count": 0},
        config=config
    )

    return result["messages"][-1].content

# Usage
response = run_agent(
    user_input="What is the capital of France?",
    user_id="user-456",
    session_id="session-001"
)
print(f"Response: {response}")

Observability is not an afterthought—it is a foundational part of building reliable agentic systems. By combining LangGraph's built-in checkpointing and state inspection with external tracing platforms like LangSmith, custom callbacks, and structured logging, you gain complete visibility into how your agents behave. This visibility is what separates prototypes from production systems. Start with basic tracing enabled from your first graph, layer in custom instrumentation as complexity grows, and use the insights from traces to continuously improve your agent's reasoning, tool usage, and performance. With the practices and patterns covered in this guide, you are well-equipped to build LangGraph applications that are not only powerful but also transparent, debuggable, and production-ready.