app.py

import os
import json
import numpy as np
import pandas as pd
import gradio as gr
from huggingface_hub import HfApi, hf_hub_download


OWNER = "inceptionai"
DATASET_REPO_ID = f"{OWNER}/requests-dataset"

HEADER = """
<center>
<h1>Archived Space</h1>
<h1>AraGen Leaderboard: Generative Tasks Evaluation of Arabic LLMs</h1>
</center>

<br></br>

<p>This leaderboard introduces generative tasks evaluation for Arabic Large Language Models (LLMs). Powered by the new <strong>3C3H</strong> evaluation measure, this framework delivers a transparent, robust, and holistic evaluation system that balances factual accuracy and usability assessment for a production ready setting.</p>

<p>For more details, please consider going through the technical blogpost <a href="https://huggingface.co./blog/leaderboard-3c3h-aragen">here</a>.</p>
"""

ABOUT_SECTION = """
## About

The AraGen Leaderboard is designed to evaluate and compare the performance of Chat Arabic Large Language Models (LLMs) on a set of generative tasks. By leveraging the new **3C3H** evaluation measure which evaluate the model's output across six dimensions —Correctness, Completeness, Conciseness, Helpfulness, Honesty, and Harmlessness— the leaderboard provides a comprehensive and holistic evaluation of a model's performance in generating human-like and ethically responsible content.

### Why Focus on Chat Models?

AraGen Leaderboard —And 3C3H in general— is specifically designed to assess **chat models**, which interact in conversational settings, intended for end user interaction and require a blend of factual accuracy and user-centric dialogue capabilities. While it is technically possible to submit foundational models, we kindly ask users to refrain from doing so. For evaluations of foundational models using likelihood accuracy based benchmarks, please refer to the [Open Arabic LLM Leaderboard (OALL)](https://huggingface.co./spaces/OALL/Open-Arabic-LLM-Leaderboard).

### How to Submit Your Model?

Navigate to the submission section below to submit your open chat model from the HuggingFace Hub for evaluation. Ensure that your model is public and the submmited metadata (precision, revision, #params) is accurate.

### Contact

For any inquiries or assistance, feel free to reach out through the community tab at [Inception AraGen Community](https://huggingface.co./spaces/inceptionai/AraGen-Leaderboard/discussions) or via [email](mailto:[email protected]).
"""

CITATION_BUTTON_LABEL = """
Copy the following snippet to cite these results
"""

CITATION_BUTTON_TEXT = """
@misc{AraGen,
  author = {El Filali, Ali and Sengupta, Neha and Abouelseoud, Arwa and Nakov, Preslav and Fourrier, Clémentine},
  title = {Rethinking LLM Evaluation with 3C3H: AraGen Benchmark and Leaderboard},
  year = {2024},
  publisher = {Inception},
  howpublished = "url{https://huggingface.co./spaces/inceptionai/AraGen-Leaderboard}"
}
"""


def load_results():
    # Get the current directory of the script and construct the path to results.json
    current_dir = os.path.dirname(os.path.abspath(__file__))
    results_file = os.path.join(current_dir, "assets", "results", "results.json")
    
    # Load the JSON data from the specified file
    with open(results_file, 'r') as f:
        data = json.load(f)
    
    # Filter out any entries that only contain '_last_sync_timestamp'
    filtered_data = []
    for entry in data:
        # If '_last_sync_timestamp' is the only key, skip it
        if len(entry.keys()) == 1 and "_last_sync_timestamp" in entry:
            continue
        filtered_data.append(entry)
    
    data = filtered_data
    
    # Lists to collect data
    data_3c3h = []
    data_tasks = []
    
    for model_data in data:
        # Extract model meta data
        meta = model_data.get('Meta', {})
        model_name = meta.get('Model Name', 'UNK')
        revision = meta.get('Revision', 'UNK')
        precision = meta.get('Precision', 'UNK')
        params = meta.get('Params', 'UNK')
        license = meta.get('License', 'UNK')
        
        # Convert "Model Size" to numeric, treating "UNK" as infinity
        try:
            model_size_numeric = float(params)
        except (ValueError, TypeError):
            model_size_numeric = np.inf
        
        # 3C3H Scores
        scores_data = model_data.get('claude-3.5-sonnet Scores', {})
        scores_3c3h = scores_data.get('3C3H Scores', {})
        scores_tasks = scores_data.get('Tasks Scores', {})
        
        # Multiply scores by 100 to get percentages (keep them as numeric values)
        formatted_scores_3c3h = {k: v*100 for k, v in scores_3c3h.items()}
        formatted_scores_tasks = {k: v*100 for k, v in scores_tasks.items()}
        
        # For 3C3H Scores DataFrame
        data_entry_3c3h = {
            'Model Name': model_name,
            'Revision': revision,
            'License': license,
            'Precision': precision,
            'Model Size': model_size_numeric,  # Numeric value for sorting
            '3C3H Score': formatted_scores_3c3h.get("3C3H Score", np.nan),
            'Correctness': formatted_scores_3c3h.get("Correctness", np.nan),
            'Completeness': formatted_scores_3c3h.get("Completeness", np.nan),
            'Conciseness': formatted_scores_3c3h.get("Conciseness", np.nan),
            'Helpfulness': formatted_scores_3c3h.get("Helpfulness", np.nan),
            'Honesty': formatted_scores_3c3h.get("Honesty", np.nan),
            'Harmlessness': formatted_scores_3c3h.get("Harmlessness", np.nan),
        }
        data_3c3h.append(data_entry_3c3h)
        
        # For Tasks Scores DataFrame
        data_entry_tasks = {
            'Model Name': model_name,
            'Revision': revision,
            'License': license,
            'Precision': precision,
            'Model Size': model_size_numeric,  # Numeric value for sorting
            **formatted_scores_tasks
        }
        data_tasks.append(data_entry_tasks)
    
    df_3c3h = pd.DataFrame(data_3c3h)
    df_tasks = pd.DataFrame(data_tasks)
    
    # Round the numeric score columns to 4 decimal places
    score_columns_3c3h = ['3C3H Score', 'Correctness', 'Completeness', 'Conciseness', 'Helpfulness', 'Honesty', 'Harmlessness']
    df_3c3h[score_columns_3c3h] = df_3c3h[score_columns_3c3h].round(4)
    
    # Replace np.inf with a large number in 'Model Size Filter' for filtering
    max_model_size_value = 1000  # Define a maximum value
    df_3c3h['Model Size Filter'] = df_3c3h['Model Size'].replace(np.inf, max_model_size_value)
    
    # Sort df_3c3h by '3C3H Score' descending if column exists
    if '3C3H Score' in df_3c3h.columns:
        df_3c3h = df_3c3h.sort_values(by='3C3H Score', ascending=False)
        df_3c3h.insert(0, 'Rank', range(1, len(df_3c3h) + 1))  # Add Rank column starting from 1
    else:
        df_3c3h.insert(0, 'Rank', range(1, len(df_3c3h) + 1))
    
    # Extract task columns
    task_columns = [col for col in df_tasks.columns if col not in ['Model Name', 'Revision', 'License', 'Precision', 'Model Size', 'Model Size Filter']]
    
    # Round the task score columns to 4 decimal places
    if task_columns:
        df_tasks[task_columns] = df_tasks[task_columns].round(4)
    
    # Replace np.inf with a large number in 'Model Size Filter' for filtering
    df_tasks['Model Size Filter'] = df_tasks['Model Size'].replace(np.inf, max_model_size_value)
    
    # Sort df_tasks by the first task column if it exists
    if task_columns:
        first_task = task_columns[0]
        df_tasks = df_tasks.sort_values(by=first_task, ascending=False)
        df_tasks.insert(0, 'Rank', range(1, len(df_tasks) + 1))  # Add Rank column starting from 1
    else:
        df_tasks = df_tasks.sort_values(by='Model Name', ascending=True)
        df_tasks.insert(0, 'Rank', range(1, len(df_tasks) + 1))
    
    return df_3c3h, df_tasks, task_columns

def load_requests(status_folder):
    api = HfApi()
    requests_data = []
    folder_path_in_repo = status_folder  # 'pending', 'finished', or 'failed'

    hf_api_token = os.environ.get('HF_API_TOKEN', None)

    try:
        # List files in the dataset repository
        files_info = api.list_repo_files(
            repo_id=DATASET_REPO_ID,
            repo_type="dataset",
            token=hf_api_token
        )
    except Exception as e:
        print(f"Error accessing dataset repository: {e}")
        return pd.DataFrame()  # Return empty DataFrame if repository not found or inaccessible

    # Filter files in the desired folder
    files_in_folder = [f for f in files_info if f.startswith(f"{folder_path_in_repo}/") and f.endswith('.json')]

    for file_path in files_in_folder:
        try:
            # Download the JSON file
            local_file_path = hf_hub_download(
                repo_id=DATASET_REPO_ID,
                filename=file_path,
                repo_type="dataset",
                token=hf_api_token
            )
            # Load JSON data
            with open(local_file_path, 'r') as f:
                request = json.load(f)
            requests_data.append(request)
        except Exception as e:
            print(f"Error loading file {file_path}: {e}")
            continue  # Skip files that can't be loaded

    df = pd.DataFrame(requests_data)
    return df

def submit_model(model_name, revision, precision, params, license):
    # Load existing evaluations
    df_3c3h, df_tasks, _ = load_results()
    existing_models_results = df_3c3h[['Model Name', 'Revision', 'Precision']]

    # Handle 'Missing' precision
    if precision == 'Missing':
        precision = None
    else:
        precision = precision.strip().lower()

    # Load pending and finished requests from the dataset repository
    df_pending = load_requests('pending')
    df_finished = load_requests('finished')

    # Check if model is already evaluated
    model_exists_in_results = ((existing_models_results['Model Name'] == model_name) &
                               (existing_models_results['Revision'] == revision) &
                               (existing_models_results['Precision'] == precision)).any()
    if model_exists_in_results:
        return f"**Model '{model_name}' with revision '{revision}' and precision '{precision}' has already been evaluated.**"

    # Check if model is in pending requests
    if not df_pending.empty:
        existing_models_pending = df_pending[['model_name', 'revision', 'precision']]
        model_exists_in_pending = ((existing_models_pending['model_name'] == model_name) &
                                   (existing_models_pending['revision'] == revision) &
                                   (existing_models_pending['precision'] == precision)).any()
        if model_exists_in_pending:
            return f"**Model '{model_name}' with revision '{revision}' and precision '{precision}' is already in the pending evaluations.**"

    # Check if model is in finished requests
    if not df_finished.empty:
        existing_models_finished = df_finished[['model_name', 'revision', 'precision']]
        model_exists_in_finished = ((existing_models_finished['model_name'] == model_name) &
                                    (existing_models_finished['revision'] == revision) &
                                    (existing_models_finished['precision'] == precision)).any()
        if model_exists_in_finished:
            return f"**Model '{model_name}' with revision '{revision}' and precision '{precision}' has already been evaluated.**"

    # Check if model exists on HuggingFace Hub
    api = HfApi()
    try:
        model_info = api.model_info(model_name)
    except Exception as e:
        return f"**Error: Could not find model '{model_name}' on HuggingFace Hub. Please ensure the model name is correct and the model is public.**"

    # Proceed with submission
    status = "PENDING"

    # Prepare the submission data
    submission = {
        "model_name": model_name,
        "license": license,
        "revision": revision,
        "precision": precision,
        "status": status,
        "params": params
    }

    # Serialize the submission to JSON
    submission_json = json.dumps(submission, indent=2)

    # Define the file path in the repository
    org_model = model_name.split('/')
    if len(org_model) != 2:
        return "**Please enter the full model name including the organization or username, e.g., 'inceptionai/jais-family-30b-8k'**"
    org, model_id = org_model
    precision_str = precision if precision else 'Missing'
    file_path_in_repo = f"pending/{org}/{model_id}_eval_request_{revision}_{precision_str}.json"

    # Upload the submission to the dataset repository
    try:
        hf_api_token = os.environ.get('HF_API_TOKEN', None)
        api.upload_file(
            path_or_fileobj=submission_json.encode('utf-8'),
            path_in_repo=file_path_in_repo,
            repo_id=DATASET_REPO_ID,
            repo_type="dataset",
            token=hf_api_token
        )
    except Exception as e:
        return f"**Error: Could not submit the model. {str(e)}**"

    return f"**Model '{model_name}' has been submitted for evaluation.**"

def main():
    df_3c3h, df_tasks, task_columns = load_results()

    # Extract unique Precision and License values for filters
    precision_options_3c3h = sorted(df_3c3h['Precision'].dropna().unique().tolist())
    precision_options_3c3h = [p for p in precision_options_3c3h if p != 'UNK']
    precision_options_3c3h.append('Missing')

    license_options_3c3h = sorted(df_3c3h['License'].dropna().unique().tolist())
    license_options_3c3h = [l for l in license_options_3c3h if l != 'UNK']
    license_options_3c3h.append('Missing')

    precision_options_tasks = sorted(df_tasks['Precision'].dropna().unique().tolist())
    precision_options_tasks = [p for p in precision_options_tasks if p != 'UNK']
    precision_options_tasks.append('Missing')

    license_options_tasks = sorted(df_tasks['License'].dropna().unique().tolist())
    license_options_tasks = [l for l in license_options_tasks if l != 'UNK']
    license_options_tasks.append('Missing')

    # Get min and max model sizes for sliders, handling 'inf' values
    min_model_size_3c3h = int(df_3c3h['Model Size Filter'].min())
    max_model_size_3c3h = int(df_3c3h['Model Size Filter'].max())

    min_model_size_tasks = int(df_tasks['Model Size Filter'].min())
    max_model_size_tasks = int(df_tasks['Model Size Filter'].max())

    # Exclude 'Model Size Filter' from column selectors
    column_choices_3c3h = [col for col in df_3c3h.columns if col != 'Model Size Filter']
    column_choices_tasks = [col for col in df_tasks.columns if col != 'Model Size Filter']

    with gr.Blocks() as demo:
        gr.HTML(HEADER)
        
        with gr.Tabs():
            with gr.Tab("Leaderboard"):
                with gr.Tabs():
                    with gr.Tab("3C3H Scores"):
                        with gr.Row():
                            search_box_3c3h = gr.Textbox(
                                placeholder="Search for models...", 
                                label="Search", 
                                interactive=True
                            )
                        with gr.Row():
                            column_selector_3c3h = gr.CheckboxGroup(
                                choices=column_choices_3c3h,
                                value=[
                                    'Rank', 'Model Name', '3C3H Score', 'Correctness', 'Completeness',
                                    'Conciseness', 'Helpfulness', 'Honesty', 'Harmlessness'
                                ],
                                label="Select columns to display",
                            )
                        with gr.Row():
                            license_filter_3c3h = gr.CheckboxGroup(
                                choices=license_options_3c3h,
                                value=license_options_3c3h.copy(),  # Default all selected
                                label="Filter by License",
                            )
                            precision_filter_3c3h = gr.CheckboxGroup(
                                choices=precision_options_3c3h,
                                value=precision_options_3c3h.copy(),  # Default all selected
                                label="Filter by Precision",
                            )
                        with gr.Row():
                            model_size_min_filter_3c3h = gr.Slider(
                                minimum=min_model_size_3c3h,
                                maximum=max_model_size_3c3h,
                                value=min_model_size_3c3h,
                                step=1,
                                label="Minimum Model Size",
                                interactive=True
                            )
                            model_size_max_filter_3c3h = gr.Slider(
                                minimum=min_model_size_3c3h,
                                maximum=max_model_size_3c3h,
                                value=max_model_size_3c3h,
                                step=1,
                                label="Maximum Model Size",
                                interactive=True
                            )
                        
                        leaderboard_3c3h = gr.Dataframe(
                            df_3c3h[['Rank', 'Model Name', '3C3H Score', 'Correctness', 'Completeness',
                                   'Conciseness', 'Helpfulness', 'Honesty', 'Harmlessness']],
                            interactive=False
                        )
                        
                        def filter_df_3c3h(search_query, selected_cols, precision_filters, license_filters, min_size, max_size):
                            filtered_df = df_3c3h.copy()
                            
                            # Ensure min_size <= max_size
                            if min_size > max_size:
                                min_size, max_size = max_size, min_size
                            
                            # Apply search filter
                            if search_query:
                                filtered_df = filtered_df[filtered_df['Model Name'].str.contains(search_query, case=False, na=False)]
                            
                            # Apply Precision filter
                            if precision_filters:
                                include_missing = 'Missing' in precision_filters
                                selected_precisions = [p for p in precision_filters if p != 'Missing']
                                if include_missing:
                                    filtered_df = filtered_df[
                                        (filtered_df['Precision'].isin(selected_precisions)) |
                                        (filtered_df['Precision'] == 'UNK') |
                                        (filtered_df['Precision'].isna())
                                    ]
                                else:
                                    filtered_df = filtered_df[filtered_df['Precision'].isin(selected_precisions)]
                            
                            # Apply License filter
                            if license_filters:
                                include_missing = 'Missing' in license_filters
                                selected_licenses = [l for l in license_filters if l != 'Missing']
                                if include_missing:
                                    filtered_df = filtered_df[
                                        (filtered_df['License'].isin(selected_licenses)) |
                                        (filtered_df['License'] == 'UNK') |
                                        (filtered_df['License'].isna())
                                    ]
                                else:
                                    filtered_df = filtered_df[filtered_df['License'].isin(selected_licenses)]
                            
                            # Apply Model Size filter
                            filtered_df = filtered_df[
                                (filtered_df['Model Size Filter'] >= min_size) &
                                (filtered_df['Model Size Filter'] <= max_size)
                            ]
                            
                            # Remove existing 'Rank' column if present
                            if 'Rank' in filtered_df.columns:
                                filtered_df = filtered_df.drop(columns=['Rank'])
                            
                            # Recalculate Rank after filtering
                            filtered_df = filtered_df.reset_index(drop=True)
                            filtered_df.insert(0, 'Rank', range(1, len(filtered_df) + 1))
                            
                            # Ensure selected columns are present
                            selected_cols = [col for col in selected_cols if col in filtered_df.columns]
                            
                            return filtered_df[selected_cols]
                        
                        # Bind the filter function to the appropriate events
                        filter_inputs_3c3h = [
                            search_box_3c3h,
                            column_selector_3c3h,
                            precision_filter_3c3h,
                            license_filter_3c3h,
                            model_size_min_filter_3c3h,
                            model_size_max_filter_3c3h
                        ]
                        search_box_3c3h.submit(
                            filter_df_3c3h,
                            inputs=filter_inputs_3c3h,
                            outputs=leaderboard_3c3h
                        )
                        
                        # Bind change events for CheckboxGroups and sliders
                        for component in filter_inputs_3c3h:
                            component.change(
                                filter_df_3c3h,
                                inputs=filter_inputs_3c3h,
                                outputs=leaderboard_3c3h
                            )
                    
                    with gr.Tab("Tasks Scores"):
                        gr.Markdown("""
                                    Note: This Table is sorted based on the First Task (Question Answering)
                                    """)
                        
                        with gr.Row():
                            search_box_tasks = gr.Textbox(
                                placeholder="Search for models...", 
                                label="Search", 
                                interactive=True
                            )
                        with gr.Row():
                            column_selector_tasks = gr.CheckboxGroup(
                                choices=column_choices_tasks,
                                value=['Rank', 'Model Name'] + task_columns,
                                label="Select columns to display",
                            )
                        with gr.Row():
                            license_filter_tasks = gr.CheckboxGroup(
                                choices=license_options_tasks,
                                value=license_options_tasks.copy(),  # Default all selected
                                label="Filter by License",
                            )
                            precision_filter_tasks = gr.CheckboxGroup(
                                choices=precision_options_tasks,
                                value=precision_options_tasks.copy(),  # Default all selected
                                label="Filter by Precision",
                            )
                        with gr.Row():
                            model_size_min_filter_tasks = gr.Slider(
                                minimum=min_model_size_tasks,
                                maximum=max_model_size_tasks,
                                value=min_model_size_tasks,
                                step=1,
                                label="Minimum Model Size",
                                interactive=True
                            )
                            model_size_max_filter_tasks = gr.Slider(
                                minimum=min_model_size_tasks,
                                maximum=max_model_size_tasks,
                                value=max_model_size_tasks,
                                step=1,
                                label="Maximum Model Size",
                                interactive=True
                            )
                        
                        leaderboard_tasks = gr.Dataframe(
                            df_tasks[['Rank', 'Model Name'] + task_columns],
                            interactive=False
                        )
                        
                        def filter_df_tasks(search_query, selected_cols, precision_filters, license_filters, min_size, max_size):
                            filtered_df = df_tasks.copy()
                            
                            # Ensure min_size <= max_size
                            if min_size > max_size:
                                min_size, max_size = max_size, min_size
                            
                            # Apply search filter
                            if search_query:
                                filtered_df = filtered_df[filtered_df['Model Name'].str.contains(search_query, case=False, na=False)]
                            
                            # Apply Precision filter
                            if precision_filters:
                                include_missing = 'Missing' in precision_filters
                                selected_precisions = [p for p in precision_filters if p != 'Missing']
                                if include_missing:
                                    filtered_df = filtered_df[
                                        (filtered_df['Precision'].isin(selected_precisions)) |
                                        (filtered_df['Precision'] == 'UNK') |
                                        (filtered_df['Precision'].isna())
                                    ]
                                else:
                                    filtered_df = filtered_df[filtered_df['Precision'].isin(selected_precisions)]
                            
                            # Apply License filter
                            if license_filters:
                                include_missing = 'Missing' in license_filters
                                selected_licenses = [l for l in license_filters if l != 'Missing']
                                if include_missing:
                                    filtered_df = filtered_df[
                                        (filtered_df['License'].isin(selected_licenses)) |
                                        (filtered_df['License'] == 'UNK') |
                                        (filtered_df['License'].isna())
                                    ]
                                else:
                                    filtered_df = filtered_df[filtered_df['License'].isin(selected_licenses)]
                            
                            # Apply Model Size filter
                            filtered_df = filtered_df[
                                (filtered_df['Model Size Filter'] >= min_size) &
                                (filtered_df['Model Size Filter'] <= max_size)
                            ]
                            
                            # Remove existing 'Rank' column if present
                            if 'Rank' in filtered_df.columns:
                                filtered_df = filtered_df.drop(columns=['Rank'])
                            
                            # Sort by the first task column if it exists
                            if task_columns:
                                first_task = task_columns[0]
                                filtered_df = filtered_df.sort_values(by=first_task, ascending=False)
                            else:
                                filtered_df = filtered_df.sort_values(by='Model Name', ascending=True)
                            
                            # Recalculate Rank after filtering
                            filtered_df = filtered_df.reset_index(drop=True)
                            filtered_df.insert(0, 'Rank', range(1, len(filtered_df) + 1))
                            
                            # Ensure selected columns are present
                            selected_cols = [col for col in selected_cols if col in filtered_df.columns]
                            
                            return filtered_df[selected_cols]
                        
                        # Bind the filter function to the appropriate events
                        filter_inputs_tasks = [
                            search_box_tasks,
                            column_selector_tasks,
                            precision_filter_tasks,
                            license_filter_tasks,
                            model_size_min_filter_tasks,
                            model_size_max_filter_tasks
                        ]
                        search_box_tasks.submit(
                            filter_df_tasks,
                            inputs=filter_inputs_tasks,
                            outputs=leaderboard_tasks
                        )
                        
                        # Bind change events for CheckboxGroups and sliders
                        for component in filter_inputs_tasks:
                            component.change(
                                filter_df_tasks,
                                inputs=filter_inputs_tasks,
                                outputs=leaderboard_tasks
                            )
            
            with gr.Tab("Submit Here"):
                gr.Markdown("# This space is a legacy leaderboard in archived mode and no longer accepts model submissions.")
                gr.Markdown("---")
                gr.Markdown(ABOUT_SECTION)
            with gr.Row():
                with gr.Accordion("📙 Citation", open=False):
                    citation_button = gr.Textbox(
                        value=CITATION_BUTTON_TEXT,
                        label=CITATION_BUTTON_LABEL,
                        lines=20,
                        elem_id="citation-button",
                        show_copy_button=True,
                    )

    demo.launch()

if __name__ == "__main__":
    main()