""" Alerting Engine Alert generation and management for drift detection. Coordinates alert creation, storage, and resolution. """ import uuid from collections import defaultdict from datetime import datetime from typing import Dict, List, Optional from backend.logging.logger import get_logger from .schemas import ( Alert, AlertSeverity, AlertSummary, AlertType, DriftDetectionResult, ) class AlertManager: """ Alert management engine. Handles: - Alert creation from drift detection - Alert storage and retrieval - Alert resolution - Alert aggregation for dashboard """ def __init__(self) -> None: """Initialize alert manager.""" self.logger = get_logger(__name__) # Active alerts by type self._active_alerts: Dict[AlertType, Alert] = {} # All alerts (including resolved) self._all_alerts: List[Alert] = [] # Alerts by model version self._alerts_by_model: Dict[str, List[Alert]] = defaultdict(list) def create_alert_from_drift( self, drift_result: DriftDetectionResult, model_version: str, ) -> Alert: """ Create an alert from drift detection result. Args: drift_result: Drift detection result model_version: Model version being monitored Returns: Created alert """ # Determine alert type from metric name alert_type = self._metric_to_alert_type(drift_result.metric_name) # Check if we already have an active alert for this metric if alert_type in self._active_alerts: existing_alert = self._active_alerts[alert_type] # Update existing alert if severity increased if drift_result.severity.value > existing_alert.severity.value: self.logger.info( "Updating existing alert with increased severity", alert_type=alert_type, old_severity=existing_alert.severity, new_severity=drift_result.severity, ) # Create new alert with updated info alert = Alert( id=str(uuid.uuid4()), alert_type=alert_type, severity=drift_result.severity, model_version=model_version, metric_name=drift_result.metric_name, baseline_value=drift_result.baseline_value, current_value=drift_result.live_value, drift_magnitude=drift_result.drift_magnitude, threshold=drift_result.threshold, timestamp=datetime.utcnow(), is_resolved=False, ) self._active_alerts[alert_type] = alert self._all_alerts.append(alert) self._alerts_by_model[model_version].append(alert) return alert # Otherwise, just update the current value and timestamp existing_alert.current_value = drift_result.live_value existing_alert.drift_magnitude = drift_result.drift_magnitude existing_alert.timestamp = datetime.utcnow() return existing_alert # Create new alert alert = Alert( id=str(uuid.uuid4()), alert_type=alert_type, severity=drift_result.severity, model_version=model_version, metric_name=drift_result.metric_name, baseline_value=drift_result.baseline_value, current_value=drift_result.live_value, drift_magnitude=drift_result.drift_magnitude, threshold=drift_result.threshold, timestamp=datetime.utcnow(), is_resolved=False, ) # Store alert self._active_alerts[alert_type] = alert self._all_alerts.append(alert) self._alerts_by_model[model_version].append(alert) self.logger.warning( "Alert created", alert_id=alert.id, alert_type=alert_type.value, severity=drift_result.severity.value, model_version=model_version, drift_magnitude=drift_result.drift_magnitude, ) return alert def resolve_alert( self, alert_type: AlertType, model_version: Optional[str] = None, ) -> bool: """ Resolve an active alert. Args: alert_type: Type of alert to resolve model_version: Optional model version to filter by Returns: True if alert was resolved, False if not found """ if alert_type not in self._active_alerts: return False alert = self._active_alerts[alert_type] # Check model version if specified if model_version and alert.model_version != model_version: return False # Mark as resolved alert.is_resolved = True alert.resolved_at = datetime.utcnow() # Remove from active alerts del self._active_alerts[alert_type] self.logger.info( "Alert resolved", alert_id=alert.id, alert_type=alert_type.value, model_version=alert.model_version, ) return True def resolve_all_alerts(self, model_version: Optional[str] = None) -> int: """ Resolve all active alerts. Args: model_version: Optional model version to filter by Returns: Number of alerts resolved """ count = 0 if model_version: # Resolve specific model's alerts alert_types_to_resolve = [ at for at, alert in self._active_alerts.items() if alert.model_version == model_version ] for alert_type in alert_types_to_resolve: if self.resolve_alert(alert_type, model_version): count += 1 else: # Resolve all count = len(self._active_alerts) for alert_type in list(self._active_alerts.keys()): self.resolve_alert(alert_type) return count def get_active_alerts(self) -> List[Alert]: """ Get all active alerts. Returns: List of active alerts """ return list(self._active_alerts.values()) def get_alerts_by_model( self, model_version: str, include_resolved: bool = False, ) -> List[Alert]: """ Get alerts for a specific model version. Args: model_version: Model version to filter by include_resolved: Include resolved alerts Returns: List of alerts """ alerts = self._alerts_by_model.get(model_version, []) if not include_resolved: alerts = [a for a in alerts if not a.is_resolved] return alerts def get_alert_summary(self, limit: int = 10) -> AlertSummary: """ Get summary of alerts for dashboard. Args: limit: Number of recent alerts to include Returns: Alert summary """ active_alerts = self.get_active_alerts() # Count by severity critical_count = sum(1 for a in active_alerts if a.severity == AlertSeverity.CRITICAL) high_count = sum(1 for a in active_alerts if a.severity == AlertSeverity.HIGH) medium_count = sum(1 for a in active_alerts if a.severity == AlertSeverity.MEDIUM) low_count = sum(1 for a in active_alerts if a.severity == AlertSeverity.LOW) # Get recent alerts (all, not just active) recent_alerts = sorted( self._all_alerts, key=lambda a: a.timestamp, reverse=True, )[:limit] return AlertSummary( total_alerts=len(active_alerts), critical_count=critical_count, high_count=high_count, medium_count=medium_count, low_count=low_count, recent_alerts=recent_alerts, ) def get_alert(self, alert_id: str) -> Optional[Alert]: """ Get a specific alert by ID. Args: alert_id: Alert ID Returns: Alert or None if not found """ for alert in self._all_alerts: if alert.id == alert_id: return alert return None def _metric_to_alert_type(self, metric_name: str) -> AlertType: """Convert metric name to alert type.""" mapping = { "hallucination": AlertType.HALLUCINATION_DRIFT, "toxicity": AlertType.TOXICITY_DRIFT, "bias": AlertType.BIAS_DRIFT, "confidence": AlertType.CONFIDENCE_COLLAPSE, "robustness": AlertType.ROBUSTNESS_COLLAPSE, } return mapping.get(metric_name, AlertType.HALLUCINATION_DRIFT) def clear_all_alerts(self) -> None: """Clear all alerts (for testing).""" self._active_alerts.clear() self._all_alerts.clear() self._alerts_by_model.clear() # Global alert manager instance _alert_manager: Optional[AlertManager] = None def get_alert_manager() -> AlertManager: """ Get the global alert manager instance. Returns: AlertManager singleton """ global _alert_manager if _alert_manager is None: _alert_manager = AlertManager() return _alert_manager __all__ = [ "AlertManager", "Alert", "AlertSummary", "AlertType", "AlertSeverity", "get_alert_manager", ]