Ch 3 — ML Monitoring & Observability¶

from fastapi import FastAPI
from prometheus_fastapi_instrumentator import Instrumentator
from prometheus_client import Counter, Gauge, Histogram

app = FastAPI()

# Custom LLM-specific counters
token_counter = Counter(
    "llm_tokens_total",
    "Total tokens processed",
    labelnames=["model", "direction"],    # direction: input | output
)
cost_histogram = Histogram(
    "llm_request_cost_usd",
    "Cost per LLM request in USD",
    buckets=[0.0001, 0.001, 0.01, 0.05, 0.1, 0.5, 1.0],
)
gpu_utilisation_gauge = Gauge("gpu_utilisation_percent", "GPU utilisation %")

# Instrument all HTTP endpoints automatically
Instrumentator().instrument(app).expose(app, endpoint="/metrics")

"""
drift_detection.py — Statistical drift tests for ML monitoring.

Requirements:
    pip install numpy scipy pandas
"""
from __future__ import annotations

import numpy as np
import pandas as pd
from scipy import stats


def compute_psi(
    expected: np.ndarray,
    actual: np.ndarray,
    n_bins: int = 10,
    epsilon: float = 1e-6,
) -> float:
    """
    Compute Population Stability Index between two 1-D numeric distributions.

    Args:
        expected: Reference distribution (e.g., training feature values).
        actual:   Current distribution (e.g., production feature values).
        n_bins:   Number of quantile-based bins.
        epsilon:  Smoothing term to avoid log(0).

    Returns:
        PSI value (float). Threshold: < 0.1 stable, 0.1-0.2 moderate, > 0.2 significant.
    """
    bin_edges = np.percentile(expected, np.linspace(0, 100, n_bins + 1))
    bin_edges[0] -= epsilon
    bin_edges[-1] += epsilon

    expected_counts, _ = np.histogram(expected, bins=bin_edges)
    actual_counts, _ = np.histogram(actual, bins=bin_edges)

    e_pct = expected_counts / len(expected) + epsilon
    a_pct = actual_counts / len(actual) + epsilon

    psi = float(np.sum((a_pct - e_pct) * np.log(a_pct / e_pct)))
    return psi

def ks_drift_test(
    reference: np.ndarray,
    current: np.ndarray,
    alpha: float = 0.05,
) -> dict:
    """
    Two-sample KS test for distribution shift on a continuous feature.

    Returns:
        dict with statistic, p_value, and drift_detected flag.
    """
    result = stats.ks_2samp(reference, current)
    return {
        "statistic": float(result.statistic),
        "p_value": float(result.pvalue),
        "drift_detected": bool(result.pvalue < alpha),
        "interpretation": (
            "Distributions differ significantly" if result.pvalue < alpha
            else "No significant difference detected"
        ),
    }

from scipy.stats import chi2_contingency


def chi2_drift_test(
    reference_series: pd.Series,
    current_series: pd.Series,
    alpha: float = 0.05,
) -> dict:
    """
    Chi-squared test for distribution shift on a categorical feature.

    Args:
        reference_series: Categorical values from the reference period.
        current_series:   Categorical values from the current period.
        alpha:            Significance level (default 5%).

    Returns:
        dict with chi2 statistic, p_value, degrees of freedom, drift_detected.
    """
    categories = set(reference_series.unique()) | set(current_series.unique())
    ref_counts = reference_series.value_counts().reindex(categories, fill_value=0)
    cur_counts = current_series.value_counts().reindex(categories, fill_value=0)

    contingency = pd.DataFrame({"reference": ref_counts, "current": cur_counts})
    chi2, p_value, dof, _ = chi2_contingency(contingency.T)

    return {
        "chi2_statistic": float(chi2),
        "p_value": float(p_value),
        "degrees_of_freedom": int(dof),
        "drift_detected": bool(p_value < alpha),
    }

import pandas as pd
from evidently.metric_preset import DataDriftPreset, DataQualityPreset
from evidently.report import Report

reference_data = pd.read_parquet("data/reference_features.parquet")
production_data = pd.read_parquet("data/production_week_24.parquet")

report = Report(metrics=[DataDriftPreset(), DataQualityPreset()])
report.run(reference_data=reference_data, current_data=production_data)
report.save_html("drift_report_week_24.html")

result = report.as_dict()
drifted = [
    m["result"]["column_name"]
    for m in result["metrics"]
    if m.get("result", {}).get("drift_detected", False)
]
print(f"Drifted features ({len(drifted)}): {drifted}")
if len(drifted) >= 3:
    trigger_retraining_pipeline()

def detect_prediction_score_drift(
    reference_scores: np.ndarray,
    current_scores: np.ndarray,
    alpha: float = 0.05,
) -> dict:
    """
    Detect drift in the distribution of model prediction probabilities.
    A shift here indicates concept drift even without ground-truth labels.
    """
    return ks_drift_test(reference_scores, current_scores, alpha=alpha)

"""
llm_observability.py — LLM tracing and quality monitoring with Langfuse.

Requirements:
    pip install langfuse anthropic
"""
from __future__ import annotations

import json
import os

import anthropic
from langfuse import Langfuse
from langfuse.decorators import langfuse_context, observe

langfuse = Langfuse(
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
    secret_key=os.environ["LANGFUSE_SECRET_KEY"],
    host=os.environ.get("LANGFUSE_HOST", "https://cloud.langfuse.com"),
)
client = anthropic.Anthropic()

FAITHFULNESS_PROMPT = """\
You evaluate whether an AI answer is faithfully grounded in the provided context.

Context:
{context}

Question: {question}
Answer: {answer}

Rate faithfulness 0.0–1.0 (1.0 = fully grounded, 0.0 = fabricated).
Respond with JSON only: {{"faithfulness": <float>, "reason": "<brief>"}}
"""


@observe(name="rag_query")
def rag_with_observability(question: str, context: str) -> str:
    """
    Execute a RAG query with full Langfuse tracing and quality scoring.

    Traces: input/output, token usage, cost, model
    Scores: faithfulness (LLM-as-judge), latency_per_token
    """
    langfuse_context.update_current_trace(
        name="RAG Pipeline",
        input={"question": question, "context_length": len(context)},
        tags=["rag", "production"],
    )

    # ── Generation ────────────────────────────────────────────────────────────
    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=512,
        messages=[
            {
                "role": "user",
                "content": (
                    f"Context:\n{context}\n\n"
                    f"Question: {question}\n\n"
                    "Answer using only the provided context. "
                    "If the context does not contain the answer, say so."
                ),
            }
        ],
    )
    answer = response.content[0].text

    # ── Faithfulness scoring (LLM-as-judge) ──────────────────────────────────
    judge_response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=128,
        messages=[
            {
                "role": "user",
                "content": FAITHFULNESS_PROMPT.format(
                    context=context, question=question, answer=answer
                ),
            }
        ],
    )

    faithfulness_score = 0.5
    try:
        score_data = json.loads(judge_response.content[0].text)
        faithfulness_score = float(score_data.get("faithfulness", 0.5))
    except (json.JSONDecodeError, KeyError, ValueError):
        pass

    # ── Cost tracking ─────────────────────────────────────────────────────────
    input_cost = response.usage.input_tokens * 0.80 / 1_000_000     # $/token (Haiku)
    output_cost = response.usage.output_tokens * 4.00 / 1_000_000
    total_cost_usd = input_cost + output_cost

    # ── Log scores to Langfuse ────────────────────────────────────────────────
    langfuse_context.score_current_trace(
        name="faithfulness",
        value=faithfulness_score,
        comment=f"cost_usd={total_cost_usd:.6f}",
    )
    langfuse_context.update_current_trace(
        output={"answer": answer},
        metadata={
            "input_tokens": response.usage.input_tokens,
            "output_tokens": response.usage.output_tokens,
            "cost_usd": total_cost_usd,
        },
    )

    return answer

# Example: weekly retraining every Sunday at 03:00 UTC
# In Kubernetes, use a CronJob:
# schedule: "0 3 * * 0"

def should_retrain_on_drift(
    reference_data: pd.DataFrame,
    current_data: pd.DataFrame,
    psi_threshold: float = 0.2,
    n_features_to_trigger: int = 3,
) -> tuple[bool, list[str]]:
    """
    Trigger retraining when PSI exceeds threshold for at least
    `n_features_to_trigger` numeric features.

    Returns:
        (trigger_retraining, list_of_drifted_features)
    """
    drifted: list[str] = []
    for col in reference_data.select_dtypes(include="number").columns:
        psi = compute_psi(reference_data[col].values, current_data[col].values)
        if psi > psi_threshold:
            drifted.append(col)
    return len(drifted) >= n_features_to_trigger, drifted

def should_retrain_on_performance(
    recent_metric: float,
    baseline_metric: float,
    relative_drop_threshold: float = 0.05,
    metric_name: str = "accuracy",
) -> bool:
    """
    Trigger retraining when a quality metric drops by more than
    `relative_drop_threshold` relative to baseline.

    Args:
        recent_metric:           Current period metric value.
        baseline_metric:         Reference (post-last-training) metric value.
        relative_drop_threshold: Fraction drop to trigger retraining (default 5%).

    Returns:
        True if retraining should be triggered.
    """
    relative_drop = (baseline_metric - recent_metric) / (baseline_metric + 1e-9)
    if relative_drop > relative_drop_threshold:
        print(
            f"RETRAIN: {metric_name} dropped {relative_drop:.1%} "
            f"({baseline_metric:.4f} → {recent_metric:.4f})"
        )
        return True
    return False