12.2 Observability, Logging & LLMOps¶

# trace_setup.py  —  one-time bootstrap for an LLM microservice
# Requirements: opentelemetry-sdk opentelemetry-exporter-otlp-proto-grpc
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

def setup_tracing(service_name: str, otlp_endpoint: str = "http://localhost:4317") -> trace.Tracer:
    """Configure an OTel TracerProvider that exports to any OTLP-compatible backend.

    Compatible backends include: Langfuse (via proxy), Jaeger, Grafana Tempo,
    Google Cloud Trace, Honeycomb, and Datadog.
    """
    provider = TracerProvider()
    exporter = OTLPSpanExporter(endpoint=otlp_endpoint, insecure=True)
    # BatchSpanProcessor buffers spans and sends them asynchronously
    # to avoid adding latency to the critical path.
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)
    return trace.get_tracer(service_name)


TRACER = setup_tracing("llm-chat-service")

# rag_pipeline.py  —  a fully-instrumented RAG request handler
import time
import hashlib
from opentelemetry import trace
from trace_setup import TRACER

def handle_request(
    user_id: str,
    query: str,
    retriever,
    llm_client,
    max_tokens: int = 512,
) -> dict:
    """
    Process one user query through a RAG pipeline, emitting OTel spans for
    every stage. The root span holds the full end-to-end latency.
    """
    # Root span for the entire request.
    with TRACER.start_as_current_span("rag_request") as root_span:
        # Attach stable, low-cardinality attributes for filtering in dashboards.
        root_span.set_attribute("user.id", user_id)
        root_span.set_attribute("query.length_chars", len(query))
        root_span.set_attribute("query.hash", hashlib.md5(query.encode()).hexdigest()[:8])

        # ── Stage 1: Retrieval ────────────────────────────────────────────────
        with TRACER.start_as_current_span("retrieval") as ret_span:
            t0 = time.perf_counter()
            docs = retriever.retrieve(query, top_k=5)
            ret_span.set_attribute("retrieval.latency_ms", (time.perf_counter() - t0) * 1000)
            ret_span.set_attribute("retrieval.num_docs", len(docs))
            ret_span.set_attribute("retrieval.top_score", docs[0]["score"] if docs else 0.0)

        # ── Stage 2: Prompt construction ─────────────────────────────────────
        with TRACER.start_as_current_span("prompt_build") as pb_span:
            context_text = "\n\n".join(d["text"] for d in docs)
            system_prompt = "You are a helpful assistant. Use the provided context."
            user_message = f"Context:\n{context_text}\n\nQuestion: {query}"
            pb_span.set_attribute("prompt.system_chars", len(system_prompt))
            pb_span.set_attribute("prompt.user_chars", len(user_message))

        # ── Stage 3: LLM inference ───────────────────────────────────────────
        with TRACER.start_as_current_span("llm_inference") as llm_span:
            t0 = time.perf_counter()
            response = llm_client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user",   "content": user_message},
                ],
                max_tokens=max_tokens,
            )
            elapsed_ms = (time.perf_counter() - t0) * 1000
            usage = response.usage
            # LLM-specific span attributes are the most valuable for cost analysis.
            # Use the OpenTelemetry GenAI semantic-convention names (gen_ai.*) rather
            # than ad-hoc keys, so any OTel-native backend can build dashboards without
            # per-app mapping rules. (The conventions are still marked "Development",
            # so pin the SDK version and expect occasional renames.)
            llm_span.set_attribute("gen_ai.operation.name",       "chat")
            llm_span.set_attribute("gen_ai.system",               "openai")   # newer revisions: gen_ai.provider.name
            llm_span.set_attribute("gen_ai.request.model",        "gpt-4o-mini")
            llm_span.set_attribute("gen_ai.request.max_tokens",   max_tokens)
            llm_span.set_attribute("gen_ai.response.model",       response.model)
            llm_span.set_attribute("gen_ai.usage.input_tokens",   usage.prompt_tokens)
            llm_span.set_attribute("gen_ai.usage.output_tokens",  usage.completion_tokens)
            # finish_reasons is an ARRAY in the spec (n>1 sampling returns several).
            llm_span.set_attribute("gen_ai.response.finish_reasons",
                                   [c.finish_reason for c in response.choices])
            llm_span.set_attribute("llm.latency_ms",              elapsed_ms)   # app-local extra

        answer = response.choices[0].message.content
        root_span.set_attribute("response.length_chars", len(answer))
        return {"answer": answer, "docs": docs}

# auto_instrument.py  —  zero-code-change tracing for OpenAI + LangChain calls
# Requirements: openinference-instrumentation-openai openinference-instrumentation-langchain
from openinference.instrumentation.openai import OpenAIInstrumentor
from openinference.instrumentation.langchain import LangChainInstrumentor
from trace_setup import setup_tracing

setup_tracing("llm-chat-service")     # installs the global TracerProvider first
OpenAIInstrumentor().instrument()     # every .chat.completions.create() now emits a span
LangChainInstrumentor().instrument()  # chains/agents emit a nested span tree

# metrics.py  —  Prometheus metrics for an LLM serving endpoint
# Requirements: prometheus_client
from prometheus_client import Counter, Histogram, Gauge, start_http_server

# ── Counters (monotonically increasing) ──────────────────────────────────────
REQUESTS_TOTAL = Counter(
    "llm_requests_total",
    "Total LLM requests",
    ["model", "status"],          # labels allow slicing by model and outcome
)
TOKENS_TOTAL = Counter(
    "llm_tokens_total",
    "Total tokens processed",
    ["model", "direction"],       # direction = 'prompt' or 'completion'
)

# ── Histograms (track distributions, not just averages) ───────────────────────
TTFT_HISTOGRAM = Histogram(
    "llm_time_to_first_token_seconds",
    "Time to first token in seconds",
    ["model"],
    # Buckets tuned for typical LLM latencies: 50ms → 10s
    buckets=[0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0],
)
LATENCY_HISTOGRAM = Histogram(
    "llm_request_latency_seconds",
    "End-to-end request latency",
    ["model"],
    buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0, 60.0],
)
COST_HISTOGRAM = Histogram(
    "llm_request_cost_usd",
    "Estimated cost per request in USD",
    ["model"],
    buckets=[0.0001, 0.001, 0.01, 0.05, 0.10, 0.50, 1.0],
)

# ── Gauges (current snapshot) ─────────────────────────────────────────────────
QUEUE_DEPTH = Gauge("llm_request_queue_depth", "Current pending request count")

# Pricing table (USD per 1M tokens); update as providers change rates
PRICING = {
    "gpt-4o-mini": {"prompt": 0.15, "completion": 0.60},
    "gpt-4o":      {"prompt": 5.00, "completion": 15.0},
}

def record_llm_call(model: str, prompt_tokens: int, completion_tokens: int,
                    latency_s: float, ttft_s: float, error: bool = False):
    """Call this after every LLM API response to update all metrics."""
    status = "error" if error else "ok"
    REQUESTS_TOTAL.labels(model=model, status=status).inc()
    TOKENS_TOTAL.labels(model=model, direction="prompt").inc(prompt_tokens)
    TOKENS_TOTAL.labels(model=model, direction="completion").inc(completion_tokens)
    TTFT_HISTOGRAM.labels(model=model).observe(ttft_s)
    LATENCY_HISTOGRAM.labels(model=model).observe(latency_s)

    if model in PRICING:
        cost = (prompt_tokens / 1e6 * PRICING[model]["prompt"]
                + completion_tokens / 1e6 * PRICING[model]["completion"])
        COST_HISTOGRAM.labels(model=model).observe(cost)

if __name__ == "__main__":
    start_http_server(9090)   # Prometheus scrapes :9090/metrics

# prometheus_alerts.yml
groups:
  - name: llm_service
    rules:
      # Alert if median TTFT exceeds 2 seconds for 5 minutes
      - alert: LLMHighTTFT
        expr: histogram_quantile(0.50, rate(llm_time_to_first_token_seconds_bucket[5m])) > 2.0
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "LLM median TTFT is {{ $value | humanizeDuration }}"

      # Alert if error rate exceeds 1% over 5 minutes
      - alert: LLMHighErrorRate
        expr: >
          rate(llm_requests_total{status="error"}[5m])
          / rate(llm_requests_total[5m]) > 0.01
        for: 5m
        labels:
          severity: critical

      # Alert if hourly token cost is trending toward budget overage
      - alert: LLMCostBudgetWarning
        expr: sum(rate(llm_tokens_total{direction="completion"}[1h])) * 3600 * 0.60 / 1e6 > 100
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "Projected hourly completion cost exceeds $100"

# selfhost_cost.py  —  derive USD/1M tokens from live throughput
GPU_HOURLY_USD = 2.00        # your actual instance price
NUM_GPUS       = 1

def usd_per_million_tokens(tokens_per_second: float) -> float:
    """Amortize GPU rent over measured throughput. Returns inf when idle."""
    if tokens_per_second <= 0:
        return float("inf")
    cost_per_second = GPU_HOURLY_USD * NUM_GPUS / 3600.0
    return cost_per_second / tokens_per_second * 1e6

# log_schema.py  —  Pydantic model for a structured LLM log record
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime
import uuid

class LLMLogRecord(BaseModel):
    # Identity
    record_id:      str       = Field(default_factory=lambda: str(uuid.uuid4()))
    trace_id:       str       # OTel trace ID — links to distributed trace
    session_id:     str       # Groups turns in a multi-turn conversation
    user_id_hashed: str       # SHA-256 of user ID — never store raw PII here

    # Request context
    timestamp_utc:  datetime
    model:          str
    prompt_version: str       # e.g. "chat-v3.2" — MUST be versioned for regression analysis

    # Content — store in encrypted, access-controlled storage (not plaintext RDBMS)
    system_prompt:  Optional[str]   = None   # May contain IP; log if needed for debugging
    user_message:   str
    assistant_reply: str

    # Usage
    prompt_tokens:      int
    completion_tokens:  int
    latency_ms:         float
    finish_reason:      str   # "stop", "length", "content_filter", etc.

    # Evaluation signals (filled in asynchronously)
    thumbs_up:          Optional[bool]  = None   # User feedback if collected
    auto_eval_score:    Optional[float] = None   # From async LLM-as-judge pipeline
    safety_flagged:     Optional[bool]  = None   # From guardrail layer

    class Config:
        json_encoders = {datetime: lambda v: v.isoformat()}

# production_eval_sampler.py
import random
from dataclasses import dataclass
from typing import Optional

@dataclass
class SamplingPolicy:
    base_rate: float          # Fraction of all traffic to evaluate (e.g. 0.05 = 5%)
    failure_rate: float       # Fraction of error/flagged requests to evaluate (e.g. 1.0 = 100%)
    new_prompt_rate: float    # Fraction of requests using a new prompt template to evaluate
    low_score_rate: float     # Fraction of requests below a score threshold to evaluate

def should_evaluate(record: dict, policy: SamplingPolicy) -> bool:
    """Return True if this record should be sent to the async eval pipeline."""
    # Always evaluate failures and safety flags
    if record.get("finish_reason") in ("content_filter", "error"):
        return random.random() < policy.failure_rate
    # Evaluate new prompt versions at higher rate to catch regressions early
    if record.get("is_new_prompt_version", False):
        return random.random() < policy.new_prompt_rate
    # Evaluate previously low-scoring records (detected drift)
    if record.get("auto_eval_score", 1.0) < 0.5:
        return random.random() < policy.low_score_rate
    # Baseline random sample for steady-state tracking
    return random.random() < policy.base_rate

POLICY = SamplingPolicy(
    base_rate=0.05,
    failure_rate=1.00,
    new_prompt_rate=0.30,
    low_score_rate=0.50,
)

# async_eval_worker.py
# Runs as a separate process / Cloud Function; reads from an eval queue
import asyncio
import json
from openai import AsyncOpenAI

client = AsyncOpenAI()

JUDGE_SYSTEM = """You are an expert evaluator. Score the assistant reply on:
1. Faithfulness (0-2): Does the reply contradict the provided context?
2. Relevance (0-2): Does the reply address the user's question?
3. Fluency (0-1): Is the reply grammatically correct?

Output ONLY valid JSON: {"faithfulness": X, "relevance": X, "fluency": X, "reason": "..."}"""

async def judge_single(record: dict) -> dict:
    """Call the judge model and parse the score, with a 30-second timeout."""
    prompt = (
        f"Context provided to assistant:\n{record.get('context', 'none')}\n\n"
        f"User question: {record['user_message']}\n\n"
        f"Assistant reply: {record['assistant_reply']}"
    )
    response = await asyncio.wait_for(
        client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": JUDGE_SYSTEM},
                {"role": "user",   "content": prompt},
            ],
            temperature=0,      # Zero temperature for reproducibility
            max_tokens=256,
        ),
        timeout=30.0,
    )
    raw = response.choices[0].message.content
    scores = json.loads(raw)
    scores["composite"] = (scores["faithfulness"] + scores["relevance"] + scores["fluency"]) / 5.0
    scores["record_id"]  = record["record_id"]
    return scores

async def eval_batch(records: list[dict]) -> list[dict]:
    """Evaluate a batch of records concurrently (up to 20 in parallel)."""
    semaphore = asyncio.Semaphore(20)
    async def bounded(r):
        async with semaphore:
            try:
                return await judge_single(r)
            except Exception as e:
                return {"record_id": r["record_id"], "error": str(e), "composite": None}
    return await asyncio.gather(*[bounded(r) for r in records])

# drift_detection.py  —  PSI on prompt embedding distributions
import numpy as np
from sklearn.decomposition import PCA

def compute_psi(reference: np.ndarray, current: np.ndarray, n_bins: int = 10) -> float:
    """
    Compute Population Stability Index on 1-D arrays.

    Typical usage: call this on the first principal component of prompt embeddings,
    comparing a rolling 24-hour window against the previous 7-day baseline.
    """
    # Use reference distribution to define bin edges (important: same bins for both)
    min_val = min(reference.min(), current.min())
    max_val = max(reference.max(), current.max())
    bins = np.linspace(min_val, max_val, n_bins + 1)

    ref_counts, _ = np.histogram(reference, bins=bins)
    cur_counts, _ = np.histogram(current,   bins=bins)

    # Add small epsilon to avoid division by zero or log(0)
    eps = 1e-6
    ref_pct = (ref_counts + eps) / (ref_counts.sum() + eps * n_bins)
    cur_pct = (cur_counts + eps) / (cur_counts.sum() + eps * n_bins)

    psi = np.sum((cur_pct - ref_pct) * np.log(cur_pct / ref_pct))
    return float(psi)


def monitor_embedding_drift(
    reference_embeddings: np.ndarray,   # shape: (N_ref, D)
    current_embeddings:   np.ndarray,   # shape: (N_cur, D)
    n_components: int = 5,
) -> dict[str, float]:
    """Project embeddings to principal components, then compute PSI on each."""
    pca = PCA(n_components=n_components).fit(reference_embeddings)
    ref_proj = pca.transform(reference_embeddings)
    cur_proj = pca.transform(current_embeddings)

    results = {}
    for i in range(n_components):
        results[f"psi_pc{i}"] = compute_psi(ref_proj[:, i], cur_proj[:, i])
    results["psi_max"] = max(results.values())
    return results

# ab_router.py  —  stateless traffic-split router for LLM variants
import hashlib
from typing import Callable

# Variant config: (weight, handler_function)
# Weights must sum to 1.0.
VARIANTS = {
    "control":     (0.80, "handle_with_gpt4o_mini"),
    "treatment_A": (0.10, "handle_with_claude3_haiku"),
    "treatment_B": (0.10, "handle_with_gpt4o_mini_v2_prompt"),
}

def route_request(session_id: str) -> str:
    """
    Route a request to a variant using a deterministic hash.

    Using a hash of session_id (not a random draw) ensures that the SAME
    user always gets the SAME variant across multiple turns — critical for
    multi-turn conversation quality evaluation.
    """
    # MD5 is fine here; we need bucketing, not cryptographic security.
    bucket = int(hashlib.md5(session_id.encode()).hexdigest(), 16) % 1000

    cumulative = 0
    for variant_name, (weight, _handler) in VARIANTS.items():
        cumulative += int(weight * 1000)
        if bucket < cumulative:
            return variant_name
    return "control"   # fallback

# langfuse_integration.py  —  SDK-level Langfuse tracing (v3 Python SDK)
# Requirements: langfuse openai
from langfuse import observe, get_client
# Drop-in wrapper: identical surface to `openai`, but every call is traced.
from langfuse.openai import OpenAI

# Langfuse reads LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, LANGFUSE_HOST from env
langfuse = get_client()
client = OpenAI()

@observe(name="rag_pipeline")          # Creates a trace named "rag_pipeline"
def run_rag(query: str, user_id: str) -> str:
    # Update the current trace with metadata (user, session, tags)
    langfuse.update_current_trace(
        user_id=user_id,
        tags=["production", "rag-v2"],
    )

    # Retrieval step — appears as a child span
    docs = retrieve_documents(query)

    # LLM call — the wrapped client records model, token usage, and cost automatically
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": build_prompt(query, docs)}],
    )
    return response.choices[0].message.content


@observe(name="retrieve_documents")    # Child span for retrieval
def retrieve_documents(query: str) -> list:
    # ... vector DB call ...
    return []

def build_prompt(query, docs):
    return f"Context: {docs}\n\nQuestion: {query}"

# Self-hosted serving with metrics + tracing turned on.
vllm serve ./stack-100m \
  --port 8000 \
  --otlp-traces-endpoint http://localhost:4317   # engine spans -> your OTel collector
# Prometheus scrape target is then http://localhost:8000/metrics

# SGLang equivalent
python -m sglang.launch_server --model-path ./stack-100m \
  --port 30000 --enable-metrics

# prometheus.yml  —  scrape a self-hosted engine alongside your app
scrape_configs:
  - job_name: vllm
    metrics_path: /metrics
    static_configs:
      - targets: ["localhost:8000"]
  - job_name: llm-app
    static_configs:
      - targets: ["localhost:9090"]