Spaces:
Runtime error
Runtime error
import glob | |
import json | |
import math | |
import os | |
from dataclasses import dataclass | |
from src.display.formatting import make_clickable_model | |
from src.display.utils import AutoEvalColumn, ModelType, Tasks, Precision, WeightType | |
from src.submission.check_validity import is_model_on_hub | |
class EvalResult: | |
"""Represents one perplexity evaluation result.""" | |
eval_name: str # org_model_precision (uid) | |
full_model: str # org/model (path on hub) | |
org: str | |
model: str | |
revision: str # commit hash, "" if main | |
results: dict | |
precision: Precision = Precision.Unknown | |
model_type: ModelType = ModelType.PT # Default to pretrained | |
weight_type: WeightType = WeightType.Original | |
architecture: str = "Unknown" | |
still_on_hub: bool = False | |
def init_from_json_file(self, json_filepath): | |
"""Inits the result from the specific model result file""" | |
with open(json_filepath) as fp: | |
data = json.load(fp) | |
config = data.get("config") | |
# Precision | |
precision = Precision.from_str(config.get("model_dtype")) | |
# Get model and org | |
org_and_model = config.get("model_name", config.get("model_args", None)) | |
org_and_model = org_and_model.split("/", 1) | |
if len(org_and_model) == 1: | |
org = None | |
model = org_and_model[0] | |
result_key = f"{model}_{precision.value.name}" | |
else: | |
org = org_and_model[0] | |
model = org_and_model[1] | |
result_key = f"{org}_{model}_{precision.value.name}" | |
full_model = "/".join(org_and_model) | |
still_on_hub, _, model_config = is_model_on_hub( | |
full_model, config.get("model_sha", "main"), trust_remote_code=True, test_tokenizer=False | |
) | |
architecture = "?" | |
if model_config is not None: | |
architectures = getattr(model_config, "architectures", None) | |
if architectures: | |
architecture = ";".join(architectures) | |
# Extract perplexity result | |
results = {} | |
if "perplexity" in data["results"]: | |
results["perplexity"] = data["results"]["perplexity"]["perplexity"] | |
return self( | |
eval_name=result_key, | |
full_model=full_model, | |
org=org, | |
model=model, | |
results=results, | |
precision=precision, | |
revision=config.get("model_sha", ""), | |
still_on_hub=still_on_hub, | |
architecture=architecture | |
) | |
def to_dict(self): | |
"""Converts the Eval Result to a dict compatible with our dataframe display""" | |
# Calculate average, handling perplexity (lower is better) | |
scores = [] | |
for task in Tasks: | |
if task.value.benchmark in self.results: | |
score = self.results[task.value.benchmark] | |
# Convert perplexity to a 0-100 scale where lower perplexity = higher score | |
# Using a log scale since perplexity can vary widely | |
# Cap at 100 for very low perplexity and 0 for very high perplexity | |
score = max(0, min(100, 100 * (1 - math.log(score) / 10))) | |
scores.append(score) | |
average = sum(scores) / len(scores) if scores else 0 | |
data_dict = { | |
"eval_name": self.eval_name, # not a column, just a save name, | |
AutoEvalColumn.precision.name: self.precision.value.name, | |
AutoEvalColumn.model_type.name: self.model_type.value.name, | |
AutoEvalColumn.model_type_symbol.name: self.model_type.value.symbol, | |
AutoEvalColumn.weight_type.name: self.weight_type.value.name, | |
AutoEvalColumn.architecture.name: self.architecture, | |
AutoEvalColumn.model.name: make_clickable_model(self.full_model), | |
AutoEvalColumn.revision.name: self.revision, | |
AutoEvalColumn.average.name: average, | |
AutoEvalColumn.still_on_hub.name: self.still_on_hub, | |
} | |
for task in Tasks: | |
benchmark = task.value.benchmark | |
if benchmark in self.results: | |
score = self.results[benchmark] | |
# Store original perplexity score (lower is better) | |
data_dict[task.value.col_name] = score | |
else: | |
data_dict[task.value.col_name] = None | |
return data_dict | |
def get_raw_eval_results(results_path: str) -> list[EvalResult]: | |
"""From the path of the results folder root, extract all perplexity results""" | |
model_result_filepaths = [] | |
for root, _, files in os.walk(results_path): | |
# We should only have json files in model results | |
if len(files) == 0 or any([not f.endswith(".json") for f in files]): | |
continue | |
for file in files: | |
model_result_filepaths.append(os.path.join(root, file)) | |
eval_results = {} | |
for model_result_filepath in model_result_filepaths: | |
# Creation of result | |
eval_result = EvalResult.init_from_json_file(model_result_filepath) | |
# Store results of same eval together | |
eval_name = eval_result.eval_name | |
if eval_name in eval_results.keys(): | |
eval_results[eval_name].results.update({k: v for k, v in eval_result.results.items() if v is not None}) | |
else: | |
eval_results[eval_name] = eval_result | |
results = [] | |
for v in eval_results.values(): | |
try: | |
v.to_dict() # we test if the dict version is complete | |
results.append(v) | |
except KeyError: # not all eval values present | |
continue | |
return results | |