Tutorials Data

Automated Metric Alerts with AI: Catch Problems Before Users Do

What is anomaly-based metric alerting and why do static thresholds fail for product metrics?

Anomaly-based metric alerting detects deviations from a dynamically computed baseline rather than comparing against a fixed threshold. Static thresholds work for stable infrastructure metrics like CPU or disk, but fail for product metrics because of seasonality (e-commerce conversion on Monday morning vs. Friday evening differs by an order of magnitude), growth trends (absolute error counts grow with traffic), and distribution skew (a 200ms mean can coexist with a 4-second p99). Anomaly detection — Z-score for stable metrics, STL decomposition for periodic ones, Isolation Forest for multi-dimensional patterns — produces alert rules that adapt to the product's actual behavior rather than the engineer's best guess at signup.

TL;DR

  • -Most production failures are discovered by users, not the engineering team — the average gap between a breakage and the first support ticket is tens of minutes; automated anomaly detection closes it to 2–5 minutes.
  • -Static thresholds fail on product metrics due to three causes: seasonality, growth trends, and distribution skew (mean vs. p99) — use Z-score for stable metrics, STL decomposition for periodic ones, and Isolation Forest for multi-dimensional anomalies.
  • -AI prompts generate complete YAML alerting configurations from a product description, cutting initial setup from days to hours — but every generated rule must be validated against historical data before deploying to production.
  • -The 12 production alerting rules in this article cover the most common failure modes: mandatory cooldowns, alerting on symptoms not causes, grouping by service, pending periods, monthly threshold review, and a dead man's switch for the alerting system itself.
  • -LLM-powered products need a separate alerting layer beyond HTTP status codes — quality score drops and cost spikes are invisible to standard infrastructure monitoring and require Langfuse-to-Prometheus export.

Most production failures are discovered by users, not the engineering team. The gap between something breaking and the first support ticket: tens of minutes. In that window, the product bleeds conversions, users leave, NPS drops. Automated metric alerts shrink detection to 2–5 minutes.

This article covers the full stack: from anomaly detection to the exact prompts that generate alerting rules. Three tools — Grafana, PostHog, custom solutions — with working configurations and the mistakes worth avoiding.

Why static thresholds don’t work for product metrics

The classic approach: “if error rate > 5%, send an alert.” This works fine for infrastructure metrics (CPU, memory, disk). For product metrics, it breaks down in three ways.

Seasonality. E-commerce conversion on Monday morning vs. Friday evening differs by an order of magnitude. A static threshold either misses real problems or fires false positives daily.

Trends. The product grows. The absolute number of errors grows with traffic. A threshold you set a month ago is already stale.

Distribution. Average API response time = 200ms. But p99 = 4 seconds. A static threshold on the mean will completely miss degradation that’s hitting 1% of your most active users.

Anomaly detection handles all three. Instead of a fixed number, the system builds a dynamic baseline and reacts to deviations from it.

Anomaly detection: three approaches

Statistical (Z-score, IQR)

The simplest approach. Compute the mean and standard deviation over a sliding window. If the current value deviates by more than N standard deviations, the alert fires.

import numpy as np

def detect_anomaly_zscore(values: list[float], threshold: float = 3.0) -> bool:
    """Z-score anomaly detection on a sliding window."""
    if len(values) < 30:
        return False

    mean = np.mean(values[:-1])
    std = np.std(values[:-1])

    if std == 0:
        return False

    z_score = abs(values[-1] - mean) / std
    return z_score > threshold

Simple to implement, minimal resources. The catch: it ignores seasonality. Use it for metrics with relatively stable behavior — error rate, latency p99.

Seasonal decomposition (STL, Prophet)

Breaks the time series into three components: trend, seasonality, residual. The alert fires on anomalies in the residual.

from statsmodels.tsa.seasonal import STL

def detect_anomaly_stl(
    series,  # pd.Series with DatetimeIndex
    period: int = 24,  # hourly seasonality
    threshold: float = 3.0
) -> bool:
    """STL decomposition + Z-score on the residual."""
    stl = STL(series, period=period, robust=True)
    result = stl.fit()

    residual = result.resid
    mean_r = residual.mean()
    std_r = residual.std()

    if std_r == 0:
        return False

    latest_residual = residual.iloc[-1]
    z_score = abs(latest_residual - mean_r) / std_r
    return z_score > threshold

The right choice for metrics with strong periodicity: DAU, conversion, revenue. Needs at least 2–3 full cycles of data to train.

ML-based (Isolation Forest, Autoencoders)

For multi-dimensional anomalies — where each metric looks normal in isolation, but the combination signals a problem.

from sklearn.ensemble import IsolationForest

def build_anomaly_detector(features_matrix):
    """Isolation Forest for multi-dimensional anomaly detection."""
    model = IsolationForest(
        n_estimators=100,
        contamination=0.01,  # expect 1% anomalies
        random_state=42
    )
    model.fit(features_matrix)
    return model

# features: [error_rate, latency_p99, conversion_rate, active_users]
# Each metric is normal individually,
# but the combination "latency rising + conversion falling" = problem

Reach for ML-based approaches when simpler methods generate too many false positives or miss complex degradation patterns.

Grafana: alerts with anomaly detection

Grafana Alerting (v9+) supports multi-dimensional alerts, contact points, and notification policies. Basic setup takes 15 minutes.

Setting up an alert rule for error rate

# grafana-alert-rule.yaml
apiVersion: 1
groups:
  - orgId: 1
    name: product-health
    folder: Alerts
    interval: 1m
    rules:
      - uid: error-rate-anomaly
        title: "Error Rate Anomaly"
        condition: C
        data:
          - refId: A
            relativeTimeRange:
              from: 3600  # last hour
              to: 0
            datasourceUid: prometheus
            model:
              expr: |
                sum(rate(http_requests_total{status=~"5.."}[5m]))
                /
                sum(rate(http_requests_total[5m]))
              intervalMs: 60000
          - refId: B
            relativeTimeRange:
              from: 604800  # last week for baseline
              to: 0
            datasourceUid: prometheus
            model:
              expr: |
                avg_over_time(
                  (sum(rate(http_requests_total{status=~"5.."}[5m]))
                   /
                   sum(rate(http_requests_total[5m]))
                  )[7d:1h]
                )
          - refId: C
            datasourceUid: "__expr__"
            model:
              type: math
              expression: "$A > ($B * 3)"  # 3x baseline
        for: 5m
        labels:
          severity: critical
          team: backend
        annotations:
          summary: "Error rate {{ $values.A }} exceeds baseline {{ $values.B }} by 3x or more"

Notification policy with escalation

# grafana-notification-policy.yaml
apiVersion: 1
policies:
  - orgId: 1
    receiver: default-slack
    group_by: ['alertname', 'team']
    group_wait: 30s
    group_interval: 5m
    repeat_interval: 4h
    routes:
      - receiver: pagerduty-critical
        matchers:
          - severity = critical
        continue: true
        group_wait: 0s
      - receiver: slack-warnings
        matchers:
          - severity = warning
        group_wait: 1m

Grafana lets you build alerts from PromQL queries with math expressions — compare the current value to last week’s baseline, yesterday at the same time, or a monthly percentile.

PostHog: alerts on product events

PostHog fits product-level metrics: funnel conversion, retention, feature adoption. It has built-in alerts on Insights.

Setting up a funnel alert

  1. Create an Insight of type Funnel: signup -> onboarding_complete -> first_action
  2. In Insight settings, select Alert threshold
  3. Set condition: conversion rate < baseline - 2 standard deviations

PostHog calculates the baseline automatically from the last 30 days.

Alert via API

import httpx

POSTHOG_HOST = "https://app.posthog.com"
POSTHOG_API_KEY = "phx_..."
PROJECT_ID = "12345"

async def create_posthog_alert(
    insight_id: int,
    threshold_type: str,  # "absolute" | "relative"
    threshold_value: float,
    notification_targets: list[dict]
):
    """Create an alert on a PostHog Insight via API."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{POSTHOG_HOST}/api/projects/{PROJECT_ID}/alerts/",
            headers={"Authorization": f"Bearer {POSTHOG_API_KEY}"},
            json={
                "insight": insight_id,
                "name": f"Alert for insight {insight_id}",
                "threshold": {
                    "type": threshold_type,
                    "value": threshold_value,
                    "configuration": {
                        "compare_to": "previous_period",
                        "direction": "decrease"
                    }
                },
                "notification_targets": notification_targets,
                "enabled": True
            }
        )
        return response.json()

# Example: alert if conversion dropped 20% relative to the previous period
await create_posthog_alert(
    insight_id=42,
    threshold_type="relative",
    threshold_value=-0.20,
    notification_targets=[
        {"type": "slack", "channel": "#product-alerts"},
        {"type": "email", "address": "[email protected]"}
    ]
)

Custom solution: Python + Prometheus + Slack

When Grafana or PostHog doesn’t cover your scenario, a custom service in 200 lines solves it. The typical case: an alert on a combination of metrics from different sources.

Architecture

┌─────────────┐     ┌─────────────────┐     ┌──────────┐
│ Prometheus   │────▶│  Alert Service   │────▶│  Slack   │
│ PostHog API  │────▶│  (Python)        │────▶│  PagerDuty│
│ Custom DB    │────▶│                  │────▶│  Telegram │
└─────────────┘     └─────────────────┘     └──────────┘

                    ┌──────┴──────┐
                    │  Rules DB   │
                    │  (YAML/DB)  │
                    └─────────────┘

Alert service

import asyncio
from dataclasses import dataclass
from enum import Enum
import httpx
import yaml

class Severity(Enum):
    INFO = "info"
    WARNING = "warning"
    CRITICAL = "critical"

@dataclass
class AlertRule:
    name: str
    metric_query: str
    source: str  # "prometheus" | "posthog" | "custom"
    condition: str  # "above" | "below" | "anomaly"
    threshold: float | None
    window_minutes: int
    severity: Severity
    channels: list[str]
    cooldown_minutes: int = 30

class MetricAlertService:
    def __init__(self, config_path: str):
        self.rules = self._load_rules(config_path)
        self.last_fired: dict[str, float] = {}
        self.prometheus_url = "http://prometheus:9090"
        self.slack_webhook = "https://hooks.slack.com/..."

    def _load_rules(self, path: str) -> list[AlertRule]:
        with open(path) as f:
            config = yaml.safe_load(f)
        return [AlertRule(**rule) for rule in config["rules"]]

    async def check_rule(self, rule: AlertRule) -> bool:
        """Check a single rule."""
        if rule.source == "prometheus":
            value = await self._query_prometheus(rule.metric_query)
        elif rule.source == "posthog":
            value = await self._query_posthog(rule.metric_query)
        else:
            value = await self._query_custom(rule.metric_query)

        if value is None:
            return False

        if rule.condition == "above":
            return value > rule.threshold
        elif rule.condition == "below":
            return value < rule.threshold
        elif rule.condition == "anomaly":
            history = await self._get_history(
                rule.source, rule.metric_query, rule.window_minutes
            )
            return detect_anomaly_zscore(history + [value])
        return False

    async def _query_prometheus(self, query: str) -> float | None:
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"{self.prometheus_url}/api/v1/query",
                params={"query": query}
            )
            data = resp.json()
            results = data.get("data", {}).get("result", [])
            if not results:
                return None
            return float(results[0]["value"][1])

    async def run_check_cycle(self):
        """One cycle checking all rules."""
        for rule in self.rules:
            triggered = await self.check_rule(rule)
            if triggered and self._cooldown_passed(rule.name, rule.cooldown_minutes):
                await self._send_alert(rule)
                self.last_fired[rule.name] = asyncio.get_event_loop().time()

    async def _send_alert(self, rule: AlertRule):
        message = f"[{rule.severity.value.upper()}] {rule.name}"
        for channel in rule.channels:
            if channel == "slack":
                await self._send_slack(message)
            elif channel == "pagerduty":
                await self._send_pagerduty(rule)

    def _cooldown_passed(self, rule_name: str, cooldown_min: int) -> bool:
        last = self.last_fired.get(rule_name, 0)
        now = asyncio.get_event_loop().time()
        return (now - last) > cooldown_min * 60

Rule configuration

# alert-rules.yaml
rules:
  - name: "API Error Rate Spike"
    metric_query: 'sum(rate(http_requests_total{status=~"5.."}[5m])) / sum(rate(http_requests_total[5m]))'
    source: prometheus
    condition: anomaly
    threshold: null
    window_minutes: 60
    severity: critical
    channels: ["slack", "pagerduty"]
    cooldown_minutes: 15

  - name: "Signup Conversion Drop"
    metric_query: "funnel/signup-to-activation"
    source: posthog
    condition: below
    threshold: 0.15  # conversion below 15%
    window_minutes: 1440  # over 24 hours
    severity: warning
    channels: ["slack"]
    cooldown_minutes: 360

  - name: "Payment Success Rate"
    metric_query: 'sum(rate(payments_total{status="success"}[10m])) / sum(rate(payments_total[10m]))'
    source: prometheus
    condition: below
    threshold: 0.95  # below 95% successful payments
    window_minutes: 30
    severity: critical
    channels: ["slack", "pagerduty"]
    cooldown_minutes: 10

  - name: "LLM Response Quality"
    metric_query: 'avg(llm_response_score{model="gpt-5.4"}[15m])'
    source: prometheus
    condition: below
    threshold: 0.7
    window_minutes: 60
    severity: warning
    channels: ["slack"]
    cooldown_minutes: 60

AI prompts for generating alerting rules

Instead of writing each rule by hand, use an LLM to generate configurations from a product description. Three prompts cover 90% of scenarios.

Prompt 1: generate rules from a product description

You are an SRE engineer. Based on the product description below, generate
a YAML configuration for the alerting system.

Product: {product_description}
Available metrics: {metrics_list}
Data sources: {datasources}

For each rule define:
- name: a human-readable name
- metric_query: exact query to the data source
- condition: above/below/anomaly
- threshold: numeric value (null for anomaly)
- severity: info/warning/critical
- window_minutes: evaluation window size
- cooldown_minutes: minimum interval between repeat alerts

Rules:
1. Critical — only for metrics directly affecting revenue or availability
2. Warning — degradation that could become critical without intervention
3. For each critical metric, add a warning at a softer threshold
4. Anomaly detection — for metrics with pronounced seasonality

Format: YAML, compatible with alert-rules.yaml

Prompt 2: analyze existing alerts and find gaps

Analyze the current alert configuration and identify gaps.

Current rules:
{current_rules_yaml}

System architecture:
{system_architecture}

Incidents over the last 3 months:
{incident_history}

Identify:
1. What types of incidents would not have been caught by current rules?
2. Which rules generate false positives (based on incident patterns)?
3. Which rules need to be added?
4. Which thresholds need adjustment?

For each identified gap, propose a concrete rule in YAML format.

Prompt 3: automatic threshold calibration

Based on historical metric data, calculate the optimal threshold for an alert.

Metric: {metric_name}
Data for 30 days (JSON): {metric_data_json}
Known incidents in this period: {incidents}
Current threshold: {current_threshold}
Current false positives per week: {false_positives_per_week}

Requirements:
- Maximum 2 false positives per week
- All known incidents must be detected
- Account for daily and weekly seasonality

Return:
1. Recommended threshold
2. Recommended window_minutes
3. Expected number of false positives
4. Which known incidents will be missed (if any)

These prompts work with GPT-5.4, Claude, and Gemini. LLM-generated rules cut alerting setup from days to hours. Every generated rule needs to be validated against historical data before it goes to production.

12 rules for production alerting

Rules proven in production. Each one prevents a distinct failure mode.

1. Cooldown is mandatory. Without one, a single incident generates dozens of alerts. Minimum: 10 minutes for critical, 30 for warning.

2. Alert on symptoms, not causes. “Conversion dropped 30%” beats “CPU > 80%”. Users don’t care about CPU.

3. Every alert needs an action. If there’s nothing to do, delete the alert. Informational signals belong in a separate channel marked INFO.

4. Group by service. A cascading failure fires alerts from 10 services at once. Grouping by service in the notification policy keeps it from becoming an alert storm.

5. Separate channels by severity. Critical in PagerDuty — the kind that wakes you up. Warning in Slack, reviewed in the morning. Info on the dashboard, checked at standup.

6. Use for (pending period). The alert fires only if the condition holds for N consecutive minutes. This cuts single-spike noise. Recommended: 3–5 minutes for critical, 10–15 for warning.

7. One alert, one owner. If it’s unclear who responds, the alert is useless. Put a team label on every rule and route it in the notification policy.

8. Review thresholds monthly. The product changes. Metrics drift. An alert configured three months ago may be pointing at the wrong number.

9. Track alert fatigue. Once the team is getting more than 5 alerts per day, they start ignoring them. Watch the firing rate and false positive percentage.

10. Test against historical data. Before deploying a new rule, run it against the last month. Did it catch all known incidents? How many false positives?

11. Write a runbook. Every alert links to a runbook: what to check, how to fix it, who to escalate to. Use the runbook_url field in Grafana alert annotations.

12. Monitor the alerting system itself. Dead man’s switch: the service sends a periodic “I’m alive” signal. If it stops arriving, your alerting is broken and you won’t know it.

# Dead man's switch in Grafana
- uid: alerting-health-check
  title: "Alerting System Health"
  condition: A
  data:
    - refId: A
      model:
        expr: 'up{job="alert-service"}'
  for: 5m
  labels:
    severity: critical
  annotations:
    summary: "Alert service is down — alerts are not being evaluated"
    runbook_url: "https://wiki.internal/runbooks/alerting-down"

Integration with LLM Observability

If your product uses LLMs, standard metrics aren’t enough. HTTP 200 doesn’t mean the model returned a useful response. You need a separate alerting layer for that.

Langfuse (see the LLM Observability guide) provides metrics you can export to Prometheus:

# Export Langfuse scores to Prometheus
from prometheus_client import Gauge
from langfuse import Langfuse

langfuse = Langfuse()
llm_quality = Gauge("llm_response_quality", "LLM response quality score", ["model", "prompt_name"])
llm_cost = Gauge("llm_cost_per_request", "LLM cost per request in USD", ["model"])
llm_latency = Gauge("llm_latency_seconds", "LLM response latency", ["model"])

async def export_langfuse_metrics():
    """Periodically export metrics from Langfuse to Prometheus."""
    traces = langfuse.fetch_traces(limit=100, order_by="timestamp.desc")

    for trace in traces.data:
        if trace.scores:
            for score in trace.scores:
                llm_quality.labels(
                    model=trace.metadata.get("model", "unknown"),
                    prompt_name=trace.metadata.get("prompt_name", "unknown")
                ).set(score.value)

        if trace.total_cost:
            llm_cost.labels(
                model=trace.metadata.get("model", "unknown")
            ).set(trace.total_cost)

Alerts for LLM metrics:

rules:
  - name: "LLM Quality Degradation"
    metric_query: 'avg(llm_response_quality{prompt_name="main-assistant"}[30m])'
    source: prometheus
    condition: below
    threshold: 0.6
    window_minutes: 30
    severity: warning
    channels: ["slack"]

  - name: "LLM Cost Spike"
    metric_query: 'sum(rate(llm_cost_per_request[1h])) * 3600'
    source: prometheus
    condition: above
    threshold: 50  # $50/hour
    window_minutes: 60
    severity: critical
    channels: ["slack", "pagerduty"]

Alerting resilience and circuit breaker

The alerting service itself can go down. Two patterns handle this.

Dead man’s switch (described above): an external service confirms the alerting system is alive.

Circuit breaker at the integration level: if the Slack API goes unresponsive, alerts fall back to another channel (email, Telegram). More on the circuit breaker pattern in the article on Deno Edge Functions.

from dataclasses import dataclass, field
from time import time

@dataclass
class CircuitBreaker:
    failure_threshold: int = 3
    reset_timeout_sec: int = 60
    failures: int = 0
    last_failure: float = 0
    state: str = "closed"  # closed | open | half-open

    def record_failure(self):
        self.failures += 1
        self.last_failure = time()
        if self.failures >= self.failure_threshold:
            self.state = "open"

    def record_success(self):
        self.failures = 0
        self.state = "closed"

    def can_execute(self) -> bool:
        if self.state == "closed":
            return True
        if self.state == "open":
            if time() - self.last_failure > self.reset_timeout_sec:
                self.state = "half-open"
                return True
            return False
        return True  # half-open: try one request

# Usage
slack_circuit = CircuitBreaker(failure_threshold=3, reset_timeout_sec=120)

async def send_alert_with_fallback(message: str):
    if slack_circuit.can_execute():
        try:
            await send_slack(message)
            slack_circuit.record_success()
        except Exception:
            slack_circuit.record_failure()
            await send_telegram(message)  # fallback
    else:
        await send_telegram(message)  # circuit open, use fallback

Implementation checklist

Minimum alerts for launch:

MetricConditionSeverityTool
Error rate (5xx)> 3x baseline over 5 minCriticalGrafana + Prometheus
Latency p99> 2x baseline over 10 minWarningGrafana + Prometheus
Signup conversion< baseline - 2 stdWarningPostHog
Payment success< 95% over 30 minCriticalGrafana + Prometheus
LLM quality score< 0.6 over 30 minWarningCustom + Langfuse
Alerting healthservice unresponsive 5 minCriticalDead man’s switch

Rollout order: error rate and latency first (Grafana + Prometheus, one hour), then payment and revenue metrics, then product metrics via PostHog, then LLM metrics if applicable. Generate additional rules with the prompts above. Monthly review: cut alerts that don’t fire usefully, recalibrate the rest.

The gap between “users are complaining” and “we’re already fixing it” decides whether those users come back tomorrow.


Need help with monitoring and alerting automation? I help startups build AI products and automate processes — belov.works.

Frequently Asked Questions

How do you set the initial thresholds for a new product with no historical data to calibrate against?
Start with conservative static thresholds based on industry benchmarks while you collect data: error rate above 2%, latency p99 above 2 seconds, payment success below 97%. Accept that you'll get some false positives for the first 2–4 weeks. Log every alert firing and whether it corresponded to a real incident. After 30 days you'll have enough data to run the threshold calibration prompt from this article, which gives you statistically grounded thresholds tuned to your actual traffic patterns. The worst approach is to start with no alerting at all while waiting for "enough data" — you'll miss real incidents during your highest-risk early launch period.
What is the practical difference between using Grafana's built-in anomaly detection versus writing a custom Python service?
Grafana's anomaly detection is fast to set up and sufficient when all your metrics are already in Prometheus or a compatible data source. The limitation is that it handles one metric at a time — it can't fire on a combination like "latency rising AND conversion falling simultaneously." The custom Python service described in this article handles cross-source multi-dimensional patterns: a single alert rule can combine Prometheus infrastructure data, PostHog product events, and custom database queries. The tradeoff is maintenance overhead. For a startup, the right sequence is to start with Grafana for infrastructure and PostHog for product metrics, and build a custom service only when you hit a specific pattern the tools can't express.
How do you prevent alert fatigue from making the team ignore the alerting system entirely?
The article's Rule 9 covers this at the measurement level. The practical intervention is a monthly alerting review meeting: look at every alert that fired in the past 30 days, classify each as true positive, false positive, or actionable but ignored, and delete or recalibrate any rule that generated more than 2 false positives per week. If the team is seeing more than 5 alerts per day, the system is already in alert fatigue — the fix is usually either raising thresholds, adding longer pending periods (the for parameter in Grafana), or splitting channels so that warning-level alerts don't wake anyone up. A well-maintained alerting system should produce 0–2 actionable alerts per day on average.