LangGraph Observability: How to Monitor and Trace State Graphs in Production

LangGraph has become the standard framework for building complex, stateful multi-agent systems in Python and TypeScript. By modeling agents as cyclical directed graphs with persistent state, LangGraph enables sophisticated reasoning loops, human-in-the-loop approvals, and multi-agent collaboration.

However, as graphs grow in complexity, debugging and monitoring LangGraph in production becomes notoriously difficult:

• Why did an execution loop 14 times through a conditional routing edge instead of terminating?
• Which node in the graph caused a $3.50 token spike?
• Why did a human-in-the-loop checkpoint fail to resume state after an asynchronous webhook?

This guide walks through building production-grade LangGraph Observability using standard OpenTelemetry (OTel) distributed tracing.

The Architecture of a LangGraph Execution Trace

In LangGraph, execution flows through Nodes (which execute python functions or LLM chains), Edges (which direct flow conditionally based on state), and State Checkpoints.

An OpenTelemetry-native trace maps directly onto these graph primitives:

Root Span: langgraph.graph_execution (id: run_01j9a)
├── Span: langgraph.node.agent_planner (duration: 820ms, cost: $0.012)
│   └── Span: llm.openai.gpt-4o (tokens: 1450, ttft: 280ms)
│
├── Span: langgraph.edge.conditional_router (decision: "tools_branch")
│
├── Span: langgraph.node.tool_executor (duration: 340ms)
│   └── Span: tool.sql_database_query (query: "SELECT ...")
│
└── Span: langgraph.node.response_generator (duration: 950ms, cost: $0.008)
    └── Span: llm.anthropic.claude-3-7 (tokens: 1100, ttft: 310ms)

1. Auto-Instrumenting LangGraph with OpenTelemetry

You can instrument LangGraph workflows in two lines using Splyntra's SDK, or manually wrap your StateGraph nodes with standard OpenTelemetry tracers:

from typing import TypedDict, List, Annotated
import operator
from langgraph.graph import StateGraph, END
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode

tracer = trace.get_tracer("splyntra.langgraph", "1.0.0")

# 1. Define Graph State Schema
class AgentState(TypedDict):
    messages: Annotated[List[dict], operator.add]
    iteration_count: int
    next_step: str

# 2. Reusable Traced Node Decorator
def traced_langgraph_node(node_name: str):
    """Wraps a LangGraph node with an OpenTelemetry span containing state metadata."""
    def decorator(func):
        def wrapper(state: AgentState):
            with tracer.start_as_current_span(
                f"langgraph.node.{node_name}",
                attributes={
                    "gen_ai.system": "langgraph",
                    "langgraph.node.name": node_name,
                    "langgraph.state.iteration": state.get("iteration_count", 0),
                    "langgraph.state.keys": list(state.keys()),
                }
            ) as span:
                try:
                    new_state = func(state)
                    span.set_attribute("langgraph.node.next_step", new_state.get("next_step", "unknown"))
                    span.set_status(Status(StatusCode.OK))
                    return new_state
                except Exception as exc:
                    span.record_exception(exc)
                    span.set_status(Status(StatusCode.ERROR, str(exc)))
                    raise exc
        return wrapper
    return decorator

2. Tracing Nodes and Conditional Routing Edges

# 3. Define Nodes with Telemetry
@traced_langgraph_node("planner")
def planner_node(state: AgentState):
    # Model reasoning step
    count = state.get("iteration_count", 0) + 1
    if count >= 3:
        return {"iteration_count": count, "next_step": "end"}
    return {"iteration_count": count, "next_step": "tools"}

@traced_langgraph_node("tool_executor")
def tool_executor_node(state: AgentState):
    # Simulated Tool Execution
    return {"messages": [{"role": "tool", "content": "Database query success"}]}

# 4. Tracing Conditional Edge Logic
def route_decision(state: AgentState) -> str:
    with tracer.start_as_current_span("langgraph.edge.routing") as span:
        next_step = state.get("next_step", "end")
        span.set_attribute("langgraph.edge.decision", next_step)
        if next_step == "tools":
            return "tools"
        return "end"

# 5. Assemble Graph
workflow = StateGraph(AgentState)
workflow.add_node("planner", planner_node)
workflow.add_node("tools", tool_executor_node)

workflow.set_entry_point("planner")
workflow.add_conditional_edges(
    "planner",
    route_decision,
    {
        "tools": "tools",
        "end": END
    }
)
workflow.add_edge("tools", "planner")

app = workflow.compile()

3. Monitoring Human-in-the-Loop (HITL) Checkpoints

LangGraph supports pausing execution for human review via interrupt_before or interrupt_after. In traditional logging, paused executions appear as dropped requests.

With OpenTelemetry distributed tracing:

1. When execution pauses, the span emits an event: langgraph.checkpoint.suspended with the thread ID.
2. When the user approves in Splyntra, execution resumes under the same traceparent context, creating a single unified trace spanning multiple hours or days.

# Thread-based state execution with checkpoint tracing
config = {"configurable": {"thread_id": "thread_user_9912"}}

with tracer.start_as_current_span(
    "langgraph.workflow_run",
    attributes={"langgraph.thread_id": "thread_user_9912"}
) as root_span:
    for event in app.stream({"iteration_count": 0, "messages": []}, config=config):
        print(f"Node Executed: {event}")

4. Key LangGraph Metrics to Track in Splyntra

Next Steps & Related Technical Guides

• How to Trace AI Agents with OpenTelemetry — Context propagation deep-dive.
• AI Agent Monitoring: Metrics You Should Track — The 12 Golden Signals.
• AI Agent Security: A Practical Guide — Defense-in-depth for stateful agents.
• How to Track LLM Costs in AI Agents — Tracking per-node token spend.