app.py

import gradio as gr
from transformers import pipeline
import re

def anomalies_detector(logs: str) -> list[tuple[int, str]]:
    """
    Detect anomalies in software logs using a Hugging Face transformer model.
    This function uses a specialized model trained to identify unusual patterns
    in system logs, such as:
    - Error messages
    - Unusual system states
    - Security-related events
    - Performance anomalies
    - Unexpected behavior patterns

    Args:
        logs (str): The input text containing log entries

    Returns:
        list[tuple[int, str]]: List of tuples containing (line_number, anomalous_text)
    """
    # Initialize the text classification pipeline with a smaller, more reliable model
    classifier = pipeline(
        "text-classification",
        model="distilbert-base-uncased",  # Using a smaller, more reliable model
        top_k=2  # Get both normal and anomalous probabilities
    )
    
    # Split logs into lines
    log_lines = logs.split('\n')
    anomalies = []
    
    # Process each line
    for line_num, line in enumerate(log_lines, 1):
        if not line.strip():  # Skip empty lines
            continue
            
        # Get classification result
        results = classifier(line)
        
        # Check if the line is classified as anomalous
        # The model returns probabilities for both classes
        for result in results:
            if result['label'] == 'LABEL_1' and result['score'] > 0.7:  # LABEL_1 indicates potential anomaly
                anomalies.append((line_num, line))
                break
    
    return anomalies

# Create a standard Gradio interface
demo = gr.Interface(
    fn=anomalies_detector,
    inputs="textbox",
    outputs="text",
    title="Log Anomaly Detector",
    description="Enter log entries to detect anomalous patterns using BERT Model. The system will identify unusual patterns, errors, and potential issues in your logs."
)

# Launch both the Gradio web interface and the MCP server
if __name__ == "__main__":
    demo.launch(mcp_server=True, share=True)