Agent Monitoring

Complete Implementation:

import asyncio
from typing import Dict, List, Any
import numpy as np
from datetime import datetime, timedelta

class LLMMonitor:
    def __init__(self):
        self.metrics_buffer = []
        self.baselines = {}
        self.alert_thresholds = {
            "response_time": 2000,  # ms
            "token_count": 4000,
            "confidence_drop": 0.3,
            "repetition_ratio": 0.4
        }
        
    async def monitor_interaction(self, 
                                 agent_id: str,
                                 request: Dict,
                                 response: Dict):
        """Monitor single LLM interaction"""
        
        # Extract metrics
        metrics = {
            "agent_id": agent_id,
            "timestamp": datetime.utcnow(),
            "request_length": len(request["prompt"]),
            "response_length": len(response["text"]),
            "response_time": response["latency"],
            "confidence": response.get("confidence", 1.0),
            "token_count": response["token_count"],
            "repetition_score": self._calculate_repetition(
                response["text"]
            ),
            "sentiment_score": self._analyze_sentiment(
                response["text"]
            ),
            "toxicity_score": self._check_toxicity(
                response["text"]
            )
        }
        
        # Check for anomalies
        anomalies = await self._detect_anomalies(metrics)
        
        # Store metrics
        await self._store_metrics(metrics)
        
        # Handle anomalies
        if anomalies:
            await self._handle_anomalies(agent_id, anomalies)
            
        return {
            "metrics": metrics,
            "anomalies": anomalies,
            "health_score": self._calculate_health_score(metrics)
        }
    
    async def _detect_anomalies(self, metrics: Dict) -> List[Dict]:
        anomalies = []
        
        # 1. Response time anomaly
        if metrics["response_time"] > self.alert_thresholds["response_time"]:
            anomalies.append({
                "type": "high_latency",
                "severity": "medium",
                "value": metrics["response_time"]
            })
        
        # 2. Token explosion
        if metrics["token_count"] > self.alert_thresholds["token_count"]:
            anomalies.append({
                "type": "token_explosion",
                "severity": "high",
                "value": metrics["token_count"]
            })
        
        # 3. Confidence drop
        baseline_confidence = self.baselines.get(
            metrics["agent_id"], {}
        ).get("confidence", 0.8)
        
        if metrics["confidence"] < baseline_confidence - self.alert_thresholds["confidence_drop"]:
            anomalies.append({
                "type": "confidence_drop",
                "severity": "high",
                "value": metrics["confidence"]
            })
        
        # 4. High repetition (possible loop)
        if metrics["repetition_score"] > self.alert_thresholds["repetition_ratio"]:
            anomalies.append({
                "type": "repetition_detected",
                "severity": "medium",
                "value": metrics["repetition_score"]
            })
        
        # 5. Prompt injection attempt
        if self._detect_injection_patterns(metrics):
            anomalies.append({
                "type": "prompt_injection_attempt",
                "severity": "critical",
                "details": "Suspicious patterns detected"
            })
        
        return anomalies
    
    def _calculate_repetition(self, text: str) -> float:
        """Calculate text repetition score"""
        words = text.lower().split()
        if len(words) < 10:
            return 0.0
            
        # N-gram analysis
        ngrams = []
        for n in [2, 3, 4]:  # 2,3,4-grams
            for i in range(len(words) - n + 1):
                ngrams.append(" ".join(words[i:i+n]))
        
        # Calculate repetition
        unique_ngrams = len(set(ngrams))
        total_ngrams = len(ngrams)
        
        if total_ngrams == 0:
            return 0.0
            
        return 1 - (unique_ngrams / total_ngrams)
    
    async def continuous_monitoring(self):
        """Main monitoring loop"""
        while True:
            # Process buffered metrics
            if self.metrics_buffer:
                await self._process_metrics_batch()
            
            # Update baselines
            await self._update_baselines()
            
            # Check system health
            await self._system_health_check()
            
            await asyncio.sleep(10)  # 10-second intervals

# Usage example
monitor = LLMMonitor()

# In your LLM serving code
async def serve_llm_request(request):
    agent_id = request["agent_id"]
    
    # Process request
    start_time = time.time()
    response = await llm_model.generate(request["prompt"])
    response["latency"] = (time.time() - start_time) * 1000
    
    # Monitor
    monitoring_result = await monitor.monitor_interaction(
        agent_id, request, response
    )
    
    # Add monitoring metadata
    response["monitoring"] = monitoring_result
    
    return response