Tracking Metrics for Individual Nodes in LangGraph

When you build a LangGraph application, you're not just calling an LLM — you're orchestrating a state machine where each node (retriever, router, tool caller, summarizer, etc.) contributes differently to the final output. End-to-end metrics like "user liked the response" are useful, but they don't tell you which node failed.

This article walks through how to instrument every node in a LangGraph graph, using a real-world Customer Support Agent as the running example.

1. Why Node-Level Metrics Matter

Consider a customer support agent with this flow:

[User Query] → Triage → Retrieve Docs → Generate Reply → (Human Escalation?) → Response

If the final answer is wrong, was it because:

The Triage node misclassified the intent?
The Retriever returned irrelevant chunks?
The Generator hallucinated despite good context?
The Escalation node failed to flag a sensitive issue?

Without per-node metrics, debugging is guesswork. With them, you can pinpoint regressions, optimize bottlenecks, and run A/B tests on individual components.

2. The Use Case: Customer Support Agent

We'll build a LangGraph agent with four nodes:

Node	Purpose	Key Metrics
`triage`	Classify intent (billing / technical / general)	Classification accuracy, latency
`retrieve`	Fetch relevant docs from vector DB	Retrieval recall@k, context relevance
`generate`	Produce the reply	Faithfulness, latency, token cost
`escalate`	Decide if human handoff is needed	Escalation rate, false positive rate

3. Setting Up the Stack

pip install langgraph langsmith langchain-openai langchain-chroma \
            opentelemetry-api opentelemetry-sdk prometheus-client

We'll use three complementary layers:

LangSmith — native LangGraph tracing (zero-code visibility).
Custom decorators — domain-specific metrics per node.
Prometheus + Grafana — real-time dashboards and alerts.

4. The Metrics Collector

First, a lightweight collector that exposes metrics to Prometheus and logs to LangSmith:

# metrics.py
from prometheus_client import Counter, Histogram, Gauge, start_http_server
from langsmith import traceable
import time
import functools

# Prometheus metrics
NODE_LATENCY = Histogram(
    "langgraph_node_latency_seconds",
    "Latency of a LangGraph node",
    ["graph", "node"]
)
NODE_CALLS = Counter(
    "langgraph_node_calls_total",
    "Total calls to a LangGraph node",
    ["graph", "node", "status"]
)
RETRIEVAL_RECALL = Histogram(
    "langgraph_retrieval_recall",
    "Recall@k of retriever nodes",
    ["graph", "node"],
    buckets=[0.0, 0.2, 0.4, 0.6, 0.8, 1.0]
)
LLM_TOKENS = Counter(
    "langgraph_llm_tokens_total",
    "Tokens consumed by LLM nodes",
    ["graph", "node", "type"]  # type = prompt | completion
)

# Start Prometheus exporter on port 8000
start_http_server(8000)


def instrument_node(graph_name: str, node_name: str, node_type: str = "generic"):
    """Decorator that records latency, call count, and traces to LangSmith."""
    def decorator(fn):
        @traceable(name=f"{graph_name}.{node_name}", run_type=node_type)
        @functools.wraps(fn)
        def wrapper(state, config=None, **kwargs):
            start = time.perf_counter()
            status = "success"
            try:
                result = fn(state, config=config, **kwargs)
                # Attach custom metadata for LangSmith UI
                if config and "callbacks" in (config or {}):
                    pass  # LangSmith picks up @traceable automatically
                return result
            except Exception as e:
                status = "error"
                raise
            finally:
                elapsed = time.perf_counter() - start
                NODE_LATENCY.labels(graph=graph_name, node=node_name).observe(elapsed)
                NODE_CALLS.labels(graph=graph_name, node=node_name, status=status).inc()
        return wrapper
    return decorator

5. Building the Graph with Instrumented Nodes

# graph.py
from typing import TypedDict, Literal, Annotated
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_chroma import Chroma
from metrics import (
    instrument_node, RETRIEVAL_RECALL, LLM_TOKENS
)

class SupportState(TypedDict):
    query: str
    intent: str
    context: list[str]
    reply: str
    escalated: bool
    ground_truth_intent: str | None  # for eval

# ---------- Node 1: Triage ----------
@instrument_node("support", "triage", node_type="llm")
def triage(state: SupportState, config) -> dict:
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    prompt = (
        "Classify the customer query into one of: billing, technical, general.\n"
        f"Query: {state['query']}\nIntent:"
    )
    resp = llm.invoke(prompt)
    return {"intent": resp.content.strip().lower()}

# ---------- Node 2: Retrieve ----------
@instrument_node("support", "retrieve", node_type="retriever")
def retrieve(state: SupportState, config) -> dict:
    db = Chroma(collection_name="docs", persist_directory="./chroma")
    docs = db.similarity_search(state["query"], k=5)
    chunks = [d.page_content for d in docs]

    # Record recall@5 if we have ground-truth doc IDs in state
    if "relevant_doc_ids" in state:
        retrieved_ids = {d.metadata.get("id") for d in docs}
        recall = len(retrieved_ids & set(state["relevant_doc_ids"])) / \
                 max(len(state["relevant_doc_ids"]), 1)
        RETRIEVAL_RECALL.labels(graph="support", node="retrieve").observe(recall)

    return {"context": chunks}

# ---------- Node 3: Generate ----------
@instrument_node("support", "generate", node_type="llm")
def generate(state: SupportState, config) -> dict:
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    ctx = "\n---\n".join(state["context"])
    prompt = f"Context:\n{ctx}\n\nCustomer: {state['query']}\nReply:"
    resp = llm.invoke(prompt)

    # Track token usage
    usage = resp.response_metadata.get("token_usage", {})
    LLM_TOKENS.labels(graph="support", node="generate", type="prompt").inc(
        usage.get("prompt_tokens", 0)
    )
    LLM_TOKENS.labels(graph="support", node="generate", type="completion").inc(
        usage.get("completion_tokens", 0)
    )
    return {"reply": resp.content}

# ---------- Node 4: Escalate ----------
@instrument_node("support", "escalate", node_type="chain")
def escalate(state: SupportState, config) -> dict:
    sensitive = any(kw in state["query"].lower()
                    for kw in ["fraud", "lawyer", "refund denied", "complaint"])
    return {"escalated": sensitive}

# ---------- Router ----------
def route_after_triage(state) -> Literal["retrieve", "escalate"]:
    return "escalate" if state["intent"] == "billing" else "retrieve"

# ---------- Build graph ----------
builder = StateGraph(SupportState)
builder.add_node("triage", triage)
builder.add_node("retrieve", retrieve)
builder.add_node("generate", generate)
builder.add_node("escalate", escalate)

builder.add_edge(START, "triage")
builder.add_conditional_edges("triage", route_after_triage)
builder.add_edge("retrieve", "generate")
builder.add_edge("generate", END)
builder.add_edge("escalate", END)

graph = builder.compile()

6. Running It and Observing Metrics

# run.py
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "support-agent-prod"

result = graph.invoke({
    "query": "I was charged twice for my subscription",
    "ground_truth_intent": "billing"
})
print(result)

Now you have three observability surfaces:

A. LangSmith Trace View

Every node appears as a span with latency, inputs/outputs, and token counts. You can drill into the retrieve span and see exactly which chunks were fetched.

B. Prometheus Metrics (scrape `http://localhost:8000/metrics`)

# HELP langgraph_node_latency_seconds
langgraph_node_latency_seconds_bucket{graph="support",node="triage",le="0.5"} 12
langgraph_node_calls_total{graph="support",node="retrieve",status="success"} 847
langgraph_retrieval_recall_bucket{graph="support",node="retrieve",le="0.8"} 312
langgraph_llm_tokens_total{graph="support",node="generate",type="completion"} 48210

C. Grafana Dashboard

Build panels for:

p95 latency per node — spot slow retrievers before users complain.
Error rate per node — catch a failing tool early.
Retrieval recall trend — detect when a new embedding model degrades search.
Token cost per node — attribute spend to the right component.

7. Real-World Insights You'll Catch

After running this in production for a week, here's what the metrics revealed in a similar setup:

Signal	What it meant	Action
`triage` p95 latency jumped from 0.4s → 2.1s	OpenAI had a latency spike on `gpt-4o`	Added fallback to `gpt-4o-mini`
`retrieve` recall@5 dropped from 0.82 → 0.54	New doc ingestion used a different chunk size	Rolled back chunker
`generate` token cost doubled	Prompt accidentally included full history	Trimmed context window
`escalate` false positive rate = 38%	Keyword "refund" triggered on benign queries	Replaced with a classifier node

None of these would have been visible from an end-to-end "user satisfaction" metric alone.

8. Beyond Basics: Evaluation as a Metric

For nodes where you can define ground truth (triage accuracy, retrieval recall, faithfulness), run an offline eval nightly using LangSmith's evaluation harness and push scores into Prometheus as Gauges:

from prometheus_client import Gauge
TRIAGE_ACC = Gauge("langgraph_triage_accuracy", "Accuracy of triage node")

# In your eval job:
TRIAGE_ACC.set(compute_triage_accuracy(test_set))

This closes the loop: production metrics catch regressions, eval metrics measure quality, and traces explain why.

9. Checklist for Your Own Graph

Wrap every node with @instrument_node (or equivalent).
Emit latency, call count, and error rate for every node.
Add domain-specific metrics (recall, accuracy, faithfulness) where ground truth exists.
Track token usage and cost per LLM node.
Export to Prometheus; visualize in Grafana.
Enable LangSmith tracing for deep-dive debugging.
Set alerts on p95 latency and error rate per node.
Run nightly evals and publish scores as Gauges.

TL;DR

Treating a LangGraph agent as a black box hides the real sources of failure. By instrumenting each node with latency, error, domain-quality, and cost metrics — and shipping them to Prometheus + LangSmith — you turn an opaque orchestration into a measurable system you can actually improve.