app.py

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'<div class="message-content">{formatted_content}</div>'
        # else:
        json_str = json.dumps(obj, indent=2)
        json_str = json_str.replace(' ', '&nbsp;')
        json_str = json_str.replace('\n', '<br>')
        return f'<div class="json-wrapper">{json_str}</div>'

    # 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'<pre class="code-block">{m.group(1)}</pre>', html_content, flags=re.DOTALL)
        
        return html_content

    formatted_info = f"""
    <style>
        .json-wrapper {{
            white-space: pre-wrap;
            word-wrap: break-word;
            font-family: monospace;
            max-height: 300px;
            overflow-y: auto;
            background-color: #f5f5f5;
            padding: 10px;
            border-radius: 5px;
        }}
        .message-content {{
            white-space: normal;
            word-wrap: break-word;
            font-family: Arial, sans-serif;
            max-height: 500px;
            overflow-y: auto;
            background-color: #ffffff;
            padding: 10px;
            border-radius: 5px;
            border: 1px solid #e0e0e0;
        }}
        .code-block {{
            background-color: #f0f0f0;
            padding: 10px;
            border-radius: 5px;
            font-family: monospace;
            white-space: pre-wrap;
            word-wrap: break-word;
        }}
    </style>

    <h3>Step {step_index + 1}: {analysis.get('headline', '')}</h3>

    <h4>Call Metadata</h4>
    <ul>
        <li><strong>Weave Task ID:</strong> {call_data['weave_task_id']}</li>
        <li><strong>Trace ID:</strong> {call_data['trace_id']}</li>
        <li><strong>Project ID:</strong> {call_data['project_id']}</li>
        <li><strong>Created Timestamp:</strong> {datetime.fromtimestamp(call_data['created_timestamp'])}</li>
        <li><strong>Model:</strong> {call_data['inputs']['model']}</li>
    </ul>

    <h4>Inputs</h4>
    {format_json(call_data['inputs'])}

    <h4>Outputs</h4>
    {format_json(call_data['outputs'])}

    <h4>Usage</h4>
    {format_json(call_data['summary'])}

    <h4>Analysis</h4>
    <ul>
        <li><strong>Description:</strong> {analysis['description']}</li>
        <li><strong>Assessment:</strong> {analysis['assessment']}</li>
        <li><strong>Success:</strong> {analysis['success']}</li>
        <li><strong>Action Type:</strong> {analysis['action_type']}</li>
    </ul>
    """
    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("""
            <style>
    .hal-header {
        color: #ecf0f1;
        border-radius: 10px;
        padding: 40px 20px;
        text-align: center;
        box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1);
    }
    .hal-title {
        font-size: 2.5em;
        font-weight: 700;
        margin: 0;
        letter-spacing: 2px;
        text-transform: uppercase;
    }
    .hal-subtitle {
        font-size: 1.2em;
        font-weight: 300;
        margin-top: 15px;
        margin-left: auto;
        margin-right: auto;
        line-height: 1.6;
        text-align: center;
    }
    .hal-highlight {
        color: #3498db;
        font-weight: 600;
    }
</style>

<header class="hal-header">
    <h1 class="hal-title">CORE-Bench Leaderboard</h1>
    <p class="hal-subtitle">
            A leaderboard evaluating agent performance on CORE-Bench.
    </p>
</header>""")
    gr.HTML("""
<style>
    .feature-row {
        display: flex;
        justify-content: space-between;
        margin-top: 20px;
        margin-bottom: 20px;
    }
    .feature-column {
        flex: 1;
        padding: 25px;
        background-color: #ffffff;
        border-radius: 10px;
        margin: 0 15px;
        text-align: center;
        box-shadow: 0 6px 12px rgba(0, 0, 0, 0.1);
        display: flex;
        flex-direction: column;
        align-items: center;
        border-top: 5px solid #3498db;
        transition: transform 0.3s ease, box-shadow 0.3s ease;
    }
    .feature-column:hover {
        transform: translateY(-5px);
        box-shadow: 0 5px 10px rgba(0, 0, 0, 0.15);
    }
    .feature-keyword {
        font-size: 1.2em;
        font-weight: bold;
        color: #1b9e77;
        margin-bottom: 15px;
        text-transform: uppercase;
        letter-spacing: 1px;
    }
    .feature-content {
        flex-grow: 1;
    }
    .feature-description {
        font-size: 0.95em;
        line-height: 1.6;
        color: #333;
    }
   .button-container {
        display: flex;
        justify-content: center;
        margin-top: 2px;
    }
    .button-container .button {
        margin: 0 10px;
        padding: 15px 25px;
        font-size: 1em;
        font-weight: bold;
        color: #fff !important; /* Force white text color */
        background-color: #3498db !important; /* Force background color */
        border: none;
        border-radius: 5px;
        text-decoration: none !important; /* Force no underline */
        text-align: center;
        transition: background-color 0.3s ease;
        cursor: pointer;
        height: 50px;
    }
    .button-container .button:hover {
        background-color: #2980b9 !important; /* Force hover color */
    }
    .button:visited {
        color: #fff; /* Keep text color white when link is visited */
    }
</style>

<div class="button-container">
    <a href="https://arxiv.org/abs/2409.11363" class="button">Paper</a>
    <a href="https://github.com/siegelz/core-bench" class="button">Github</a>
    <a href="https://huggingface.co/datasets/siegelz/core-bench" class="button">Dataset</a>
</div>
            
</br>
<h2 class="section-heading" id="leaderboards">Leaderboards</h2>
<p>
   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:
</p>
<p>
    <i><b>CORE-Bench-Hard:</b></i> 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.
</p>
<p>        
    <i><b>CORE-Bench-Medium:</b></i> 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.
</p>
<p>         
    <i><b>CORE-Bench-Easy:</b></i> 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.
</p>
""")
    
    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("""<h2 class="section-heading" id="agent-submission">How can I submit agents to the leaderboard?</h2>""")
    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())