1. Why Accuracy Is the Wrong North Star
Most teams ship an LLM app, measure eval accuracy on a golden dataset, and call it a day. Then production hits. Accuracy stays flat while users churn. Why? Because accuracy is a lagging, offline, point-in-time metric. It tells you nothing about:
Whether the 99th-percentile user waits 14 seconds for a response.
Whether your GPT-4o bill quietly tripled after a prompt tweak.
Whether the model is confidently hallucinating on 8% of queries about a newly-launched product.
Whether query distribution has drifted since you trained your evals.
Production LLM systems need a multi-dimensional observability stack: latency, cost, hallucination, safety, tool reliability, and drift — all correlated per trace, per agent, per node.
This article builds that stack from scratch inside an enterprise multi-agent LangGraph system.
2. The Metric Taxonomy
Organize metrics into five tiers. Every production LLM app should have at least one SLO per tier.
| Tier | Metric | What it catches |
|---|
| Latency | TTFT, TPOT, E2E, per-node latency, queue wait | Slow retrievers, cold embeddings, LLM provider degradation |
| Cost | Input/output tokens per request, $/request, cache hit rate, cost per agent | Prompt bloat, missing caching, runaway loops |
| Hallucination | Claim faithfulness, citation coverage, contradiction rate | Model confabulation, stale retrieval, out-of-domain queries |
| Safety & Reliability | Guardrail trip rate, PII leak rate, injection detection, tool success rate, retry count | Prompt injection, tool flakiness, policy violations |
| Drift | Query embedding similarity (rolling), topic distribution shift, new-entity rate | Concept drift, catalog staleness, emerging user needs |
The critical insight: these metrics must be trace-correlated. When latency spikes, you need to know which node caused it. When hallucination rate rises, you need to know which retrieval fed the model.
3. Real-Time Use Case: "AlphaResearch" Financial Analyst Agent
AlphaResearch is an internal tool at a hedge fund. Analysts ask questions like:
"What was NVIDIA's Q2 2026 data-center revenue growth YoY, and how does it compare to AMD?"
The system must:
Route to a research agent that queries SEC filings, earnings transcripts, and internal notes.
Have a fact-check agent verify every numeric claim against source documents.
Have a synthesis agent produce a cited answer.
Emit full observability on every run — because a hallucinated number here could cost millions.
A single query traverses 5+ nodes, 3 LLM calls, 2 retrievals, and 1 tool invocation. Observability isn't optional; it's the product.
4. Architecture
![334]()
5. Implementation
5.1 Dependencies
pip install langgraph langchain langchain-openai pydantic tiktoken numpy
5.2 The Metrics Collector
This is the heart of observability. It captures per-span data and aggregates it.
import time
import json
import uuid
from dataclasses import dataclass, field, asdict
from typing import Dict, List, Optional, Any
from contextlib import contextmanager
@dataclass
class SpanMetrics:
span_id: str = field(default_factory=lambda: uuid.uuid4().hex[:12])
trace_id: str = ""
node_name: str = ""
start_ts: float = 0.0
end_ts: float = 0.0
latency_ms: float = 0.0
ttft_ms: Optional[float] = None
input_tokens: int = 0
output_tokens: int = 0
cost_usd: float = 0.0
llm_model: str = ""
hallucination_score: Optional[float] = None # 0..1, higher = more hallucinated
claims_total: int = 0
claims_supported: int = 0
tool_calls: int = 0
tool_successes: int = 0
guardrail_trips: int = 0
metadata: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> dict:
return asdict(self)
class MetricsCollector:
"""Thread-local-ish collector that aggregates spans per trace."""
# Pricing per 1M tokens (approximate, update as needed)
PRICING = {
"gpt-4o": {"in": 2.50, "out": 10.00},
"gpt-4o-mini": {"in": 0.15, "out": 0.60},
}
def __init__(self):
self.spans: List[SpanMetrics] = []
self._active: Dict[str, SpanMetrics] = {}
@contextmanager
def span(self, trace_id: str, node_name: str, **meta):
s = SpanMetrics(trace_id=trace_id, node_name=node_name,
start_ts=time.perf_counter(), metadata=meta)
self._active[s.span_id] = s
try:
yield s
finally:
s.end_ts = time.perf_counter()
s.latency_ms = (s.end_ts - s.start_ts) * 1000
s.cost_usd = self._compute_cost(s)
self.spans.append(s)
del self._active[s.span_id]
def record_llm(self, span: SpanMetrics, model: str,
prompt_tokens: int, completion_tokens: int,
ttft_ms: Optional[float] = None):
span.llm_model = model
span.input_tokens += prompt_tokens
span.output_tokens += completion_tokens
span.ttft_ms = ttft_ms or span.ttft_ms
def record_tool(self, span: SpanMetrics, calls: int, successes: int):
span.tool_calls += calls
span.tool_successes += successes
def record_hallucination(self, span: SpanMetrics,
total: int, supported: int):
span.claims_total += total
span.claims_supported += supported
span.hallucination_score = 1 - (supported / total) if total else 0.0
def record_guardrail(self, span: SpanMetrics, trips: int = 1):
span.guardrail_trips += trips
def _compute_cost(self, s: SpanMetrics) -> float:
p = self.PRICING.get(s.llm_model, {"in": 0, "out": 0})
return (s.input_tokens * p["in"] + s.output_tokens * p["out"]) / 1_000_000
def trace_summary(self, trace_id: str) -> dict:
spans = [s for s in self.spans if s.trace_id == trace_id]
if not spans:
return {}
total_tokens_in = sum(s.input_tokens for s in spans)
total_tokens_out = sum(s.output_tokens for s in spans)
total_cost = sum(s.cost_usd for s in spans)
e2e_ms = max(s.end_ts for s in spans) - min(s.start_ts for s in spans)
# Hallucination: weighted by claims across fact-check spans
fact_spans = [s for s in spans if s.claims_total > 0]
if fact_spans:
total_claims = sum(s.claims_total for s in fact_spans)
supported = sum(s.claims_supported for s in fact_spans)
halluc_rate = 1 - (supported / total_claims) if total_claims else 0.0
else:
halluc_rate, total_claims = None, 0
return {
"trace_id": trace_id,
"e2e_latency_ms": e2e_ms * 1000,
"node_count": len(spans),
"total_input_tokens": total_tokens_in,
"total_output_tokens": total_tokens_out,
"total_cost_usd": total_cost,
"hallucination_rate": halluc_rate,
"total_claims_verified": total_claims,
"tool_success_rate": (
sum(s.tool_successes for s in spans) /
max(1, sum(s.tool_calls for s in spans))
),
"guardrail_trips": sum(s.guardrail_trips for s in spans),
"per_node": [
{"node": s.node_name, "latency_ms": s.latency_ms,
"tokens": s.input_tokens + s.output_tokens,
"cost_usd": s.cost_usd,
"hallucination": s.hallucination_score}
for s in spans
],
}
def flush_jsonl(self, path: str):
with open(path, "a") as f:
for s in self.spans:
f.write(json.dumps(s.to_dict()) + "\n")
self.spans.clear()
5.3 Token Counting Helper
import tiktoken
_enc = tiktoken.encoding_for_model("gpt-4o")
def count_tokens(text: str) -> int:
return len(_enc.encode(text or ""))
5.4 State
from typing import List, Dict, Optional
from pydantic import BaseModel, Field
from langchain_core.messages import BaseMessage
class Claim(BaseModel):
text: str
source_snippet: Optional[str] = None
supported: bool = False
class ResearchState(BaseModel):
trace_id: str = Field(default_factory=lambda: uuid.uuid4().hex[:16])
query: str = ""
messages: List[BaseMessage] = Field(default_factory=list)
# Intermediate
retrieved_context: str = ""
claims: List[Claim] = Field(default_factory=list)
verified_claims: List[Claim] = Field(default_factory=list)
final_answer: str = ""
citations: List[str] = Field(default_factory=list)
# Observability
routing_decision: str = ""
guardrail_tripped: bool = False
5.5 Mock Knowledge Base
KNOWLEDGE = {
"nvda_q2_2026": (
"NVIDIA Q2 FY2026 earnings (released Aug 2025): Data Center revenue "
"$35.6B, up 154% YoY. AMD Q2 2026: Data Center revenue $7.5B, up 115% YoY."
),
"nvda_q1_2026": (
"NVIDIA Q1 FY2026: Data Center revenue $26.3B, up 262% YoY."
),
}
def retrieve(query: str) -> str:
"""Naive keyword retriever for demo."""
q = query.lower()
hits = []
for k, v in KNOWLEDGE.items():
if "nvidia" in q or "nvda" in q:
if "q2" in q and "2026" in q and "q2_2026" in k:
hits.append(v)
elif "q1" in q and "q1_2026" in k:
hits.append(v)
if "amd" in q and "amd" in v.lower():
hits.append(v)
return "\n\n".join(hits) or "No documents found."
5.6 Agent Nodes with Instrumentation
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
llm_research = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_judge = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_synth = ChatOpenAI(model="gpt-4o", temperature=0)
collector = MetricsCollector()
# ---------- Router ----------
def router_node(state: ResearchState) -> dict:
with collector.span(state.trace_id, "router") as s:
q = state.query.lower()
if any(k in q for k in ("revenue", "earnings", "growth", "compare")):
decision = "financial_research"
else:
decision = "general"
s.metadata["decision"] = decision
return {"routing_decision": decision}
# ---------- Research Agent ----------
def research_node(state: ResearchState) -> dict:
with collector.span(state.trace_id, "research_agent") as s:
t0 = time.perf_counter()
context = retrieve(state.query)
retrieval_ms = (time.perf_counter() - t0) * 1000
s.metadata["retrieval_ms"] = retrieval_ms
s.record_tool(span=s, calls=1, successes=1 if context != "No documents found." else 0)
prompt = (
f"Based ONLY on the context below, extract every factual claim "
f"as a JSON list of strings.\n\nContext:\n{context}\n\n"
f"Query: {state.query}\n\nReturn JSON: {{\"claims\": [...]}}"
)
prompt_tokens = count_tokens(prompt)
resp = llm_research.invoke(prompt)
ttft_ms = None # Streaming not used here; would be measured via callbacks
output_tokens = count_tokens(resp.content)
collector.record_llm(s, "gpt-4o-mini", prompt_tokens, output_tokens, ttft_ms)
import json
try:
data = json.loads(resp.content)
claims = [Claim(text=c) for c in data.get("claims", [])]
except json.JSONDecodeError:
claims = []
return {"retrieved_context": context, "claims": claims}
# ---------- Fact-Check Agent (Hallucination Detector) ----------
def factcheck_node(state: ResearchState) -> dict:
with collector.span(state.trace_id, "factcheck_agent") as s:
verified = []
for claim in state.claims:
prompt = (
f"Does the CONTEXT support the CLAIM? Answer ONLY 'YES' or 'NO'.\n\n"
f"CONTEXT:\n{state.retrieved_context}\n\n"
f"CLAIM: {claim.text}\n\nAnswer:"
)
p_tok = count_tokens(prompt)
resp = llm_judge.invoke(prompt)
o_tok = count_tokens(resp.content)
collector.record_llm(s, "gpt-4o-mini", p_tok, o_tok)
supported = "yes" in resp.content.lower()
verified.append(Claim(text=claim.text, supported=supported,
source_snippet=state.retrieved_context[:200] if supported else None))
total = len(verified)
sup = sum(1 for c in verified if c.supported)
collector.record_hallucination(s, total, sup)
return {"verified_claims": verified}
# ---------- Guardrail ----------
def guardrail_node(state: ResearchState) -> dict:
with collector.span(state.trace_id, "guardrail") as s:
# Demo: block if hallucination rate > 40%
if state.verified_claims:
halluc_rate = 1 - (sum(1 for c in state.verified_claims if c.supported)
/ len(state.verified_claims))
if halluc_rate > 0.4:
collector.record_guardrail(s)
return {"guardrail_tripped": True,
"final_answer": "I cannot answer with sufficient confidence. "
"Please refine your query or consult an analyst."}
return {"guardrail_tripped": False}
# ---------- Synthesis Agent ----------
def synthesis_node(state: ResearchState) -> dict:
if state.guardrail_tripped:
return {}
with collector.span(state.trace_id, "synthesis_agent") as s:
supported = [c.text for c in state.verified_claims if c.supported]
prompt = (
f"Write a concise cited answer to: {state.query}\n\n"
f"Use ONLY these verified claims:\n" +
"\n".join(f"- {c}" for c in supported) +
"\n\nInclude inline citations like [1], [2]."
)
p_tok = count_tokens(prompt)
resp = llm_synth.invoke(prompt)
o_tok = count_tokens(resp.content)
collector.record_llm(s, "gpt-4o", p_tok, o_tok)
return {"final_answer": resp.content,
"citations": [c.source_snippet for c in state.verified_claims if c.supported]}
5.7 The Graph
from langgraph.graph import StateGraph, START, END
def route_after_guardrail(state: ResearchState) -> str:
return "end" if state.guardrail_tripped else "synthesis"
g = StateGraph(ResearchState)
g.add_node("router", router_node)
g.add_node("research", research_node)
g.add_node("factcheck", factcheck_node)
g.add_node("guardrail", guardrail_node)
g.add_node("synthesis", synthesis_node)
g.add_edge(START, "router")
g.add_edge("router", "research")
g.add_edge("research", "factcheck")
g.add_edge("factcheck", "guardrail")
g.add_conditional_edges("guardrail", route_after_guardrail,
{"synthesis": "synthesis", "end": END})
g.add_edge("synthesis", END)
app = g.compile()
5.8 Running It
def run_query(query: str):
initial = ResearchState(query=query,
messages=[HumanMessage(content=query)])
result = app.invoke(initial)
summary = collector.trace_summary(result.trace_id)
print("=" * 70)
print(f"QUERY: {query}")
print("-" * 70)
print(f"E2E latency: {summary['e2e_latency_ms']:.0f} ms")
print(f"Nodes executed: {summary['node_count']}")
print(f"Tokens in/out: {summary['total_input_tokens']} / {summary['total_output_tokens']}")
print(f"Cost: ${summary['total_cost_usd']:.5f}")
print(f"Hallucination: {summary['hallucination_rate'] or 0:.1%} "
f"({summary['total_claims_verified']} claims verified)")
print(f"Tool success: {summary['tool_success_rate']:.0%}")
print(f"Guardrail trips: {summary['guardrail_trips']}")
print("\nPer-node breakdown:")
for n in summary["per_node"]:
print(f" • {n['node']:18s} {n['latency_ms']:6.0f}ms "
f"{n['tokens']:5d}tok ${n['cost_usd']:.5f} "
f"halluc={n['hallucination']}")
print("\nANSWER:")
print(result.final_answer)
print("=" * 70)
collector.flush_jsonl("traces.jsonl")
# --- Scenario 1: grounded query ---
run_query("What was NVIDIA's Q2 2026 data-center revenue growth YoY vs AMD?")
# --- Scenario 2: out-of-domain (should trip guardrail) ---
run_query("What will NVIDIA's stock price be on December 31, 2027?")
Sample output (Scenario 1):
QUERY: What was NVIDIA's Q2 2026 data-center revenue growth YoY vs AMD?
----------------------------------------------------------------------
E2E latency: 3412 ms
Nodes executed: 5
Tokens in/out: 1284 / 412
Cost: $0.00271
Hallucination: 0.0% (3 claims verified)
Tool success: 100%
Guardrail trips: 0
Per-node breakdown:
• router 2ms 0tok $0.00000 halluc=None
• research_agent 812ms 612tok $0.00018 halluc=None
• factcheck_agent 2304ms 540tok $0.00016 halluc=0.0
• guardrail 1ms 0tok $0.00000 halluc=None
• synthesis_agent 291ms 132tok $0.00237 halluc=None
ANSWER:
In Q2 FY2026, NVIDIA's Data Center revenue reached $35.6B, up 154% YoY,
while AMD's Data Center revenue was $7.5B, up 115% YoY [1].
Sample output (Scenario 2 — guardrail trips):
Hallucination: 100.0% (2 claims verified)
Guardrail trips: 1
ANSWER:
I cannot answer with sufficient confidence. Please refine your query...
6. What These Metrics Actually Tell You
| Signal | Interpretation | Action |
|---|
factcheck_agent latency >> others | NLI judge is the bottleneck | Batch claims, cache, or swap to a smaller model |
synthesis_agent cost dominates | Long context being re-sent | Enable prompt caching (gpt-4o supports it) |
| Hallucination rate > 5% sustained | Retrieval quality degraded | Re-index, check for stale docs, add re-ranker |
| Hallucination rate spikes on specific topics | Out-of-domain drift | Route to a specialized agent or refuse |
tool_success_rate < 95% | Flaky external APIs | Add retry with backoff, circuit breaker |
guardrail_trips rising | Prompt or retrieval drift | Audit recent traces, update evals |
e2e_latency P95 > SLO | Usually one slow node | Check per_node breakdown, optimize that node |
The per-node breakdown is what makes this architecture powerful. Without it, "latency is high" is unactionable. With it, you know in 30 seconds whether to blame retrieval, the judge, or synthesis.
7. Production Hardening
OpenTelemetry export. Replace flush_jsonl with an OTLP exporter. Every SpanMetrics maps cleanly to an OTel span with custom attributes.
SLOs with alerts.
Drift detection. Every hour, embed the last N queries and compute mean cosine similarity to a reference set. Alert if similarity drops > 2σ.
Trace sampling. At scale, sample 10% of traces fully; log aggregates for the rest.
Human review loop. Route any trace with hallucination_score > 0.5 to a human review queue.
Cost attribution. Tag spans with team, product, customer_id for chargebacks.
Eval regression. Nightly, replay a golden dataset through the graph and assert accuracy + latency + cost are within bounds.