Spaces:
Running
Running
import os | |
import json | |
import datetime | |
from email.utils import parseaddr | |
import gradio as gr | |
import pandas as pd | |
from datasets import load_dataset | |
from evaluation.evaluator import question_scorer as eval_scorer | |
from apscheduler.schedulers.background import BackgroundScheduler | |
from huggingface_hub import HfApi | |
from content import format_error, format_warning, format_log, TITLE | |
# Placeholder for the question_scorer function | |
def question_scorer(prediction, gold_answer): | |
acc, has_ans = eval_scorer(prediction, gold_answer) | |
return acc, has_ans | |
# Constants and Configuration | |
TOKEN = os.environ.get("TOKEN", None) | |
OWNER = "Ori" | |
DATA_DATASET = f"Ori/AssistantBench_V1.0" | |
RESULTS_DATASET = f"Ori/results" | |
SUBMISSION_DATASET = f"AssistantBench/submissions" | |
LEADERBOARD_PATH = f"{OWNER}/leaderboard" | |
api = HfApi() | |
YEAR_VERSION = "default" | |
os.makedirs("scored", exist_ok=True) | |
# Load datasets | |
eval_results = load_dataset(RESULTS_DATASET, token=TOKEN, download_mode="force_redownload", | |
ignore_verifications=True, trust_remote_code=True) | |
gold_results = load_dataset(DATA_DATASET, token=TOKEN, trust_remote_code=True) | |
gold_answers = {split: {row["id"]: row["answer"] for row in gold_results[split]} for split in ["test"]} | |
gold_difficulties = {split: {row["id"]: row["difficulty"] for row in gold_results[split]} for split in ["test"]} | |
# Function to get dataframe from results | |
def get_dataframe_from_results(eval_results, split): | |
local_df = eval_results[split] | |
df = pd.DataFrame(local_df) | |
df = df.sort_values(by=["Accuracy"], ascending=False) | |
numeric_cols = [c for c in local_df.column_names if "score" in c] | |
df[numeric_cols] = df[numeric_cols].multiply(100).round(decimals=2) | |
return df | |
# Update function to format dataframe | |
def format_dataframe(df): | |
df["Accuracy"] = df["Accuracy"].apply(lambda x: f"**{x:.2f}**") | |
if "URL" in df.columns: | |
df["Model Name"] = df.apply(lambda row: f"[{row['Model Name']}]({row['URL']})", axis=1) | |
df = df.drop(columns=["URL"]) | |
df = df.rename(columns={"Model Family": "Base Model"}) | |
df = df[["Model Name", "Accuracy", "Accuracy (easy)", "Accuracy (medium)", "Accuracy (hard)", "Answer rate", "Precision", "EM", "Base Model", "Organization"]] | |
return df | |
eval_dataframe_test = get_dataframe_from_results(eval_results=eval_results, split="test") | |
eval_dataframe_test = format_dataframe(eval_dataframe_test) | |
# Function to restart the space | |
def restart_space(): | |
api.restart_space(repo_id=LEADERBOARD_PATH, token=TOKEN) | |
TYPES = ["markdown", "markdown", "number", "number", "number", "number", "number", "number", "str", "str"] | |
# Function to add a new evaluation | |
def add_new_eval( | |
model_name: str, | |
model_family: str, | |
url: str, | |
path_to_file: str, | |
organization: str, | |
mail: str, | |
): | |
_, parsed_mail = parseaddr(mail) | |
if "@" not in parsed_mail: | |
return format_warning("Please provide a valid email address.") | |
print("Adding new eval") | |
if model_name.lower() in set( | |
[m.lower() for m in eval_results["test"]["Model Name"]]) and organization.lower() in set( | |
[o.lower() for o in eval_results["test"]["Organization"]]): | |
return format_warning("This model has already been submitted.") | |
if path_to_file is None: | |
return format_warning("Please attach a file.") | |
api.upload_file( | |
repo_id=SUBMISSION_DATASET, | |
path_or_fileobj=path_to_file.name, | |
path_in_repo=f"{organization}/{model_name}/{YEAR_VERSION}_test_raw_{datetime.datetime.today()}.jsonl", | |
repo_type="dataset", | |
token=TOKEN | |
) | |
file_path = path_to_file.name | |
scores = 0 | |
num_questions = 0 | |
difficulty_scores = {"Easy": 0, "Medium": 0, "Hard": 0} | |
difficulty_counts = {"Easy": 0, "Medium": 0, "Hard": 0} | |
with open(f"scored/{organization}_{model_name}.jsonl", "w") as scored_file: | |
with open(file_path, 'r') as f: | |
for ix, line in enumerate(f): | |
try: | |
task = json.loads(line) | |
except Exception: | |
return format_error(f"Line {ix} is incorrectly formatted. Please fix it and resubmit your file.") | |
if "answer" not in task: | |
return format_error( | |
f"Line {ix} contains no answer key. Please fix it and resubmit your file.") | |
answer = task["answer"] | |
task_id = task["id"] | |
if task_id not in gold_answers["test"]: | |
return format_error( | |
f"{task_id} not found in test set. Are you sure you submitted the correct file?") | |
score, has_ans = question_scorer(task['answer'], gold_answers["test"][task_id]) | |
difficulty = gold_difficulties["test"][task_id] | |
scored_file.write( | |
json.dumps({ | |
"id": task_id, | |
"model_answer": answer, | |
"score": score, | |
"has_ans": has_ans | |
}) + "\n" | |
) | |
scores += score | |
num_questions += 1 | |
difficulty_scores[difficulty] += score | |
difficulty_counts[difficulty] += 1 | |
accuracy_easy = difficulty_scores["Easy"] / difficulty_counts["Easy"] if difficulty_counts["Easy"] > 0 else 0 | |
accuracy_medium = difficulty_scores["Medium"] / difficulty_counts["Medium"] if difficulty_counts["Medium"] > 0 else 0 | |
accuracy_hard = difficulty_scores["Hard"] / difficulty_counts["Hard"] if difficulty_counts["Hard"] > 0 else 0 | |
api.upload_file( | |
repo_id=SUBMISSION_DATASET, | |
path_or_fileobj=f"scored/{organization}_{model_name}.jsonl", | |
path_in_repo=f"{organization}/{model_name}/{YEAR_VERSION}_test_scored_{datetime.datetime.today()}.jsonl", | |
repo_type="dataset", | |
token=TOKEN | |
) | |
accuracy = float("{:.1f}".format(np.average([x["acc"] for x in scored_file]) * 100)) | |
coverage = float("{:.1f}".format(np.average([x["has_ans"] for x in scored_file]))) | |
em = float("{:.1f}".format(np.average([1 if x["acc"] == 1 else 0 for x in scored_file]))) | |
precision = float("{:.1f}".format(np.average([x["acc"] for x in scored_file if x["has_ans"] == 1]))) | |
eval_entry = { | |
"Model Name": model_name, | |
"Base Model": model_family, | |
"URL": url, | |
"Organization": organization, | |
"Accuracy": accuracy, | |
"Accuracy (easy)": accuracy_easy, | |
"Accuracy (medium)": accuracy_medium, | |
"Accuracy (hard)": accuracy_hard, | |
"Answer rate": coverage, | |
"Precision": precision, | |
"EM": em | |
} | |
eval_results["test"] = eval_results["test"].add_item(eval_entry) | |
eval_results.push_to_hub(RESULTS_DATASET, config_name=YEAR_VERSION, token=TOKEN) | |
return format_log( | |
f"Model {model_name} submitted by {organization} successfully.\nPlease wait a few hours and refresh the leaderboard to see your score displayed.") | |
# Function to refresh the results | |
def refresh(): | |
eval_results = load_dataset(RESULTS_DATASET, YEAR_VERSION, token=TOKEN, download_mode="force_redownload", | |
ignore_verifications=True, trust_remote_code=True) | |
eval_dataframe_test = get_dataframe_from_results(eval_results=eval_results, split="test") | |
eval_dataframe_test = format_dataframe(eval_dataframe_test) | |
return eval_dataframe_test | |
# Gradio interface | |
demo = gr.Blocks() | |
with demo: | |
gr.HTML("<h1>AssistantBench</h1>") | |
gr.Markdown(""" | |
AssistantBench aims to evaluate the ability of web agents to assist with real and time-consuming tasks. | |
For more information, please check out our paper or the official website. | |
To download AssistantBench, press [here](https://huggingface.co/datasets/Ori/AssistantBench_V1.0). | |
""") | |
gr.HTML("<h2>AssistantBench Leaderboard</h2>") | |
with gr.Tab("Results: Test"): | |
leaderboard_table_test = gr.Dataframe( | |
value=eval_dataframe_test, datatype=TYPES, interactive=False, | |
column_widths=["20%"] | |
) | |
refresh_button = gr.Button("Refresh") | |
refresh_button.click( | |
refresh, | |
inputs=[], | |
outputs=[ | |
leaderboard_table_test, | |
], | |
) | |
gr.HTML("<h2>Making a New Submission</h2>") | |
with gr.Accordion("Submit a new model for evaluation"): | |
with gr.Row(): | |
gr.Markdown(""" | |
To make a new submission, upload a predictions file. Our scoring function can be found [here](https://huggingface.co/spaces/AssistantBench/leaderboard/blob/main/scorer.py). We support JSONL files with the following format: | |
``` | |
{"id": "task_id_1", "answer": "Answer 1 from your model"} | |
{"id": "task_id_2", "answer": "Answer 2 from your model"} | |
``` | |
""") | |
with gr.Row(): | |
with gr.Column(): | |
model_name_textbox = gr.Textbox(label="Model Name") | |
model_family_textbox = gr.Textbox(label="Base Model") | |
url_textbox = gr.Textbox(label="URL to Model Information") | |
with gr.Column(): | |
organization = gr.Textbox(label="Organization") | |
mail = gr.Textbox( | |
label="Contact Email (will be stored privately & used if there is an issue with your submission)") | |
file_output = gr.File() | |
submit_button = gr.Button("Submit Eval") | |
submission_result = gr.Markdown() | |
submit_button.click( | |
add_new_eval, | |
[ | |
model_name_textbox, | |
model_family_textbox, | |
url_textbox, | |
file_output, | |
organization, | |
], | |
submission_result, | |
) | |
with gr.Row(): | |
with gr.Accordion("📙 Citation", open=False): | |
citation_text = """@article{yoran-etal-2024-assistantbench, | |
title={AssistantBench: Can Web Agents Solve Realistic and Time-Consuming Tasks?}, | |
author={Ori Yoran and Samuel Amouyal and Chaitanya Malaviya and Ben Bogin and Ofir Press and Jonathan Berant}, | |
year={2024}, | |
eprint={?}, | |
archivePrefix={arXiv}, | |
primaryClass={cs.CL} | |
}""" | |
citation_button = gr.Textbox( | |
value=citation_text, | |
label="Citation", | |
lines=20, | |
elem_id="citation-button", | |
show_copy_button=True | |
) | |
gr.HTML( | |
"<p>We would like to thank the GAIA team for sharing the source code for their leaderboard which we used as a template and HuggingFace for hosting the leaderboard.</p>") | |
scheduler = BackgroundScheduler() | |
scheduler.add_job(restart_space, "interval", seconds=3600) | |
scheduler.start() | |
demo.launch(debug=True) | |