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
- Create an Insight of type Funnel: signup -> onboarding_complete -> first_action
- In Insight settings, select Alert threshold
- 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:
| 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 | service unresponsive 5 min | Critical | 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.
Frequently Asked Questions
How do you set the initial thresholds for a new product with no historical data to calibrate against?
What is the practical difference between using Grafana's built-in anomaly detection versus writing a custom Python service?
How do you prevent alert fatigue from making the team ignore the alerting system entirely?
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.