Spaces:
Build error
Build error
File size: 4,638 Bytes
52e8fba |
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 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 |
import json
import os
from datetime import datetime
from typing import Dict, Any, List
import pandas as pd
class MonitoringSystem:
"""
System for monitoring agent performance and tracking metrics.
"""
def __init__(self, log_file="logs/monitoring_logs.json"):
self.log_file = log_file
self._ensure_log_directory()
def _ensure_log_directory(self):
"""Create logs directory if it doesn't exist."""
os.makedirs(os.path.dirname(self.log_file), exist_ok=True)
# Initialize log file if it doesn't exist
if not os.path.exists(self.log_file):
with open(self.log_file, 'w') as f:
json.dump([], f)
def log_generation(self, data: Dict[str, Any]):
"""
Log a code generation event.
Args:
data: Dictionary containing generation details
"""
try:
# Read existing logs
with open(self.log_file, 'r') as f:
logs = json.load(f)
# Add timestamp if not present
if 'timestamp' not in data:
data['timestamp'] = datetime.now().isoformat()
# Add to logs
logs.append(data)
# Write back (limit to last 100 entries)
with open(self.log_file, 'w') as f:
json.dump(logs[-100:], f, indent=2)
except Exception as e:
print(f"Error logging generation: {e}")
def get_metrics(self, n_days: int = 7) -> List[Dict]:
"""
Get metrics from the last n days.
Args:
n_days: Number of days to look back
Returns:
List of log entries
"""
try:
with open(self.log_file, 'r') as f:
logs = json.load(f)
# Filter by date if requested
if n_days:
cutoff_date = datetime.now().timestamp() - (n_days * 24 * 60 * 60)
filtered_logs = [
log for log in logs
if datetime.fromisoformat(log['timestamp']).timestamp() > cutoff_date
]
return filtered_logs
else:
return logs
except Exception as e:
print(f"Error reading metrics: {e}")
return []
def calculate_statistics(self) -> Dict[str, Any]:
"""
Calculate statistics from logs.
Returns:
Dictionary with statistics
"""
logs = self.get_metrics(n_days=None)
if not logs:
return {
"total_generations": 0,
"average_score": 0,
"success_rate": 0
}
# Convert to DataFrame for easier analysis
df = pd.DataFrame(logs)
stats = {
"total_generations": len(df),
"models_used": df['model'].value_counts().to_dict() if 'model' in df.columns else {},
"unique_prompts": df['prompt'].nunique() if 'prompt' in df.columns else 0
}
# Calculate average scores if available
if 'review_score' in df.columns:
stats['average_review_score'] = df['review_score'].mean()
stats['best_score'] = df['review_score'].max()
stats['worst_score'] = df['review_score'].min()
if 'test_score' in df.columns:
test_scores = df['test_score'].dropna()
if len(test_scores) > 0:
stats['average_test_score'] = test_scores.mean()
# Calculate success rate (assuming any generation with a score is successful)
successful = len(df[df['review_score'] > 0]) if 'review_score' in df.columns else 0
stats['success_rate'] = (successful / len(df)) * 100 if len(df) > 0 else 0
return stats
def export_logs(self, format: str = "csv") -> str:
"""
Export logs in specified format.
Args:
format: Output format (csv, json)
Returns:
Path to exported file
"""
logs = self.get_metrics(n_days=None)
df = pd.DataFrame(logs)
if format == "csv":
export_path = self.log_file.replace('.json', '.csv')
df.to_csv(export_path, index=False)
return export_path
elif format == "json":
return self.log_file
else:
raise ValueError(f"Unsupported format: {format}") |