import json
from pathlib import Path
from typing import Any, Dict, List
import gradio as gr
import pandas as pd
from shared import (
CATEGORY_TABLE_CSS,
DEFAULT_WEIGHTS,
LEADERBOARD_DARK_MODE_CSS,
LEADERBOARD_TABLE_CSS,
build_pareto_figure,
build_weights,
create_empty_pareto_figure,
create_weight_bar_html,
escape_html,
get_color_for_accuracy,
sort_by_overall_score,
)
# ── Data loading ────────────────────────────────────────────────────────
_DATA_DIR = Path(__file__).parent / "data"
def _load_metrics() -> Dict[str, Any]:
metrics_path = _DATA_DIR / "content_moderation_metrics.json"
if not metrics_path.exists():
print(f"Warning: {metrics_path} not found")
return {}
with open(metrics_path, "r") as f:
return json.load(f)
METRICS_DATA = _load_metrics()
# ── Constants ───────────────────────────────────────────────────────────
CATEGORY_LABELS = {
"hate speech": "Hate Speech",
"harmful manipulation": "Harmful Manipulation",
"privacy": "Privacy",
"cyber": "Cyber",
"physical harm": "Physical Harm",
"sexual content": "Sexual Content",
"integrity & quality violations": "Integrity & Quality",
"cbrn": "CBRN",
"illegal activities": "Illegal Activities",
"self-harm": "Self-Harm",
"harm to minors": "Harm to Minors",
}
DATASET_DESCRIPTIONS: Dict[str, Dict[str, Any]] = {
"input": {
"url": "https://huggingface.co/datasets/centrepourlasecuriteia/content-moderation-input-dataset",
"label": "BELLS-O Content Moderation Input Dataset",
"summary": "This dataset contains 1,400 samples across 11 harm categories for evaluating input content moderation. It includes 300 benign prompts (for FPR evaluation) and 1,100 harmful prompts (100 per category). Samples were constructed using three methods: AI-generated (27%), extracted from existing datasets (44%), and handcrafted by experts (29%).",
"categories": dict.fromkeys(CATEGORY_LABELS, ""),
"accuracy_categories": list(CATEGORY_LABELS.keys()),
"fpr_categories": ["benign"],
},
"output": {
"url": "https://huggingface.co/datasets/centrepourlasecuriteia/content-moderation-output-dataset",
"label": "BELLS-O Content Moderation Output Dataset",
"summary": "This dataset contains 1,400 samples across 11 harm categories for evaluating output content moderation. It includes 300 benign outputs (for FPR evaluation) and 1,100 harmful outputs (100 per category). Samples were constructed using three methods: AI-generated (27%), extracted from existing datasets (44%), and handcrafted by experts (29%).",
"categories": dict.fromkeys(CATEGORY_LABELS, ""),
"accuracy_categories": list(CATEGORY_LABELS.keys()),
"fpr_categories": ["benign"],
},
}
def create_dataset_info_html(dataset_type: str) -> str:
"""Create an HTML info box describing the selected dataset."""
info = DATASET_DESCRIPTIONS.get(dataset_type)
if not info:
return ""
ds_label = info.get("label", dataset_type)
url = info.get("url", "")
summary = info.get("summary", "")
categories = info.get("categories", {})
accuracy_categories = info.get("accuracy_categories", [])
fpr_categories = info.get("fpr_categories", [])
title_html = f'{ds_label}' if url else ds_label
html = """"""
html += f'
{title_html}
'
if summary:
html += f'
{summary}
'
# Only show categories list if descriptions are non-empty
non_empty_cats = {k: v for k, v in categories.items() if v}
if non_empty_cats:
html += '
Categories:'
html += '
'
for cat_key, cat_desc in non_empty_cats.items():
cat_label = CATEGORY_LABELS.get(cat_key, cat_key)
html += f'- {cat_label}: {cat_desc}
'
html += "
"
if accuracy_categories:
cat_names = ", ".join(CATEGORY_LABELS.get(c, c) for c in accuracy_categories)
html += f'
Detection Rate from: {cat_names}
'
if fpr_categories:
cat_names = ", ".join(CATEGORY_LABELS.get(c, c) for c in fpr_categories)
html += f'
FPR from: {cat_names}
'
html += "
"
return html
# ── Data helpers ────────────────────────────────────────────────────────
def filter_metrics_by_dataset(dataset_type: str) -> Dict[str, Any]:
matching_names = {
f"bells-o-project-content-moderation-{dataset_type}",
f"centrepourlasecuriteia-content-moderation-{dataset_type}-dataset",
}
return {k: v for k, v in METRICS_DATA.items() if v.get("dataset_name") in matching_names}
def prepare_leaderboard_data(selected_categories: List[str] | None = None, dataset_type: str = "input") -> pd.DataFrame:
metrics_data = filter_metrics_by_dataset(dataset_type)
if not metrics_data:
return pd.DataFrame(
columns=[
"Model Snapshot",
"Model Developer",
"Provider",
"Detection Rate (%)",
"FPR (%)",
"Latency CI 95% (ms)",
"Mean Latency (ms)",
"Compute Access",
"Total Cost",
"Cost per 1M units",
"Cost per h",
"Cost Additional Info",
"Model Type",
"Execution Info",
"_accuracy",
"_fpr",
"_mean_latency",
"_total_cost",
]
)
all_categories = set()
for data in metrics_data.values():
all_categories.update(data.get("accuracy_per_category", {}).keys())
use_overall = (
selected_categories is None or len(selected_categories) == 0 or set(selected_categories) == all_categories
)
rows = []
for key, data in metrics_data.items():
latency_ci = data.get("latency_ci_95", {})
cost_info = data.get("cost_info", {})
latency_ci_str = (
f"[{int(round(latency_ci.get('lower', 0), 3) * 1000)}, {int(round(latency_ci.get('upper', 0), 3) * 1000)}]"
)
cost_input = cost_info.get("cost_per_1M_input_tokens", "N/A")
cost_output = cost_info.get("cost_per_1M_output_tokens", "N/A")
if cost_input == "N/A" or cost_output == "N/A":
cost_str = "Unknown"
else:
cost_str = f"Input: ${round(cost_input, 2)}/1M, Output: ${round(cost_output, 2)}/1M"
if use_overall:
accuracy = data.get("accuracy", 0)
else:
accuracy_per_category = data.get("accuracy_per_category", {})
selected_accuracies = [
accuracy_per_category.get(cat, 0) for cat in selected_categories if cat in accuracy_per_category
]
accuracy = sum(selected_accuracies) / len(selected_accuracies) if selected_accuracies else 0
accuracy_pct = accuracy * 100
fpr_pct = data.get("fpr", 0) * 100
cost_per_h_value = data["cost_info"].get("cost_per_h", "Unknown")
cost_per_h_str = f"${cost_per_h_value}" if cost_per_h_value != "N/A" else cost_per_h_value
rows.append(
{
"Model Snapshot": data.get("model_name", ""),
"Model URL": data.get("model_url", ""),
"Model Developer": data.get("model_developer", ""),
"Provider": data.get("provider", ""),
"Detection Rate (%)": round(accuracy_pct, 2),
"FPR (%)": round(fpr_pct, 2),
"Latency CI 95% (ms)": latency_ci_str,
"Mean Latency (ms)": int(round(data.get("mean_latency", 0), 3) * 1000),
"Compute Access": data["execution_specifications"].get("type", "Unknown"),
"Total Cost": f"{round(data['cost_info'].get('total_cost', 'Unknown') * 100, 1)} ct",
"Cost per 1M units": cost_str,
"Cost per h": cost_per_h_str,
"Cost Additional Info": data["cost_info"].get("cost_additional_info", ""),
"Model Type": data.get("model_type", "Unknown"),
"Execution Info": data["execution_specifications"].get("details", "Unknown"),
"_accuracy": accuracy,
"_fpr": data.get("fpr", 0),
"_mean_latency": int(round(data.get("mean_latency", 0), 3) * 1000),
"_total_cost": round(data["cost_info"].get("total_cost", "Unknown") * 100, 1),
}
)
return pd.DataFrame(rows)
# ── Leaderboard HTML ────────────────────────────────────────────────────
def create_leaderboard_html(
sort_by: str = "Detection Rate (%)",
selected_categories: List[str] | None = None,
dataset_type: str = "input",
weights: Dict[str, float] | None = None,
) -> str:
leaderboard_df = prepare_leaderboard_data(selected_categories=selected_categories, dataset_type=dataset_type)
sort_mapping = {
"Detection Rate (%)": "_accuracy",
"FPR (%)": "_fpr",
"Mean Latency (ms)": "_mean_latency",
"Total Cost": "_total_cost",
}
sort_column = sort_mapping.get(sort_by, "_accuracy")
if sort_by == "Overall Score":
sorted_df = sort_by_overall_score(leaderboard_df, weights).copy()
elif sort_by == "Detection Rate (%)":
sorted_df = leaderboard_df.sort_values(by=sort_column, ascending=False).copy()
elif sort_by in ["FPR (%)", "Mean Latency (ms)", "Total Cost"]:
sorted_df = leaderboard_df.sort_values(by=[sort_column, "_accuracy"], ascending=[True, False]).copy()
else:
sorted_df = leaderboard_df.sort_values(by=sort_column, ascending=False).copy()
sorted_df = sorted_df.reset_index(drop=True)
html = LEADERBOARD_TABLE_CSS
html += """
| Rank |
Model Snapshot |
Model Developer |
Provider |
Model Type 1 |
Detection Rate (%) 2 |
FPR (%) 3 |
Latency CI 95% (ms) |
Mean Latency (ms) |
Compute Access 4 |
Total Cost 5 |
Cost per 1M units 6 |
Cost per h 7 |
Cost Additional Info |
Execution Info |
"""
for idx, row in sorted_df.iterrows():
rank = idx + 1
detection_rate = row["Detection Rate (%)"]
fpr = row["FPR (%)"]
latency = row["Mean Latency (ms)"]
model_type = row["Model Type"]
type_class = "type-specialized" if model_type == "specialized" else "type-generalist"
model_name = escape_html(row["Model Snapshot"])
model_url = str(row["Model URL"]) if row["Model URL"] else ""
developer = escape_html(row["Model Developer"])
provider = escape_html(row["Provider"])
compute_access = escape_html(row["Compute Access"])
total_cost = escape_html(row["Total Cost"])
cost_per_1m = escape_html(row["Cost per 1M units"])
cost_per_h = escape_html(row["Cost per h"])
cost_add_info = escape_html(row["Cost Additional Info"])
latency_ci = escape_html(row["Latency CI 95% (ms)"])
exec_info = escape_html(row["Execution Info"])
if model_url and model_url != "None":
model_name_html = f'{model_name}'
else:
model_name_html = model_name
html += f"""
| {rank} |
{model_name_html} |
{developer} |
{provider} |
{model_type.title()} |
{detection_rate:.2f}% |
{fpr:.2f}% |
{latency_ci} |
{latency:d} |
{compute_access} |
{total_cost} |
{cost_per_1m} |
{cost_per_h} |
{cost_add_info} |
{exec_info} |
"""
html += """
"""
html += LEADERBOARD_DARK_MODE_CSS
return html
# ── Category table ──────────────────────────────────────────────────────
def create_category_accuracy_table_html(selected_models: List[str] | None = None, dataset_type: str = "input") -> str:
metrics_data = filter_metrics_by_dataset(dataset_type)
if selected_models is None or len(selected_models) == 0:
selected_models = [data.get("model_name", "") for data in metrics_data.values()]
all_categories = set()
for data in metrics_data.values():
all_categories.update(data.get("accuracy_per_category", {}).keys())
category_order = list(CATEGORY_LABELS.keys())
sorted_categories = [cat for cat in category_order if cat in all_categories]
sorted_categories.extend([cat for cat in all_categories if cat not in category_order])
html = CATEGORY_TABLE_CSS
html += ''
html += "| Model | "
for category in sorted_categories:
category_label = CATEGORY_LABELS.get(category, category.title())
html += f"{category_label} | "
html += "
"
html += ""
for key, data in metrics_data.items():
model_name = data.get("model_name", "")
if model_name not in selected_models:
continue
html += f"| {escape_html(model_name)} | "
accuracy_per_category = data.get("accuracy_per_category", {})
for category in sorted_categories:
accuracy = accuracy_per_category.get(category, 0)
acc_pct = round(accuracy * 100, 1)
bg_color = get_color_for_accuracy(accuracy)
html += f'{acc_pct}% | '
html += "
"
html += "
"
return html
# ── Helpers ─────────────────────────────────────────────────────────────
def get_available_categories(dataset_type: str = "input") -> List[tuple]:
metrics_data = filter_metrics_by_dataset(dataset_type)
all_categories = set()
for data in metrics_data.values():
all_categories.update(data.get("accuracy_per_category", {}).keys())
category_order = list(CATEGORY_LABELS.keys())
sorted_categories = [cat for cat in category_order if cat in all_categories]
sorted_categories.extend([cat for cat in all_categories if cat not in category_order])
return [(CATEGORY_LABELS.get(cat, cat.title()), cat) for cat in sorted_categories]
def get_available_models(dataset_type: str = "input") -> List[str]:
metrics_data = filter_metrics_by_dataset(dataset_type)
return sorted([data.get("model_name", "") for data in metrics_data.values()])
# ── Pareto ──────────────────────────────────────────────────────────────
_PARETO_METRIC_MAP = {
"Detection Rate in %": ("_accuracy", False),
"FPR in %": ("_fpr", True),
"Mean Latency in ms": ("_mean_latency", True),
"Total Cost in ct": ("_total_cost", True),
}
def create_pareto_plot_interactive(x_metric, y_metric, dataset_type="input"):
df = prepare_leaderboard_data(None, dataset_type)
return build_pareto_figure(df, x_metric, y_metric, _PARETO_METRIC_MAP)
# ── Tab builder ─────────────────────────────────────────────────────────
def build_cm_tab(dataset_type: str):
"""Build the three sub-tabs (Leaderboard, Category Performance, Pareto Frontier)
for a given content-moderation dataset type ('input' or 'output').
"""
label = "Input" if dataset_type == "input" else "Output"
with gr.Column():
gr.Markdown(f"### {label} Dataset")
gr.Markdown(
f"This benchmark uses the [BELLS-O {label} Dataset](https://huggingface.co/datasets/bells-o-project/content-moderation-{dataset_type}) "
f"to evaluate supervisor performance on {'input prompt' if dataset_type == 'input' else 'model output'} content moderation."
)
with gr.Tabs():
# ── Leaderboard sub-tab ─────────────────────────────────────
with gr.Tab("Leaderboard"):
with gr.Column():
gr.Markdown("### Interactive Leaderboard")
gr.Markdown(
"Sort the leaderboard by different metrics to compare model performance. "
"Use the dropdown below to change the sorting order. "
"Select specific categories to see detection rates calculated only for those categories."
)
category_choices = get_available_categories(dataset_type)
with gr.Row():
sort_metric = gr.Dropdown(
choices=[
"Overall Score",
"Detection Rate (%)",
"FPR (%)",
"Mean Latency (ms)",
"Total Cost",
],
value="Overall Score",
label="Sort by Metric",
interactive=True,
info="Higher detection rate is better. Lower FPR and latency are better.",
)
weight_hider = gr.HTML(value="") # inject CSS to hide/show weight row
with gr.Column(elem_id=f"weight-row-cm-{dataset_type}"):
with gr.Row():
weight_detection = gr.Number(
value=DEFAULT_WEIGHTS["Detection Rate (%)"],
label="Detection Rate weight",
minimum=0.0,
interactive=True,
)
weight_fpr = gr.Number(
value=DEFAULT_WEIGHTS["FPR (%)"],
label="FPR weight",
minimum=0.0,
interactive=True,
)
weight_latency = gr.Number(
value=DEFAULT_WEIGHTS["Mean Latency (ms)"],
label="Latency weight",
minimum=0.0,
interactive=True,
)
weight_cost = gr.Number(
value=DEFAULT_WEIGHTS["Total Cost"],
label="Cost weight",
minimum=0.0,
interactive=True,
)
weight_bar = gr.HTML(value=create_weight_bar_html(DEFAULT_WEIGHTS))
weight_inputs = [weight_detection, weight_fpr, weight_latency, weight_cost]
_weight_elem_id = f"weight-row-cm-{dataset_type}"
gr.HTML(value=create_dataset_info_html(dataset_type))
with gr.Row():
category_selector = gr.CheckboxGroup(
choices=category_choices,
value=[cat for _, cat in category_choices],
label="Select Categories for Detection Rate",
interactive=True,
info="Select categories to calculate detection rate. By default, all categories are selected (overall detection rate).",
)
leaderboard_html = gr.HTML(
value=create_leaderboard_html(
sort_by="Overall Score", dataset_type=dataset_type, weights=DEFAULT_WEIGHTS
),
label="Model Rankings",
)
def update_leaderboard(sort_by, selected_categories, w_det, w_fpr, w_lat, w_cost):
weights = build_weights(w_det, w_fpr, w_lat, w_cost)
html = create_leaderboard_html(
sort_by=sort_by,
selected_categories=selected_categories,
dataset_type=dataset_type,
weights=weights,
)
bar = create_weight_bar_html(weights)
return html, bar
def _weight_css(show: bool) -> str:
if show:
return f""
return f""
def on_sort_change(sort_by, selected_categories, w_det, w_fpr, w_lat, w_cost):
weights = build_weights(w_det, w_fpr, w_lat, w_cost)
new_html = create_leaderboard_html(
sort_by=sort_by,
selected_categories=selected_categories,
dataset_type=dataset_type,
weights=weights,
)
bar = create_weight_bar_html(weights)
css = _weight_css(sort_by == "Overall Score")
return css, new_html, bar
all_inputs = [sort_metric, category_selector] + weight_inputs
sort_metric.change(
fn=on_sort_change,
inputs=all_inputs,
outputs=[weight_hider, leaderboard_html, weight_bar],
)
category_selector.change(
fn=update_leaderboard,
inputs=all_inputs,
outputs=[leaderboard_html, weight_bar],
)
for w_input in weight_inputs:
w_input.change(
fn=update_leaderboard,
inputs=all_inputs,
outputs=[leaderboard_html, weight_bar],
)
# ── Category Performance sub-tab ────────────────────────────
with gr.Tab("Category Performance"):
with gr.Column():
gr.Markdown("### Accuracy per Category by Model")
gr.Markdown(
"Compare how different models perform across the 11 harm categories. "
"Each cell shows the detection rate (accuracy) for that model-category combination. "
"Cell colors indicate performance: darker green = higher accuracy (closer to 100%), lighter colors = lower accuracy."
)
gr.HTML(value=create_dataset_info_html(dataset_type))
model_choices = get_available_models(dataset_type)
model_selector = gr.CheckboxGroup(
choices=model_choices,
value=model_choices,
label="Select Models to Compare",
interactive=True,
info="Select one or more models to compare their performance across categories.",
)
category_table_html = gr.HTML(
value=create_category_accuracy_table_html(model_choices, dataset_type=dataset_type),
label="Category Accuracy Comparison",
)
model_selector.change(
fn=lambda models: create_category_accuracy_table_html(models, dataset_type=dataset_type),
inputs=model_selector,
outputs=category_table_html,
)
# ── Pareto Frontier sub-tab ─────────────────────────────────
with gr.Tab("Pareto Frontier"):
with gr.Column():
gr.Markdown("### Pareto Frontier Analysis")
gr.Markdown(
"Visualize trade-offs between different metrics. Models on the Pareto frontier "
"represent optimal trade-offs where improving one metric would require sacrificing another. "
"Hover over points to see the model name.\n\n"
"*NOTE: The automatic pareto curve plotting is experimental*"
)
with gr.Row():
pareto_x_metric = gr.Dropdown(
choices=["Detection Rate in %", "FPR in %", "Mean Latency in ms", "Total Cost in ct"],
value="Detection Rate in %",
label="X-Axis Metric",
interactive=True,
)
pareto_y_metric = gr.Dropdown(
choices=["Detection Rate in %", "FPR in %", "Mean Latency in ms", "Total Cost in ct"],
value="FPR in %",
label="Y-Axis Metric",
interactive=True,
)
with gr.Column():
gr.HTML('', visible=False)
pareto_plot = gr.Plot(
value=create_pareto_plot_interactive("Detection Rate in %", "FPR in %", dataset_type)
)
gr.HTML("
", visible=False)
def _update_pareto(x_metric, y_metric):
if x_metric == y_metric:
return create_empty_pareto_figure()
return create_pareto_plot_interactive(x_metric, y_metric, dataset_type)
pareto_x_metric.change(
fn=_update_pareto,
inputs=[pareto_x_metric, pareto_y_metric],
outputs=pareto_plot,
)
pareto_y_metric.change(
fn=_update_pareto,
inputs=[pareto_x_metric, pareto_y_metric],
outputs=pareto_plot,
)