app.py

import pandas as pd
import gradio as gr
import os
import re
import requests
from dotenv import load_dotenv
from matplotlib.colors import LinearSegmentedColormap
import plotly.express as px
import plotly.graph_objects as go
from sklearn.linear_model import LinearRegression
import numpy as np
from huggingface_hub import HfApi
from huggingface_hub.hf_api import HTTPError
from huggingface_hub.utils._errors import GatedRepoError

load_dotenv()
webhook_url = os.environ.get("WEBHOOK_URL")

file_name_list = [
    '9b',
    '7b',
    '3b',
    '1b5',
]

sheet_name_list = [
    'cr',
    'bpc',
    'bpb',
]

metric_list = [
    'Compression Rate (%)',
    'Bits Per Character (BPC)',
    'Bits Per Byte (BPB)',
]

model_size_list = [
    '~9B',
    '~7B',
    '~3B',
    '~1.5B',
]

metric_to_sheet = {
    'Compression Rate (%)': 'cr',
    'Bits Per Character (BPC)': 'bpc',
    'Bits Per Byte (BPB)': 'bpb',
}

model_size_to_file_name = {
    '~9B': '9b',
    '~7B': '7b',
    '~3B': '3b',
    '~1.5B': '1b5',
}

css = """

.gr-dataframe table {

    table-layout: fixed;

    width: 100%; /* Ensures the table fills its container */

}

.gr-dataframe th, .gr-dataframe td {

    width: 100px;  /* Set the exact width of each cell */

    overflow: hidden;  /* Ensures the content doesn't overflow */

    text-overflow: ellipsis;  /* Adds an ellipsis (...) if the text overflows */

    white-space: nowrap;  /* Keeps the content on a single line */

}

"""

about_md = """

# Uncheatable Eval



GitHub page: [https://github.com/Jellyfish042/uncheatable_eval](https://github.com/Jellyfish042/uncheatable_eval)



## Introduction

Traditional LLM benchmarks are easily compromised by unintentional or intentional data leakage, making many benchmarks unreliable and unable to truly reflect the capabilities of LLMs.



Uncheatable Eval addresses this issue by testing LLMs on real-time, newly generated data from the internet, 

ensuring that the evaluation is immune to data leaks and cannot be gamed.



## How?

Uncheatable Eval assesses the language modeling capabilities of LLMs on new data from various sources such as recent papers on arXiv, new projects on GitHub, news articles, and more. Since this data is brand new (e.g., from the past 1-2 weeks), it is impossible for these data to be included in the training sets of publicly released models, thus avoiding the impact of unintentional or intentional data leaks.



Specifically, we calculate the sum of negative log probabilities of the models on these texts. In other words, models that are more likely to generate these texts are considered better.



*Note* : Uncheatable Eval only tests base models.



## Q&A

### Why Calculate the Sum of Negative Log Probabilities?

First, the goal of language models, at least today's language models, is to generate text that is as realistic as possible, maximizing the probability of real text. They are trained and designed to do exactly this. Calculating the sum of negative log probabilities on real text is the most direct way to test this capability.



Second, from the perspective of "compression is intelligence," a good way to test a language model would be to use the model with an entropy coding algorithm for compression and test the model's compression rate [[1]](https://arxiv.org/abs/2309.10668)[[2]](https://arxiv.org/abs/2402.00861). A model with a lower compression rate is considered better. Using a language model + arithmetic coding as an example, it is easy to prove that a model's ability to compress a piece of text is proportional to the sum of its negative log probabilities on that text (see [proof](#proof-of-the-equivalence-between-compression-capability-and-negative-log-probability-sum)).

Therefore, the compression rate of a model can be directly calculated through the sum of negative log probabilities, and the method for this has been provided in `show_results_v2.ipynb`.

### Can Models Using Different Tokenizers Be Directly Compared?

Yes. When calculating the sum of negative log probabilities, we essentially treat the model + tokenizer as a single entity or system. As long as this system has a high probability of generating real text, we consider it better. From the perspective of compression, you can choose any tokenizer. From the compression rate perspective, we don't care; we only care about whether your system can compress the text more effectively.



### Is It Really Uncheatable? Can't I train my model on a large number of arXiv papers to improve its test performance on arXiv papers?

Uncheatable Eval's data sources currently include new arXiv papers, new GitHub projects, BBC news, AO3 fanfictions, and new Wikipedia entries, with more sources to be added in the future. If you genuinely achieve excellent results across these data by training extensively on these sources, I would consider you to have developed a genuinely good language model rather than cheating.



From my test results, accurately modeling these data is very challenging. I believe Uncheatable Eval more accurately reflects the value of every bit of data and computing you invest compared to other benchmarks. Models trained with more data and computing are almost always better, and there are no shortcuts. This is a key strength of Uncheatable Eval.



### Is This Too "Random"? Why Consider Random Texts from the Internet as Ground Truth?

This is why we choose rigorous and verified texts such as arXiv papers and news reports, which typically have better quality. Additionally, a round of Uncheatable Eval evaluates a model over millions of tokens, increasing the reliability of the results.



In fact, the model rankings obtained through Uncheatable Eval are very stable. For instance, the model ranked first in January's data is highly likely to remain first in February, March, April, May, and June, indicating that the data obtained through this method is sufficiently representative.

"""


def rename_columns(df):
    df.columns = [col.rsplit('_', maxsplit=1)[0] for col in df.columns]
    return df


def get_folders_matching_format(directory):
    pattern = re.compile(r'^\d{4}-\d{2}$')
    folders = []

    if not os.path.exists(directory):
        return folders

    for item in os.listdir(directory):
        full_path = os.path.join(directory, item)
        if os.path.isdir(full_path) and pattern.match(item):
            folders.append(full_path)

    return folders


def get_unique_column_names(all_data):
    # column_names = {}
    #
    # for folder_name, files in all_data.items():
    #     for file_name, sheets in files.items():
    #         for sheet_name, dataframe in sheets.items():
    #             for column in dataframe.columns:
    #                 if column not in ['Name', 'Average (The lower the better)', 'Parameters Count (B)']:
    #                     column_names[column] = None
    #
    # return list(column_names.keys())

    return ['ao3_\u200benglish', 'bbc_\u200bnews', 'wikipedia_\u200benglish', 'arxiv_\u200bcomputer_\u200bscience',
            'arxiv_\u200bphysics', 'github_\u200bcpp', 'github_\u200bpython']


def color_cell(value):
    return 'background-color: #fffdd0' if pd.notna(value) else 'default'


def update_table(period: str,

                 models: list,

                 metric: str,

                 visible_columns: list,

                 color_columns: list,

                 sort_by: str = 'Average (The lower the better)',

                 ascending: bool = True):
    target_data = all_data[period]
    target_metric = metric_to_sheet[metric]

    if models:
        target_model_size = [model_size_to_file_name[model] for model in models]
        combined_data = pd.concat([target_data[model][target_metric] for model in target_model_size], axis=0)
        combined_data['Name'] = combined_data['Name'].apply(lambda x: x.replace('.pth', ''))

        combined_data.reset_index(drop=True, inplace=True)

        if 'Average (The lower the better)' in combined_data.columns:
            relevant_columns = [col for col in visible_columns if
                                col not in ['Name', 'Parameters Count (B)', 'Average (The lower the better)']]
            combined_data['Average (The lower the better)'] = round(combined_data[relevant_columns].mean(axis=1), 3)

        sorted_data = combined_data.sort_values(by=sort_by, ascending=ascending)
        sorted_data = sorted_data.rename(columns={'Average (The lower the better)': 'Average (lower=better)'})
        visible_columns = ['Name', 'Parameters Count (B)', 'Average (lower=better)'] + visible_columns
        filtered_data = sorted_data[visible_columns]

        filtered_data.columns = [col.replace('_', ' ') for col in filtered_data.columns]

        formatter = {col: "{:.3f}" for col in filtered_data.columns if
                     filtered_data[col].dtype in ['float64', 'float32']}

        def color_column(s):
            return ['background-color: #fffdd0' if pd.notna(x) else 'default' for x in s]

        # color gradient
        colors = ["#63be7b", "#ffffff", "#f8696b"]
        cmap = LinearSegmentedColormap.from_list("custom_cmap", colors)
        target_color_columns = []
        if 'Average' in color_columns:
            target_color_columns.append('Average (lower=better)')
        if 'Individual Tests' in color_columns:
            target_color_columns.extend([col for col in filtered_data.columns if
                                         col not in ['Name', 'Parameters Count (B)', 'Average (lower=better)']])

        # styler = filtered_data.style.format(formatter).background_gradient(
        #     cmap=cmap,
        #     subset=target_color_columns,
        #     vmin=min_value,
        #     vmax=max_value
        # ).apply(color_column, subset=['Parameters Count (B)'])

        # for better visualization
        vmin = {}
        vmax = {}
        for column in filtered_data.columns:
            if column in ['Name', 'Parameters Count (B)']:
                continue
            col_values = filtered_data[column]
            second_largest = col_values.nlargest(2).iloc[-1]
            vmin[column] = col_values.min()
            vmax[column] = second_largest

        target_color_columns = []
        if 'Average' in color_columns:
            target_color_columns.append('Average (lower=better)')
        if 'Individual Tests' in color_columns:
            target_color_columns.extend([col for col in filtered_data.columns if
                                         col not in ['Name', 'Parameters Count (B)', 'Average (lower=better)']])

        styler = filtered_data.style.format(formatter).map(color_cell, subset=['Parameters Count (B)'])
        for column in target_color_columns:
            styler = styler.background_gradient(cmap=cmap, subset=[column], vmin=vmin[column], vmax=vmax[column])

        return styler
    else:
        return pd.DataFrame()


def check_model_exists(model_id):
    api = HfApi()
    try:
        model_info = api.model_info(model_id)
        return "Exists and is accessible"
    except GatedRepoError:
        return "Exists but is restricted"
    except HTTPError as e:
        if e.response.status_code == 404:
            return "Does not exist"
        else:
            return "Error: " + str(e)


def submit_model(name):
    if 'Exists' not in check_model_exists(name):
        return f"# ERROR: Model {name} does not exist on Hugging Face!"

    try:
        response = requests.post(webhook_url, json={"content": name})
        if response.status_code == 200:
            response_data = response.json()
            if response_data.get('status') == 'success':
                return "# SUCCESS: We will check the model as soon as possible. Thank you for your submission!"
            else:
                return f"# ERROR: {response_data.get('message', 'Unknown error')}"
        else:
            return f"# ERROR: Failed to submit model {name}. Server returned status code {response.status_code}."
    except requests.exceptions.HTTPError:
        return "# ERROR: Network error while contacting queue. Please try again in a few minutes."
    except Exception as e:
        print(e)
        return "ERROR: Unexpected error. Please try again later."


all_data = {}
time_list = []
for folder in get_folders_matching_format('data'):
    folder_name = os.path.basename(folder)
    time_list.append(folder_name)
    if all_data.get(folder) is None:
        all_data[folder_name] = {}
    for file_name in file_name_list:
        if all_data.get(file_name) is None:
            all_data[folder_name][file_name] = {}
        for sheet_name in sheet_name_list:
            final_file_name = os.path.join(folder, file_name)
            all_data[folder_name][file_name][sheet_name] = rename_columns(
                pd.read_excel(final_file_name + '.xlsx', sheet_name=sheet_name))


# def create_scaling_plot(all_data, period):
#     selected_columns = ['Name', 'Parameters Count (B)', 'Average (The lower the better)']
#     target_data = all_data[period]
#     new_df = pd.DataFrame()
#
#     for size in target_data.keys():
#         new_df = pd.concat([new_df, target_data[size]['cr'].loc[:, selected_columns]], axis=0)
#
#     new_df.rename(columns={
#         'Parameters Count (B)': 'Params(B)',
#         'Average (The lower the better)': 'Compression Rate (%)'
#     }, inplace=True)
#
#     fig = px.scatter(new_df,
#                      x='Params(B)',
#                      y='Compression Rate (%)',
#                      title='Compression Rate Scaling Law',
#                      hover_name='Name'
#                      )
#     fig.update_traces(marker=dict(size=12))
#     return fig


def create_scaling_plot(all_data, period):
    selected_columns = ['Name', 'Parameters Count (B)', 'Average (The lower the better)']
    target_data = all_data[period]
    new_df = pd.DataFrame()

    for size in target_data.keys():
        new_df = pd.concat([new_df, target_data[size]['cr'].loc[:, selected_columns]], axis=0)

    new_df.rename(columns={
        'Parameters Count (B)': 'Params(B)',
        'Average (The lower the better)': 'Compression Rate (%)'
    }, inplace=True)

    new_df['Log Params(B)'] = np.log(new_df['Params(B)'])
    new_df['Log Compression Rate (%)'] = np.log(new_df['Compression Rate (%)'])

    fig = px.scatter(new_df,
                     x='Log Params(B)',
                     y='Log Compression Rate (%)',
                     title='Compression Rate Scaling Law',
                     hover_name='Name',
                     custom_data=['Params(B)', 'Compression Rate (%)']
                     )

    fig.update_traces(
        hovertemplate="<b>%{hovertext}</b><br>Params(B): %{customdata[0]:.2f} B<br>Compression Rate (%): %{customdata[1]:.2f}<extra></extra>"
    )

    names_to_connect = ['Meta-Llama-3-8B',
                        'stablelm-3b-4e1t',
                        'Qwen2-1.5B',
                        'TinyLlama-1.1B-intermediate-step-1431k-3T']
    connection_points = new_df[new_df['Name'].isin(names_to_connect)]

    new_df['Color'] = new_df['Name'].apply(lambda name: '#39C5BB' if name in names_to_connect else '#636efa')

    fig.update_traces(marker=dict(color=new_df['Color']))

    X = connection_points['Log Params(B)'].values.reshape(-1, 1)
    y = connection_points['Log Compression Rate (%)'].values
    model = LinearRegression().fit(X, y)

    x_min = connection_points['Log Params(B)'].min()
    x_max = connection_points['Log Params(B)'].max()
    extended_x = np.linspace(x_min, x_max * 1.25, 100)
    extended_x_original = np.exp(extended_x)
    trend_line_y = model.predict(extended_x.reshape(-1, 1))
    trend_line_y_original = np.exp(trend_line_y)

    trend_line = go.Scatter(
        x=extended_x,
        y=trend_line_y,
        mode='lines',
        line=dict(color='skyblue', dash='dash'),
        name='Trend Line',
        hovertemplate='<b>Params(B):</b> %{customdata[0]:.2f}<br>' +
                      '<b>Compression Rate (%):</b> %{customdata[1]:.2f}<extra></extra>',
        customdata=np.stack((extended_x_original, trend_line_y_original), axis=-1)
    )

    fig.add_trace(trend_line)

    x_min = new_df['Params(B)'].min()
    x_max = new_df['Params(B)'].max()
    x_tick_vals = np.geomspace(x_min, x_max, num=5)
    x_tick_text = [f"{val:.1f}" for val in x_tick_vals]

    y_min = new_df['Compression Rate (%)'].min()
    y_max = new_df['Compression Rate (%)'].max()
    y_tick_vals = np.geomspace(y_min, y_max, num=5)
    y_tick_text = [f"{val:.1f}" for val in y_tick_vals]

    fig.update_xaxes(tickvals=np.log(x_tick_vals), ticktext=x_tick_text, title='Params(B)')
    fig.update_yaxes(tickvals=np.log(y_tick_vals), ticktext=y_tick_text, title='Compression Rate (%)', autorange='reversed')

    fig.update_layout(
        xaxis=dict(showgrid=True, zeroline=False),
        yaxis=dict(showgrid=True, zeroline=False)
    )

    fig.update_traces(marker=dict(size=12))

    return fig


initial_fig = create_scaling_plot(all_data, time_list[-1])

initial_period = time_list[-1]
initial_models = model_size_list
initial_metric = metric_list[0]
initial_columns = get_unique_column_names(all_data)
initial_colors = ['Average']

initial_data = update_table(initial_period, initial_models, initial_metric, initial_columns, initial_colors)

css = '''

.gradio-container {

    max-width: 95% !important;

}

.tab-buttons button {

    font-size: 1.3em;

}

.gr-dataframe th {

    white-space: normal;

    word-break: break-word;

}



'''

with gr.Blocks(css=css) as demo:
    gr.HTML('<h1 style="text-align:center"><span style="font-size:1.3em">🏆 LLM Compression Leaderboard</span></h1>')
    gr.HTML(
        "<h1 style='text-align:center'><span style='font-size:0.8em'>Welcome to Uncheatable Eval LLM Compression Leaderboard, where fancy fine-tuning and cheating won’t work 🚫; only compute 💻, data 📊, and real innovation 🔥 can prevail!</span></h1>")
    with gr.Tabs() as tabs:
        with gr.Tab("🏆 Leaderboard"):
            with gr.Row():
                with gr.Column():
                    period_selector = gr.Dropdown(label="Period", choices=time_list, value=time_list[0])
                    model_selector = gr.CheckboxGroup(label="Model", choices=model_size_list, value=model_size_list)
                    metric_selector = gr.Dropdown(label="Metric", choices=metric_list, value=metric_list[0])
                with gr.Column():
                    color_selector = gr.CheckboxGroup(label="Colored Columns",
                                                      choices=['Average', 'Individual Tests'],
                                                      value=['Average'])
                    colfilter = gr.CheckboxGroup(label="Data Source",
                                                 choices=get_unique_column_names(all_data),
                                                 value=get_unique_column_names(all_data))

            table = gr.Dataframe(initial_data, column_widths=[130, 60, 60, 35, 35, 40, 40, 35, 35, 35],
                                 wrap=True,
                                 height=800,
                                 )

            period_selector.change(update_table,
                                   inputs=[period_selector, model_selector, metric_selector, colfilter, color_selector],
                                   outputs=table)
            model_selector.change(update_table,
                                  inputs=[period_selector, model_selector, metric_selector, colfilter, color_selector],
                                  outputs=table)
            metric_selector.change(update_table,
                                   inputs=[period_selector, model_selector, metric_selector, colfilter, color_selector],
                                   outputs=table)
            colfilter.change(update_table,
                             inputs=[period_selector, model_selector, metric_selector, colfilter, color_selector],
                             outputs=table)
            color_selector.change(update_table,
                                  inputs=[period_selector, model_selector, metric_selector, colfilter, color_selector],
                                  outputs=table)

        with gr.Tab("🌍 MultiLang"):
            gr.Markdown("## Coming soon...")
        with gr.Tab("📈 Scaling Law"):
            period_selector_2 = gr.Dropdown(label="Period", choices=time_list, value=time_list[0])

            def update_plot(period):
                new_fig = create_scaling_plot(all_data, period)
                return new_fig


            plot = gr.Plot(initial_fig)
            period_selector_2.change(update_plot, inputs=period_selector_2, outputs=plot)

        with gr.Tab("ℹ️ About"):
            gr.Markdown(about_md)
        with gr.Tab("🚀 Submit"):
            with gr.Group():
                with gr.Row():
                    model_name = gr.Textbox(max_lines=1,
                                            placeholder="Enter model name...",
                                            show_label=False,
                                            scale=4)
                    submit = gr.Button("Submit", variant="primary", scale=0)
            output = gr.Markdown(
                "# Enter a public HF repo id, then hit Submit to add it to the evaluation queue.")

            submit.click(fn=submit_model, inputs=model_name, outputs=output)

demo.launch()