Beginner

Building Your First Automation

A comprehensive guide to building your first automation within ai automation fundamentals. Covers core concepts, practical implementation, code examples, and best practices.

Hands-On: Building Your First AI Automation

Theory is valuable, but nothing replaces the experience of building a working automation from scratch. In this lesson, we will build a complete, practical AI automation that you can adapt to your own use cases. We will automate the process of monitoring a data source, detecting anomalies, classifying issues, and sending intelligent notifications.

This project demonstrates several key AI automation concepts: scheduled execution, data processing, ML-based classification, conditional logic, and integration with external systems.

Project Architecture

Our automation will follow this flow:

  1. Fetch metrics data from a simulated data source on a schedule
  2. Run anomaly detection to identify unusual patterns
  3. Classify the type and severity of any anomalies found
  4. Generate a human-readable summary using templates
  5. Route notifications to the appropriate channel based on severity
Python - Complete Automation
import numpy as np
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import List, Optional
import json

@dataclass
class MetricPoint:
    timestamp: datetime
    value: float
    metric_name: str

@dataclass
class Anomaly:
    metric: MetricPoint
    severity: str  # low, medium, high, critical
    category: str  # spike, drop, trend, pattern
    z_score: float
    message: str

class AnomalyDetector:
    """Simple statistical anomaly detection."""

    def __init__(self, window_size: int = 30, threshold: float = 2.5):
        self.window_size = window_size
        self.threshold = threshold
        self.history: List[float] = []

    def detect(self, point: MetricPoint) -> Optional[Anomaly]:
        self.history.append(point.value)
        if len(self.history) < self.window_size:
            return None

        window = self.history[-self.window_size:]
        mean = np.mean(window[:-1])
        std = np.std(window[:-1])

        if std == 0:
            return None

        z_score = abs(point.value - mean) / std

        if z_score > self.threshold:
            severity = self._classify_severity(z_score)
            category = "spike" if point.value > mean else "drop"
            return Anomaly(
                metric=point,
                severity=severity,
                category=category,
                z_score=round(z_score, 2),
                message=f"{point.metric_name}: {category} detected "
                        f"(value={point.value:.1f}, expected={mean:.1f})"
            )
        return None

    def _classify_severity(self, z_score: float) -> str:
        if z_score > 5.0: return "critical"
        elif z_score > 4.0: return "high"
        elif z_score > 3.0: return "medium"
        else: return "low"

class NotificationRouter:
    """Route notifications based on severity."""

    def route(self, anomaly: Anomaly) -> dict:
        routing = {
            "critical": {"channel": "pagerduty", "escalate": True},
            "high": {"channel": "slack_urgent", "escalate": False},
            "medium": {"channel": "slack_alerts", "escalate": False},
            "low": {"channel": "email_digest", "escalate": False},
        }
        destination = routing.get(anomaly.severity, routing["low"])
        return {
            "destination": destination,
            "payload": {
                "severity": anomaly.severity,
                "message": anomaly.message,
                "timestamp": anomaly.metric.timestamp.isoformat(),
                "z_score": anomaly.z_score,
            }
        }

# Run the automation
detector = AnomalyDetector(window_size=30, threshold=2.5)
router = NotificationRouter()

# Simulate metric stream with an anomaly
np.random.seed(42)
normal_values = np.random.normal(100, 5, 35)
normal_values[32] = 150  # Inject anomaly

for i, val in enumerate(normal_values):
    point = MetricPoint(
        timestamp=datetime.now() - timedelta(minutes=35-i),
        value=val,
        metric_name="api_latency_ms"
    )
    anomaly = detector.detect(point)
    if anomaly:
        notification = router.route(anomaly)
        print(json.dumps(notification, indent=2, default=str))

Key Design Patterns

This simple automation demonstrates several patterns you will use repeatedly:

1. The Detector-Classifier-Router Pattern

Most AI automations follow this three-stage pattern: detect an event or condition, classify its type and urgency, and route it to the appropriate handler. This separation of concerns makes each component independently testable and replaceable.

2. Graceful Degradation

Notice how the anomaly detector requires a minimum window of historical data before making predictions. During the warmup period, it simply returns None rather than making unreliable predictions. Always design your automations to degrade gracefully when conditions are not ideal.

3. Configuration-Driven Behavior

Thresholds, routing rules, and window sizes are parameterized rather than hardcoded. This makes it easy to tune the automation without changing code.

💡
Test with real data early: Simulated data is useful for development, but test with real data as soon as possible. Real data has quirks, gaps, and patterns that simulations miss. You will discover edge cases that your automation needs to handle.

Extending the Automation

To make this production-ready, you would add:

  • Persistent state: Store historical data in a database rather than in-memory lists
  • Scheduling: Use Airflow, Prefect, or a cron job to run the detection on a regular interval
  • Multiple metrics: Run detectors for CPU, memory, latency, error rate, and business metrics simultaneously
  • Correlation: Detect when multiple metrics are anomalous simultaneously, which often indicates a systemic issue
  • Feedback loop: Track whether alerts were true positives and use that feedback to tune thresholds automatically
⚠
Alert fatigue kills automation: If your automation sends too many false positive alerts, people will start ignoring them. Tune your thresholds aggressively and implement alert deduplication and suppression from the start.