Monitoring & Detection: Catching Issues in Production

Overview

You can't fix what you don't know is broken. Monitoring and detection catch issues before users see them.

---

Safety Metrics

class SafetyMonitor:
    def track(self, agent_run: dict):
        """Track safety metrics for each run"""

        metrics = {
            'run_id': agent_run['id'],
            'violations': len(agent_run['policy_violations']),
            'injections_blocked': len(agent_run['blocked_inputs']),
            'harmful_outputs_filtered': len(agent_run['filtered_outputs']),
            'timeout_events': agent_run.get('timeouts', 0),
            'execution_cost': agent_run['cost'],
            'error_rate': agent_run['errors'] / agent_run['total_calls']
        }

        self.publish_metrics(metrics)

        # Alert on anomalies
        if metrics['violations'] > VIOLATION_THRESHOLD:
            self.alert('High policy violations detected')

        if metrics['error_rate'] > ERROR_THRESHOLD:
            self.alert('High error rate detected')

---

Anomaly Detection

class AnomalyDetector:
    def detect(self, metrics: dict) -> List[str]:
        """Detect anomalous behavior"""

        anomalies = []

        # Compare to baseline
        baseline = self.get_baseline(metrics['agent_id'])

        # Check for sudden spikes
        if metrics['violations'] > baseline['violations'] * 3:
            anomalies.append('Violation spike')

        if metrics['cost'] > baseline['cost'] * 5:
            anomalies.append('Unexpected cost increase')

        if metrics['error_rate'] > baseline['error_rate'] * 2:
            anomalies.append('Error rate spike')

        return anomalies

---

Alerts & Escalation

class AlertSystem:
    def alert(self, severity: str, message: str, context: dict):
        """Send alert based on severity"""

        if severity == 'CRITICAL':
            # Immediate escalation
            self.notify_security_team(message, context)
            self.disable_agent(context['agent_id'])

        elif severity == 'HIGH':
            # Alert ops team
            self.notify_ops_team(message, context)
            self.throttle_agent(context['agent_id'])

        elif severity == 'MEDIUM':
            # Log and monitor
            self.log_warning(message, context)

---

3 Warnings ⚠️

Warning 1: Alert Fatigue

# ❌ WRONG
if error_count > 1:
    alert()  # Too many alerts!

# Users ignore alerts → defeats monitoring

# ✅ RIGHT
if error_count > THRESHOLD:
    alert()  # Alert only on real issues
if error_count_unexpected():
    alert()  # Or on anomalies

Warning 2: Monitoring Blind Spots

# ❌ WRONG
monitor_api_calls()
# But don't monitor:
# - Token usage
# - Cost
# - Policy violations

# Discover issues too late

# ✅ RIGHT
monitor_api_calls()
monitor_token_usage()
monitor_cost()
monitor_policy_violations()
monitor_error_rates()

Warning 3: No Action on Alerts

# ❌ WRONG
if anomaly_detected():
    log_alert()  # Log it... and then what?

# Alerts are useless without response

# ✅ RIGHT
if anomaly_detected():
    log_alert()
    if severity == 'CRITICAL':
        disable_agent()
        notify_team()
    elif severity == 'HIGH':
        throttle_agent()
        investigate()

---

Last Updated: August 9, 2026