Your SIEM fires 3,000 alerts on a Tuesday night. Your on-call analyst acknowledges 40 of them. The other 2,960? Noise — mostly. This is the alert fatigue problem, and it gets worse as environments scale. A language model can automate the first-pass triage that currently eats analyst hours, and it can do it with enough context to be genuinely useful rather than adding yet another filter layer.
This post walks through a working Python implementation that feeds SIEM alerts to an LLM, enriches them with threat intel context, and outputs a prioritized triage report.
Most SIEM rules are threshold-based: if more than N failed logins happen in M minutes from the same IP, fire an alert. This works for known attack patterns but produces two failure modes at scale:
Failed SSH login from 185.220.101.47
. It doesn't tell you that this IP is a known Tor exit node, that the targeted user has admin privileges, and that this is the third attempt in 6 hours from that subnet.A language model can synthesize these facts and produce a verdict. It won't replace a human for complex incidents, but it handles the 80% that are obviously benign or obviously high-priority.
The pipeline has four steps:
We'll use Python with the openai
SDK (works with any OpenAI-compatible endpoint, including local models), and the output schema is enforced via structured outputs so downstream code doesn't have to parse free text.
Before touching the LLM, build good context. Garbage in, garbage out.
import json
import ipaddress
from dataclasses import dataclass, asdict
from typing import Optional
@dataclass
class AlertContext:
raw_alert: dict
user_role: Optional[str]
user_dept: Optional[str]
asset_criticality: str # "low", "medium", "high", "critical"
ip_reputation: Optional[str]
related_alerts_24h: int
is_business_hours: bool
def enrich_alert(alert: dict, user_db: dict, asset_db: dict) -> AlertContext:
user = user_db.get(alert.get("username", ""), {})
asset = asset_db.get(alert.get("hostname", ""), {})
src_ip = alert.get("src_ip", "")
ip_rep = None
try:
addr = ipaddress.ip_address(src_ip)
if addr.is_private:
ip_rep = "internal"
else:
ip_rep = "unknown_external"
except ValueError:
pass
return AlertContext(
raw_alert=alert,
user_role=user.get("role"),
user_dept=user.get("department"),
asset_criticality=asset.get("criticality", "medium"),
ip_reputation=ip_rep,
related_alerts_24h=count_related_alerts(alert, window_hours=24),
is_business_hours=is_business_hours(alert.get("timestamp")),
)
def count_related_alerts(alert: dict, window_hours: int) -> int:
return 0
def is_business_hours(ts: Optional[str]) -> bool:
if not ts:
return False
from datetime import datetime, timezone
dt = datetime.fromisoformat(ts).astimezone(timezone.utc)
return 8 <= dt.hour < 18 and dt.weekday() < 5
The AlertContext
object gives the model everything it needs without dumping a 10 KB log blob into the prompt.
Now the core piece: sending the enriched alert to a language model and getting a structured verdict back.
from openai import OpenAI
from pydantic import BaseModel
from enum import Enum
class Severity(str, Enum):
CRITICAL = "critical"
HIGH = "high"
MEDIUM = "medium"
LOW = "low"
INFO = "informational"
class Disposition(str, Enum):
ESCALATE = "escalate"
MONITOR = "monitor"
CLOSE = "close"
class TriageResult(BaseModel):
severity: Severity
disposition: Disposition
confidence: float # 0.0 to 1.0
reasoning: str # one concise paragraph
recommended_actions: list[str]
false_positive_indicators: list[str]
TRIAGE_SYSTEM_PROMPT = """You are a senior SOC analyst performing first-pass alert triage.
Given an enriched security alert, output a structured triage verdict.
Rules:
- If asset_criticality is "critical", escalate unless you have strong FP evidence.
- Off-hours activity on privileged accounts increases severity by one level.
- Known internal IPs with a single failed login are almost always false positives.
- confidence reflects certainty, not severity.
- reasoning: one paragraph, max 80 words, no bullet points.
- recommended_actions: max 3 concrete next steps for an analyst.
- false_positive_indicators: specific facts suggesting benign activity."""
def triage_alert(ctx: AlertContext, client: OpenAI, model: str = "gpt-4o-mini") -> TriageResult:
payload = json.dumps(asdict(ctx), indent=2, default=str)
response = client.beta.chat.completions.parse(
model=model,
messages=[
{"role": "system", "content": TRIAGE_SYSTEM_PROMPT},
{"role": "user", "content": f"Triage this alert:\n\n{payload}"}
],
response_format=TriageResult,
temperature=0.1,
)
return response.choices[0].message.parsed
client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
sample_alert = {
"alert_id": "A-2026-04421",
"rule": "Multiple Failed SSH Logins",
"username": "j.martin",
"hostname": "prod-db-01",
"src_ip": "185.220.101.47",
"timestamp": "2026-07-28T02:14:33+00:00",
"failed_attempts": 12,
}
user_db = {"j.martin": {"role": "DBA", "department": "Engineering"}}
asset_db = {"prod-db-01": {"criticality": "critical"}}
ctx = enrich_alert(sample_alert, user_db, asset_db)
result = triage_alert(ctx, client)
print(f"[{result.severity.upper()}] → {result.disposition}")
print(f"Confidence: {result.confidence:.0%}")
print(f"Reasoning: {result.reasoning}")
for action in result.recommended_actions:
print(f" • {action}")
For this sample alert — Tor exit node, DBA account, production database, 2 AM — a properly prompted model outputs severity=critical, disposition=escalate
with clear reasoning about the combination of off-hours timing, privileged account, and known malicious IP source.
Individual triage is useful; batch processing is where it delivers ROI. Process the overnight alert queue before the morning shift arrives:
import asyncio
async def batch_triage(
alerts: list[dict],
client: OpenAI,
user_db: dict,
asset_db: dict,
concurrency: int = 5,
) -> list[tuple[dict, TriageResult]]:
semaphore = asyncio.Semaphore(concurrency)
async def triage_one(alert: dict) -> tuple[dict, TriageResult]:
async with semaphore:
ctx = enrich_alert(alert, user_db, asset_db)
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(None, triage_alert, ctx, client)
return alert, result
return await asyncio.gather(*[triage_one(a) for a in alerts])
async def main():
alerts = load_alerts_from_siem() # Your SIEM API or export
results = await batch_triage(alerts, client, user_db, asset_db)
to_escalate = [(a, r) for a, r in results if r.disposition == Disposition.ESCALATE]
to_close = [(a, r) for a, r in results if r.disposition == Disposition.CLOSE]
print(f"Total: {len(alerts)}")
print(f"Escalate: {len(to_escalate)} ({len(to_escalate)/len(alerts):.0%})")
print(f"Auto-close: {len(to_close)} ({len(to_close)/len(alerts):.0%})")
for alert, result in to_escalate:
create_ticket(alert, result)
for alert, result in to_close:
auto_close_alert(alert, result)
asyncio.run(main())
The concurrency=5
cap prevents rate limiting on the LLM API. Tune it based on your provider's limits or the hardware running your local model.
Don't deploy this without a feedback loop. Track three numbers:
Feed disagreements back into your system prompt as concrete examples. After two weeks of tuning on your environment's specific noise profile, 80% auto-close rates with under 1% false negative rate are realistic — this matches results from teams running similar pipelines in production.
For a baseline of what "good" triage looks like per alert category, the security hardening checklists we publish include SOC triage playbooks you can adapt directly as prompt templates.
LLM-powered alert triage is one of the highest-ROI applications of language models in security operations right now. The reason is straightforward: alert triage is fundamentally a classification and reasoning task over structured context, which is exactly what these models do well.
The implementation here runs entirely on a local model if you can't send alert data to an external API — just point base_url
at your Ollama or vLLM endpoint. Start with a pilot on a single rule category like failed logins, measure results for two weeks, then expand.
The hard part isn't the code. It's building the enrichment layer with meaningful context, writing a system prompt that reflects your actual triage playbooks, and creating the feedback loop that catches model mistakes before they become missed incidents.
I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.