|
``` |
|
import pandas as pd |
|
from transformers import pipeline |
|
import logging |
|
|
|
logger = logging.getLogger(__name__) |
|
|
|
def detect_anomalies(df): |
|
"""Detect anomalies in log data using a Hugging Face model.""" |
|
logger.info("Detecting anomalies...") |
|
try: |
|
detector = pipeline( |
|
"text-classification", |
|
model="prajjwal1/bert-tiny", |
|
tokenizer="prajjwal1/bert-tiny", |
|
clean_up_tokenization_spaces=True |
|
) |
|
df["text"] = df["status"] + " Usage:" + df["usage_count"].astype(str) |
|
results = detector(df["text"].tolist()) |
|
df["anomaly"] = [r["label"] for r in results] |
|
anomalies = df[df["anomaly"] == "POSITIVE"] |
|
logger.info(f"Detected {len(anomalies)} anomalies.") |
|
return anomalies |
|
except Exception as e: |
|
logger.error(f"Failed to detect anomalies: {e}") |
|
raise |
|
``` |