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

Component
Responsibility


Log Collector (Filebeat / Fluent Bit)
Ships raw log lines to a Kafka topic.


Kafka → Python Consumer
Buffers logs, batches them, and calls the Flask /predict endpoint.


Flask Inference Service
Loads the Claude‑tuned model, runs inference, returns anomaly score & label.


Prometheus Exporter
Exposes latency, error‑rate, and anomaly‑rate metrics.


Grafana Dashboard
Visualises spikes, sends alerts via Alertmanager.

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

# 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

# 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

# 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)

# 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)

# 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"]

# 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

# 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 .

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

# 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.