Building an AI agent for anomaly detection in dashboards

Most dashboard pipelines silently drop outliers into a graph and call it monitoring. This tutorial builds an AI agent anomaly detection dashboards system that catches metric deviations and explains them in plain language using an LLM-backed reasoning loop. We keep the statistical detector dumb and let the model handle narrative.

Prerequisites

• Python 3.11 or newer
• pip install pandas numpy openai python-dotenv requests
• A time-series metrics export. We use a local metrics.csv with columns ts,metric,value (timestamp ISO, metric name, float).
• An OpenAI-compatible API key. We route through n4n.ai’s endpoint to get automatic fallback across 240+ models and per-token metering, so set LLM_API_KEY and LLM_BASE=https://api.n4n.ai/v1.

Create a .env file:

echo "LLM_API_KEY=sk-..." > .env
echo "LLM_BASE=https://api.n4n.ai/v1" >> .env

Step 1: Ingest and resample metrics

Irregular sampling destroys rolling statistics. Load, parse, and resample to a fixed 1-minute grid per metric.

import pandas as pd

df = pd.read_csv("metrics.csv", parse_dates=["ts"])
df = df.set_index("ts").sort_index()
# Resample to 1-minute buckets, mean per metric
resampled = (
    df.groupby("metric")["value"]
    .resample("1min")
    .mean()
    .reset_index()
)
# Forward-fill small gaps up to 2 minutes
resampled["value"] = resampled.groupby("metric")["value"].ffill(limit=2)
print(resampled.head())

Expected output:

                 ts metric     value
0 2024-01-01 00:00    cpu  23.400000
1 2024-01-01 00:01    cpu  24.100000
2 2024-01-01 00:02    cpu  22.900000
3 2024-01-01 00:03    cpu  23.000000

Step 2: Statistical anomaly detection

Do not ask an LLM to find outliers. A rolling z-score is deterministic, debuggable, and costs zero tokens. Flag points where |z| > 3 over a 30-sample window.

import numpy as np

def flag_anomalies(group, window=30, threshold=3.0):
    g = group.sort_values("ts").copy()
    roll = g["value"].rolling(window, min_periods=window // 2)
    mean = roll.mean()
    std = roll.std()
    g["z"] = (g["value"] - mean) / std
    g["anomaly"] = g["z"].abs() > threshold
    return g

labeled = resampled.groupby("metric", group_keys=False).apply(flag_anomalies)
anoms = labeled[labeled["anomaly"]]
print(f"Found {len(anoms)} anomalies")
print(anoms[["ts", "metric", "value", "z"]].head())

Checkpoint output:

Found 4 anomalies
                 ts metric   value        z
12 2024-01-01 00:12    cpu  98.200  4.2123
45 2024-01-01 00:45   mem  91.400  3.8891

Step 3: Build the agent context

The AI agent anomaly detection dashboards pattern separates detection from reasoning. The model receives only the anomaly plus a 10-minute context window, not the full series.

def build_prompt(row, context):
    ctx = context[
        (context["metric"] == row["metric"]) &
        (context["ts"] >= row["ts"] - pd.Timedelta("10min")) &
        (context["ts"] <= row["ts"])
    ]
    lines = "\n".join(f"  {r.ts:%H:%M} {r.value:.1f}" for r in ctx.itertuples())
    return f"""Metric '{row.metric}' spiked to {row.value:.1f} (z={row.z:.1f}) at {row.ts:%Y-%m-%d %H:%M}.
Recent values:
{lines}

Hypothesize the most likely root cause in one sentence. If you need another metric, reply 'QUERY: <metric_name>'."""

Step 4: Agent loop with LLM fallback

We call the model, parse the response, and stub the “QUERY:” branch. Routing through the gateway gives us provider fallback without code changes.

from openai import OpenAI
from dotenv import load_dotenv
import os

load_dotenv()
client = OpenAI(
    api_key=os.environ["LLM_API_KEY"],
    base_url=os.environ["LLM_BASE"],
)

def ask_agent(prompt):
    try:
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.2,
            extra_headers={"x-routing": "cost-optimized"}  # honored by n4n.ai
        )
        return resp.choices[0].message.content.strip()
    except Exception as e:
        return f"ERROR: {e}"

for _, row in anoms.iterrows():
    prompt = build_prompt(row, labeled)
    answer = ask_agent(prompt)
    if answer.startswith("QUERY:"):
        print(f"Agent requests extra metric: {answer}")
    else:
        print(f"[{row.metric} @ {row.ts:%H:%M}] {answer}")

Expected output:

[cpu @ 00:12] Likely a batch job scheduled at midnight saturated the host CPU.
[mem @ 00:45] Possible memory leak in the ingestion worker after config reload.

Step 5: Push to the dashboard

An anomaly with no visibility is useless. Emit a structured event your dashboard can render. Replace the print with a real webhook.

import json
import requests

def emit(event):
    # requests.post("https://dash.example.com/api/events", json=event)
    print(json.dumps(event))

for _, row in anoms.iterrows():
    prompt = build_prompt(row, labeled)
    explanation = ask_agent(prompt)
    emit({
        "ts": row.ts.isoformat(),
        "metric": row.metric,
        "value": float(row.value),
        "zscore": float(row.z),
        "explanation": explanation,
        "source": "ai-agent-anomaly-detection-dashboards"
    })

The source tag lets you filter these from raw threshold alerts in Grafana or Superset.

Step 6: Schedule and operate

Wrap the pipeline and run it on a cron. The statistical gatekeeper runs every minute; the LLM runs only on the few rows that pass.

def run_pipeline():
    df = pd.read_csv("metrics.csv", parse_dates=["ts"])
    resampled = df.groupby("metric")["value"].resample("1min").mean().reset_index()
    labeled = resampled.groupby("metric", group_keys=False).apply(flag_anomalies)
    anoms = labeled[labeled["anomaly"]]
    for _, row in anoms.iterrows():
        prompt = build_prompt(row, labeled)
        emit({/* ... */})

if __name__ == "__main__":
    run_pipeline()

A production AI agent anomaly detection dashboards deployment should cache prompt prefixes and forward provider cache-control hints to cut token cost. The gateway we used forwards those headers automatically.

Evaluating the agent

Before trusting it in prod, log every explanation with a later human label.

# log to sqlite or just append to a file
with open("agent_feedback.csv", "a") as f:
    f.write(f"{row.ts},{row.metric},{explanation}\n")

After a week, compute precision: how many explanations matched the post-incident root cause. If it drops below 70%, tighten the z-score threshold or add more context metrics.

What you skipped

We did not build a vector store, a multi-agent debate, or a fine-tuned classifier. For dashboard anomaly triage, those are premature. Start with z-scores and one LLM call per anomaly; add complexity only when the false-positive rate demands it.