File size: 2,050 Bytes
dec98e0 4e2190a dec98e0 4e2190a dec98e0 4e2190a dec98e0 4e2190a dec98e0 4e2190a dec98e0 4e2190a dec98e0 4e2190a dec98e0 4e2190a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 |
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 model specialized in log analysis
classifier = pipeline(
"text-classification",
model="microsoft/codebert-base", # Using CodeBERT which is better for technical text
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
# CodeBERT returns probabilities for both classes
for result in results:
if result['label'] == 'LABEL_1' and result['score'] > 0.7:
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 AI. 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)
|