import gradio as gr from gradio_leaderboard import Leaderboard, SelectColumns, ColumnFilter import config from envs import RESULTS_REPO_ID, REPO_ID, API, HF_TOKEN from pathlib import Path import pandas as pd import os import json from utils.viz import create_scatter_plot, create_flow_chart, create_bar_chart, create_task_success_heatmap, create_leaderboard from utils.processing import check_and_process_uploads from huggingface_hub import snapshot_download from apscheduler.schedulers.background import BackgroundScheduler from datetime import datetime import json import re import markdown import asyncio from apscheduler.schedulers.asyncio import AsyncIOScheduler import weave from utils.db import TracePreprocessor from gradio.themes.soft import Soft preprocessor = TracePreprocessor() from datetime import datetime abs_path = Path(__file__).parent def restart_space(): API.restart_space(repo_id=REPO_ID, token=HF_TOKEN) # New function to download results def download_latest_results(): print("Downloading latest results...") snapshot_download(RESULTS_REPO_ID, local_dir= "evals_upload", repo_type='dataset', tqdm_class=None, etag_timeout=30, max_workers=4, ) print("Download complete.") def get_analyzed_traces(agent_name, benchmark_name): return preprocessor.get_analyzed_traces(agent_name, benchmark_name) def get_failure_report(agent_name, benchmark_name): return preprocessor.get_failure_report(agent_name, benchmark_name) def parse_json_files(folder_path, benchmark_name, aggregate=True): return preprocessor.get_parsed_results(benchmark_name, aggregate=aggregate) def update_agent_dropdown(benchmark_name, metric): df = parse_json_files(os.path.join(abs_path, "evals_live"), benchmark_name) agents = df['Agent Name'].tolist() best_agent = get_best_agent(benchmark_name, metric) return gr.Dropdown(choices=agents, value=best_agent, label="Select Agent") def get_best_agent(benchmark_name, metric): df = parse_json_files(os.path.join(abs_path, "evals_live"), benchmark_name) return df.loc[df[metric].idxmax()]['Agent Name'] def update_task_analysis(benchmark_name, agent_name): if not agent_name: return "Please select an agent.", None, None, "" analyzed_traces = get_analyzed_traces(agent_name, benchmark_name) if not analyzed_traces: return f"No analysis available for agent: {agent_name}", None, None, "" task_ids = list(analyzed_traces.keys()) overview, flow_chart, _ = update_task_details(benchmark_name, agent_name, task_ids[0]) return overview, flow_chart, gr.Dropdown(choices=task_ids, value=task_ids[0], label="Select Task"), "" def update_task_details(benchmark_name, agent_name, task_id): if not task_id: return "Please select a task.", None, "" analyzed_traces = get_analyzed_traces(agent_name, benchmark_name) if not analyzed_traces or task_id not in analyzed_traces: return f"No analysis available for task: {task_id}", None, "" analysis = analyzed_traces[task_id] summary = analysis.get('task_analysis', {}) overview = f"### Summary\n\n{summary.get('overview', 'No overview available.')}\n\n" # overview += f"### Successes\n{summary.get('key_successes', 'No successes listed.')}\n\n" # overview += f"### Challenges\n{summary.get('main_challenges', 'No challenges listed.')}\n\n" # overview += f"### Overall Assessment\n{summary.get('overall_assessment', 'No assessment available.')}\n\n" if summary.get('overview', 'No overview available.') != "Not available": flow_chart = create_flow_chart(analysis['steps']) else: flow_chart = None return overview, flow_chart, "" def format_call_info(step, step_index): call_data = step['call_data'] analysis = step['analysis'] def format_json(obj): # if isinstance(obj, dict) and 'choices' in obj: # # Special handling for message content # formatted_content = format_message_content(obj['choices'][0]) # return f'
{formatted_content}
' # else: json_str = json.dumps(obj, indent=2) json_str = json_str.replace(' ', ' ') json_str = json_str.replace('\n', '
') return f'
{json_str}
' # Currently not used but we can enable it to format message content def format_message_content(content): # Convert Markdown to HTML html_content = markdown.markdown(content) # Replace ``` code blocks with styled pre blocks html_content = re.sub(r'```python\n(.*?)```', lambda m: f'
{m.group(1)}
', html_content, flags=re.DOTALL) return html_content formatted_info = f"""

Step {step_index + 1}: {analysis.get('headline', '')}

Call Metadata

Inputs

{format_json(call_data['inputs'])}

Outputs

{format_json(call_data['outputs'])}

Usage

{format_json(call_data['summary'])}

Analysis

""" return formatted_info def update_failure_report(agent_name, benchmark_name): failure_report = get_failure_report(agent_name, benchmark_name) if not failure_report: return "No failure report available for this agent.", None # Create overview of failure categories categories_overview = "### Failure Categories:\n\n" for category in failure_report['failure_categories']: categories_overview += f"#### {category['category_name']}\n" categories_overview += f"{category['description']}\n\n" # Count tasks affected by each category category_counts = {} for task, classification in failure_report['task_classifications'].items(): category_id = classification['category_id'] category_counts[category_id] = category_counts.get(category_id, 0) + 1 # Prepare data for bar chart categories = [cat['category_name'] for cat in failure_report['failure_categories']] counts = [category_counts.get(str(i+1), 0) for i in range(len(categories))] # Create bar chart chart = create_bar_chart(categories, counts, "Failure Categories", "Number of Affected Tasks", "Failure Categories Distribution") return categories_overview, chart from gradio.themes.utils import colors, fonts, sizes from typing import Iterable class MyTheme(Soft): def __init__( self, *, primary_hue: colors.Color | str = colors.blue, text_size: sizes.Size | str = sizes.text_lg, font: fonts.Font | str | Iterable[fonts.Font | str] = ( fonts.GoogleFont("Lato"), "ui-sans-serif", "sans-serif", ), font_mono: fonts.Font | str | Iterable[fonts.Font | str] = ( fonts.GoogleFont("IBM Plex Mono"), "ui-monospace", "monospace", ), ): super().__init__( primary_hue=primary_hue, text_size=text_size, font=font, font_mono=font_mono, ) my_theme = MyTheme() with gr.Blocks(theme=my_theme, css='css.css', title="CORE-Bench Leaderboard") as demo: # gr.Markdown((Path(__file__).parent / "header.md").read_text(), elem_classes=["text-large"]) gr.HTML("""

CORE-Bench Leaderboard

A leaderboard evaluating agent performance on CORE-Bench.

""") gr.HTML("""
Paper Github Dataset

Leaderboards

CORE-Bench evaluates the ability of agents to computationally reproduce the results of published scientific papers. Agents are given the codebase of a paper and must install all libraries and dependencies, run the code, and read through the output and figures to answer questions about the paper. The benchmark has tasks at three difficulty levels:

CORE-Bench-Hard: The agent is given the codebase of the paper and must install all libraries and dependencies, run the code, and read through the output and figures to answer questions about the paper. This level is most akin to fully reproducing a paper and is the most realistic and challenging level.

CORE-Bench-Medium: The agent is given a Dockerfile and instructions on how to use the Dockerfile to fully reproduce the paper. This level mainly evaluates agents ability to use and interact with the terminal. The agent must then answer questions about the output of the code, as in the above level.

CORE-Bench-Easy: The agent is given the output of the code and must answer questions about the output without running any code. To answer questions, agents must navigate through the terminal output as well as files and figures generated by the code.

""") with gr.Tabs() as tabs: with gr.Tab("CORE-Bench-Hard"): with gr.Row(): with gr.Column(scale=2): Leaderboard( value=create_leaderboard(parse_json_files(os.path.join(abs_path, "evals_live"), 'corebench_hard'), ci_metrics=["Accuracy", "Total Cost"]), select_columns=SelectColumns( default_selection=config.COREBENCH_ON_LOAD_COLUMNS + ["Verified"], cant_deselect=["Agent Name"], label="Select Columns to Display:", ), hide_columns=config.COREBENCH_HIDE_COLUMNS, search_columns=config.COREBENCH_SEARCH_COLUMNS, ) # gr.Markdown("""*Error ranges span from the lowest to highest observed values in repeated runs.*""", elem_classes=["text-right"]) with gr.Row(): gr.Markdown("### Accuracy vs. Cost on CORE-Bench-Hard") with gr.Row(): scatter_plot = gr.Plot(create_scatter_plot(parse_json_files(os.path.join(abs_path, "evals_live"), 'corebench_hard', aggregate=False), "Total Cost", "Accuracy", "Total Cost (in USD)", "Accuracy", ["Agent Name"])) with gr.Tab("CORE-Bench-Medium"): with gr.Row(): with gr.Column(scale=2): Leaderboard( value=create_leaderboard(parse_json_files(os.path.join(abs_path, "evals_live"), 'corebench_medium'), ci_metrics=["Accuracy", "Total Cost"]), select_columns=SelectColumns( default_selection=config.COREBENCH_ON_LOAD_COLUMNS + ["Verified"], cant_deselect=["Agent Name"], label="Select Columns to Display:", ), hide_columns=config.COREBENCH_HIDE_COLUMNS, search_columns=config.COREBENCH_SEARCH_COLUMNS, ) # gr.Markdown("""*Error ranges span from the lowest to highest observed values in repeated runs.*""", elem_classes=["text-right"]) with gr.Row(): gr.Markdown("### Accuracy vs. Cost on CORE-Bench-Medium") with gr.Row(): scatter_plot = gr.Plot(create_scatter_plot(parse_json_files(os.path.join(abs_path, "evals_live"), 'corebench_medium', aggregate=False), "Total Cost", "Accuracy", "Total Cost (in USD)", "Accuracy", ["Agent Name"])) with gr.Tab("CORE-Bench-Easy"): with gr.Row(): with gr.Column(scale=2): Leaderboard( value=create_leaderboard(parse_json_files(os.path.join(abs_path, "evals_live"), 'corebench_easy'), ci_metrics=["Accuracy", "Total Cost"]), select_columns=SelectColumns( default_selection=config.COREBENCH_ON_LOAD_COLUMNS + ["Verified"], cant_deselect=["Agent Name"], label="Select Columns to Display:", ), hide_columns=config.COREBENCH_HIDE_COLUMNS, search_columns=config.COREBENCH_SEARCH_COLUMNS, ) # gr.Markdown("""*Error ranges span from the lowest to highest observed values in repeated runs.*""", elem_classes=["text-right"]) with gr.Row(): gr.Markdown("### Accuracy vs. Cost on CORE-Bench-Easy") with gr.Row(): scatter_plot = gr.Plot(create_scatter_plot(parse_json_files(os.path.join(abs_path, "evals_live"), 'corebench_easy', aggregate=False), "Total Cost", "Accuracy", "Total Cost (in USD)", "Accuracy", ["Agent Name"])) # Will trigger autoscaling of plots when tabs are switched tabs.select(fn=None, inputs=None, outputs=None, js=""" function() { setTimeout(function() { window.dispatchEvent(new Event('resize')); }, 100); } """) gr.HTML("""

How can I submit agents to the leaderboard?

""") gr.Markdown((Path(__file__).parent / "agent_submission.md").read_text()) async def main(): # Preprocess traces # preprocessor = TracePreprocessor() # preprocessor.preprocess_traces('evals_live') # preprocessor = TracePreprocessor() # Download the results from the Hugging Face Hub # await asyncio.to_thread(download_latest_results) # # Check for new uploads and process them # await check_and_process_uploads() scheduler = AsyncIOScheduler() scheduler.add_job(restart_space, "interval", hours=1) # scheduler.add_job(download_latest_results, "interval", hours=1) # scheduler.add_job(check_and_process_uploads, "interval", hours=1) scheduler.start() await demo.launch(favicon_path="hal.png") if __name__ == "__main__": weave.init(f'leaderboard_{datetime.now().strftime("%Y%m%d%H%M%S")}') asyncio.run(main())