# Automated Metric Alerts with AI: Catch Problems Before Users Do

> Anomaly detection, AI prompts for configuring alerts, and tools (Grafana, PostHog, custom). Practical rules and configs for product metrics.
> Author: Roman Belov · Published: 2026-06-05 · Source: https://futurecraft.pro/blog/automated-metric-alerts/

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 severalfold. 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.

```python
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.

```python
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.

```python
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

```yaml
# 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

```yaml
# 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: trends, funnels (steps and trends views), and SQL insights.

### Setting up a funnel alert

1. Create an Insight of type Funnel: signup -> onboarding_complete -> first_action
2. In the insight, open Actions -> Alerts -> New alert
3. Pick the series to monitor and the condition type: has value (absolute threshold), increases by, or decreases by (as an absolute value or a percentage)
4. Set the check frequency (from real-time on higher tiers to monthly) and the recipients

Instead of a fixed threshold you can enable anomaly detection: PostHog builds the baseline from historical data itself, with a configurable detector (Z-score or MAD, for example), sensitivity, and window. Notifications reach subscribed users in-app and by email, plus Slack, Discord, Microsoft Teams, or a webhook. Alerts can also be managed programmatically (Alerts API), but the UI fully covers the typical scenarios.

## 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

```python
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

```yaml
# 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 — but not inside the same system. An `up{job="alert-service"}` rule in the same Grafana fails exactly when you need it: if evaluation stops, nothing computes the condition and nothing sends the notification. The check has to live outside — an independent monitor (healthchecks.io, UptimeRobot) or a second stack instance on separate infrastructure. The scheme is inverted: the service pings the monitor, not the other way around. When pings stop arriving, the external monitor sends the alert itself.

```python
# Dead man's switch via healthchecks.io:
# every successful rule check cycle ends with a ping
async def run_forever(service: MetricAlertService):
    async with httpx.AsyncClient() as client:
        while True:
            await service.run_check_cycle()
            # no ping past period + grace time —
            # healthchecks.io sends the alert itself (email, Slack, PagerDuty)
            await client.get("https://hc-ping.com/<your-check-uuid>")
            await asyncio.sleep(60)
```

## 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](/blog/llm-observability-langfuse/)) provides metrics you can export to Prometheus:

```python
# 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:

```yaml
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](/blog/circuit-breaker-deno-edge-functions/).

```python
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:

| Metric | Condition | Severity | Tool |
|--------|-----------|----------|------|
| Error rate (5xx) | > 3x baseline over 5 min | Critical | Grafana + Prometheus |
| Latency p99 | > 2x baseline over 10 min | Warning | Grafana + Prometheus |
| Signup conversion | < baseline - 2 std | Warning | PostHog |
| Payment success | < 95% over 30 min | Critical | Grafana + Prometheus |
| LLM quality score | < 0.6 over 30 min | Warning | Custom + Langfuse |
| Alerting health | no ping for 5+ min | Critical | External dead 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](https://belov.works).*