model-evaluator / utils.py
Tristan's picture
Tristan
HF staff
add zero shot classification task (#45)
d68b7d5 unverified
import inspect
import uuid
from typing import Dict, List, Union
import jsonlines
import requests
import streamlit as st
from evaluate import load
from huggingface_hub import HfApi, ModelFilter, Repository, dataset_info, list_metrics
from tqdm import tqdm
AUTOTRAIN_TASK_TO_HUB_TASK = {
"binary_classification": "text-classification",
"multi_class_classification": "text-classification",
"natural_language_inference": "text-classification",
"entity_extraction": "token-classification",
"extractive_question_answering": "question-answering",
"translation": "translation",
"summarization": "summarization",
"image_binary_classification": "image-classification",
"image_multi_class_classification": "image-classification",
"text_zero_shot_classification": "text-generation",
}
HUB_TASK_TO_AUTOTRAIN_TASK = {v: k for k, v in AUTOTRAIN_TASK_TO_HUB_TASK.items()}
LOGS_REPO = "evaluation-job-logs"
def get_auth_headers(token: str, prefix: str = "Bearer"):
return {"Authorization": f"{prefix} {token}"}
def http_post(path: str, token: str, payload=None, domain: str = None, params=None) -> requests.Response:
"""HTTP POST request to the AutoNLP API, raises UnreachableAPIError if the API cannot be reached"""
try:
response = requests.post(
url=domain + path,
json=payload,
headers=get_auth_headers(token=token),
allow_redirects=True,
params=params,
)
except requests.exceptions.ConnectionError:
print("❌ Failed to reach AutoNLP API, check your internet connection")
response.raise_for_status()
return response
def http_get(path: str, domain: str, token: str = None, params: dict = None) -> requests.Response:
"""HTTP POST request to `path`, raises UnreachableAPIError if the API cannot be reached"""
try:
response = requests.get(
url=domain + path,
headers=get_auth_headers(token=token),
allow_redirects=True,
params=params,
)
except requests.exceptions.ConnectionError:
print(f"❌ Failed to reach {path}, check your internet connection")
response.raise_for_status()
return response
def get_metadata(dataset_name: str, token: str) -> Union[Dict, None]:
data = dataset_info(dataset_name, token=token)
if data.cardData is not None and "train-eval-index" in data.cardData.keys():
return data.cardData["train-eval-index"]
else:
return None
def get_compatible_models(task: str, dataset_ids: List[str]) -> List[str]:
"""
Returns all model IDs that are compatible with the given task and dataset names.
Args:
task (`str`): The task to search for.
dataset_names (`List[str]`): A list of dataset names to search for.
Returns:
A list of model IDs, sorted alphabetically.
"""
compatible_models = []
# Allow any summarization model to be used for summarization tasks
# and allow any text-generation model to be used for text_zero_shot_classification
if task in ("summarization", "text_zero_shot_classification"):
model_filter = ModelFilter(
task=AUTOTRAIN_TASK_TO_HUB_TASK[task],
library=["transformers", "pytorch"],
)
compatible_models.extend(HfApi().list_models(filter=model_filter))
# Include models trained on SQuAD datasets, since these can be evaluated on
# other SQuAD-like datasets
if task == "extractive_question_answering":
dataset_ids.extend(["squad", "squad_v2"])
# TODO: relax filter on PyTorch models if TensorFlow supported in AutoTrain
for dataset_id in dataset_ids:
model_filter = ModelFilter(
task=AUTOTRAIN_TASK_TO_HUB_TASK[task],
trained_dataset=dataset_id,
library=["transformers", "pytorch"],
)
compatible_models.extend(HfApi().list_models(filter=model_filter))
return sorted(set([model.modelId for model in compatible_models]))
def get_key(col_mapping, val):
for key, value in col_mapping.items():
if val == value:
return key
return "key doesn't exist"
def format_col_mapping(col_mapping: dict) -> dict:
for k, v in col_mapping["answers"].items():
col_mapping[f"answers.{k}"] = f"answers.{v}"
del col_mapping["answers"]
return col_mapping
def commit_evaluation_log(evaluation_log, hf_access_token=None):
logs_repo_url = f"https://huggingface.co/datasets/autoevaluate/{LOGS_REPO}"
logs_repo = Repository(
local_dir=LOGS_REPO,
clone_from=logs_repo_url,
repo_type="dataset",
private=True,
use_auth_token=hf_access_token,
)
logs_repo.git_pull()
with jsonlines.open(f"{LOGS_REPO}/logs.jsonl") as r:
lines = []
for obj in r:
lines.append(obj)
lines.append(evaluation_log)
with jsonlines.open(f"{LOGS_REPO}/logs.jsonl", mode="w") as writer:
for job in lines:
writer.write(job)
logs_repo.push_to_hub(
commit_message=f"Evaluation submitted with project name {evaluation_log['payload']['proj_name']}"
)
print("INFO -- Pushed evaluation logs to the Hub")
@st.experimental_memo
def get_supported_metrics():
"""Helper function to get all metrics compatible with evaluation service.
Requires all metric dependencies installed in the same environment, so wait until
https://github.com/huggingface/evaluate/issues/138 is resolved before using this.
"""
metrics = [metric.id for metric in list_metrics()]
supported_metrics = []
for metric in tqdm(metrics):
# TODO: this currently requires all metric dependencies to be installed
# in the same environment. Refactor to avoid needing to actually load
# the metric.
try:
print(f"INFO -- Attempting to load metric: {metric}")
metric_func = load(metric)
except Exception as e:
print(e)
print("WARNING -- Skipping the following metric, which cannot load:", metric)
continue
argspec = inspect.getfullargspec(metric_func.compute)
if "references" in argspec.kwonlyargs and "predictions" in argspec.kwonlyargs:
# We require that "references" and "predictions" are arguments
# to the metric function. We also require that the other arguments
# besides "references" and "predictions" have defaults and so do not
# need to be specified explicitly.
defaults = True
for key, value in argspec.kwonlydefaults.items():
if key not in ("references", "predictions"):
if value is None:
defaults = False
break
if defaults:
supported_metrics.append(metric)
return supported_metrics
def get_dataset_card_url(dataset_id: str) -> str:
"""Gets the URL to edit the dataset card for the given dataset ID."""
if "/" in dataset_id:
return f"https://huggingface.co/datasets/{dataset_id}/edit/main/README.md"
else:
return f"https://github.com/huggingface/datasets/edit/master/datasets/{dataset_id}/README.md"
def create_autotrain_project_name(dataset_id: str, dataset_config: str) -> str:
"""Creates an AutoTrain project name for the given dataset ID."""
# Project names cannot have "/", so we need to format community datasets accordingly
dataset_id_formatted = dataset_id.replace("/", "__")
dataset_config_formatted = dataset_config.replace("--", "__")
# Project names need to be unique, so we append a random string to guarantee this while adhering to naming rules
basename = f"eval-{dataset_id_formatted}-{dataset_config_formatted}"
basename = basename[:60] if len(basename) > 60 else basename # Hub naming limitation
return f"{basename}-{str(uuid.uuid4())[:6]}"
def get_config_metadata(config: str, metadata: List[Dict] = None) -> Union[Dict, None]:
"""Gets the dataset card metadata for the given config."""
if metadata is None:
return None
config_metadata = [m for m in metadata if m["config"] == config]
if len(config_metadata) >= 1:
return config_metadata[0]
else:
return None