AI-Enhanced Log Analysis and Anomaly Alert System — Part 5: Exposing the Model via a Flask Inference API

⏱ 8 min read  |  ~1596 words

AI‑Enhanced Log Analysis and Anomaly Alert System — Part 5: Exposing the Model via a Flask Inference API

Quick recap (Parts 1‑4): We started by collecting raw syslog and application‑level logs, transformed them into token‑level time‑series, and trained a lightweight transformer (Claude 4.6 Opus fine‑tuned on our domain data). In Part 4 we wrapped the model in a reusable Python class, added seasonal‑adjusted post‑processing, and stored the artifact in an mlflow model registry.

Based on my technical understanding as a Lead Programmer Analyst (PHP, Perl, Python, Shell) and the latest advances in GPT‑5.4 Pro Parallel Agents and Claude 4.6 Opus agentic workflows, this part shows how to expose that model as a production‑ready Flask inference service. You’ll get a full‑stack example that can be dropped into a Kubernetes pod, a Docker container, or a simple VM, and that plays nicely with Grafana, Prometheus, or any downstream alerting engine.

Why a Flask API?

• Flask is minimal yet extensible – perfect for a model‑centric microservice.
• It integrates seamlessly with PyTorch and 🤗 Transformers without the ceremony of larger frameworks.
• When paired with Gunicorn and parallel inference workers, you can fully exploit GPT‑5.4’s multi‑agent parallelism for sub‑second latency.

High‑level Architecture

Prerequisites

• Python 3.11+ (the official runtime for GPT‑5.4 agents).
• CUDA‑enabled GPU or an Intel Xeon with AVX‑512 for optimal transformer inference.
• Docker 24+ (optional but recommended for reproducibility).
• Access token for Anthropic’s Claude 4.6 Opus API (or a locally hosted checkpoint if you have a licensed copy).

Step 1 – Project Layout

ai_log_anomaly/
├── app/
│   ├── __init__.py
│   ├── inference.py          # Model wrapper
│   └── api.py                # Flask routes
├── tests/
│   └── test_api.py
├── Dockerfile
├── requirements.txt
└── run.py                    # Entrypoint for local dev

Step 2 – Dependency Manifest

# requirements.txt
flask==3.0.2
gunicorn==22.0.0
torch==2.4.0
transformers==4.44.0
accelerate==0.34.2
anthropic==0.7.2          # Claude 4.6 Opus SDK
pydantic==2.8.2
prometheus-client==0.20.0

Step 3 – Model Wrapper (app/inference.py)

The wrapper abstracts loading, tokenisation, and post‑processing. It also supports parallel agent execution via torch.compile and the new torch.run API introduced in PyTorch 2.4.

# app/inference.py
import os
import json
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM
from anthropic import Anthropic
from pydantic import BaseModel, ValidationError
from typing import List, Dict, Any

class LogRecord(BaseModel):
    timestamp: str
    host: str
    service: str
    level: str
    message: str

class AnomalyResult(BaseModel):
    anomaly_score: float
    label: str
    reason: str | None = None

class LogAnomalyModel:
    """Singleton that loads the Claude‑tuned checkpoint once per process."""
    _instance = None

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            cls._instance = super(LogAnomalyModel, cls).__new__(cls)
        return cls._instance

    def __init__(self):
        if getattr(self, "_initialized", False):
            return
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        # ------------------------------------------------------------
        # 1️⃣ Load tokenizer & model – we are using a locally saved
        #    Claude‑4.6‑Opus checkpoint that was exported from the
        #    MLflow registry in Part 4.
        # ------------------------------------------------------------
        model_path = os.getenv("MODEL_PATH", "./model_checkpoint")
        self.tokenizer = AutoTokenizer.from_pretrained(model_path)
        self.model = AutoModelForCausalLM.from_pretrained(
            model_path,
            torch_dtype=torch.float16 if self.device.type == "cuda" else torch.float32,
            device_map="auto"
        )
        self.model.eval()
        # Optional: compile for extra speed on recent GPUs
        if self.device.type == "cuda":
            self.model = torch.compile(self.model, mode="max-autotune")
        self._initialized = True

    def _prepare_prompt(self, logs: List[LogRecord]) -> str:
        """Serialize a batch of logs into the prompt format Claude expects."""
        prompt_lines = ["You are a log‑anomaly detector. Return JSON per line."]
        for rec in logs:
            line = json.dumps(rec.dict())
            prompt_lines.append(line)
        # Add explicit instruction for score range 0‑1
        prompt_lines.append(
            "For each line output a JSON object with keys: "
            "\"anomaly_score\" (0‑1), \"label\" (\"normal\"|\"anomaly\"), "
            "\"reason\" (optional)."
        )
        return "\n".join(prompt_lines)

    def predict(self, logs: List[Dict[str, Any]]) -> List[AnomalyResult]:
        # Validate incoming payload
        validated = [LogRecord(**log) for log in logs]

        # Build prompt
        prompt = self._prepare_prompt(validated)

        # ------------------------------------------------------------
        # 2️⃣ Run inference – we use Anthropic's client for the hosted
        #    Claude 4.6 Opus endpoint. If you have a local checkpoint,
        #    replace the call with self.model.generate().
        # ------------------------------------------------------------
        client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        response = client.completions.create(
            model="claude-4.6-opus",
            max_tokens=1024,
            temperature=0.0,
            stream=False,
            prompt=prompt
        )
        # The response is a single string with one JSON per line.
        results = []
        for line in response.completion.strip().split("\n"):
            try:
                data = json.loads(line)
                results.append(AnomalyResult(**data))
            except (json.JSONDecodeError, ValidationError):
                # Fallback – mark as uncertain anomaly
                results.append(AnomalyResult(
                    anomaly_score=0.5,
                    label="anomaly",
                    reason="Failed to parse model output"
                ))
        return results

Step 4 – Flask API (app/api.py)

The API validates JSON, forwards the request to LogAnomalyModel, and returns a structured response. It also emits Prometheus metrics for observability.

# app/api.py
import os
import time
from flask import Flask, request, jsonify
from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST
from .inference import LogAnomalyModel

app = Flask(__name__)

# -------------------- Prometheus Metrics --------------------
REQUEST_COUNT = Counter(
    "log_anomaly_requests_total",
    "Total number of inference requests",
    ["method", "endpoint", "http_status"]
)
REQUEST_LATENCY = Histogram(
    "log_anomaly_request_latency_seconds",
    "Latency of inference requests",
    ["endpoint"]
)
ANOMALY_DETECTED = Counter(
    "log_anomaly_detected_total",
    "Number of logs flagged as anomalies",
    ["service"]
)

model = LogAnomalyModel()

@app.route("/healthz", methods=["GET"])
def health_check():
    return "OK", 200

@app.route("/metrics")
def metrics():
    return generate_latest(), 200, {"Content-Type": CONTENT_TYPE_LATEST}

@app.route("/predict", methods=["POST"])
def predict():
    start = time.time()
    try:
        payload = request.get_json(force=True)
        if not isinstance(payload, list):
            raise ValueError("Payload must be a JSON array of log records.")
        # Forward to model
        results = model.predict(payload)
        # Count anomalies per service
        for rec, res in zip(payload, results):
            if res.label == "anomaly":
                ANOMALY_DETECTED.labels(service=rec.get("service", "unknown")).inc()
        response = [r.dict() for r in results]
        status = 200
    except Exception as exc:
        response = {"error": str(exc)}
        status = 400
    finally:
        latency = time.time() - start
        REQUEST_LATENCY.labels(endpoint="/predict").observe(latency)
        REQUEST_COUNT.labels(
            method=request.method,
            endpoint="/predict",
            http_status=str(status)
        ).inc()
    return jsonify(response), status

if __name__ == "__main__":
    # Development server – use Gunicorn in production
    app.run(host="0.0.0.0", port=int(os.getenv("PORT", 8080)), debug=False)

Step 5 – Local Development Entrypoint (run.py)

# run.py
import os
from app.api import app

if __name__ == "__main__":
    # Enable hot‑reload for rapid iteration
    app.run(host="0.0.0.0", port=int(os.getenv("PORT", 8080)), debug=True)

Step 6 – Dockerisation

Containerising the service guarantees that the exact Python runtime, CUDA drivers, and model checkpoint travel together.

# Dockerfile
FROM python:3.11-slim

# ---- Install system deps (curl, gnupg) ---------------------------------
RUN apt-get update && apt-get install -y --no-install-recommends \
    curl gnupg && rm -rf /var/lib/apt/lists/*

# ---- Set up a non‑root user ---------------------------------------------
ARG UID=1000
ARG GID=1000
RUN groupadd -g ${GID} appgroup && \
    useradd -m -u ${UID} -g ${GID} -s /bin/bash appuser

# ---- Python environment --------------------------------------------------
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# ---- Copy source ---------------------------------------------------------
COPY . /app

# ---- Model checkpoint (assume you copy it in CI) -------------------------
# In CI you would `COPY model_checkpoint/ ./model_checkpoint/`

# ---- Runtime user --------------------------------------------------------
USER appuser

# ---- Expose Flask port ---------------------------------------------------
EXPOSE 8080

# ---- Entrypoint ---------------------------------------------------------
CMD ["gunicorn", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "app.api:app", "--bind", "0.0.0.0:8080"]

Step 7 – Orchestrating with Kubernetes (Optional)

If you’re running a fleet of micro‑services, spin up the Flask pod behind a ClusterIP service and let the Kafka consumer hit http://log-anomaly-svc:8080/predict. Below is a minimal deployment.yaml snippet.

# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: log-anomaly-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: log-anomaly
  template:
    metadata:
      labels:
        app: log-anomaly
    spec:
      containers:
      - name: api
        image: yourrepo/ai_log_anomaly:latest
        ports:
        - containerPort: 8080
        env:
        - name: ANTHROPIC_API_KEY
          valueFrom:
            secretKeyRef:
              name: anthropic-secret
              key: api-key
        - name: MODEL_PATH
          value: "/app/model_checkpoint"
        resources:
          limits:
            cpu: "2000m"
            memory: "4Gi"
          requests:
            cpu: "500m"
            memory: "1Gi"
---
apiVersion: v1
kind: Service
metadata:
  name: log-anomaly-svc
spec:
  selector:
    app: log-anomaly
  ports:
  - protocol: TCP
    port: 80
    targetPort: 8080

Step 8 – Testing the Endpoint

Use curl or httpie to send a batch of log entries. Below is a ready‑to‑run example that mirrors the log format shown in Maximilian Oliver’s Medium post “Detecting Anomalies in Real‑Time Logs Using AI”.

# test_payload.json
[
  {
    "timestamp": "2026-09-18T12:34:56Z",
    "host": "web‑01",
    "service": "nginx",
    "level": "INFO",
    "message": "GET /api/v1/users 200 123ms"
  },
  {
    "timestamp": "2026-09-18T12:35:01Z",
    "host": "db‑02",
    "service": "postgres",
    "level": "ERROR",
    "message": "deadlock detected on transaction 219"
  }
]

# curl request
curl -X POST http://localhost:8080/predict \
     -H "Content-Type: application/json" \
     -d @test_payload.json | jq .

Typical response (formatted with jq) looks like:

[
  {
    "anomaly_score": 0.02,
    "label": "normal",
    "reason": null
  },
  {
    "anomaly_score": 0.94,
    "label": "anomaly",
    "reason": "High‑severity ERROR with deadlock pattern"
  }
]

Step 9 – Scaling with GPT‑5.4 Parallel Agents

When the log ingress rate climbs to >10 k logs/sec (a realistic scenario in large‑scale Kubernetes clusters, as highlighted in the DevOps.com 2026 article), a single Flask process becomes a bottleneck. GPT‑5.4 Pro Parallel Agents allow you to fan out inference across multiple GPU shards without manually handling thread pools.

• Wrap the LogAnomalyModel.predict call in a ParallelAgent object.
• Configure max_concurrency=8 (or as many GPU streams as your hardware supports).
• Replace the synchronous call in /predict with an await on the agent’s run_async method.

Below is a concise snippet that you can drop into app/inference.py. It uses the openai SDK’s AsyncClient as a stand‑in; the same pattern applies to Anthropic’s async API once it ships.

# Async parallel wrapper (optional)
import asyncio
from openai import AsyncClient

class ParallelClaudeAgent:
    def __init__(self, max_concurrency: int = 8):
        self.client = AsyncClient(api_key=os.getenv("ANTHROPIC_API_KEY"))
        self.semaphore = asyncio.Semaphore(max_concurrency)

    async def _run_one(self, prompt: str) -> str:
        async with self.semaphore:
            resp = await self.client.completions.create(
                model="claude-4.6-opus",
                max_tokens=1024,
                temperature=0.0,
                prompt=prompt,
                stream=False,
            )
            return resp.completion

    async def run_batch(self, prompts: List[str]) -> List[str]:
        tasks = [self._run_one(p) for p in prompts]
        return await asyncio.gather(*tasks)

# Usage inside LogAnomalyModel.predict
# Replace the sync client call with:
#   async_results = await ParallelClaudeAgent().run_batch([prompt])
#   # Then parse async_results[0] as before.

Deploy the Flask app with uvicorn workers (instead of Gunicorn) to enable native async handling, or keep Gunicorn with the uvicorn.work