Build an AI Log Pattern Classifier with Claude API

pip install anthropic python-dotenv requests

import anthropic
import json
import os
import sys
from dataclasses import dataclass
from typing import Optional
 
 
@dataclass
class LogEntry:
    line: str
    timestamp: Optional[str] = None
    service: Optional[str] = None
    level: Optional[str] = None
 
 
def parse_log_line(raw: str) -> LogEntry:
    """
    Try to extract timestamp, service, level from common log formats.
    Falls back to treating the whole line as content.
    """
    import re
 
    # Common patterns: [2026-07-11T02:14:33Z] ERROR service-name: message
    pattern = r"(\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}[^\s]*)\s+(ERROR|WARN|INFO|DEBUG|CRITICAL|FATAL)?\s*([^\s:]+)?:?\s*(.*)"
    match = re.match(pattern, raw.strip(), re.IGNORECASE)
 
    if match:
        return LogEntry(
            line=raw.strip(),
            timestamp=match.group(1),
            level=match.group(2),
            service=match.group(3),
        )
 
    return LogEntry(line=raw.strip())
 
 
def read_logs_from_file(filepath: str) -> list[LogEntry]:
    with open(filepath, "r") as f:
        return [parse_log_line(line) for line in f if line.strip()]
 
 
def read_logs_from_stdin() -> list[LogEntry]:
    return [parse_log_line(line) for line in sys.stdin if line.strip()]

client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
 
CATEGORIES = {
    "critical": "Service down, data loss, security breach, unrecoverable error",
    "error": "Operation failed but service still running, needs attention soon",
    "warning": "Degraded performance, retry succeeded, approaching limits",
    "anomaly": "Unusual pattern, unexpected behavior, not necessarily an error",
    "noise": "Normal operation, health checks, routine info",
}
 
 
def classify_log_batch(entries: list[LogEntry]) -> list[dict]:
    """
    Send a batch of log lines to Claude for classification.
    Returns list of {line, category, severity, explanation, action_needed}
    """
    log_text = "\n".join(
        f"[{i}] {e.line}" for i, e in enumerate(entries)
    )
 
    prompt = (
        "You are a senior SRE classifying log lines for on-call triage.\n\n"
        "Categories:\n"
        + "\n".join(f"- {k}: {v}" for k, v in CATEGORIES.items())
        + "\n\n"
        "Log lines to classify:\n"
        + log_text
        + "\n\n"
        "Return a JSON array. Each item must have:\n"
        '{"index": 0, "category": "critical|error|warning|anomaly|noise", '
        '"severity": 1-10, "explanation": "why", "action_needed": true/false, '
        '"suggested_action": "what to check if action_needed"}\n\n'
        "Return ONLY the JSON array, no markdown."
    )
 
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=2000,
        messages=[{"role": "user", "content": prompt}]
    )
 
    text = response.content[0].text.strip()
    if text.startswith("```"):
        text = text.split("```")[1]
        if text.startswith("json"):
            text = text[4:]
 
    return json.loads(text)
 
 
def classify_logs(entries: list[LogEntry], batch_size: int = 50) -> list[dict]:
    """Process logs in batches to stay within token limits."""
    results = []
 
    for i in range(0, len(entries), batch_size):
        batch = entries[i:i + batch_size]
        print(f"Classifying batch {i // batch_size + 1} ({len(batch)} lines)...")
 
        try:
            batch_results = classify_log_batch(batch)
            # Add original log line back
            for result in batch_results:
                idx = result.get("index", 0)
                if idx < len(batch):
                    result["original_line"] = batch[idx].line
                    result["service"] = batch[idx].service
            results.extend(batch_results)
        except (json.JSONDecodeError, Exception) as e:
            print(f"Batch {i // batch_size + 1} failed: {e}")
 
    return results

def generate_report(results: list[dict]) -> str:
    """Generate a human-readable triage report."""
    from collections import defaultdict
 
    by_category = defaultdict(list)
    for r in results:
        by_category[r.get("category", "noise")].append(r)
 
    lines = [
        "\n" + "=" * 60,
        "  LOG TRIAGE REPORT",
        "=" * 60,
        f"\nTotal logs analyzed: {len(results)}",
        f"  🔴 Critical:  {len(by_category['critical'])}",
        f"  🟠 Errors:    {len(by_category['error'])}",
        f"  🟡 Warnings:  {len(by_category['warning'])}",
        f"  🔵 Anomalies: {len(by_category['anomaly'])}",
        f"  ⚪ Noise:     {len(by_category['noise'])}",
    ]
 
    # Show critical and errors first
    for category in ["critical", "error", "warning", "anomaly"]:
        items = by_category[category]
        if not items:
            continue
 
        icon = {"critical": "🔴", "error": "🟠", "warning": "🟡", "anomaly": "🔵"}[category]
        lines.append(f"\n{icon} {category.upper()} ({len(items)})")
        lines.append("-" * 40)
 
        for item in sorted(items, key=lambda x: -x.get("severity", 0)):
            lines.append(f"\nSeverity {item.get('severity', '?')}/10")
            lines.append(f"Log: {item.get('original_line', '')[:120]}")
            lines.append(f"Why: {item.get('explanation', '')}")
            if item.get("action_needed"):
                lines.append(f"Action: {item.get('suggested_action', '')}")
 
    # Action required summary
    action_items = [r for r in results if r.get("action_needed")]
    if action_items:
        lines.append(f"\n{'='*60}")
        lines.append(f"ACTION REQUIRED ({len(action_items)} items)")
        lines.append("=" * 60)
        for item in action_items[:10]:
            lines.append(f"• {item.get('suggested_action', '')}")
 
    return "\n".join(lines)
 
 
def main():
    import argparse
 
    parser = argparse.ArgumentParser(description="AI Log Pattern Classifier")
    parser.add_argument("logfile", nargs="?", help="Log file to analyze (stdin if omitted)")
    parser.add_argument("--batch-size", type=int, default=30, help="Lines per Claude API call")
    parser.add_argument("--json", action="store_true", help="Output raw JSON instead of report")
    args = parser.parse_args()
 
    from dotenv import load_dotenv
    load_dotenv()
 
    print("Reading logs...")
    if args.logfile:
        entries = read_logs_from_file(args.logfile)
    else:
        entries = read_logs_from_stdin()
 
    print(f"Loaded {len(entries)} log lines")
 
    results = classify_logs(entries, batch_size=args.batch_size)
 
    if args.json:
        print(json.dumps(results, indent=2))
    else:
        print(generate_report(results))
 
 
if __name__ == "__main__":
    main()

# From a log file
python log_classifier.py app.log
 
# From kubectl logs
kubectl logs -n production deployment/payment-api --tail=500 | python log_classifier.py
 
# From docker
docker logs my-container 2>&1 | python log_classifier.py
 
# JSON output for further processing
python log_classifier.py app.log --json | jq '.[] | select(.category == "critical")'

# alert_on_critical.py
import subprocess
import requests
 
def send_slack_alert(message: str, webhook_url: str):
    requests.post(webhook_url, json={"text": message})
 
result = subprocess.run(
    ["python", "log_classifier.py", "--json", "app.log"],
    capture_output=True, text=True
)
results = json.loads(result.stdout)
criticals = [r for r in results if r["category"] == "critical"]
 
if criticals:
    msg = f"🚨 *{len(criticals)} CRITICAL log patterns detected*\n"
    for c in criticals[:3]:
        msg += f"• {c['explanation']}\n"
    send_slack_alert(msg, os.getenv("SLACK_WEBHOOK_URL"))