Module 1.3: Observability, LangFuse & Tracing Hierarchies

Curriculum Alignment: [docs/plan/01_phase1_engine_and_prompting.md](file:///Users/huychau/Documents/working/training/ai/docs/plan/01_phase1_engine_and_prompting.md)
Topic Scope: OpenTelemetry Mapping, Traces vs Spans vs Generations, Latency & Token Economics, LangFuse Instrumentation
Level: Senior Architect / Advanced AI Engineering

---

1. Systems Analogy: Why Classical APM Fails for LLMs

In microservices architectures, classical Application Performance Monitoring (APM) tools (Datadog, Prometheus, New Relic) measure:

• Throughput (RPS)
• Error Rate (HTTP 5xx status codes)
• Latency percentiles ($p50,p95,p99$)

In Generative AI systems, HTTP 200 OK does not mean success. A model can return an HTTP 200 in 1.2s while outputting a catastrophic hallucination, violating safety policies, or blowing the token budget on a recursive loop.

To debug and govern AI systems, we need Semantic Observability:

1. What was the exact prompt and system context at step $N$?
2. How many prompt tokens vs completion tokens were consumed?
3. What was the cost of this specific transaction?
4. What was the intermediate reasoning trajectory before the final answer?

---

2. The Tracing Hierarchy: Mapping OpenTelemetry to GenAI

Modern AI observability frameworks (like LangFuse and LangSmith) follow OpenTelemetry distributed tracing conventions:

Hierarchy Breakdown:

• Trace: The top-level root container representing the complete end-to-end user transaction or API request.
• Span: A nested execution block representing a distinct programmatic step (e.g. data fetching, embedding calculation, regex validation).
• Generation: The atomic unit representing the actual LLM API invocation. Captures model parameters (temperature, top_p), token usage (prompt, completion, total), and exact input/output payloads.
• Score / Metric: Quantitative feedback attached to a trace or generation (e.g., user_feedback=1.0, toxicity=0.01, latency_ok=True).

---

3. Token Economics & Profiling Metrics

When profiling AI microservices, track these key economic and operational metrics:

1. Time-to-First-Token (TTFT): The latency from request dispatch until the first streamed token arrives (reflects prompt processing and server queuing).
2. Inter-Token Latency (ITL): The generation time per subsequent token (reflects model parameter bandwidth and GPU decoding speed).
3. Prompt Cache Hit Rate: When sending repetitive system prompts, modern LLM APIs (Gemini, Claude) cache the prefix. Observability must track cached prompt tokens vs novel prompt tokens to monitor cost savings.

---

4. Production Python Instrumentation: LangFuse SDK

Here is a production-grade Python implementation instrumenting a classification pipeline using the official LangFuse SDK:

"""
LangFuse Tracing Pipeline Example
Demonstrates Parent-Child Span propagation, Generation tracking, and error scoring.
"""
import os
import time
from typing import Any
from langfuse import Langfuse
from langfuse.decorators import langfuse_context, observe
from pydantic import BaseModel

# Initialize client using environment variables
langfuse = Langfuse(
    public_key=os.environ.get("LANGFUSE_PUBLIC_KEY", "pk-dummy"),
    secret_key=os.environ.get("LANGFUSE_SECRET_KEY", "sk-dummy"),
    host=os.environ.get("LANGFUSE_HOST", "https://cloud.langfuse.com")
)

class InputPayload(BaseModel):
    user_id: str
    raw_query: str

@observe(name="normalize_input")
def sanitize_input(text: str) -> str:
    """Simulates text normalization and malicious pattern stripping."""
    time.sleep(0.02)  # Simulating processing
    return text.strip()

@observe(as_type="generation")
def call_mock_llm(rendered_prompt: str, temperature: float = 0.0) -> dict[str, Any]:
    """Simulates an atomic LLM generation span with metadata and token usage."""
    start_time = time.time()
    time.sleep(0.15)  # Simulate network latency

    # Mock output
    output_text = '{"intent": "BILLING_INQUIRY", "confidence": 0.98}'
    
    # Update generation metadata directly in LangFuse context
    langfuse_context.update_current_observation(
        input=rendered_prompt,
        output=output_text,
        model="gemini-2.5-flash",
        model_parameters={"temperature": temperature},
        usage={
            "input": 120,
            "output": 24,
            "total": 144
        }
    )
    return {"raw_response": output_text}

@observe(name="triage_pipeline")
def execute_triage_pipeline(payload: InputPayload) -> dict[str, Any]:
    """Root trace orchestrating nested spans and generations."""
    # Set user ID and session tags on the root trace
    langfuse_context.update_current_trace(
        user_id=payload.user_id,
        tags=["phase_1", "triage_service"],
        metadata={"client_version": "1.0.4"}
    )

    clean_text = sanitize_input(payload.raw_query)
    prompt = f"System: Classify input.\nInput: {clean_text}"
    
    llm_result = call_mock_llm(prompt, temperature=0.0)
    
    # Attach a custom evaluation score to the root trace
    langfuse_context.score_current_trace(
        name="schema_validity",
        value=1.0,
        comment="Successfully parsed valid JSON"
    )
    
    return {"status": "success", "result": llm_result}

if __name__ == "__main__":
    test_payload = InputPayload(user_id="user_492", raw_query="How do I change my subscription billing cycle?")
    print("Executing traced pipeline...")
    result = execute_triage_pipeline(test_payload)
    print("Result:", result)
    # Flush events to remote LangFuse server
    langfuse.flush()

---

5. Curated Reading & Canonical References

Resource	Canonical Reference & Link	Specific Focus Areas
Primary Observability Tool	LangFuse Tracing Documentation	Trace hierarchy, @observe() decorator mechanics, and generation token/cost attribution.
LangFuse Python SDK	LangFuse Python SDK Reference	Context managers, manual span manipulation, and score_current_trace().
Production Architecture Template	FastAPI LangGraph Production Template (GitHub)	Canonical folder structure, telemetry integration, and middleware lifecycle.
OpenTelemetry Standard	OTel GenAI Semantic Conventions	Industry-standard span attributes (gen_ai.usage.prompt_tokens, gen_ai.system).
LangSmith Tracing	LangSmith Conceptual Overview	Nested execution trees, parent-child run propagation, and latency profiling.

---

6. Hands-on Engineering Exercises

Exercise 1.3: Latency & Cost Breakdown Tracer

• Task: Build a Python module using @observe that wraps an LLM classification workflow.
• Requirements:
  • Track three distinct spans: input_sanitization, llm_inference, and pydantic_parsing.
  • Calculate and log the percentage of total trace duration consumed by llm_inference vs. deterministic Python code.