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:
![30]()
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.