AI agents for automated log triage and alert dedup

Pager fatigue is a silent killer of engineering velocity. An AI agent log triage alert deduplication pipeline lets you compress hundreds of redundant pages into a single actionable incident with a synthesized root-cause hypothesis. This guide shows how to build one from scratch using standard LLM APIs and a few hundred lines of Python.

Step 1: Normalize alerts into a single schema

Most teams already emit alerts from Prometheus, CloudWatch, or custom apps, but each source uses a different shape. Before any clustering can happen, you need one canonical record. I use a strict Pydantic model and a thin adapter per source.

from pydantic import BaseModel
from datetime import datetime

class Alert(BaseModel):
    id: str
    source: str
    labels: dict
    text: str
    ts: datetime

def from_alertmanager(am_payload: dict) -> Alert:
    return Alert(
        id=am_payload["fingerprint"],
        source="alertmanager",
        labels=am_payload.get("labels", {}),
        text=am_payload.get("annotations", {}).get("description", ""),
        ts=datetime.now()
    )

The text field is what the embedding model will see. Keep it dense: include the error message, the service name from labels, and any cardinality that matters. Do not dump the raw JSON blob—noise in, noise out. Push these normalized objects into a Redis list or Kafka topic so downstream workers can pull a batch every 30 seconds.

Step 2: Embed and cluster with cosine similarity

Once you have a batch of normalized alerts, the next move in AI agent log triage alert deduplication is to group near-identical signals. A lightweight embedding model is enough; you do not need a fine-tuned SRE model on day one.

import numpy as np
from openai import OpenAI

client = OpenAI()  # expects OPENAI_API_KEY

def embed(texts: list[str]) -> np.ndarray:
    resp = client.embeddings.create(model="text-embedding-3-small", input=texts)
    return np.array([d.embedding for d in resp.data])

def cluster(alerts: list[Alert], threshold: float = 0.86):
    vectors = embed([a.text for a in alerts])
    clusters = []
    for i, vec in enumerate(vectors):
        best = None
        best_sim = 0.0
        for c in clusters:
            sim = np.dot(vec, c["centroid"]) / (np.linalg.norm(vec) * np.linalg.norm(c["centroid"]))
            if sim > best_sim:
                best_sim, best = sim, c
        if best and best_sim > threshold:
            best["items"].append(alerts[i])
        else:
            clusters.append({"centroid": vec, "items": [alerts[i]]})
    return clusters

Tune the threshold against your own history. Start at 0.85 and inspect the clusters manually for a week. If unrelated alerts merge, raise it; if identical ones split, lower it. For volumes above ~10k alerts/minute, move the vectors into a dedicated ANN index like FAISS, but the greedy loop above is correct and runs fine at small scale.

Step 3: Summarize clusters with an LLM agent

Now you have clusters of 1 to N alerts. This is where the AI agent log triage alert deduplication logic earns its keep: instead of paging a human with 40 copies of “Redis timeout”, you send one message with a summary, a suspected cause, and a severity.

Point your OpenAI client at an OpenAI-compatible gateway such as n4n.ai, which exposes 240+ models behind one endpoint and automatically falls back when a provider is rate-limited. That removes the need to write your own retry mesh across vendors.

import json
from openai import OpenAI

client = OpenAI(base_url="https://api.n4n.ai/v1", api_key="YOUR_KEY")

def summarize_cluster(items: list[Alert]) -> dict:
    lines = "\n".join(f"- {a.text}" for a in items)
    prompt = (
        "You are an SRE assistant. Given these alerts, respond ONLY with JSON "
        "containing keys: summary (string), suspected_root_cause (string), "
        "severity (int 1-5), recommended_first_action (string).\n\n"
        f"{lines}"
    )
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        response_format={"type": "json_object"},
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1
    )
    return json.loads(resp.choices[0].message.content)

The gateway forwards provider cache-control hints, so if the same cluster text appears within a short window, the prompt hits cache and you pay fewer tokens. Keep temperature low; this is not a creative task. Validate the returned JSON against a schema before trusting it.

Step 4: Deduplicate against open incidents

Summaries are useless if you still open a new ticket for every cluster. Maintain a state store keyed by cluster centroid hash or a stable label set. When a new cluster arrives, check if an incident with the same signature is already open.

def incident_key(items: list[Alert]) -> str:
    svc = items[0].labels.get("service", "unknown")
    return f"{svc}:{items[0].labels.get('alertname', 'generic')}"

def upsert_incident(key: str, summary: dict):
    if existing := find_open_incident(key):
        update_incident(existing.id, summary)
        return existing.id
    return create_incident(key, summary)

This is the core of alert deduplication: the first cluster opens the incident, subsequent ones within the same window just append count and refresh the timestamp. Your on-call gets one page, not fifty. If the incident closes and the same key recurs, treat it as new.

Step 5: Capture human corrections

An agent that never learns is a liability. When an engineer acknowledges or resolves an incident, record whether the LLM summary matched reality. Store the cluster text, the model output, and the human label in a simple table.

def record_feedback(incident_id: str, model_output: dict, human_note: str, accurate: bool):
    db.feedback.insert({
        "incident_id": incident_id,
        "model_output": model_output,
        "human_note": human_note,
        "accurate": accurate,
        "ts": datetime.now()
    })

After a few hundred labeled examples, you can prompt-engineer against real failures or fine-tune a small model. Either way, the feedback loop is what separates a demo from a system that survives production.

Verify success

You cannot ship this blind. Validate with a controlled test:

1. Spin up a staging Alertmanager or a script that emits 100 copies of the same synthetic alert (“API-123: 500 from payments-service”).
2. Run the pipeline end to end. Confirm exactly one incident is created, not 100.
3. Inspect the LLM summary: it should mention payments-service and suggest checking downstream dependencies.
4. Emit a different alert from another service and confirm a second incident, not a merge.
5. Check your token metering: the second batch of identical clusters should show reduced token usage if cache-control is honored.

If those five hold, you have a working AI agent log triage alert deduplication pipeline. Roll it to a single low-risk service, watch the feedback table for a week, then expand. The goal is not zero alerts—it is zero meaningless alerts.