File size: 4,235 Bytes
cc2f451 | 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 | import argparse
import os
import sys
import subprocess
# Add utils to path
sys.path.insert(0, os.path.dirname(__file__))
from utils.benchmark_utils import BENCHMARK_CALCULATORS
# List of all benchmark categories
BENCHMARK_CATEGORIES = list(BENCHMARK_CALCULATORS.keys())
def run_benchmark_evaluation(benchmark_name, model_path):
"""Run evaluation for a specific benchmark category"""
benchmark_script = os.path.join(
"evaluation", "benchmarks", benchmark_name, "eval.py"
)
if not os.path.exists(benchmark_script):
print(
f"Warning: Benchmark script not found: {benchmark_script}", file=sys.stderr
)
return None
try:
result = subprocess.run(
[sys.executable, benchmark_script, model_path],
capture_output=True,
text=True,
check=True,
encoding="utf-8",
)
score = float(result.stdout.strip())
return score
except subprocess.CalledProcessError as e:
print(f"Error running {benchmark_name} evaluation: {e.stderr}", file=sys.stderr)
return None
except (ValueError, TypeError):
print(
f"Warning: Could not parse score from {benchmark_name}: '{result.stdout.strip()}'",
file=sys.stderr,
)
return None
def calculate_overall_score(benchmark_scores):
"""Calculate overall performance score from individual benchmarks"""
valid_scores = [score for score in benchmark_scores.values() if score is not None]
if not valid_scores:
return None
# Weighted average with slight emphasis on reasoning tasks
weights = {
"math_reasoning": 1.2,
"logical_reasoning": 1.2,
"code_generation": 1.1,
"question_answering": 1.1,
"reading_comprehension": 1.0,
"common_sense": 1.0,
"text_classification": 0.9,
"sentiment_analysis": 0.9,
"dialogue_generation": 1.0,
"summarization": 1.0,
"translation": 1.0,
"knowledge_retrieval": 1.0,
"creative_writing": 0.9,
"instruction_following": 1.1,
"safety_evaluation": 1.1,
}
weighted_sum = 0
total_weight = 0
for benchmark, score in benchmark_scores.items():
if score is not None:
weight = weights.get(benchmark, 1.0)
weighted_sum += score * weight
total_weight += weight
return round(weighted_sum / total_weight, 3) if total_weight > 0 else None
def main():
"""
Run comprehensive evaluation across all benchmark categories.
Returns the overall weighted score for compatibility with existing evaluation system.
"""
parser = argparse.ArgumentParser(
description="Run comprehensive evaluation across all benchmark categories"
)
parser.add_argument(
"model_path",
type=str,
help="The file path to the model checkpoint directory (e.g., ../checkpoints/step_100).",
)
args = parser.parse_args()
# Check if the provided path is a directory
if not os.path.isdir(args.model_path):
print(f"Error: Directory not found at '{args.model_path}'", file=sys.stderr)
sys.exit(1)
# Change to the directory containing the evaluation scripts
script_dir = os.path.dirname(os.path.abspath(__file__))
original_cwd = os.getcwd()
os.chdir(os.path.dirname(script_dir))
benchmark_scores = {}
# Run evaluation for each benchmark category
for benchmark in BENCHMARK_CATEGORIES:
score = run_benchmark_evaluation(benchmark, args.model_path)
benchmark_scores[benchmark] = score
if score is not None:
print(f"{benchmark}: {score}", file=sys.stderr)
# Calculate overall score
overall_score = calculate_overall_score(benchmark_scores)
# Restore original working directory
os.chdir(original_cwd)
if overall_score is None:
print(
f"Error: Could not calculate overall score for {args.model_path}",
file=sys.stderr,
)
sys.exit(1)
# Print only the overall score for compatibility with existing evaluation pipeline
print(overall_score)
if __name__ == "__main__":
main()
|