File size: 6,369 Bytes
8c404fc |
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 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 |
import json
import os
from typing import Dict, List, Tuple
class RAGScoreCalculator:
"""
Dynamic RAG Score calculator that calculates scores at runtime
without modifying the original JSON files.
"""
def __init__(self, retrieval_dir: str = "result/retrieval"):
self.retrieval_dir = retrieval_dir
self.stats = None
self.all_data = None
self._load_and_analyze()
def _load_and_analyze(self):
"""Load all retrieval detail files and calculate normalization statistics."""
self.all_data = []
detail_files = [f for f in os.listdir(self.retrieval_dir) if f.startswith('detail_')]
if not detail_files:
print("Warning: No detail files found in retrieval directory")
return
for filename in detail_files:
filepath = os.path.join(self.retrieval_dir, filename)
try:
with open(filepath, 'r') as f:
data = json.load(f)
self.all_data.append(data)
except Exception as e:
print(f"Error loading {filename}: {e}")
continue
if not self.all_data:
print("Warning: No valid data loaded from detail files")
return
# Calculate normalization statistics
self._calculate_stats()
def _calculate_stats(self):
"""Calculate min/max statistics for normalization."""
if not self.all_data:
return
# Extract values for analysis
rag_success_rates = [d.get('RAG_success_rate', 0) for d in self.all_data]
max_correct_refs = [d.get('max_correct_references', 0) for d in self.all_data]
false_positives = [d.get('total_false_positives', 0) for d in self.all_data]
missed_refs = [d.get('total_missed_references', 0) for d in self.all_data]
# Calculate min/max for normalization
self.stats = {
'rag_success_rate': {
'min': min(rag_success_rates),
'max': max(rag_success_rates)
},
'max_correct_references': {
'min': min(max_correct_refs),
'max': max(max_correct_refs)
},
'total_false_positives': {
'min': min(false_positives),
'max': max(false_positives)
},
'total_missed_references': {
'min': 0, # Fixed minimum value
'max': 7114 # Fixed maximum value
}
}
def normalize_value(self, value, min_val, max_val, higher_is_better=True):
"""Normalize a value to 0-1 range."""
if max_val == min_val:
return 1.0 # If all values are the same, return 1
normalized = (value - min_val) / (max_val - min_val)
if not higher_is_better:
normalized = 1 - normalized # Flip for "lower is better" metrics
return normalized
def calculate_rag_score(self, data: Dict) -> float:
"""Calculate the RAG score for a single model's data."""
if not self.stats:
print("Warning: No statistics available for normalization")
return 0.0
# Extract values with defaults
rag_success_rate = data.get('RAG_success_rate', 0)
max_correct_refs = data.get('max_correct_references', 0)
false_positives = data.get('total_false_positives', 0)
missed_refs = data.get('total_missed_references', 0)
# Normalize values (0-1)
norm_max_correct = self.normalize_value(
max_correct_refs,
self.stats['max_correct_references']['min'],
self.stats['max_correct_references']['max'],
higher_is_better=True
)
norm_false_positives = self.normalize_value(
false_positives,
self.stats['total_false_positives']['min'],
self.stats['total_false_positives']['max'],
higher_is_better=False # Lower is better
)
norm_missed_refs = self.normalize_value(
missed_refs,
self.stats['total_missed_references']['min'],
self.stats['total_missed_references']['max'],
higher_is_better=False # Lower is better
)
# Calculate weighted score
# Weights: rag_success_rate=0.9, false_positives=0.9, max_correct=0.1, missed_refs=0.1
rag_score = (
0.9 * rag_success_rate +
0.9 * norm_false_positives +
0.1 * norm_max_correct +
0.1 * norm_missed_refs
) / 2.0 # Divide by 2 since total weights = 2.0
return round(rag_score, 4)
def get_normalization_info(self) -> Dict:
"""Get current normalization statistics for debugging."""
return {
'stats': self.stats,
'total_files': len(self.all_data) if self.all_data else 0,
'retrieval_dir': self.retrieval_dir
}
def refresh_stats(self):
"""Refresh statistics by reloading data - call this when new data is added."""
print("Refreshing RAG Score normalization statistics...")
self._load_and_analyze()
return self.stats is not None
def main():
"""Main function for testing RAG score calculations."""
calculator = RAGScoreCalculator()
print("RAG Score Calculator (Runtime Only)")
print("===================================")
# Show normalization info
info = calculator.get_normalization_info()
print(f"Total files: {info['total_files']}")
print(f"Retrieval directory: {info['retrieval_dir']}")
if info['stats']:
print("\nNormalization ranges:")
for metric, data in info['stats'].items():
print(f" {metric}: {data['min']} - {data['max']}")
print("\nSample RAG Score calculations:")
for i, data in enumerate(calculator.all_data[:5]): # Show first 5
rag_score = calculator.calculate_rag_score(data)
model_name = data.get('model_name', 'Unknown')
print(f" {model_name}: {rag_score}")
else:
print("\n❌ No statistics available for normalization")
if __name__ == "__main__":
main() |