🧠 LangGraph & LLM Observability — Master Guide

import os
from langchain_core.tracers import LangChainTracer

# Method 1: env vars (zero code change)
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls__your_key"
os.environ["LANGCHAIN_PROJECT"] = "my-agent-prod"

# Method 2: explicit tracer with metadata
tracer = LangChainTracer(
    project_name="my-agent-prod",
    tags=["production", "v2"],
    metadata={
        "team": "platform",
        "version": "2.1.0"
    }
)
config = {"callbacks": [tracer]}
result = graph.invoke(input, config=config)

from langfuse import Langfuse
from langfuse.callback import CallbackHandler

# Init client
langfuse = Langfuse(
    public_key="pk-lf-...",
    secret_key="sk-lf-...",
    host="https://cloud.langfuse.com"  # or self-hosted
)

# Create handler with session/user context
handler = CallbackHandler(
    session_id="session_abc",   # group runs by session
    user_id="user_123",         # attribute cost to user
    trace_name="research-agent",
    tags=["prod", "v1"],
    metadata={"feature": "doc_summarizer"}
)

config = {"callbacks": [handler]}
result = graph.invoke(input, config=config)

# Flush before process exits (important!)
langfuse.flush()

from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.graph import StateGraph, END
import psycopg

DB_URI = "postgresql://user:pass@localhost/mydb"

# Setup checkpointer
with psycopg.connect(DB_URI) as conn:
    checkpointer = PostgresSaver(conn)
    checkpointer.setup()  # creates tables on first run

# Compile with checkpointer
graph = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["approval_node"]  # HITL breakpoint
)

# Each run needs a unique thread_id
config = {"configurable": {"thread_id": "user_abc_run_001"}}

# Run — state auto-saved after each node
result = graph.invoke({"messages": [...]}, config=config)

# Resume later (after human approval)
graph.invoke(None, config=config)  # None = resume from checkpoint

# Time travel: inspect history
history = list(graph.get_state_history(config))
for state in history:
    print(f"Step {state.metadata['step']}: {state.next}")

import time
from dataclasses import dataclass, field
from typing import Dict, List

@dataclass
class TelemetrySpan:
    name: str
    start: float = field(default_factory=time.time)
    prompt_tokens: int = 0
    completion_tokens: int = 0
    latency_ms: float = 0
    status: str = "running"

class AnomalyDetector:
    BASELINES = {
        "latency_ms": 2000,    # alert if > 2x this
        "prompt_tokens": 8000, # alert if > this
        "cost_usd": 0.01       # alert if > this per run
    }
    COST_PER_1K = {"input": 0.003, "output": 0.015}  # adjust per model

    def __init__(self):
        self.spans: List[TelemetrySpan] = []
        self.alerts: List[str] = []

    def start_span(self, name: str) -> TelemetrySpan:
        span = TelemetrySpan(name=name)
        self.spans.append(span)
        return span

    def end_span(self, span: TelemetrySpan, prompt_tok: int, comp_tok: int):
        span.latency_ms = (time.time() - span.start) * 1000
        span.prompt_tokens = prompt_tok
        span.completion_tokens = comp_tok
        span.status = "done"
        self._check_anomalies(span)

    def _check_anomalies(self, span: TelemetrySpan):
        if span.latency_ms > self.BASELINES["latency_ms"] * 2:
            self.alerts.append(
                f"🚨 [{span.name}] Latency spike: {span.latency_ms:.0f}ms"
            )
        if span.prompt_tokens > self.BASELINES["prompt_tokens"]:
            self.alerts.append(
                f"⚠️ [{span.name}] Large context: {span.prompt_tokens} tokens"
            )
        cost = (span.prompt_tokens / 1000 * self.COST_PER_1K["input"] +
                span.completion_tokens / 1000 * self.COST_PER_1K["output"])
        if cost > self.BASELINES["cost_usd"]:
            self.alerts.append(f"💸 [{span.name}] High cost: ${cost:.4f}")

    def report(self) -> Dict:
        total_cost = sum(
            (s.prompt_tokens/1000 * self.COST_PER_1K["input"] +
             s.completion_tokens/1000 * self.COST_PER_1K["output"])
            for s in self.spans
        )
        return {
            "spans": len(self.spans),
            "total_latency_ms": sum(s.latency_ms for s in self.spans),
            "total_tokens": sum(s.prompt_tokens + s.completion_tokens
                                for s in self.spans),
            "total_cost_usd": round(total_cost, 6),
            "alerts": self.alerts
        }

import json
from anthropic import Anthropic

client = Anthropic()

JUDGE_SYSTEM = """You evaluate AI answers for hallucinations.
Respond ONLY in JSON with keys:
faithfulness (0-10), factual_accuracy (0-10),
hallucination_risk (0-10), verdict (PASS/WARN/FAIL),
issues (list), reasoning (string)"""

def hallucination_check(reference: str, answer: str) -> dict:
    """
    Run LLM-as-judge evaluation.
    Use this in your LangSmith/Langfuse eval pipeline.
    """
    resp = client.messages.create(
        model="claude-haiku-4-5",  # cheap model for evals!
        max_tokens=500,
        system=JUDGE_SYSTEM,
        messages=[{
            "role": "user",
            "content": f"REFERENCE:\n{reference}\n\nANSWER TO EVALUATE:\n{answer}"
        }]
    )
    raw = resp.content[0].text
    try:
        return json.loads(raw)
    except:
        return {"verdict": "ERROR", "reasoning": raw}

# Hook into LangSmith as a run evaluator:
def langsmith_evaluator(run, example) -> dict:
    result = hallucination_check(
        reference=example.outputs["expected"],
        answer=run.outputs["output"]
    )
    return {
        "key": "hallucination_score",
        "score": result.get("faithfulness", 0) / 10,
        "comment": result.get("reasoning", "")
    }

from langgraph.graph import StateGraph, END
from langgraph.types import interrupt
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    draft: str
    approved: bool
    human_notes: str

def draft_node(state: AgentState) -> AgentState:
    """Node 1: Generate draft content"""
    # ... call LLM to generate draft
    return {"draft": "Generated draft here..."}

def review_node(state: AgentState) -> AgentState:
    """Node 2: HITL interrupt — waits for human"""
    # This pauses execution and saves checkpoint
    human_decision = interrupt({
        "draft": state["draft"],
        "question": "Please review this draft. Approve, edit, or reject."
    })
    return {
        "approved": human_decision.get("approved", False),
        "human_notes": human_decision.get("notes", ""),
        "draft": human_decision.get("edited_draft", state["draft"])
    }

def finalize_node(state: AgentState) -> AgentState:
    """Node 3: Only runs after human approval"""
    if not state["approved"]:
        return {"messages": [{"role": "assistant",
                              "content": "Draft rejected. Starting over."}]}
    # ... finalize with human notes incorporated
    return {"messages": [{"role": "assistant",
                          "content": f"Finalized: {state['draft']}"}]}

# Build graph
workflow = StateGraph(AgentState)
workflow.add_node("draft", draft_node)
workflow.add_node("review", review_node)
workflow.add_node("finalize", finalize_node)
workflow.set_entry_point("draft")
workflow.add_edge("draft", "review")
workflow.add_edge("review", "finalize")
workflow.add_edge("finalize", END)

graph = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["review"]  # pause BEFORE review node
)

# Run 1: executes until interrupt
config = {"configurable": {"thread_id": "run_001"}}
graph.invoke({"messages": []}, config)

# Later: resume with human input
graph.invoke(
    {"approved": True, "human_notes": "Looks good, just shorter please"},
    config
)

from langfuse import Langfuse
from functools import wraps
import time

langfuse = Langfuse()

# Cost rates per model (update as pricing changes)
MODEL_COSTS = {
    "claude-haiku-4-5":   {"in": 0.0008, "out": 0.004},
    "claude-sonnet-4-6":  {"in": 0.003,  "out": 0.015},
    "claude-opus-4-6":    {"in": 0.015,  "out": 0.075},
}

BUDGET_ALERTS = {
    "per_run_usd": 0.05,      # alert if single run > $0.05
    "per_user_daily_usd": 1.00, # alert if user > $1/day
}

def traced_llm_call(model: str, feature: str, user_id: str):
    """Decorator to auto-trace any LLM call with cost tracking"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            trace = langfuse.trace(
                name=func.__name__,
                metadata={
                    "feature": feature,
                    "user_id": user_id,
                    "model": model
                }
            )
            span = trace.span(name="llm_call", start_time=time.time())
            try:
                result = func(*args, **kwargs)
                # Extract usage from response
                usage = result.usage
                cost = (
                    usage.input_tokens * MODEL_COSTS[model]["in"] / 1000 +
                    usage.output_tokens * MODEL_COSTS[model]["out"] / 1000
                )
                span.end(
                    output=result.content[0].text,
                    usage={
                        "input": usage.input_tokens,
                        "output": usage.output_tokens,
                        "unit": "TOKENS"
                    },
                    metadata={"cost_usd": cost}
                )
                if cost > BUDGET_ALERTS["per_run_usd"]:
                    trace.event(
                        name="budget_alert",
                        level="WARNING",
                        metadata={"cost": cost, "threshold": BUDGET_ALERTS["per_run_usd"]}
                    )
                return result
            except Exception as e:
                span.end(level="ERROR", status_message=str(e))
                raise
        return wrapper
    return decorator

import json
from datetime import datetime

def stream_with_telemetry(graph, input_data, config):
    """
    Stream a LangGraph run with live telemetry logging.
    Shows each node as it executes — great for debugging.
    """
    run_log = []
    start = datetime.now()

    print(f"\n{'='*50}")
    print(f"🚀 Run started: {start.strftime('%H:%M:%S.%f')[:-3]}")
    print(f"{'='*50}")

    for chunk in graph.stream(input_data, config, stream_mode="all"):
        ts = (datetime.now() - start).total_seconds()

        if isinstance(chunk, tuple):
            event_type, data = chunk

            if event_type == "updates":
                for node, updates in data.items():
                    print(f"\n[{ts:.2f}s] 📦 NODE: {node}")
                    # Log state changes
                    for key, val in updates.items():
                        if key == "messages":
                            print(f"  messages: +{len(val)} new message(s)")
                        else:
                            val_str = str(val)[:80] + "..." if len(str(val)) > 80 else val
                            print(f"  {key}: {val_str}")
                    run_log.append({
                        "ts": ts, "node": node, "updates": list(updates.keys())
                    })

            elif event_type == "debug":
                if data.get("type") == "checkpoint":
                    print(f"[{ts:.2f}s] 💾 CHECKPOINT saved (step {data.get('step', '?')})")

    print(f"\n{'='*50}")
    print(f"✅ Run complete in {ts:.2f}s | {len(run_log)} nodes executed")
    print(json.dumps({"timeline": run_log}, indent=2))
    return run_log

# Usage:
# run_log = stream_with_telemetry(graph, {"messages": [...]}, config)