Monitoring Ml Models

Ml Monitoring

Monitor ML model performance, data drift, LLM observability, and production metrics

Monitor machine learning models in production for performance degradation, data drift, latency issues, and cost tracking.

Process

1. Review the task requirements.
2. Apply the skill's methodology.
3. Validate the output against the defined criteria.

Step 1: Model Performance Monitoring

# performance_monitoring.py
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List
from dataclasses import dataclass

@dataclass
class PredictionRecord:
    timestamp: datetime
    input: str
    prediction: str
    actual: str = None
    latency_ms: float = None
    model_version: str = None

class ModelPerformanceMonitor:
    """Monitor model performance metrics."""

    def __init__(self, window_size: int = 1000):
        self.window_size = window_size
        self.predictions: List[PredictionRecord] = []
        self.metrics_history = []

    def record_prediction(self, record: PredictionRecord):
        """Record a prediction."""
        self.predictions.append(record)

        # Keep only recent predictions
        if len(self.predictions) > self.window_size:
            self.predictions = self.predictions[-self.window_size:]

    def calculate_accuracy(self, window_minutes: int = 60) -> float:
        """Calculate accuracy over time window."""
        cutoff = datetime.now() - timedelta(minutes=window_minutes)
        recent = [
            p for p in self.predictions
            if p.timestamp >= cutoff and p.actual is not None
        ]

        if not recent:
            return None

        correct = sum(1 for p in recent if p.prediction == p.actual)
        return correct / len(recent)

    def detect_accuracy_drift(
        self,
        baseline_accuracy: float,
        threshold: float = 0.05,
    ) -> bool:
        """Detect if accuracy has drifted significantly."""
        current_accuracy = self.calculate_accuracy()

        if current_accuracy is None:
            return False

        drift = baseline_accuracy - current_accuracy
        return drift > threshold

    def get_latency_stats(self) -> Dict:
        """Get latency statistics."""
        latencies = [p.latency_ms for p in self.predictions if p.latency_ms]

        if not latencies:
            return {}

        return {
            "mean": np.mean(latencies),
            "median": np.median(latencies),
            "p95": np.percentile(latencies, 95),
            "p99": np.percentile(latencies, 99),
            "max": np.max(latencies),
        }

    def get_performance_report(self) -> Dict:
        """Generate performance report."""
        return {
            "total_predictions": len(self.predictions),
            "accuracy": self.calculate_accuracy(),
            "latency_stats": self.get_latency_stats(),
            "predictions_per_minute": self._calculate_throughput(),
        }

    def _calculate_throughput(self) -> float:
        """Calculate predictions per minute."""
        if len(self.predictions) < 2:
            return 0.0

        time_span = (
            self.predictions[-1].timestamp - self.predictions[0].timestamp
        ).total_seconds() / 60

        if time_span == 0:
            return 0.0

        return len(self.predictions) / time_span

Step 2: Data Drift Detection with Evidently AI

# data_drift_detection.py
from evidently import ColumnMapping
from evidently.report import Report
from evidently.metrics import (
    DataDriftTable,
    DatasetDriftMetric,
    ColumnDriftMetric,
)
import pandas as pd

class DataDriftMonitor:
    """Monitor data drift using Evidently AI."""

    def __init__(self, reference_data: pd.DataFrame):
        self.reference_data = reference_data
        self.column_mapping = ColumnMapping(
            target=None,
            prediction="prediction",
            numerical_features=list(reference_data.select_dtypes(include=[np.number]).columns),
            categorical_features=list(reference_data.select_dtypes(include=["object"]).columns),
        )

    def check_drift(
        self,
        current_data: pd.DataFrame,
        threshold: float = 0.1,
    ) -> Dict:
        """Check for data drift."""

        # Create drift report
        report = Report(metrics=[
            DatasetDriftMetric(),
            DataDriftTable(),
        ])

        report.run(
            reference_data=self.reference_data,
            current_data=current_data,
            column_mapping=self.column_mapping,
        )

        # Extract drift information
        result = report.as_dict()
        dataset_drift = result["metrics"][0]["result"]["dataset_drift"]
        drift_score = result["metrics"][0]["result"]["drift_score"]

        # Get column-level drift
        drifted_columns = []
        if len(result["metrics"]) > 1:
            for col_result in result["metrics"][1]["result"]["drift_by_columns"]:
                if col_result["drift_detected"]:
                    drifted_columns.append({
                        "column": col_result["column_name"],
                        "drift_score": col_result["drift_score"],
                    })

        return {
            "dataset_drift_detected": dataset_drift,
            "drift_score": drift_score,
            "drifted_columns": drifted_columns,
            "threshold": threshold,
            "action_required": drift_score > threshold,
        }

    def generate_drift_report_html(
        self,
        current_data: pd.DataFrame,
        output_path: str = "drift_report.html",
    ):
        """Generate HTML drift report."""
        report = Report(metrics=[
            DatasetDriftMetric(),
            DataDriftTable(),
            ColumnDriftMetric(column_name="feature_1"),
        ])

        report.run(
            reference_data=self.reference_data,
            current_data=current_data,
            column_mapping=self.column_mapping,
        )

        report.save_html(output_path)
        return output_path

# Usage
reference_data = pd.read_csv("reference_data.csv")
monitor = DataDriftMonitor(reference_data)

# Check drift periodically
current_data = pd.read_csv("current_data.csv")
drift_result = monitor.check_drift(current_data, threshold=0.15)

if drift_result["action_required"]:
    print(f"Drift detected! Score: {drift_result['drift_score']}")
    monitor.generate_drift_report_html(current_data)

Step 3: LLM Observability with Langfuse

# langfuse_monitoring.py
from langfuse import Langfuse
from langfuse.decorators import langfuse_context, observe
from typing import Optional
import os

# Initialize Langfuse
langfuse = Langfuse(
    public_key=os.getenv("LANGFUSE_PUBLIC_KEY"),
    secret_key=os.getenv("LANGFUSE_SECRET_KEY"),
    host=os.getenv("LANGFUSE_HOST", "https://cloud.langfuse.com"),
)

class LLMMonitor:
    """Monitor LLM usage and performance."""

    @observe(name="llm_generation")
    def generate_with_monitoring(
        self,
        prompt: str,
        model: str,
        **kwargs,
    ) -> str:
        """Generate with Langfuse tracking."""

        # Set trace metadata
        langfuse_context.update_current_trace(
            name="llm_inference",
            user_id="user_123",
            session_id="session_456",
            metadata={
                "model": model,
                "environment": "production",
            },
        )

        # Set generation parameters
        langfuse_context.update_current_generation(
            model=model,
            model_parameters={
                "temperature": kwargs.get("temperature", 0.7),
                "max_tokens": kwargs.get("max_tokens", 512),
            },
            input=prompt,
        )

        # Simulate LLM call
        response = self._call_llm(prompt, model, **kwargs)

        # Record output
        langfuse_context.update_current_generation(
            output=response,
            usage={
                "prompt_tokens": self._count_tokens(prompt),
                "completion_tokens": self._count_tokens(response),
                "total_tokens": self._count_tokens(prompt) + self._count_tokens(response),
            },
        )

        return response

    def _call_llm(self, prompt: str, model: str, **kwargs) -> str:
        """Call LLM (placeholder)."""
        # Replace with actual LLM call
        return f"Response to: {prompt}"

    def _count_tokens(self, text: str) -> int:
        """Count tokens (simplified)."""
        return len(text.split()) * 1.3  # Rough estimate

# Track custom events
def track_custom_event(event_name: str, metadata: dict):
    """Track custom event."""
    langfuse.event(
        name=event_name,
        metadata=metadata,
    )

# Score generations
def score_generation(
    trace_id: str,
    score_name: str,
    value: float,
    comment: Optional[str] = None,
):
    """Score a generation."""
    langfuse.score(
        trace_id=trace_id,
        name=score_name,
        value=value,
        comment=comment,
    )

Step 4: A/B Testing Setup

# ab_testing.py
import random
from typing import Dict, List
from dataclasses import dataclass
from datetime import datetime

@dataclass
class ABTestResult:
    model_version: str
    prompt: str
    response: str
    latency_ms: float
    user_rating: float = None
    timestamp: datetime = None

class ABTestManager:
    """Manage A/B testing between model versions."""

    def __init__(self, variants: Dict[str, float]):
        """
        Args:
            variants: Dict mapping model version to traffic percentage
        """
        self.variants = variants
        self.results: List[ABTestResult] = []
        self._validate_traffic_split()

    def _validate_traffic_split(self):
        """Validate traffic split sums to 100%."""
        total = sum(self.variants.values())
        if abs(total - 1.0) > 0.01:
            raise ValueError(f"Traffic split must sum to 1.0, got {total}")

    def select_variant(self) -> str:
        """Select model variant based on traffic split."""
        rand = random.random()
        cumulative = 0.0

        for variant, percentage in self.variants.items():
            cumulative += percentage
            if rand <= cumulative:
                return variant

        return list(self.variants.keys())[-1]

    def record_result(self, result: ABTestResult):
        """Record A/B test result."""
        result.timestamp = datetime.now()
        self.results.append(result)

    def get_variant_stats(self, variant: str) -> Dict:
        """Get statistics for a variant."""
        variant_results = [
            r for r in self.results if r.model_version == variant
        ]

        if not variant_results:
            return {}

        latencies = [r.latency_ms for r in variant_results]
        ratings = [r.user_rating for r in variant_results if r.user_rating]

        return {
            "count": len(variant_results),
            "avg_latency_ms": sum(latencies) / len(latencies),
            "avg_rating": sum(ratings) / len(ratings) if ratings else None,
            "p95_latency_ms": sorted(latencies)[int(len(latencies) * 0.95)],
        }

    def compare_variants(self) -> Dict:
        """Compare all variants."""
        comparison = {}

        for variant in self.variants.keys():
            comparison[variant] = self.get_variant_stats(variant)

        return comparison

# Usage
ab_manager = ABTestManager({
    "model-v1": 0.5,
    "model-v2": 0.5,
})

variant = ab_manager.select_variant()
result = ABTestResult(
    model_version=variant,
    prompt="Hello",
    response="Hi there",
    latency_ms=150.0,
    user_rating=4.5,
)
ab_manager.record_result(result)

Step 5: Token Usage and Cost Tracking

# cost_tracking.py
from typing import Dict, List
from datetime import datetime, timedelta
from dataclasses import dataclass

@dataclass
class TokenUsage:
    timestamp: datetime
    model: str
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    cost_usd: float

# Pricing per 1M tokens (example)
MODEL_PRICING = {
    "gpt-4": {"input": 30.0, "output": 60.0},
    "gpt-3.5-turbo": {"input": 0.5, "output": 1.5},
    "claude-3-opus": {"input": 15.0, "output": 75.0},
    "claude-3-sonnet": {"input": 3.0, "output": 15.0},
}

class CostTracker:
    """Track token usage and costs."""

    def __init__(self):
        self.usage_history: List[TokenUsage] = []

    def record_usage(
        self,
        model: str,
        prompt_tokens: int,
        completion_tokens: int,
    ):
        """Record token usage."""
        total_tokens = prompt_tokens + completion_tokens

        # Calculate cost
        pricing = MODEL_PRICING.get(model, {"input": 0, "output": 0})
        cost = (
            (prompt_tokens / 1_000_000) * pricing["input"] +
            (completion_tokens / 1_000_000) * pricing["output"]
        )

        usage = TokenUsage(
            timestamp=datetime.now(),
            model=model,
            prompt_tokens=prompt_tokens,
            completion_tokens=completion_tokens,
            total_tokens=total_tokens,
            cost_usd=cost,
        )

        self.usage_history.append(usage)

    def get_daily_cost(self, date: datetime = None) -> float:
        """Get total cost for a day."""
        if date is None:
            date = datetime.now()

        start = date.replace(hour=0, minute=0, second=0, microsecond=0)
        end = start + timedelta(days=1)

        daily_usage = [
            u for u in self.usage_history
            if start <= u.timestamp < end
        ]

        return sum(u.cost_usd for u in daily_usage)

    def get_model_breakdown(self, days: int = 7) -> Dict:
        """Get cost breakdown by model."""
        cutoff = datetime.now() - timedelta(days=days)
        recent = [u for u in self.usage_history if u.timestamp >= cutoff]

        breakdown = {}
        for usage in recent:
            if usage.model not in breakdown:
                breakdown[usage.model] = {
                    "cost": 0.0,
                    "tokens": 0,
                    "requests": 0,
                }

            breakdown[usage.model]["cost"] += usage.cost_usd
            breakdown[usage.model]["tokens"] += usage.total_tokens
            breakdown[usage.model]["requests"] += 1

        return breakdown

    def check_budget_alert(self, daily_budget: float) -> bool:
        """Check if daily budget exceeded."""
        daily_cost = self.get_daily_cost()
        return daily_cost > daily_budget

# Usage
tracker = CostTracker()
tracker.record_usage("gpt-4", prompt_tokens=1000, completion_tokens=500)
daily_cost = tracker.get_daily_cost()
breakdown = tracker.get_model_breakdown(days=7)

Step 6: Alerting Rules

# alerting.py
from typing import Callable, Optional
from dataclasses import dataclass
import smtplib
from email.mime.text import MIMEText

@dataclass
class AlertRule:
    name: str
    condition: Callable[[], bool]
    severity: str  # "critical", "warning", "info"
    message: str
    enabled: bool = True

class AlertManager:
    """Manage alerting rules."""

    def __init__(self):
        self.rules: List[AlertRule] = []
        self.alert_history = []

    def add_rule(self, rule: AlertRule):
        """Add alert rule."""
        self.rules.append(rule)

    def check_rules(self):
        """Check all alert rules."""
        triggered = []

        for rule in self.rules:
            if not rule.enabled:
                continue

            try:
                if rule.condition():
                    triggered.append(rule)
                    self._send_alert(rule)
            except Exception as e:
                print(f"Error checking rule {rule.name}: {e}")

        return triggered

    def _send_alert(self, rule: AlertRule):
        """Send alert (placeholder)."""
        # Implement email, Slack, PagerDuty, etc.
        print(f"ALERT [{rule.severity.upper()}]: {rule.name} - {rule.message}")
        self.alert_history.append({
            "rule": rule.name,
            "timestamp": datetime.now(),
            "severity": rule.severity,
        })

# Usage
alert_manager = AlertManager()

# Add rules
alert_manager.add_rule(AlertRule(
    name="accuracy_drift",
    condition=lambda: monitor.detect_accuracy_drift(0.85, threshold=0.1),
    severity="critical",
    message="Model accuracy has dropped below threshold",
))

alert_manager.add_rule(AlertRule(
    name="high_latency",
    condition=lambda: monitor.get_latency_stats().get("p95", 0) > 1000,
    severity="warning",
    message="P95 latency exceeds 1 second",
))

alert_manager.add_rule(AlertRule(
    name="budget_exceeded",
    condition=lambda: tracker.check_budget_alert(daily_budget=100.0),
    severity="critical",
    message="Daily budget exceeded",
))

# Check rules periodically
triggered = alert_manager.check_rules()

Output

• Model performance dashboards
• Data drift detection reports
• LLM observability traces
• A/B test comparison results
• Cost tracking and budget alerts

Best Practices

• Monitor accuracy, latency, and throughput continuously
• Set up data drift detection with Evidently
• Use Langfuse for LLM observability
• Track token usage and costs per model
• Set up alerting for critical metrics
• Run A/B tests before full rollout
• Keep reference datasets for drift comparison

Related

• Skill: model-serving
• Skill: deploying-ml-models
• Skill: optimizing-ai-costs

When to Use

This skill should be used when strict adherence to the defined process is required.

Prerequisites

• Basic understanding of the agent factory context.
• Access to the necessary tools and resources.