app.py

import gradio as gr
import pandas as pd

from src.about import CITATION_BUTTON_LABEL, CITATION_BUTTON_TEXT, INTRODUCTION_TEXT, LLM_BENCHMARKS_TEXT, TITLE
from src.display.css_html_js import custom_css
from src.display.utils import COLS, TS_COLS, TYPES, AutoEvalColumn, fields
from src.envs import CRM_RESULTS_PATH
from src.populate import get_leaderboard_df_crm

original_df = get_leaderboard_df_crm(CRM_RESULTS_PATH, COLS, TS_COLS)

leaderboard_df = original_df.copy()
# leaderboard_df = leaderboard_df.style.format({"accuracy_metric_average": "{0:.2f}"})


# Searching and filtering
def update_table(
    hidden_df: pd.DataFrame,
    columns: list,
    llm_query: list,
    llm_provider_query: list,
    accuracy_method_query: str,
    accuracy_threshold_query: str,
    use_case_area_query: list,
    use_case_query: list,
    use_case_type_query: list,
    metric_area_query: list,
):
    filtered_df = filter_llm_func(hidden_df, llm_query)
    filtered_df = filter_llm_provider_func(filtered_df, llm_provider_query)
    filtered_df = filter_accuracy_method_func(filtered_df, accuracy_method_query)
    filtered_df["Accuracy Threshold"] = filter_accuracy_threshold_func(filtered_df, accuracy_threshold_query)
    filtered_df = filtered_df[filtered_df["Accuracy Threshold"]]
    filtered_df["Use Case Area"] = filtered_df["Use Case Name"].apply(lambda x: x.split(": ")[0])
    filtered_df = filter_use_case_area_func(filtered_df, use_case_area_query)
    filtered_df = filter_use_case_func(filtered_df, use_case_query)
    filtered_df = filter_use_case_type_func(filtered_df, use_case_type_query)
    # Filtering by metric area
    metric_area_maps = {
        "Cost": ["Cost Band"],
        "Accuracy": ["Accuracy", "Instruction Following", "Conciseness", "Completeness", "Factuality"],
        "Speed (Latency)": ["Response Time (Sec)", "Mean Output Tokens"],
        "Trust & Safety": ["Trust & Safety", "Safety", "Privacy", "Truthfulness", "CRM Bias"],
    }
    all_metric_cols = []
    for area in metric_area_maps:
        all_metric_cols = all_metric_cols + metric_area_maps[area]

    columns_to_keep = list(set(columns).difference(set(all_metric_cols)))
    for area in metric_area_query:
        columns_to_keep = columns_to_keep + metric_area_maps[area]
    columns = list(set(columns).intersection(set(columns_to_keep)))

    df = select_columns(filtered_df, columns)

    return df.style.map(highlight_cost_band_low, props="background-color: #b3d5a4")


# def highlight_cols(x):
#     df = x.copy()
#     df.loc[:, :] = "color: black"
#     df.loc[, ["Accuracy"]] = "background-color: #b3d5a4"
#     return df


def highlight_cost_band_low(s, props=""):

    return props if s == "Low" else None


def init_leaderboard_df(
    leaderboard_df: pd.DataFrame,
    columns: list,
    llm_query: list,
    llm_provider_query: list,
    accuracy_method_query: str,
    accuracy_threshold_query: str,
    use_case_area_query: list,
    use_case_query: list,
    use_case_type_query: list,
    metric_area_query: list,
):

    # Applying the style function
    # return df.style.apply(highlight_cols, axis=None)
    return update_table(
        leaderboard_df,
        columns,
        llm_query,
        llm_provider_query,
        accuracy_method_query,
        accuracy_threshold_query,
        use_case_area_query,
        use_case_query,
        use_case_type_query,
        metric_area_query,
    )


def filter_accuracy_method_func(df: pd.DataFrame, accuracy_method_query: str) -> pd.DataFrame:
    return df[df["Accuracy Method"] == accuracy_method_query]


def filter_accuracy_threshold_func(df: pd.DataFrame, accuracy_threshold_query: str) -> pd.DataFrame:
    accuracy_cols = ["Instruction Following", "Conciseness", "Completeness", "Accuracy"]
    return (df.loc[:, accuracy_cols] >= float(accuracy_threshold_query)).all(axis=1)


def filter_use_case_area_func(df: pd.DataFrame, use_case_area_query: list) -> pd.DataFrame:
    return df[
        df["Use Case Area"].apply(
            lambda x: len(set([_.strip() for _ in x.split("&")]).intersection(use_case_area_query))
        )
        > 0
    ]


def filter_use_case_func(df: pd.DataFrame, use_case_query: list) -> pd.DataFrame:
    return df[df["Use Case Name"].isin(use_case_query)]


def filter_use_case_type_func(df: pd.DataFrame, use_case_type_query: list) -> pd.DataFrame:
    return df[df["Use Case Type"].isin(use_case_type_query)]


def filter_llm_func(df: pd.DataFrame, llm_query: list) -> pd.DataFrame:
    return df[df["Model Name"].isin(llm_query)]


def filter_llm_provider_func(df: pd.DataFrame, llm_provider_query: list) -> pd.DataFrame:
    return df[df["LLM Provider"].isin(llm_provider_query)]


def filter_metric_area_func(df: pd.DataFrame, metric_area_query: list) -> pd.DataFrame:
    return df[df["Metric Area"].isin(metric_area_query)]


def select_columns(df: pd.DataFrame, columns: list) -> pd.DataFrame:
    always_here_cols = [
        AutoEvalColumn.model.name,
    ]
    # We use COLS to maintain sorting
    filtered_df = df[always_here_cols + [c for c in COLS if c in df.columns and c in columns]]
    return filtered_df


demo = gr.Blocks(css=custom_css)
with demo:
    gr.HTML(TITLE)
    gr.Markdown(INTRODUCTION_TEXT, elem_classes="markdown-text")

    with gr.Tabs(elem_classes="tab-buttons") as tabs:
        with gr.TabItem("🏅 LLM Benchmark", elem_id="llm-benchmark-tab-table", id=0):
            with gr.Row():
                shown_columns = gr.CheckboxGroup(
                    choices=[c.name for c in fields(AutoEvalColumn) if not c.hidden and not c.never_hidden],
                    value=[
                        c.name
                        for c in fields(AutoEvalColumn)
                        if c.displayed_by_default and not c.hidden and not c.never_hidden
                    ],
                    label="Select columns to show",
                    elem_id="column-select",
                    interactive=True,
                )
            with gr.Row():
                with gr.Column():
                    filter_llm = gr.CheckboxGroup(
                        choices=list(original_df["Model Name"].unique()),
                        value=list(original_df["Model Name"].unique()),
                        label="Model Name",
                        info="",
                        interactive=True,
                    )
                with gr.Column():
                    with gr.Row():
                        filter_llm_provider = gr.CheckboxGroup(
                            choices=list(original_df["LLM Provider"].unique()),
                            value=list(original_df["LLM Provider"].unique()),
                            label="LLM Provider",
                            info="",
                            interactive=True,
                        )
                    with gr.Row():
                        filter_metric_area = gr.CheckboxGroup(
                            choices=["Accuracy", "Speed (Latency)", "Trust & Safety", "Cost"],
                            value=["Accuracy", "Speed (Latency)", "Trust & Safety", "Cost"],
                            label="Metric Area",
                            info="",
                            interactive=True,
                        )
            with gr.Row():
                filter_use_case = gr.CheckboxGroup(
                    choices=list(original_df["Use Case Name"].unique()),
                    value=list(original_df["Use Case Name"].unique()),
                    label="Use Case",
                    info="",
                    # multiselect=True,
                    interactive=True,
                )
            with gr.Row():
                with gr.Column():
                    filter_use_case_area = gr.CheckboxGroup(
                        choices=["Service", "Sales"],
                        value=["Service", "Sales"],
                        label="Use Case Area",
                        info="",
                        interactive=True,
                    )
                with gr.Column():
                    filter_use_case_type = gr.CheckboxGroup(
                        choices=["Summary", "Generation"],
                        value=["Summary", "Generation"],
                        label="Use Case Type",
                        info="",
                        interactive=True,
                    )
                # with gr.Column():
                #     filter_use_case = gr.Dropdown(
                #         choices=list(original_df["Use Case Name"].unique()),
                #         value=list(original_df["Use Case Name"].unique()),
                #         label="Use Case",
                #         info="",
                #         multiselect=True,
                #         interactive=True,
                #     )
                with gr.Column():
                    filter_accuracy_method = gr.Radio(
                        choices=["Manual", "Auto"],
                        value="Manual",
                        label="Accuracy Method",
                        info="",
                        interactive=True,
                    )
                with gr.Column():
                    filter_accuracy_threshold = gr.Number(
                        value="0",
                        label="Accuracy Threshold",
                        info="Range: 0.0 to 4.0",
                        interactive=True,
                    )

            leaderboard_table = gr.components.Dataframe(
                # value=leaderboard_df[[c.name for c in fields(AutoEvalColumn) if c.never_hidden] + shown_columns.value],
                value=init_leaderboard_df(
                    leaderboard_df,
                    shown_columns.value,
                    filter_llm.value,
                    filter_llm_provider.value,
                    filter_accuracy_method.value,
                    filter_accuracy_threshold.value,
                    filter_use_case_area.value,
                    filter_use_case.value,
                    filter_use_case_type.value,
                    filter_metric_area.value,
                ),
                headers=[c.name for c in fields(AutoEvalColumn) if c.never_hidden] + shown_columns.value,
                datatype=TYPES,
                elem_id="leaderboard-table",
                interactive=False,
                visible=True,
            )

            # Dummy leaderboard for handling the case when the user uses backspace key
            hidden_leaderboard_table_for_search = gr.components.Dataframe(
                value=original_df[COLS],
                headers=COLS,
                datatype=TYPES,
                visible=False,
            )
            for selector in [
                shown_columns,
                filter_llm,
                filter_llm_provider,
                filter_accuracy_method,
                filter_accuracy_threshold,
                filter_use_case_area,
                filter_use_case,
                filter_use_case_type,
                filter_metric_area,
            ]:
                selector.change(
                    update_table,
                    [
                        hidden_leaderboard_table_for_search,
                        shown_columns,
                        filter_llm,
                        filter_llm_provider,
                        filter_accuracy_method,
                        filter_accuracy_threshold,
                        filter_use_case_area,
                        filter_use_case,
                        filter_use_case_type,
                        filter_metric_area,
                    ],
                    leaderboard_table,
                    queue=True,
                )
        with gr.TabItem("📝 About", elem_id="llm-benchmark-tab-table", id=3):
            gr.Markdown(LLM_BENCHMARKS_TEXT, elem_classes="markdown-text")

    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,
            )

# scheduler = BackgroundScheduler()
# scheduler.add_job(restart_space, "interval", seconds=1800)
# scheduler.start()
demo.queue(default_concurrency_limit=40).launch()