app.py

import os
import gradio as gr
import pandas as pd
import plotly.express as px
from apscheduler.schedulers.background import BackgroundScheduler

from src.assets.css_html_js import custom_css
from src.assets.text_content import (
    TITLE,
    INTRODUCTION_TEXT,
    ABOUT_TEXT,
    EXAMPLE_CONFIG_TEXT,
    CITATION_BUTTON_LABEL,
    CITATION_BUTTON_TEXT,
)
from src.utils import (
    restart_space,
    load_dataset_repo,
    process_model_name,
    process_model_type,
)

HARDWARES = ["A100-80GB", "RTX4090-24GB"]
LLM_PERF_LEADERBOARD_REPO = "optimum/llm-perf-leaderboard"
LLM_PERF_DATASET_REPO = "optimum/llm-perf-dataset"
OPTIMUM_TOKEN = os.environ.get("OPTIMUM_TOKEN", None)

ALL_COLUMNS_MAPPING = {
    "backend.name": "Backend 🏭",
    "backend.torch_dtype": "Dtype 📥",
    "optimizations": "Optimizations 🛠️",
    "quantization": "Quantization 🗜️",
    #
    "weight_class": "Class 🏋️",
    "model_type": "Type 🤗",
    #
    "generate.peak_memory(MB)": "Memory (MB) ⬇️",
    "generate.throughput(tokens/s)": "Throughput (tokens/s) ⬆️",
    "generate.energy_consumption(kWh/token)": "Energy (kWh/token) ⬇️",
    "best_score": "Best Score (%) ⬆️",
    #
    "best_scored_model": "Best Scored LLM 🏆",
}
ALL_COLUMNS_DATATYPES = [
    "str",
    "str",
    "str",
    "str",
    #
    "str",
    "str",
    #
    "number",
    "number",
    "number",
    "str",
    #
    "markdown",
]
NO_DUPLICATES_COLUMNS = [
    "backend.name",
    "backend.torch_dtype",
    "optimizations",
    "quantization",
    #
    "weight_class",
    "model_type",
]
SORTING_COLUMN = ["best_score"]
SORTING_ASCENDING = [False]

llm_perf_dataset_repo = load_dataset_repo(LLM_PERF_DATASET_REPO, OPTIMUM_TOKEN)


def get_benchmark_df(benchmark="Succeeded-1xA100-80GB"):
    if llm_perf_dataset_repo:
        llm_perf_dataset_repo.git_pull()

    # load data
    benchmark_df = pd.read_csv(f"./llm-perf-dataset/reports/{benchmark}.csv")
    clusters_df = pd.read_csv("./llm-perf-dataset/Clustered-Open-LLM-Leaderboard.csv")
    # merge on model
    merged_df = benchmark_df.merge(
        clusters_df, left_on="model", right_on="best_scored_model"
    )
    # fix energy consumption nans
    merged_df["generate.energy_consumption(kWh/token)"].fillna("N/A", inplace=True)

    # add optimizations
    merged_df["optimizations"] = merged_df["backend.bettertransformer"].apply(
        lambda x: "BetterTransformer" if x else "None"
    )
    # add quantization scheme
    merged_df["quantization"] = merged_df["backend.quantization_strategy"].apply(
        lambda x: "BnB.4bit" if x == "bnb" else ("GPTQ.4bit" if x == "gptq" else "None")
    )
    # # distance to 100% score
    # score_distance = 100 - merged_df["best_score"]
    # # distance to 0s latency
    # latency_distance = merged_df["generate.latency(s)"]
    # # distance to 0MB memory
    # memory_distance = merged_df["forward.peak_memory(MB)"]
    # # add perf distance
    # merged_df["perf_distance"] = (
    #     score_distance**2 + latency_distance**2 + memory_distance**2
    # ) ** 0.5

    # sort
    merged_df.sort_values(by=SORTING_COLUMN, ascending=SORTING_ASCENDING, inplace=True)
    # drop duplicates
    merged_df.drop_duplicates(subset=NO_DUPLICATES_COLUMNS, inplace=True)
    return merged_df


def get_benchmark_table(bench_df):
    copy_df = bench_df.copy()
    # adding ** to quantized models score since we can't garantee the score is the same
    copy_df["best_score"] = copy_df.apply(
        lambda x: f"{x['best_score']}**"
        if x["backend.quantization_strategy"] in ["bnb", "gptq"]
        else x["best_score"],
        axis=1,
    )
    # filter
    copy_df = copy_df[list(ALL_COLUMNS_MAPPING.keys())]
    # rename
    copy_df.rename(columns=ALL_COLUMNS_MAPPING, inplace=True)
    # transform
    copy_df["Type 🤗"] = copy_df["Type 🤗"].apply(process_model_type)
    copy_df["Best Scored LLM 🏆"] = copy_df["Best Scored LLM 🏆"].apply(
        process_model_name
    )

    return copy_df


def get_benchmark_plot(bench_df):
    # filter latency bigger than 150s
    bench_df = bench_df[bench_df["generate.latency(s)"] <= 150]

    fig = px.scatter(
        bench_df,
        y="best_score",
        x="generate.latency(s)",
        size="generate.peak_memory(MB)",
        color="model_type",
        custom_data=list(ALL_COLUMNS_MAPPING.keys()),
        color_discrete_sequence=px.colors.qualitative.Light24,
    )
    fig.update_layout(
        title={
            "text": "Latency vs. Score vs. Memory",
            "y": 0.95,
            "x": 0.5,
            "xanchor": "center",
            "yanchor": "top",
        },
        xaxis_title="Generation Throughput (tokens/s)",
        yaxis_title="Open LLM Score (%)",
        legend_title="LLM Type",
        width=1200,
        height=600,
    )
    fig.update_traces(
        hovertemplate="<br>".join(
            [
                f"<b>{ALL_COLUMNS_MAPPING[key]}:</b> %{{customdata[{i}]}}"
                for i, key in enumerate(ALL_COLUMNS_MAPPING.keys())
            ]
        )
    )
    return fig


def filter_query(
    text,
    backends,
    datatypes,
    optimizations,
    quantization_scheme,
    score,
    memory,
    benchmark,
):
    raw_df = get_benchmark_df(benchmark=benchmark)
    filtered_df = raw_df[
        raw_df["best_scored_model"].str.lower().str.contains(text.lower())
        & raw_df["backend.name"].isin(backends)
        & raw_df["backend.torch_dtype"].isin(datatypes)
        & (
            pd.concat(
                [
                    raw_df["optimizations"].str.contains(optimization)
                    for optimization in optimizations
                ],
                axis=1,
            ).any(axis="columns")
            if len(optimizations) > 0
            else True
        )
        & (
            pd.concat(
                [
                    raw_df["quantization"] == quantization
                    for quantization in quantization_scheme
                ],
                axis=1,
            ).any(axis="columns")
            if len(quantization_scheme) > 0
            else True
        )
        & (raw_df["best_score"] >= score)
        & (raw_df["forward.peak_memory(MB)"] <= memory)
    ]
    filtered_table = get_benchmark_table(filtered_df)
    filtered_plot = get_benchmark_plot(filtered_df)
    return filtered_table, filtered_plot


# Demo interface
demo = gr.Blocks(css=custom_css)
with demo:
    # leaderboard title
    gr.HTML(TITLE)
    # introduction text
    gr.Markdown(INTRODUCTION_TEXT, elem_classes="descriptive-text")

    with gr.Tabs(elem_classes="leaderboard-tabs"):
        hardware_plots = {}
        hardware_learboards = {}
        ####################### HARDWARE TABS #######################
        for hardware in ["A100-80GB", "RTX4090-24GB"]:
            hardware_df = get_benchmark_df(benchmark=f"Succeeded-1x{hardware}")
            hardware_learboards[hardware] = get_benchmark_table(hardware_df)
            hardware_plots[hardware] = get_benchmark_plot(hardware_df)
            del hardware_df
            with gr.TabItem(f"{hardware} 🖥️", id=hardware):
                with gr.Tabs(elem_classes="hardware-tabs"):
                    with gr.TabItem("Leaderboard 🏅", id=0):
                        gr.HTML(
                            "👉 Scroll to the right 👉 for additional columns.",
                            elem_id="descriptive-text",
                        )
                        # Original leaderboard table
                        hardware_leaderboard = gr.components.Dataframe(
                            value=hardware_learboards[hardware],
                            headers=list(ALL_COLUMNS_MAPPING.values()),
                            datatype=ALL_COLUMNS_DATATYPES,
                            elem_id="hardware-leaderboard",
                            # show_label=False,
                        )
                    with gr.TabItem("Plot 📊", id=1):
                        gr.HTML(
                            "👆 Hover over the points 👆 for additional information.",
                            elem_id="descriptive-text",
                        )
                        # Original leaderboard plot
                        hardware_plotly = gr.components.Plot(
                            value=hardware_plots[hardware],
                            elem_id="hardware-plot",
                            show_label=False,
                        )

        ####################### CONTROL PANEL #######################
        with gr.TabItem("Control Panel 🎛️", id=2):
            gr.HTML(
                "Use this control panel to filter the leaderboard's table and plot.",  # noqa: E501
                elem_id="descriptive-text",
            )
            with gr.Row():
                with gr.Column(scale=1):
                    search_bar = gr.Textbox(
                        label="Model 🤗",
                        info="🔍 Search for a model name",
                        elem_id="search-bar",
                    )
                with gr.Column(scale=1):
                    with gr.Box():
                        score_slider = gr.Slider(
                            label="Open LLM Score 📈",
                            info="🎚️ Slide to minimum Open LLM score",
                            value=0,
                            elem_id="threshold-slider",
                        )
                with gr.Column(scale=1):
                    with gr.Box():
                        memory_slider = gr.Slider(
                            label="Peak Memory (MB) 📈",
                            info="🎚️ Slide to maximum Peak Memory",
                            minimum=0,
                            maximum=80 * 1024,
                            value=80 * 1024,
                            elem_id="memory-slider",
                        )
            with gr.Row():
                with gr.Column(scale=1):
                    backend_checkboxes = gr.CheckboxGroup(
                        label="Backends 🏭",
                        choices=["pytorch", "onnxruntime"],
                        value=["pytorch", "onnxruntime"],
                        info="☑️ Select the backends",
                        elem_id="backend-checkboxes",
                    )
                with gr.Column(scale=1):
                    datatype_checkboxes = gr.CheckboxGroup(
                        label="Dtypes 📥",
                        choices=["float32", "float16"],
                        value=["float32", "float16"],
                        info="☑️ Select the load dtypes",
                        elem_id="dtype-checkboxes",
                    )
                with gr.Column(scale=1):
                    optimizations_checkboxes = gr.CheckboxGroup(
                        label="Optimizations 🛠️",
                        choices=["None", "BetterTransformer"],
                        value=["None", "BetterTransformer"],
                        info="☑️ Select the optimizations",
                        elem_id="optimizations-checkboxes",
                    )
                with gr.Column(scale=1):
                    quantization_checkboxes = gr.CheckboxGroup(
                        label="Quantization 🗜️",
                        choices=["None", "BnB.4bit", "GPTQ.4bit"],
                        value=["None", "BnB.4bit", "GPTQ.4bit"],
                        info="☑️ Select the quantization schemes",
                        elem_id="quantization-checkboxes",
                    )
            with gr.Row():
                filter_button = gr.Button(
                    value="Filter 🚀",
                    elem_id="filter-button",
                )
            for hardware in HARDWARES:
                filter_button.click(
                    filter_query,
                    [
                        search_bar,
                        backend_checkboxes,
                        datatype_checkboxes,
                        optimizations_checkboxes,
                        quantization_checkboxes,
                        score_slider,
                        memory_slider,
                    ],
                    [hardware_learbo