WanIrfan/Atlas
0
1import time
2from datetime import datetime
3from collections import defaultdict
4import json
5import os
6
7class MetricsTracker:
8 """
9 Tracks system performance metrics across queries.
10 """
11
12 def __init__(self, save_path="metrics_data.json"):
13 self.save_path = save_path
14 self.metrics = {
15 "total_queries": 0,
16 "rag_success": 0,
17 "web_search_fallback": 0,
18 "trusted_search_used": 0,
19 "general_search_used": 0,
20 "complexity_distribution": {
21 "simple": 0,
22 "moderate": 0,
23 "complex": 0
24 },
25 "domain_usage": {
26 "medical": 0,
27 "islamic": 0,
28 "insurance": 0
29 },
30 "response_times": [],
31 "worker_contributions": {
32 "dense_semantic": 0,
33 "bm25_keyword": 0
34 },
35 "validation_stats": {
36 "valid": 0,
37 "invalid": 0,
38 "skipped": 0
39 },
40 "query_history": [] # Store recent queries for analysis
41 }
42
43 # Load existing metrics if available
44 self.load_metrics()
45
46 def start_query(self):
47 """Start timing a query."""
48 return time.time()
49
50 def end_query(self, start_time):
51 """Calculate and store query response time."""
52 response_time = time.time() - start_time
53 self.metrics["response_times"].append(response_time)
54 return response_time
55
56 def log_query(self, query, domain, source, complexity=None,
57 validation=None, response_time=None, answer_preview=None):
58 """
59 Log a complete query with all its metadata.
60
61 Args:
62 query (str): User's query
63 domain (str): Domain (medical, islamic, insurance)
64 source (str): Where answer came from (RAG, WebSearch, etc.)
65 complexity (dict): Complexity analysis result
66 validation (tuple): (is_valid, reason)
67 response_time (float): Time taken in seconds
68 answer_preview (str): First 100 chars of answer
69 """
70 self.metrics["total_queries"] += 1
71
72 # Track domain usage
73 if domain in self.metrics["domain_usage"]:
74 self.metrics["domain_usage"][domain] += 1
75
76 # Track source usage
77 if "RAG" in source or "Database" in source:
78 self.metrics["rag_success"] += 1
79 elif "Trusted" in source:
80 self.metrics["trusted_search_used"] += 1
81 self.metrics["web_search_fallback"] += 1
82 elif "Etiqa" in source:
83 self.metrics["web_search_fallback"] += 1
84 elif "Web" in source or "Search" in source:
85 self.metrics["general_search_used"] += 1
86 self.metrics["web_search_fallback"] += 1
87
88 # Track complexity distribution
89 if complexity and "complexity" in complexity:
90 comp_level = complexity["complexity"]
91 if comp_level in self.metrics["complexity_distribution"]:
92 self.metrics["complexity_distribution"][comp_level] += 1
93
94 # Track validation
95 if validation:
96 is_valid, reason = validation
97 if "skip" in reason.lower():
98 self.metrics["validation_stats"]["skipped"] += 1
99 elif is_valid:
100 self.metrics["validation_stats"]["valid"] += 1
101 else:
102 self.metrics["validation_stats"]["invalid"] += 1
103
104 # Store query history (last 50 queries)
105 query_record = {
106 "timestamp": datetime.now().isoformat(),
107 "query": query[:100], # Truncate long queries
108 "domain": domain,
109 "source": source,
110 "complexity": complexity.get("complexity") if complexity else None,
111 "k_used": complexity.get("k") if complexity else None,
112 "response_time": round(response_time, 2) if response_time else None,
113 "validated": is_valid if validation else None,
114 "answer_preview": answer_preview[:100] if answer_preview else None
115 }
116
117 self.metrics["query_history"].append(query_record)
118
119 # Keep only last 50 queries
120 if len(self.metrics["query_history"]) > 50:
121 self.metrics["query_history"] = self.metrics["query_history"][-50:]
122
123 # Auto-save after each query
124 self.save_metrics()
125
126 def log_worker_contribution(self, worker_stats):
127 """
128 Log which swarm workers contributed to the final answer.
129
130 Args:
131 worker_stats (dict): e.g., {"dense_semantic": 5, "bm25_keyword": 3}
132 """
133 for worker, count in worker_stats.items():
134 if worker in self.metrics["worker_contributions"]:
135 self.metrics["worker_contributions"][worker] += count
136
137 def get_stats(self):
138 """Get current statistics."""
139 total = self.metrics["total_queries"]
140
141 if total == 0:
142 return {
143 "total_queries": 0,
144 "rag_success_rate": 0,
145 "web_search_rate": 0,
146 "avg_response_time": 0,
147 "complexity_distribution": self.metrics["complexity_distribution"],
148 "domain_usage": self.metrics["domain_usage"]
149 }
150
151 # Calculate averages and percentages
152 avg_response_time = (
153 sum(self.metrics["response_times"]) / len(self.metrics["response_times"])
154 if self.metrics["response_times"] else 0
155 )
156
157 stats = {
158 "total_queries": total,
159 "rag_success_rate": round((self.metrics["rag_success"] / total) * 100, 1),
160 "web_search_rate": round((self.metrics["web_search_fallback"] / total) * 100, 1),
161 "trusted_search_rate": round((self.metrics["trusted_search_used"] / total) * 100, 1),
162 "general_search_rate": round((self.metrics["general_search_used"] / total) * 100, 1),
163 "avg_response_time": round(avg_response_time, 2),
164 "median_response_time": self._get_median(self.metrics["response_times"]),
165 "complexity_distribution": self.metrics["complexity_distribution"],
166 "domain_usage": self.metrics["domain_usage"],
167 "worker_contributions": self.metrics["worker_contributions"],
168 "validation_stats": self.metrics["validation_stats"],
169 "recent_queries": self.metrics["query_history"][-10:] # Last 10 queries
170 }
171
172 return stats
173
174 def _get_median(self, values):
175 """Calculate median of a list."""
176 if not values:
177 return 0
178 sorted_values = sorted(values)
179 n = len(sorted_values)
180 mid = n // 2
181 if n % 2 == 0:
182 return round((sorted_values[mid-1] + sorted_values[mid]) / 2, 2)
183 return round(sorted_values[mid], 2)
184
185 def save_metrics(self):
186 """Save metrics to JSON file."""
187 try:
188 with open(self.save_path, 'w') as f:
189 json.dump(self.metrics, f, indent=2)
190 except Exception as e:
191 print(f"Warning: Could not save metrics: {e}")
192
193 def load_metrics(self):
194 """Load metrics from JSON file if it exists."""
195 if os.path.exists(self.save_path):
196 try:
197 with open(self.save_path, 'r') as f:
198 self.metrics = json.load(f)
199 print(f"✅ Loaded existing metrics from {self.save_path}")
200 except Exception as e:
201 print(f"Warning: Could not load metrics: {e}")
202
203 def reset_metrics(self):
204 """Reset all metrics (useful for testing)."""
205 self.__init__(self.save_path)
206 self.save_metrics()