8. Project: AI-Specific Monitoring Layer

Description

This project closes Module 6. You take everything you built — prompt quality tracking, token usage monitoring, model drift detection, hallucination detection, LangSmith integration, and custom metrics with dashboards — and integrate it into a functional AI-Specific Monitoring Layer.

This monitoring layer is what differentiates "we have observability" from "we have observability for AI". Modules 1-5 could apply (with adjustments) to any software system. Module 6 is exclusively AI — without it, your observability stack doesn't understand what your LLM says, whether it says it well, or whether it changed how it says it.

The deliverable is an integrated system you can point at your AI system in production. Detailed token tracking, quality scoring, hallucination detection configured, and everything correlated with OTel.

Connection with the module: Capsules 02-07 built individual components. This project integrates them into a single cohesive system.

---

Objective

Build a complete AI-Specific Monitoring Layer that monitors the semantic dimension of your AI system.

By the end of this project you will be able to:

• ✅ Have an integrated monitoring layer with all of the module's components
• ✅ Prompt quality tracking with baselines and drift detection
• ✅ Token usage monitoring with drill-down, outliers, and budget enforcement
• ✅ Model drift detection with baseline comparison and alerts
• ✅ Hallucination detection with heuristics and LLM-as-judge
• ✅ Custom AI-specific metrics with a complete dashboard
• ✅ Correlation with OTel to integrate with the existing stack
• ✅ A simulation that validates the behavior of the complete system

---

What You Will Build

AI-SPECIFIC MONITORING LAYER — Deliverable
├── 1. AIMonitoringLayer (Integrated System)
│   ├── Orchestrates all components
│   ├── Processes each request through all checks
│   ├── Generates consolidated alerts
│   └── Produces a unified dashboard
│
├── 2. Components
│   ├── PromptQualityTracker (capsule 02)
│   ├── TokenUsageMonitor (capsule 03)
│   ├── DriftDetector (capsule 04)
│   ├── HallucinationDetector (capsule 05)
│   └── AIMetricsCollector (capsule 07)
│
├── 3. OTel Integration
│   ├── Span attributes for each component
│   ├── Custom metrics (gauges, counters, histograms)
│   └── Events for alerts
│
└── 4. Complete Simulation
    ├── Phase 1: Normal traffic → establish baselines
    ├── Phase 2: Model drift → detect the change
    ├── Phase 3: Hallucination spike → detect degradation
    ├── Phase 4: Recovery → confirm stabilization
    └── Final dashboard with the system state

---

Technical Specifications

Stack

python >= 3.10
opentelemetry-api
opentelemetry-sdk
tiktoken (optional, for precise token counting)

Conceptual structure

ai-monitoring-layer/
├── monitoring_layer.py       # Integrated AIMonitoringLayer
├── prompt_tracker.py         # PromptQualityTracker
├── token_monitor.py          # TokenUsageMonitor
├── drift_detector.py         # DriftDetector
├── hallucination_detector.py # HallucinationDetector
├── metrics_collector.py      # AIMetricsCollector
├── otel_integration.py       # OTel spans + metrics
└── simulate.py               # Complete simulation

---

Step by Step

Step 1: Shared data models

import re
import time
import random
import hashlib
import statistics
import json
from dataclasses import dataclass, field
from datetime import datetime, timezone
from collections import deque, defaultdict
from typing import Optional
from enum import Enum


class AlertSeverity(Enum):
    INFO = "info"
    WARNING = "warning"
    CRITICAL = "critical"


class FidelityLevel(Enum):
    FAITHFUL = 0
    MINOR_IMPRECISION = 1
    IRRELEVANT = 2
    OUTDATED = 3
    DISTORTED = 4
    FABRICATED = 5


@dataclass
class AIRequest:
    """A complete request that flows through the whole monitoring layer."""
    timestamp: float
    endpoint: str
    model: str
    user_id: str
    system_prompt: str
    user_prompt: str
    output_text: str
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    cost_usd: float
    latency_seconds: float
    context: str = ""
    success: bool = True
    is_retry: bool = False
    retry_reason: str = ""


@dataclass
class MonitoringAlert:
    """A unified alert from any component."""
    source: str
    alert_type: str
    severity: AlertSeverity
    message: str
    value: float
    threshold: float
    endpoint: str
    metadata: dict = field(default_factory=dict)
    timestamp: str = ""

    def __post_init__(self):
        if not self.timestamp:
            self.timestamp = datetime.now(timezone.utc).isoformat()


@dataclass
class MonitoringResult:
    """The complete result of processing a request."""
    request_id: str
    prompt_analysis: dict
    token_analysis: dict
    hallucination_score: Optional[float]
    drift_score: Optional[float]
    quality_score: Optional[float]
    alerts: list[MonitoringAlert]
    otel_attributes: dict

Step 2: Integrated components (compact version)

class PromptTracker:
    """Prompt quality tracking (integrated version)."""

    def __init__(self, max_instructions: int = 8):
        self.max_instructions = max_instructions
        self.history: deque[dict] = deque(maxlen=5000)
        self.baselines: dict[str, dict] = {}
        self.template_hashes: dict[str, str] = {}

    def analyze(self, system_prompt: str, user_prompt: str, endpoint: str) -> tuple[dict, list[MonitoringAlert]]:
        full = system_prompt + "\n" + user_prompt
        normalized = re.sub(r"\s+", " ", system_prompt.strip().lower())
        template_hash = hashlib.sha256(normalized.encode()).hexdigest()[:12]

        instruction_count = len(re.findall(
            r"(?:must|always|never|respond|include|avoid)\b",
            system_prompt, re.IGNORECASE,
        ))

        analysis = {
            "prompt_length": len(full),
            "system_length": len(system_prompt),
            "user_length": len(user_prompt),
            "instruction_count": max(instruction_count, 1),
            "template_hash": template_hash,
            "estimated_tokens": len(full) // 4,
        }
        self.history.append({**analysis, "endpoint": endpoint})

        alerts = []
        baseline = self.baselines.get(endpoint)
        if baseline:
            std = baseline.get("length_std", 1) or 1
            z = (analysis["prompt_length"] - baseline["length_mean"]) / std
            if abs(z) > 2.0:
                direction = "longer" if z > 0 else "shorter"
                alerts.append(MonitoringAlert(
                    source="prompt_tracker", alert_type="length_drift",
                    severity=AlertSeverity.WARNING,
                    message=f"Prompt {direction}: {analysis['prompt_length']} chars (baseline: {baseline['length_mean']:.0f})",
                    value=analysis["prompt_length"], threshold=baseline["length_mean"],
                    endpoint=endpoint, metadata={"z_score": round(z, 2)},
                ))

            registered = self.template_hashes.get(endpoint)
            if registered and template_hash != registered:
                alerts.append(MonitoringAlert(
                    source="prompt_tracker", alert_type="template_change",
                    severity=AlertSeverity.CRITICAL,
                    message=f"Template changed in {endpoint}",
                    value=0, threshold=1, endpoint=endpoint,
                    metadata={"old_hash": registered, "new_hash": template_hash},
                ))

        if analysis["instruction_count"] > self.max_instructions:
            alerts.append(MonitoringAlert(
                source="prompt_tracker", alert_type="complexity_creep",
                severity=AlertSeverity.WARNING,
                message=f"{analysis['instruction_count']} instructions (max: {self.max_instructions})",
                value=analysis["instruction_count"], threshold=self.max_instructions,
                endpoint=endpoint,
            ))

        return analysis, alerts

    def establish_baseline(self, endpoint: str):
        ep_data = [h for h in self.history if h["endpoint"] == endpoint]
        if len(ep_data) < 50:
            return
        lengths = [h["prompt_length"] for h in ep_data[-200:]]
        hashes = [h["template_hash"] for h in ep_data[-200:]]
        most_common = max(set(hashes), key=hashes.count)

        self.baselines[endpoint] = {
            "length_mean": statistics.mean(lengths),
            "length_std": statistics.stdev(lengths) if len(lengths) > 1 else 0,
        }
        self.template_hashes[endpoint] = most_common


class TokenTracker:
    """Token usage monitoring (integrated version)."""

    def __init__(self, outlier_multiplier: float = 10.0, budgets: dict = None):
        self.outlier_multiplier = outlier_multiplier
        self.budgets = budgets or {}
        self.by_endpoint: dict[str, list[int]] = defaultdict(list)
        self.by_model: dict[str, int] = defaultdict(int)
        self.by_user: dict[str, int] = defaultdict(int)
        self.endpoint_totals: dict[str, int] = defaultdict(int)
        self.baselines: dict[str, dict] = {}

    def record(self, req: AIRequest) -> tuple[dict, list[MonitoringAlert]]:
        self.by_endpoint[req.endpoint].append(req.total_tokens)
        self.by_model[req.model] += req.total_tokens
        self.by_user[req.user_id] += req.total_tokens
        self.endpoint_totals[req.endpoint] += req.total_tokens

        analysis = {
            "total_tokens": req.total_tokens,
            "prompt_tokens": req.prompt_tokens,
            "completion_tokens": req.completion_tokens,
            "cost_usd": req.cost_usd,
            "endpoint_total": self.endpoint_totals[req.endpoint],
        }

        alerts = []
        baseline = self.baselines.get(req.endpoint)
        if baseline and baseline["mean"] > 0:
            if req.total_tokens > baseline["mean"] * self.outlier_multiplier:
                alerts.append(MonitoringAlert(
                    source="token_tracker", alert_type="token_outlier",
                    severity=AlertSeverity.WARNING,
                    message=f"Outlier: {req.total_tokens:,} tokens ({req.total_tokens / baseline['mean']:.1f}x avg)",
                    value=req.total_tokens, threshold=baseline["mean"] * self.outlier_multiplier,
                    endpoint=req.endpoint,
                    metadata={"user_id": req.user_id, "model": req.model},
                ))

        budget = self.budgets.get(req.endpoint)
        if budget and self.endpoint_totals[req.endpoint] >= budget:
            alerts.append(MonitoringAlert(
                source="token_tracker", alert_type="budget_exceeded",
                severity=AlertSeverity.CRITICAL,
                message=f"Token budget exceeded in {req.endpoint}: {self.endpoint_totals[req.endpoint]:,} / {budget:,}",
                value=self.endpoint_totals[req.endpoint], threshold=budget,
                endpoint=req.endpoint,
            ))

        return analysis, alerts

    def establish_baseline(self, endpoint: str):
        values = self.by_endpoint.get(endpoint, [])
        if len(values) < 50:
            return
        self.baselines[endpoint] = {
            "mean": statistics.mean(values[-200:]),
            "std": statistics.stdev(values[-200:]) if len(values) > 1 else 0,
        }


class DriftChecker:
    """Drift detection (integrated version)."""

    def __init__(self, threshold_sigma: float = 2.0):
        self.threshold = threshold_sigma
        self.output_lengths: dict[str, deque] = defaultdict(lambda: deque(maxlen=5000))
        self.baselines: dict[str, dict] = {}

    def record(self, req: AIRequest) -> tuple[dict, list[MonitoringAlert]]:
        key = f"{req.endpoint}:{req.model}"
        output_len = len(req.output_text)
        self.output_lengths[key].append(output_len)

        analysis = {"output_length": output_len, "key": key}
        alerts = []

        baseline = self.baselines.get(key)
        if baseline and baseline["std"] > 0:
            recent = list(self.output_lengths[key])[-50:]
            if len(recent) >= 30:
                current_mean = statistics.mean(recent)
                z = (current_mean - baseline["mean"]) / baseline["std"]
                change_pct = ((current_mean - baseline["mean"]) / baseline["mean"]) * 100

                analysis["drift_z_score"] = round(z, 2)
                analysis["drift_change_pct"] = round(change_pct, 1)

                if abs(z) > self.threshold:
                    direction = "shorter" if z < 0 else "longer"
                    alerts.append(MonitoringAlert(
                        source="drift_checker", alert_type="model_drift",
                        severity=AlertSeverity.WARNING,
                        message=f"Drift: {direction} responses ({change_pct:+.0f}%) in {req.model}",
                        value=current_mean, threshold=baseline["mean"],
                        endpoint=req.endpoint,
                        metadata={"model": req.model, "z_score": round(z, 2)},
                    ))

        return analysis, alerts

    def establish_baseline(self, endpoint: str, model: str):
        key = f"{endpoint}:{model}"
        values = list(self.output_lengths.get(key, []))
        if len(values) < 50:
            return
        self.baselines[key] = {
            "mean": statistics.mean(values[-200:]),
            "std": statistics.stdev(values[-200:]) if len(values) > 1 else 0,
        }


class HallucinationChecker:
    """Hallucination detection (integrated version)."""

    OVERCONFIDENCE = [r"\b(definitely|absolutely|exactly|without a doubt)\b"]
    HEDGE = [r"\b(maybe|perhaps|i think|might|not sure)\b"]
    SPECIFIC = r"\b\d{2,}\.\d+%\b|\b\d{4,}\b"

    def __init__(self, heuristic_threshold: float = 0.4):
        self.threshold = heuristic_threshold
        self.history: deque[float] = deque(maxlen=10_000)

    def evaluate(self, req: AIRequest) -> tuple[dict, list[MonitoringAlert]]:
        text = req.output_text
        words = max(len(text.split()), 1)

        overconf = sum(len(re.findall(p, text, re.I)) for p in self.OVERCONFIDENCE)
        hedge = sum(len(re.findall(p, text, re.I)) for p in self.HEDGE)
        specific = len(re.findall(self.SPECIFIC, text))

        overconf_score = min(overconf / max(words / 50, 1), 1.0)
        hedge_ratio = hedge / words
        spec_score = min(specific / max(words / 100, 1), 1.0)

        heuristic = (
            overconf_score * 0.35
            + spec_score * 0.30
            + (1.0 - min(hedge_ratio * 20, 1.0)) * 0.20
        )

        context_overlap = 0.0
        if req.context:
            ctx_words = set(req.context.lower().split())
            out_words = set(text.lower().split())
            context_overlap = len(ctx_words & out_words) / max(len(out_words), 1)
            faithfulness = context_overlap * 0.7 + (1 - heuristic) * 0.3
            score = 1.0 - faithfulness
        else:
            score = heuristic * 0.6

        score = round(max(0.0, min(1.0, score)), 3)
        self.history.append(score)

        analysis = {
            "hallucination_score": score,
            "heuristic_score": round(heuristic, 3),
            "overconfidence": round(overconf_score, 3),
            "context_overlap": round(context_overlap, 3),
            "is_hallucination": score >= 0.5,
        }

        alerts = []
        if len(self.history) >= 50:
            recent = list(self.history)[-50:]
            halluc_rate = sum(1 for s in recent if s >= 0.5) / len(recent)
            if halluc_rate > 0.10:
                alerts.append(MonitoringAlert(
                    source="hallucination_checker", alert_type="hallucination_rate_high",
                    severity=AlertSeverity.WARNING,
                    message=f"Hallucination rate: {halluc_rate*100:.1f}% (last 50 requests)",
                    value=halluc_rate, threshold=0.10,
                    endpoint=req.endpoint,
                ))

        return analysis, alerts

Step 3: AIMonitoringLayer — Integrated System

class AIMonitoringLayer:
    """AI-specific monitoring layer that integrates all components.

    Processes each request through:
    1. Prompt quality tracking
    2. Token usage monitoring
    3. Drift detection
    4. Hallucination detection
    5. Aggregated metrics
    """

    def __init__(
        self,
        token_budgets: dict = None,
        outlier_multiplier: float = 10.0,
        drift_sigma: float = 2.0,
        hallucination_threshold: float = 0.4,
    ):
        self.prompt_tracker = PromptTracker()
        self.token_tracker = TokenTracker(
            outlier_multiplier=outlier_multiplier,
            budgets=token_budgets or {},
        )
        self.drift_checker = DriftChecker(threshold_sigma=drift_sigma)
        self.hallucination_checker = HallucinationChecker(
            heuristic_threshold=hallucination_threshold,
        )

        self.all_alerts: list[MonitoringAlert] = []
        self.all_results: list[MonitoringResult] = []
        self.request_count: int = 0

        self.quality_scores: deque[float] = deque(maxlen=10_000)
        self.cost_by_endpoint: dict[str, float] = defaultdict(float)
        self.requests_by_endpoint: dict[str, int] = defaultdict(int)

    def process(self, req: AIRequest) -> MonitoringResult:
        """Processes a request through the entire monitoring layer."""
        self.request_count += 1
        request_id = f"req-{self.request_count:06d}"
        all_alerts = []

        prompt_analysis, prompt_alerts = self.prompt_tracker.analyze(
            req.system_prompt, req.user_prompt, req.endpoint,
        )
        all_alerts.extend(prompt_alerts)

        token_analysis, token_alerts = self.token_tracker.record(req)
        all_alerts.extend(token_alerts)

        drift_analysis, drift_alerts = self.drift_checker.record(req)
        all_alerts.extend(drift_alerts)

        halluc_analysis, halluc_alerts = self.hallucination_checker.evaluate(req)
        all_alerts.extend(halluc_alerts)

        quality_score = None
        if req.success and not halluc_analysis["is_hallucination"]:
            quality_score = max(0, 1.0 - halluc_analysis["hallucination_score"])
            self.quality_scores.append(quality_score)

        self.cost_by_endpoint[req.endpoint] += req.cost_usd
        self.requests_by_endpoint[req.endpoint] += 1

        otel_attributes = {
            "ai.prompt.length": prompt_analysis["prompt_length"],
            "ai.prompt.template_hash": prompt_analysis["template_hash"],
            "ai.prompt.instruction_count": prompt_analysis["instruction_count"],
            "ai.tokens.total": req.total_tokens,
            "ai.tokens.prompt": req.prompt_tokens,
            "ai.tokens.completion": req.completion_tokens,
            "ai.hallucination.score": halluc_analysis["hallucination_score"],
            "ai.hallucination.is_hallucination": halluc_analysis["is_hallucination"],
            "ai.quality.score": quality_score or 0,
            "ai.drift.z_score": drift_analysis.get("drift_z_score", 0),
            "ai.cost.usd": req.cost_usd,
            "ai.model": req.model,
        }

        result = MonitoringResult(
            request_id=request_id,
            prompt_analysis=prompt_analysis,
            token_analysis=token_analysis,
            hallucination_score=halluc_analysis["hallucination_score"],
            drift_score=drift_analysis.get("drift_z_score"),
            quality_score=quality_score,
            alerts=all_alerts,
            otel_attributes=otel_attributes,
        )

        self.all_alerts.extend(all_alerts)
        self.all_results.append(result)
        return result

    def establish_baselines(self, endpoints: list[str], models: list[str]):
        """Establishes baselines for all components."""
        for ep in endpoints:
            self.prompt_tracker.establish_baseline(ep)
            self.token_tracker.establish_baseline(ep)
            for model in models:
                self.drift_checker.establish_baseline(ep, model)

    def dashboard(self) -> dict:
        """Complete dashboard of the monitoring layer state."""
        quality_list = list(self.quality_scores)

        halluc_scores = list(self.hallucination_checker.history)
        halluc_rate = (
            sum(1 for s in halluc_scores[-100:] if s >= 0.5) / max(len(halluc_scores[-100:]), 1)
            if halluc_scores else 0
        )

        alert_counts = defaultdict(int)
        for a in self.all_alerts:
            alert_counts[a.source] += 1

        return {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "overview": {
                "total_requests": self.request_count,
                "total_alerts": len(self.all_alerts),
                "alerts_by_source": dict(alert_counts),
            },
            "quality": {
                "mean": round(statistics.mean(quality_list), 4) if quality_list else None,
                "recent_mean": round(statistics.mean(quality_list[-100:]), 4) if len(quality_list) >= 100 else None,
                "samples": len(quality_list),
            },
            "hallucination": {
                "rate": round(halluc_rate, 4),
                "mean_score": round(statistics.mean(halluc_scores[-100:]), 4) if halluc_scores else 0,
            },
            "cost": {
                "total_usd": round(sum(self.cost_by_endpoint.values()), 4),
                "by_endpoint": {
                    ep: round(c, 4) for ep, c in self.cost_by_endpoint.items()
                },
            },
            "endpoints": {
                ep: {"requests": self.requests_by_endpoint[ep]}
                for ep in self.requests_by_endpoint
            },
        }

Step 4: Complete simulation

def generate_request(
    i: int,
    base_time: float,
    phase: str,
    endpoints: list[str],
    models: dict[str, str],
) -> AIRequest:
    """Generates a simulated request according to the phase."""
    ep = random.choice(endpoints)
    model = models[ep]

    base_system = "You are a technical assistant. Respond concisely and precisely."
    context = "Python was created in 1991 by Guido van Rossum. It is an interpreted language."

    if phase == "normal":
        user_prompt = f"Technical question #{i}: " + "context " * random.randint(3, 8)
        output = "Python is an interpreted language created " + "information " * random.randint(10, 30)
        prompt_tok = random.randint(200, 500)
        comp_tok = random.randint(100, 300)
    elif phase == "drift":
        user_prompt = f"Question #{i}: " + "context " * random.randint(3, 8)
        output = "Yes. " + "ok " * random.randint(3, 8)
        prompt_tok = random.randint(200, 500)
        comp_tok = random.randint(20, 80)
    elif phase == "hallucination":
        user_prompt = f"Question #{i}: " + "context " * random.randint(3, 8)
        if random.random() < 0.30:
            output = f"Definitely Python was created in {random.randint(1980, 2010)} with exactly {random.random()*100:.1f}% adoption without a doubt"
        else:
            output = "Python is a language created approximately in 1991 by Guido van Rossum"
        prompt_tok = random.randint(200, 500)
        comp_tok = random.randint(100, 400)
    else:
        user_prompt = f"Question #{i}: " + "context " * random.randint(3, 8)
        output = "Python is an interpreted language created in 1991 by Guido van Rossum. " + "detail " * random.randint(10, 25)
        prompt_tok = random.randint(200, 500)
        comp_tok = random.randint(100, 300)

    cost_rate = 0.005 if model == "gpt-4o" else 0.0003
    cost = (prompt_tok + comp_tok) / 1000 * cost_rate

    return AIRequest(
        timestamp=base_time + i * 5,
        endpoint=ep, model=model, user_id=f"user_{random.randint(1, 20)}",
        system_prompt=base_system, user_prompt=user_prompt,
        output_text=output, prompt_tokens=prompt_tok,
        completion_tokens=comp_tok, total_tokens=prompt_tok + comp_tok,
        cost_usd=cost, latency_seconds=max(0.1, random.gauss(1.2, 0.3)),
        context=context,
    )


def run_simulation():
    random.seed(42)
    endpoints = ["/chat", "/analyze"]
    models = {"/chat": "gpt-4o-mini", "/analyze": "gpt-4o"}

    layer = AIMonitoringLayer(
        token_budgets={"/analyze": 200_000},
        outlier_multiplier=8.0,
        drift_sigma=2.0,
        hallucination_threshold=0.35,
    )

    print("=" * 65)
    print("PROJECT: AI-Specific Monitoring Layer — Complete Simulation")
    print("=" * 65)

    # --- PHASE 1: Normal traffic → baselines ---
    print("\n─── PHASE 1: Normal traffic (200 requests) → Baselines ───")
    for i in range(200):
        req = generate_request(i, 0.0, "normal", endpoints, models)
        result = layer.process(req)
        if result.alerts:
            for a in result.alerts:
                print(f"  [{a.severity.value:8}] {a.source:25} | {a.message[:50]}...")

    layer.establish_baselines(endpoints, list(set(models.values())))
    print(f"\n  Baselines established")
    print(f"  Requests: {layer.request_count}")
    print(f"  Phase 1 alerts: {len(layer.all_alerts)}")

    d = layer.dashboard()
    print(f"  Quality: {d['quality']['mean']}")
    print(f"  Hallucination rate: {d['hallucination']['rate']}")

    # --- PHASE 2: Model drift ---
    print("\n─── PHASE 2: Model drift (80 requests) → Short responses ───")
    alerts_before = len(layer.all_alerts)
    for i in range(80):
        req = generate_request(i, 1000.0, "drift", endpoints, models)
        result = layer.process(req)
        if result.alerts:
            for a in result.alerts:
                print(f"  [{a.severity.value:8}] {a.source:25} | {a.message[:50]}...")

    new_alerts = len(layer.all_alerts) - alerts_before
    print(f"\n  New alerts: {new_alerts}")
    d = layer.dashboard()
    print(f"  Quality: {d['quality'].get('recent_mean', 'N/A')}")

    # --- PHASE 3: Hallucination spike ---
    print("\n─── PHASE 3: Hallucination spike (80 requests) → Fabricated outputs ───")
    alerts_before = len(layer.all_alerts)
    for i in range(80):
        req = generate_request(i, 1400.0, "hallucination", endpoints, models)
        result = layer.process(req)
        if result.alerts:
            for a in result.alerts:
                print(f"  [{a.severity.value:8}] {a.source:25} | {a.message[:50]}...")

    new_alerts = len(layer.all_alerts) - alerts_before
    print(f"\n  New alerts: {new_alerts}")
    d = layer.dashboard()
    print(f"  Hallucination rate: {d['hallucination']['rate']}")

    # --- PHASE 4: Recovery ---
    print("\n─── PHASE 4: Recovery (100 requests) → Normal traffic ───")
    alerts_before = len(layer.all_alerts)
    for i in range(100):
        req = generate_request(i, 1800.0, "recovery", endpoints, models)
        result = layer.process(req)
        if result.alerts:
            for a in result.alerts:
                print(f"  [{a.severity.value:8}] {a.source:25} | {a.message[:50]}...")

    new_alerts = len(layer.all_alerts) - alerts_before
    print(f"\n  New alerts: {new_alerts}")

    # --- FINAL DASHBOARD ---
    print("\n" + "=" * 65)
    print("FINAL DASHBOARD — AI-Specific Monitoring Layer")
    print("=" * 65)
    final = layer.dashboard()
    print(json.dumps(final, indent=2, default=str))

    # --- ALERT SUMMARY ---
    print("\n" + "=" * 65)
    print("ALERT SUMMARY")
    print("=" * 65)
    by_source = defaultdict(list)
    for a in layer.all_alerts:
        by_source[a.source].append(a)

    for source, alerts in by_source.items():
        print(f"\n  {source}:")
        by_type = defaultdict(int)
        for a in alerts:
            by_type[a.alert_type] += 1
        for atype, count in sorted(by_type.items(), key=lambda x: x[1], reverse=True):
            print(f"    {atype:30} × {count}")

    print(f"\n  Total alerts: {len(layer.all_alerts)}")
    print(f"  Total requests: {layer.request_count}")
    print(f"  Alert rate: {len(layer.all_alerts) / layer.request_count * 100:.1f}%")

    # --- OTel ATTRIBUTES SAMPLE ---
    print("\n" + "=" * 65)
    print("OTEL ATTRIBUTES — Sample of the last request")
    print("=" * 65)
    if layer.all_results:
        last = layer.all_results[-1]
        for key, value in last.otel_attributes.items():
            print(f"  {key:40} = {value}")


run_simulation()

---

Completeness Checklist

Integrated System

• AIMonitoringLayer processes requests through all components
• Consolidated alerts from all sources
• Unified dashboard with the complete state
• OTel attributes for each request

Prompt Quality

• Length drift detection with baselines
• Template change detection with a hash
• Complexity tracking

Token Usage

• Tracking per endpoint and model
• Outlier detection (anomalous requests)
• Budget enforcement per endpoint

Drift Detection

• Output length baseline per endpoint/model
• Z-score comparison to detect changes
• Alerts when drift exceeds the threshold

Hallucination Detection

• Heuristic checks (overconfidence, specificity)
• Context overlap scoring
• Rate tracking over time
• Alerts when the rate rises

Dashboard and Metrics

• Overview (requests, alerts, cost)
• Quality metrics (mean, trend)
• Hallucination rate
• Cost by endpoint
• Alert summary by source

Simulation

• Normal phase → baselines with no spurious alerts
• Drift phase → drift detected
• Hallucination phase → high rate detected
• Recovery phase → system stabilizes
• Final dashboard shows the correct state

---

Comparison: Before and After

BEFORE (Without an AI monitoring layer)
─────────────────────────────────────────────
"How do you know your model is still giving good answers?"
→ "Users aren't complaining..."

"Would you detect it if OpenAI updates the model?"
→ "Eventually, when someone notices..."

"What is your hallucination rate?"
→ "We don't measure it..."

AFTER (With an AI monitoring layer)
─────────────────────────────────────────────
"How do you know your model is still giving good answers?"
→ "Quality score at 0.84, stable over the last 4 hours,
   hallucination rate at 3%, drift score at 0.15."

"Would you detect it if OpenAI updates the model?"
→ "The drift detector compares the output distribution every 50
   requests against the baseline. If output length changes > 2σ,
   it alerts in <10 minutes."

"What is your hallucination rate?"
→ "3.2% overall, 1.1% in /chat, 5.8% in /analyze.
   Alert threshold at 10%. Heuristic + context overlap."

---

Next Steps

This project leaves you with:

1. An integrated monitoring layer that understands the semantic dimension of your AI system
2. Established baselines for prompt quality, token usage, output distribution
3. Automated detection of drift, hallucinations, token outliers, prompt degradation
4. An AI-specific dashboard with metrics that only exist in AI systems
5. OTel attributes ready to integrate with your tracing backend

The connection with the rest of the guide:

What you have (Module 6)             Next step
──────────────────────────────────────────────────────
The AI monitoring layer detects      → Module 7: Debugging
anomalies: drift, hallucinations,      Why did the drift happen?
token outliers, prompt issues.         What caused the hallucinations?
                                       Investigation paths and runbooks.

AI-specific alerts configured        → Module 7: Detailed runbooks
with thresholds and baselines.         for each type of AI anomaly.

An AI dashboard with quality,        → Module 8: Integrate everything
hallucination, drift, tokens.          OTel + infra dashboards + AI
                                       monitoring in a production-ready stack.

---

Success Criteria

Your project is complete when:

• ✅ The monitoring layer processes requests and produces analysis of the 4 dimensions (prompts, tokens, drift, hallucinations)
• ✅ The baselines are established automatically with the first N requests
• ✅ The drift simulation generates drift_checker alerts
• ✅ The hallucination simulation generates hallucination_rate_high alerts
• ✅ Normal post-baseline traffic doesn't generate spurious alerts
• ✅ The dashboard shows metrics consistent with what happened in the simulation
• ✅ The OTel attributes are ready to be sent to the tracing backend
• ✅ You can explain each AI-specific metric and what action to take when it changes

---

Summary

• This project integrates all of Module 6: prompt tracking, token monitoring, drift detection, hallucination detection, custom metrics.
• The AIMonitoringLayer centralizes the processing: each request passes through all components and produces a unified result.
• Baselines are the foundation. Without a normal reference period, there's no anomaly detection.
• The simulation validates 4 phases: normal (baselines), drift (short responses), hallucination (fabricated outputs), recovery (stabilization).
• An AI-specific dashboard with metrics that only exist in AI systems: quality score, hallucination rate, drift score.
• OTel attributes in each request let you integrate with Jaeger/Tempo/Datadog for searching and correlation.
• The difference: you went from "we monitor latency and cost" to "we monitor what our LLM says, how well it says it, and whether it changed how it says it."

---

Additional Resources

1. OpenTelemetry Semantic Conventions for GenAI — Standard conventions for instrumenting LLMs
2. LangSmith Documentation — A complementary tool for debugging
3. Evidently AI — Open-source ML monitoring framework
4. Arize Phoenix — Open-source LLM observability
5. Ragas — Evaluation framework for RAG
6. Chen et al. — How is ChatGPT's behavior changing over time? — Academic evidence of model drift
7. Eugene Yan — LLM Patterns — Production patterns for LLM systems
8. Google SRE — Monitoring Distributed Systems — Monitoring principles

---