Monitoring ML Pipelines #

class SpeechRecognitionMonitor:
    def __init__(self):
        self.metrics = {
            'vad_false_positive_rate': [],
            'vad_false_negative_rate': [],
            'asr_accuracy_per_vad_decision': [],
            'end_to_end_latency_ms': [],
            'audio_quality_confidence': []
        }

    def monitor_vad_component(self, audio_segment, vad_prediction, ground_truth_speech):
        """Monitor Voice Activity Detection component."""
        if vad_prediction and not ground_truth_speech:
            self.metrics['vad_false_positive_rate'].append(1)
        elif not vad_prediction and ground_truth_speech:
            self.metrics['vad_false_negative_rate'].append(1)

    def monitor_downstream_effect(self, vad_output, asr_accuracy):
        """Track how VAD decisions affect ASR performance."""
        self.metrics['asr_accuracy_per_vad_decision'].append(asr_accuracy)

    def get_vad_performance(self):
        """Aggregate VAD component metrics."""
        total_decisions = len(self.metrics['vad_false_positive_rate']) + len(self.metrics['vad_false_negative_rate'])
        total_errors = len(self.metrics['vad_false_positive_rate']) + len(self.metrics['vad_false_negative_rate'])

        return {
            'error_rate': total_errors / total_decisions if total_decisions > 0 else 1.0,
            'false_positive_rate': len(self.metrics['vad_false_positive_rate']) / total_decisions if total_decisions > 0 else 1.0,
            'false_negative_rate': len(self.metrics['vad_false_negative_rate']) / total_decisions if total_decisions > 0 else 1.0
        }

# Usage
speech_monitor = SpeechRecognitionMonitor()

# In production pipeline
for audio, ground_truth, vad_pred, asr_acc in pipeline_data:
    speech_monitor.monitor_vad_component(audio, vad_pred, ground_truth)
    speech_monitor.monitor_downstream_effect(vad_pred, asr_acc)

    if speech_monitor.get_vad_performance()['error_rate'] > 0.1:  # 10% error threshold
        logger.warning("VAD component showing elevated error rates")

class RecommendationPipelineMonitor:
    def __init__(self):
        self.user_profile_metrics = []
        self.recommendation_metrics = []
        self.pipeline_latency = []

    def monitor_user_profiling(self, user_events, profile_confidence):
        """Track user profile generation quality."""
        self.user_profile_metrics.append({
            'events_processed': len(user_events),
            'profile_confidence': profile_confidence,
            'timestamp': datetime.now()
        })

    def monitor_recommendations(self, profile_data, recommendations, user_feedback):
        """Monitor recommendation quality and relevance."""
        precision = len(set(recommendations) & set(user_feedback)) / len(recommendations) if recommendations else 0

        self.recommendation_metrics.append({
            'recommendations_made': len(recommendations),
            'precision_at_k': precision,
            'profile_age_hours': (datetime.now() - profile_data.get('last_updated', datetime.now())).total_seconds() / 3600,
            'timestamp': datetime.now()
        })

    def detect_cascading_failure(self):
        """Detect if profile degradation affects recommendations."""
        recent_profiles = [m['profile_confidence'] for m in self.user_profile_metrics[-100:]]
        recent_recommendations = [m['precision_at_k'] for m in self.recommendation_metrics[-100:]]

        # Correlation between profile confidence and recommendation quality
        if len(recent_profiles) > 10 and len(recent_recommendations) > 10:
            correlation = np.corrcoef(recent_profiles, recent_recommendations)[0, 1]
            if correlation < 0.3:  # Weak correlation suggests profile issues
                return True
        return False

# Alert setup
rec_monitor = RecommendationPipelineMonitor()
# ... monitoring logic ...
if rec_monitor.detect_cascading_failure():
    trigger_alert("Profile degradation affecting recommendation quality")

def create_pipeline_health_dashboard(pipeline_monitor, components):
    """Create comprehensive pipeline monitoring dashboard."""
    import plotly.graph_objects as go
    from plotly.subplots import make_subplots

    fig = make_subplots(
        rows=2, cols=2,
        subplot_titles=('Component Latencies', 'Error Correlation',
                       'Data Flow Rates', 'Alert History'),
        specs=[[{'type': 'histogram'}, {'type': 'scatter'}],
               [{'type': 'scatter'}, {'type': 'table'}]]
    )

    # Component latency histogram
    for component in components:
        latencies = pipeline_monitor.get_component_metrics(component)['latency_samples'][-1000:]
        fig.add_trace(go.Histogram(x=latencies, name=component), row=1, col=1)

    # Error correlation scatter
    error_rates = [pipeline_monitor.get_component_metrics(c)['error_rate'] for c in components]
    latencies = [pipeline_monitor.get_component_metrics(c)['avg_latency'] for c in components]
    fig.add_trace(go.Scatter(x=error_rates, y=latencies, mode='markers',
                            text=components, name='Error vs Latency'), row=1, col=2)

    # Data flow rates over time
    timestamps = [m['timestamp'] for m in pipeline_monitor.get_flow_metrics()]
    rates = [m['records_per_second'] for m in pipeline_monitor.get_flow_metrics()]
    fig.add_trace(go.Scatter(x=timestamps, y=rates, mode='lines',
                            name='Data Flow Rate'), row=2, col=1)

    # Alert table
    alerts = pipeline_monitor.get_recent_alerts()
    fig.add_trace(go.Table(
        header=dict(values=['Timestamp', 'Component', 'Alert Type', 'Severity']),
        cells=dict(values=[[str(a['timestamp']) for a in alerts],
                          [a['component'] for a in alerts],
                          [a['type'] for a in alerts],
                          [a['severity'] for a in alerts]])
    ), row=2, col=2)

    fig.update_layout(height=800, title_text="Pipeline Health Dashboard")
    fig.show()