Upload 2 files

anomaly_detector.py

@@ -0,0 +1,63 @@
+import numpy as np
+from collections import deque
+
+class AnomalyDetector:
+    def __init__(self, window_size=20, threshold=2.5):
+        """
+        Physics-based Anomaly Detection.
+        We keep a 'window' of recent data to establish a baseline.
+        If a new data point deviates by > 2.5 Standard Deviations (Z-Score),
+        we flag it as an anomaly.
+        
+        :param window_size: How many recent data points to remember.
+        :param threshold: The Z-Score limit (Physics standard is usually 3-sigma, we use 2.5).
+        """
+        self.window_size = window_size
+        self.threshold = threshold
+        # A deque is like a list that automatically drops old data when full
+        self.data_window = deque(maxlen=window_size)
+
+    def update(self, value):
+        """
+        Ingest new data, update statistics, and check for drift.
+        Returns: (is_anomaly, message, current_z_score)
+        """
+        # 1. Add new value to history
+        self.data_window.append(value)
+        
+        # 2. Need enough data to establish a baseline (at least 5 points)
+        if len(self.data_window) < 5:
+            return False, "Initializing baseline...", 0.0
+
+        # 3. Calculate Physics Metrics (Mean & Standard Deviation)
+        mean = np.mean(self.data_window)
+        std_dev = np.std(self.data_window)
+
+        # Avoid division by zero if flatline
+        if std_dev == 0:
+            return False, "Stable", 0.0
+
+        # 4. Calculate Z-Score (How far is this point from the 'Normal'?)
+        z_score = (value - mean) / std_dev
+
+        # 5. Check Threshold
+        if abs(z_score) > self.threshold:
+            return True, f"CRITICAL: Anomaly Detected! Value {value} is {z_score:.2f}σ deviation.", z_score
+        
+        return False, "Normal", z_score
+
+# --- Quick Test Block (Runs only if you play this file directly) ---
+if __name__ == "__main__":
+    print("🔬 Initializing Sentinel Detector...")
+    detector = AnomalyDetector()
+    
+    # 1. Simulate Normal Traffic (Values around 50)
+    normal_data = [50, 52, 49, 51, 50, 48, 53, 50, 51, 49]
+    for val in normal_data:
+        detector.update(val)
+        print(f"Input: {val} | Status: OK")
+
+    # 2. Simulate a SUDDEN SPIKE (Anomaly)
+    spike = 120 # Massive jump
+    is_anomaly, msg, z = detector.update(spike)
+    print(f"\n⚠️ Input: {spike} | {msg}")

main.py

@@ -0,0 +1,68 @@
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel
+import uvicorn
+import logging
+
+# Import our custom modules
+from anomaly_detector import AnomalyDetector
+from rag_agent import SentinelAgent
+
+# 1. Initialize the App
+app = FastAPI(
+    title="Sentinel MLOps Agent",
+    description="Autonomous Anomaly Detection & RAG Investigation API",
+    version="1.0"
+)
+
+# 2. Load the Engines
+print("🔋 Starting Sentinel Engines...")
+detector = AnomalyDetector()
+agent = SentinelAgent()
+
+# 3. Define the Data Format (Validation)
+class MetricData(BaseModel):
+    timestamp: str
+    service_name: str
+    cpu_usage: float
+
+# 4. The API Endpoint
+@app.post("/monitor")
+async def monitor_system(data: MetricData):
+    """
+    Receives live system data.
+    - If Normal: Returns OK.
+    - If Anomaly: Triggers Gemini 2.5 to investigate.
+    """
+    # Step A: Check for Physics/Math Anomaly
+    is_anomaly, msg, z_score = detector.update(data.cpu_usage)
+    
+    if not is_anomaly:
+        return {
+            "status": "Healthy", 
+            "message": msg, 
+            "z_score": z_score
+        }
+    
+    # Step B: If Anomaly, Wake up the Agent
+    print(f"🚨 ALERT: Anomaly on {data.service_name} detected!")
+    
+    report = agent.investigate(
+        anomaly_value=data.cpu_usage, 
+        z_score=z_score
+    )
+    
+    # Return the Full Incident Report
+    return {
+        "status": "CRITICAL",
+        "service": data.service_name,
+        "deviation": f"{z_score:.2f} sigma",
+        "investigation_report": report
+    }
+
+@app.get("/")
+def home():
+    return {"message": "Sentinel AI Agent is Active. Send POST to /monitor"}
+
+# 5. Run the Server (Mobile Friendly)
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=8000)