import pandas as pd
import gradio as gr
import os
import html
import requests
from dotenv import load_dotenv
from matplotlib.colors import LinearSegmentedColormap
import plotly.graph_objects as go
import numpy as np
from scipy.optimize import curve_fit
from huggingface_hub import HfApi
from huggingface_hub.utils import GatedRepoError, HfHubHTTPError
from gradio_rangeslider import RangeSlider
import datetime
from title import css, TITLE_HTML, SUBTITLE_HTML, LINKS_HTML
from data_manager import DataManager, LongContextDataManager
from longctx_utils import *
load_dotenv()
webhook_url = os.environ.get("WEBHOOK_URL")
metric_list = [
"Compression Ratio (%)",
"Bits Per Character (BPC)",
"Bits Per Byte (BPB)",
]
model_size_list = [
">20B",
"~14B",
# "~9B",
"~7B",
"~3B",
"~1.5B",
"Other",
]
metric_to_sheet = {
"Compression Ratio (%)": "cr",
"Bits Per Character (BPC)": "bpc",
"Bits Per Byte (BPB)": "bpb",
}
model_size_to_file_name = {
">20B": "20b+",
"~14B": "14b",
# "~9B": "9b",
"~7B": "7b",
"~3B": "3b",
"~1.5B": "1b5",
"Other": "other",
}
SCALING_EXTRAPOLATE_MAX_B = 10000
SCALING_FIT_POINTS = 200
SCALING_PLOT_WIDTH = 1000
SCALING_PLOT_HEIGHT = 620
SCALING_PLOT_MARGIN = dict(l=70, r=35, t=70, b=65)
MODEL_NAME_DISPLAY_MAX_CHARS = 28
FRONTIER_TABLE_COLUMNS = ["params", "model", "ratio%", "vs fit%", "row_bg", "row_hover_bg"]
FIT_RESIDUAL_GOOD_COLOR = "#2CA25F"
FIT_RESIDUAL_NEUTRAL_COLOR = "#F7F7F7"
FIT_RESIDUAL_BAD_COLOR = "#DE2D26"
FIT_LINE_COLOR = "#4B5563"
def format_fit_equation(a, b, c):
if abs(c) < 0.01:
return f"y = {a:.2f} × x^{b:.3f}"
return f"y = {a:.2f} × x^{b:.3f} + {c:.2f}"
def build_fit_line_hovertemplate(label, a, b, c, raw_rmse, log_rmse):
equation = format_fit_equation(a, b, c)
return (
f"{label}
"
f"{equation}
"
f"Raw RMSE: {raw_rmse:.2f}
"
f"Log-RMSE: {log_rmse:.3f}
"
"Params: %{x:.2f}B
"
"Predicted CR: %{y:.2f}%"
)
def build_fit_summary_legend_text(label, a, b, c, raw_rmse, log_rmse, include_label=True):
parts = []
if include_label and label:
parts.append(label)
parts.append(format_fit_equation(a, b, c))
parts.append(f"Raw RMSE: {raw_rmse:.2f}")
parts.append(f"Log-RMSE: {log_rmse:.3f}")
return "
".join(parts)
def read_about_md():
with open("about.md", "r", encoding="utf-8") as f:
return f.read()
def read_longctx_about_md():
with open("longctx_about.md", "r", encoding="utf-8") as f:
return f.read()
def update_table(
data_manager: DataManager,
period: str,
models_size: list,
metric: str,
visible_columns: list,
color_columns: list,
size_range: list,
midpoint: float = 0.5,
ascending: bool = True,
request: gr.Request = None,
):
is_dark_mode = request.is_dark if request else False
print(
f"Updating - time: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}, period: {period}, models: {models_size}, metric: {metric}, visible_columns: {visible_columns}, color_columns: {color_columns}, size_range: {size_range}, ascending: {ascending}, is_dark: {is_dark_mode}\n"
)
target_file_name = [model_size_to_file_name[model] for model in models_size]
metric_code = metric_to_sheet[metric]
# 过滤掉不在当前 period 可用列中的列名,避免错误
if visible_columns:
available_columns = data_manager.get_available_columns(period)
visible_columns = [col for col in visible_columns if col in available_columns]
filtered_data = data_manager.query(
period=period,
metric_code=metric_code,
param_range=(size_range[0], size_range[1]),
model_groups=target_file_name,
visible_columns=visible_columns,
)
if len(filtered_data) == 0:
return "No data available for the selected models and period."
colors = ["#2ca02c", "#2b2b2b", "#d62728"] if is_dark_mode else ["#63be7b", "#ffffff", "#f8696b"]
vmin, vmax, vmid = {}, {}, {}
for column in filtered_data.columns:
if column in ["Name", "Params (B)"]:
continue
col_values = filtered_data[column].dropna()
if len(col_values) > 1:
sorted_values = np.sort(col_values)
vmin[column] = sorted_values.min()
vmax[column] = sorted_values.max()
idx = int(len(sorted_values) * midpoint)
vmid[column] = sorted_values[idx]
def custom_background_gradient(series, cmap, vmin_val, vmax_val, vmid_val):
if len(series) == 0:
return series
def normalize(x):
if pd.isna(x):
return 0.5 # Neutral for NaN
if vmid_val == vmin_val and x <= vmid_val:
return 0.0
if vmid_val == vmax_val and x >= vmid_val:
return 1.0
if vmid_val == vmin_val or vmid_val == vmax_val:
return 0.5
if x <= vmid_val:
return 0.5 * (x - vmin_val) / (vmid_val - vmin_val)
else:
return 0.5 + 0.5 * (x - vmid_val) / (vmax_val - vmid_val)
normed = series.apply(normalize)
cmap_colors = [cmap(x) for x in normed]
return ["background-color: rgba({}, {}, {}, {}); color: black;".format(*[int(255 * c) for c in color[:3]], color[3]) for color in 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", "Params (B)", "Average (lower=better)"]])
def color_params_column_dynamic(value):
if not pd.notna(value):
return "default"
if is_dark_mode:
return "background-color: #4b4936; color: #f0f0f0;"
else:
return "background-color: #fffdd0; color: black;"
tooltip_map_by_column = {}
for column in filtered_data.columns:
if column in ["Name", "Params (B)"]:
continue
valid_mask = filtered_data[column].notna()
valid_count = int(valid_mask.sum())
if valid_count == 0:
continue
ranks = filtered_data[column].rank(method="min", ascending=ascending, na_option="bottom").astype("Int64")
tooltip_map_by_column[column] = {
value: f"Rank: {int(rank)}/{valid_count}" for value, rank in zip(filtered_data.loc[valid_mask, column], ranks.loc[valid_mask])
}
def format_cell_content(text, tooltip=None, extra_classes=None):
safe_text = html.escape(str(text))
classes = ["cell-tooltip-trigger"]
if extra_classes:
classes.extend(extra_classes)
tooltip_attr = ""
if tooltip:
tooltip_attr = f' data-tooltip="{html.escape(str(tooltip))}"'
return f'{safe_text}'
def truncate_model_name(text):
if len(text) <= MODEL_NAME_DISPLAY_MAX_CHARS:
return text
cutoff = text.rfind("-", 0, MODEL_NAME_DISPLAY_MAX_CHARS + 1)
if cutoff > 0:
return text[:cutoff]
return text[:MODEL_NAME_DISPLAY_MAX_CHARS]
def format_model_name(value):
if not pd.notna(value):
return ""
full_value = str(value)
safe_value = html.escape(full_value)
safe_display_value = html.escape(truncate_model_name(full_value))
return (
f''
f'{safe_display_value}'
""
)
formatter = {}
for column in filtered_data.columns:
if column == "Name":
formatter[column] = format_model_name
elif filtered_data[column].dtype in ["float64", "float32"]:
if column == "Params (B)":
formatter[column] = lambda value: "" if not pd.notna(value) else f"{value:.3f}"
else:
tooltip_map = tooltip_map_by_column.get(column, {})
def make_numeric_formatter(current_tooltip_map):
def format_numeric(value):
if not pd.notna(value):
return ""
return format_cell_content(f"{value:.3f}", tooltip=current_tooltip_map.get(value))
return format_numeric
formatter[column] = make_numeric_formatter(tooltip_map)
styler = filtered_data.style.format(formatter)
styler = styler.map(color_params_column_dynamic, subset=["Params (B)"])
for column in target_color_columns:
if column in vmin:
custom_cmap = LinearSegmentedColormap.from_list("custom_cmap", colors)
styler = styler.apply(
custom_background_gradient, cmap=custom_cmap, vmin_val=vmin[column], vmax_val=vmax[column], vmid_val=vmid[column], subset=[column]
)
styler = styler.hide(axis="index")
widths = [250, 85, 85, 70, 70, 70, 70, 70, 70, 70, 70, 70, 70, 70, 70, 70, 70, 70]
hover_filter = "brightness(1.06) saturate(1.03)" if is_dark_mode else "brightness(0.97) saturate(1.05)"
hover_shadow = "inset 0 0 0 2px rgba(255, 255, 255, 0.45)" if is_dark_mode else "inset 0 0 0 2px rgba(0, 0, 0, 0.16)"
table_styles = []
table_styles.append(
{
"selector": "th",
"props": [
("background-color", "var(--background-fill-secondary)"),
("color", "var(--body-text-color)"),
("padding", "6px 4px"),
("font-weight", "bold"),
("font-size", "12px"),
("line-height", "1.2"),
("white-space", "normal"),
("word-break", "break-word"),
("overflow-wrap", "anywhere"),
],
}
)
table_styles.append(
{
"selector": "td",
"props": [
("position", "relative"),
("transition", "filter 0.12s ease, box-shadow 0.12s ease"),
],
}
)
table_styles.append(
{
"selector": "td:hover",
"props": [
("filter", hover_filter),
("box-shadow", hover_shadow),
("z-index", "1"),
],
}
)
for i, w in enumerate(widths):
table_styles.append(
{
"selector": f"th.col{i}, td.col{i}",
"props": [
("min-width", f"{w}px"),
("max-width", f"{w}px"),
("text-align", "center"),
("border", f"1px solid var(--border-color-primary)"),
],
}
)
table_styles.append(
{
"selector": "td.col0 .model-name-cell",
"props": [
("display", "block"),
("width", "100%"),
("overflow", "hidden"),
("text-overflow", "ellipsis"),
("white-space", "nowrap"),
],
}
)
table_styles.append(
{
"selector": "td .cell-tooltip-trigger",
"props": [
("display", "block"),
("position", "relative"),
("width", "100%"),
],
}
)
table_styles.append(
{
"selector": "td:hover .cell-tooltip-trigger[data-tooltip]::after",
"props": [
("content", "attr(data-tooltip)"),
("position", "absolute"),
("top", "calc(100% + 6px)"),
("left", "50%"),
("transform", "translateX(-50%)"),
("z-index", "10000"),
("background-color", "var(--background-fill-secondary)"),
("color", "var(--body-text-color)"),
("border", "1px solid var(--border-color-primary)"),
("border-radius", "6px"),
("padding", "4px 6px"),
("font-size", "12px"),
("white-space", "nowrap"),
("pointer-events", "none"),
("box-shadow", "0 4px 14px rgba(0, 0, 0, 0.14)"),
],
}
)
table_styles.append(
{
"selector": "td:last-child:hover .cell-tooltip-trigger[data-tooltip]::after",
"props": [
("left", "auto"),
("right", "0"),
("transform", "none"),
],
}
)
table_styles.append(
{
"selector": "tbody tr:last-child td:hover .cell-tooltip-trigger[data-tooltip]::after",
"props": [
("top", "auto"),
("bottom", "calc(100% + 6px)"),
],
}
)
styler = styler.set_table_styles(table_styles)
styler = styler.set_table_attributes(
'style="border-collapse: collapse; border: 1px solid var(--border-color-primary); width: max-content !important; min-width: 100% !important; margin-left: 0 !important; margin-right: 0 !important;"'
)
table_html = styler.to_html()
return (
'
'
f"{table_html}"
"
"
)
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 HfHubHTTPError 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."
def power_law_with_offset(x, a, b, c):
"""带偏置的幂律函数: y = a * x^b + c"""
return a * np.power(x, b) + c
def calculate_fit_delta_percent(x_values, y_values, params):
"""Return percent delta from fit; negative is better because lower CR is better."""
x_arr = np.array(x_values, dtype=float)
y_arr = np.array(y_values, dtype=float)
fit_y = power_law_with_offset(x_arr, *params)
with np.errstate(divide="ignore", invalid="ignore"):
delta = ((y_arr - fit_y) / fit_y) * 100
delta[~np.isfinite(delta)] = np.nan
return delta.tolist()
def create_fit_delta_colorscale(cmin, cmax):
if cmax <= 0:
return [[0.0, FIT_RESIDUAL_NEUTRAL_COLOR], [1.0, FIT_RESIDUAL_GOOD_COLOR]]
if cmin >= 0:
return [[0.0, FIT_RESIDUAL_BAD_COLOR], [1.0, FIT_RESIDUAL_NEUTRAL_COLOR]]
mapped_cmin = -cmax
mapped_cmax = -cmin
zero_position = (0 - mapped_cmin) / (mapped_cmax - mapped_cmin)
return [
[0.0, FIT_RESIDUAL_BAD_COLOR],
[zero_position, FIT_RESIDUAL_NEUTRAL_COLOR],
[1.0, FIT_RESIDUAL_GOOD_COLOR],
]
def create_fit_delta_color_values(delta_values):
return [-float(v) if np.isfinite(v) else np.nan for v in delta_values]
def get_fit_delta_bounds(delta_values):
finite_values = [float(v) for v in delta_values if np.isfinite(v)]
if not finite_values:
return -1.0, 1.0
cmin = min(finite_values)
cmax = max(finite_values)
if cmin == cmax:
if cmin < 0:
cmax = 0.0
elif cmin > 0:
cmin = 0.0
else:
cmin, cmax = -1.0, 1.0
else:
cmin = min(cmin, 0.0)
cmax = max(cmax, 0.0)
return cmin, cmax
def _hex_to_rgb(hex_color):
hex_color = hex_color.lstrip("#")
return tuple(int(hex_color[i : i + 2], 16) for i in (0, 2, 4))
def _interpolate_rgb(start_rgb, end_rgb, t):
t = min(max(float(t), 0.0), 1.0)
return tuple(int(round(start + (end - start) * t)) for start, end in zip(start_rgb, end_rgb))
def fit_delta_to_rgba(delta_value, cmin, cmax, alpha=0.22):
if not np.isfinite(delta_value):
return "transparent"
good_rgb = _hex_to_rgb(FIT_RESIDUAL_GOOD_COLOR)
neutral_rgb = _hex_to_rgb(FIT_RESIDUAL_NEUTRAL_COLOR)
bad_rgb = _hex_to_rgb(FIT_RESIDUAL_BAD_COLOR)
if cmax <= 0:
denom = cmax - cmin
t = 1.0 if denom == 0 else (cmax - float(delta_value)) / denom
rgb = _interpolate_rgb(neutral_rgb, good_rgb, t)
elif cmin >= 0:
denom = cmax - cmin
t = 1.0 if denom == 0 else (float(delta_value) - cmin) / denom
rgb = _interpolate_rgb(neutral_rgb, bad_rgb, t)
elif float(delta_value) <= 0:
denom = 0.0 - cmin
t = 1.0 if denom == 0 else abs(float(delta_value)) / denom
rgb = _interpolate_rgb(neutral_rgb, good_rgb, t)
else:
denom = cmax - 0.0
t = 1.0 if denom == 0 else float(delta_value) / denom
rgb = _interpolate_rgb(neutral_rgb, bad_rgb, t)
return f"rgba({rgb[0]}, {rgb[1]}, {rgb[2]}, {alpha:.3f})"
def create_fit_delta_coloraxis(delta_values):
cmin, cmax = get_fit_delta_bounds(delta_values)
tick_values = np.linspace(cmax, cmin, 5)
return dict(
colorscale=create_fit_delta_colorscale(cmin, cmax),
cmin=-cmax,
cmax=-cmin,
showscale=False,
colorbar=dict(
title="vs fit",
tickmode="array",
tickvals=[-float(v) for v in tick_values],
ticktext=[f"{float(v):+.1f}%" for v in tick_values],
),
)
def filter_pareto_frontier(x_values, y_values, names):
"""
筛选帕累托前沿的点
对于每一个数据点 (x_i, y_i),如果不存在另一个点 (x_j, y_j) 满足 x_j <= x_i 且 y_j < y_i,
那么这个点 (x_i, y_i) 就属于帕累托前沿。
参数:
x_values: 参数量列表
y_values: 压缩比列表
names: 模型名称列表
返回:
(pareto_x, pareto_y, pareto_names): 帕累托前沿的点
"""
points = list(zip(x_values, y_values, names))
pareto_points = []
for i, (xi, yi, ni) in enumerate(points):
is_pareto = True
for j, (xj, yj, _) in enumerate(points):
if i != j:
# 如果存在另一个点,参数量更小或相等,且压缩比更低,则当前点不在帕累托前沿
if xj <= xi and yj < yi:
is_pareto = False
break
if is_pareto:
pareto_points.append((xi, yi, ni))
if pareto_points:
pareto_x, pareto_y, pareto_names = zip(*pareto_points)
return list(pareto_x), list(pareto_y), list(pareto_names)
else:
return [], [], []
def _build_frontier_table_rows(x_values, y_values, names, use_pareto=False):
valid_data = []
for x, y, n in zip(x_values, y_values, names):
try:
x_float = float(x)
y_float = float(y)
except (TypeError, ValueError):
continue
if x_float > 0 and y_float > 0 and not np.isnan(x_float) and not np.isnan(y_float):
valid_data.append((x_float, y_float, str(n)))
if not valid_data:
return [], []
valid_x, valid_y, valid_names = zip(*valid_data)
valid_x, valid_y, valid_names = list(valid_x), list(valid_y), list(valid_names)
pareto_x, pareto_y, pareto_names = filter_pareto_frontier(valid_x, valid_y, valid_names)
fit_x, fit_y, fit_names = (pareto_x, pareto_y, pareto_names) if use_pareto else (valid_x, valid_y, valid_names)
if len(fit_x) < 2:
return [], []
params, _, _, _, _ = fit_power_law_with_offset(
fit_x,
fit_y,
extrapolate_max_b=SCALING_EXTRAPOLATE_MAX_B,
)
point_delta_values = calculate_fit_delta_percent(valid_x, valid_y, params)
pareto_delta_values = calculate_fit_delta_percent(pareto_x, pareto_y, params)
rows = [
{
"params": round(float(param), 3),
"model": _format_frontier_model_name(model),
"ratio%": round(float(ratio), 3),
"vs fit%": float(delta),
}
for param, ratio, model, delta in zip(pareto_x, pareto_y, pareto_names, pareto_delta_values)
]
return sorted(rows, key=lambda row: (-row["params"], row["model"])), point_delta_values
def _format_frontier_model_name(model_name: str):
if not isinstance(model_name, str):
return str(model_name)
lower_name = model_name.lower()
ctx_suffix = "-ctx8192"
if lower_name.startswith("rwkv") and lower_name.endswith(ctx_suffix):
return model_name[: -len(ctx_suffix)]
return model_name
def create_scaling_frontier_table(
data_manager: DataManager,
period: str,
mode: str,
selected_datasets: list,
display_mode: str,
use_pareto: bool = False,
):
new_df = data_manager.query(
period=period,
metric_code="cr",
param_range=(0, 40),
model_groups=None,
visible_columns=None,
)
if len(new_df) == 0:
return pd.DataFrame(columns=FRONTIER_TABLE_COLUMNS)
rows = []
all_delta_values = []
is_by_dataset = "By Dataset" in str(mode)
if not is_by_dataset:
series_rows, series_deltas = _build_frontier_table_rows(
new_df["Params (B)"].astype(float).tolist(),
new_df["Average (lower=better)"].astype(float).tolist(),
new_df["Name"].tolist(),
use_pareto=use_pareto,
)
rows.extend(series_rows)
all_delta_values.extend(series_deltas)
return _finalize_frontier_table(rows, all_delta_values)
selected_datasets = selected_datasets or []
selected_datasets = [dataset for dataset in selected_datasets if dataset in new_df.columns]
if not selected_datasets:
return pd.DataFrame(columns=FRONTIER_TABLE_COLUMNS)
x_values = new_df["Params (B)"].astype(float).tolist()
names = new_df["Name"].tolist()
if display_mode == "average":
avg_y_values = []
for i in range(len(new_df)):
valid_values = []
for dataset in selected_datasets:
val = new_df[dataset].iloc[i]
if pd.notna(val) and val > 0:
valid_values.append(val)
avg_y_values.append(float(np.mean(valid_values)) if valid_values else np.nan)
series_rows, series_deltas = _build_frontier_table_rows(x_values, avg_y_values, names, use_pareto=use_pareto)
rows.extend(series_rows)
all_delta_values.extend(series_deltas)
else:
for dataset in selected_datasets:
series_rows, series_deltas = _build_frontier_table_rows(
x_values,
new_df[dataset].astype(float).tolist(),
names,
use_pareto=use_pareto,
)
rows.extend(series_rows)
all_delta_values.extend(series_deltas)
best_rows = {}
for row in rows:
row_key = (row["params"], row["model"])
if row_key not in best_rows or row["ratio%"] < best_rows[row_key]["ratio%"]:
best_rows[row_key] = row
rows = sorted(best_rows.values(), key=lambda row: (-row["params"], row["model"]))
return _finalize_frontier_table(rows, all_delta_values)
def _finalize_frontier_table(rows, all_delta_values):
if not rows:
return pd.DataFrame(columns=FRONTIER_TABLE_COLUMNS)
cmin, cmax = get_fit_delta_bounds(all_delta_values)
for row in rows:
delta = row.get("vs fit%", np.nan)
row["row_bg"] = fit_delta_to_rgba(delta, cmin, cmax, alpha=0.22)
row["row_hover_bg"] = fit_delta_to_rgba(delta, cmin, cmax, alpha=0.32)
return pd.DataFrame(rows, columns=FRONTIER_TABLE_COLUMNS)
def render_scaling_frontier_table(frontier_df):
if frontier_df is None or len(frontier_df) == 0:
rows_html = '| No Pareto frontier models |
'
else:
rows = []
for _, row in frontier_df.iterrows():
params = html.escape(f'{float(row["params"]):.3f}')
model = html.escape(str(row["model"]))
ratio = html.escape(f'{float(row["ratio%"]):.3f}')
fit_delta = row.get("vs fit%", np.nan)
fit_delta_text = html.escape(f"{float(fit_delta):+.2f}%") if np.isfinite(fit_delta) else "--"
row_bg = html.escape(str(row.get("row_bg", "transparent")), quote=True)
row_hover_bg = html.escape(str(row.get("row_hover_bg", row_bg)), quote=True)
rows.append(
f''
f'| {params} | '
f'{model} | '
f'{ratio} | '
f'{fit_delta_text} | '
f"
"
)
rows_html = "\n".join(rows)
return f"""
"""
def fit_power_law_with_offset(x_values, y_values, extrapolate_max_b=None, num_points=SCALING_FIT_POINTS):
"""
使用带偏置的幂律拟合原始数据
返回: (params, raw_rmse, log_rmse, fit_x, fit_y)
"""
x_arr = np.array(x_values)
y_arr = np.array(y_values)
# 初始参数估计
# 使用简单的幂律拟合作为初始值
log_x = np.log10(x_arr)
log_y = np.log10(y_arr)
slope, intercept = np.polyfit(log_x, log_y, 1)
a_init = 10**intercept
b_init = slope
c_init = 0 # 偏置初始值设为0
x_min, x_max = x_arr.min(), x_arr.max()
x_start = max(x_min * 0.8, np.finfo(float).tiny)
x_end = x_max * 1.2
if extrapolate_max_b is not None:
x_end = max(x_end, extrapolate_max_b)
fit_x = np.logspace(np.log10(x_start), np.log10(x_end), num_points)
try:
# 使用curve_fit进行非线性拟合
params, _ = curve_fit(power_law_with_offset, x_arr, y_arr, p0=[a_init, b_init, c_init], maxfev=10000)
a, b, c = params
# 计算预测值
y_pred = power_law_with_offset(x_arr, a, b, c)
# 计算原始空间 RMSE
raw_rmse = np.sqrt(np.mean((y_arr - y_pred) ** 2))
# 计算对数空间 RMSE
log_y_actual = np.log10(y_arr)
log_y_pred = np.log10(y_pred)
log_rmse = np.sqrt(np.mean((log_y_actual - log_y_pred) ** 2))
# 生成拟合曲线的点
fit_y = power_law_with_offset(fit_x, a, b, c)
return params, raw_rmse, log_rmse, fit_x, fit_y
except Exception as e:
print(f"Fitting failed: {e}")
# 如果拟合失败,返回简单幂律拟合结果
a = a_init
b = b_init
c = 0
params = (a, b, c)
y_pred = a * np.power(x_arr, b)
# 计算原始空间 RMSE
raw_rmse = np.sqrt(np.mean((y_arr - y_pred) ** 2))
# 计算对数空间 RMSE
log_y_actual = np.log10(y_arr)
log_y_pred = np.log10(y_pred)
log_rmse = np.sqrt(np.mean((log_y_actual - log_y_pred) ** 2))
fit_y = a * np.power(fit_x, b)
return params, raw_rmse, log_rmse, fit_x, fit_y
def create_scaling_plot(data_manager: DataManager, period: str, use_pareto: bool = False):
new_df = data_manager.query(
period=period,
metric_code="cr",
param_range=(0, 40),
model_groups=None,
visible_columns=None,
)
if len(new_df) == 0:
fig = go.Figure()
fig.update_layout(
title={"text": "Compression Ratio Scaling Law", "x": 0.5},
width=SCALING_PLOT_WIDTH,
height=SCALING_PLOT_HEIGHT,
margin=SCALING_PLOT_MARGIN,
)
return fig
x_values = new_df["Params (B)"].astype(float).tolist()
y_values = new_df["Average (lower=better)"].astype(float).tolist()
names = new_df["Name"].tolist()
# 过滤掉无效值(NaN, 0, 负数)
valid_data = [(x, y, n) for x, y, n in zip(x_values, y_values, names) if x > 0 and y > 0 and not np.isnan(x) and not np.isnan(y)]
if len(valid_data) == 0:
fig = go.Figure()
fig.update_layout(
title={"text": "Compression Ratio Scaling Law", "x": 0.5},
width=SCALING_PLOT_WIDTH,
height=SCALING_PLOT_HEIGHT,
margin=SCALING_PLOT_MARGIN,
)
return fig
x_values, y_values, names = zip(*valid_data)
x_values, y_values, names = list(x_values), list(y_values), list(names)
# 如果选择帕累托前沿,筛选数据点
if use_pareto:
fit_x_values, fit_y_values, fit_names = filter_pareto_frontier(x_values, y_values, names)
if len(fit_x_values) == 0:
fig = go.Figure()
fig.update_layout(
title={"text": "Compression Ratio Scaling Law - No Pareto Frontier", "x": 0.5},
width=SCALING_PLOT_WIDTH,
height=SCALING_PLOT_HEIGHT,
margin=SCALING_PLOT_MARGIN,
)
return fig
else:
fit_x_values, fit_y_values, fit_names = x_values, y_values, names
x_min_val = min(x_values)
x_max_val = max(x_values)
x_axis_max = x_max_val
# 使用筛选后的数据进行拟合
params, raw_rmse, log_rmse, fit_x, fit_y = fit_power_law_with_offset(
fit_x_values,
fit_y_values,
extrapolate_max_b=SCALING_EXTRAPOLATE_MAX_B,
)
a, b, c = params
y_min_val = min(y_values)
y_max_val = max(y_values)
positive_fit_y = fit_y[fit_y > 0]
if positive_fit_y.size:
y_min_val = min(y_min_val, float(positive_fit_y.min()))
y_max_val = max(y_max_val, float(positive_fit_y.max()))
x_min = np.log10(x_min_val)
x_max = np.log10(x_axis_max)
y_min = np.log10(y_min_val)
y_max = np.log10(y_max_val)
x_dtick = (x_max - x_min) / 4
y_dtick = (y_max - y_min) / 4
fig = go.Figure()
point_delta_values = calculate_fit_delta_percent(x_values, y_values, params)
# Pareto 模式下,普通散点仅显示非 Pareto 点,避免同一点重复绘制
base_points = list(zip(x_values, y_values, names, point_delta_values))
if use_pareto:
pareto_points = set(zip(fit_x_values, fit_y_values, fit_names))
base_points = [p for p in base_points if (p[0], p[1], p[2]) not in pareto_points]
if base_points:
base_x_values, base_y_values, base_names, base_delta_values = zip(*base_points)
fig.add_trace(
go.Scatter(
x=list(base_x_values),
y=list(base_y_values),
mode="markers",
name="Non-Pareto Models" if use_pareto else "All Models",
marker=dict(size=12, color=create_fit_delta_color_values(base_delta_values), coloraxis="coloraxis", opacity=0.85),
text=list(base_names),
customdata=list(zip(base_x_values, base_y_values, base_delta_values)),
hovertemplate=(
"%{text}
"
+ "Params: %{customdata[0]:.2f}B
"
+ "Compression Ratio: %{customdata[1]:.2f}%
"
+ "vs Fit: %{customdata[2]:+.2f}%
"
+ ""
),
)
)
# 如果使用帕累托前沿,高亮显示帕累托前沿的点
if use_pareto:
pareto_delta_values = calculate_fit_delta_percent(fit_x_values, fit_y_values, params)
fig.add_trace(
go.Scatter(
x=fit_x_values,
y=fit_y_values,
mode="markers",
name="Pareto Frontier",
marker=dict(
size=14,
color=create_fit_delta_color_values(pareto_delta_values),
coloraxis="coloraxis",
symbol="diamond",
opacity=1.0,
line=dict(color="#263238", width=1),
),
text=fit_names,
customdata=list(zip(fit_x_values, fit_y_values, pareto_delta_values)),
hovertemplate=(
"%{text} (Pareto)
"
+ "Params: %{customdata[0]:.2f}B
"
+ "Compression Ratio: %{customdata[1]:.2f}%
"
+ "vs Fit: %{customdata[2]:+.2f}%
"
+ ""
),
)
)
# 添加拟合曲线
fit_type = "Pareto Fit" if use_pareto else "Fit"
fig.add_trace(
go.Scatter(
x=fit_x.tolist(),
y=fit_y.tolist(),
mode="lines",
name=build_fit_summary_legend_text(fit_type, a, b, c, raw_rmse, log_rmse),
line=dict(color=FIT_LINE_COLOR, width=2, dash="dash"),
hovertemplate=build_fit_line_hovertemplate(fit_type, a, b, c, raw_rmse, log_rmse),
)
)
title_suffix = " (Pareto Frontier)" if use_pareto else ""
fig.update_layout(
title={"text": f"Compression Ratio Scaling Law{title_suffix}", "x": 0.5, "xanchor": "center", "yanchor": "top"},
width=SCALING_PLOT_WIDTH,
height=SCALING_PLOT_HEIGHT,
margin=SCALING_PLOT_MARGIN,
showlegend=True,
legend=dict(
yanchor="top",
y=0.99,
xanchor="left",
x=0.01,
bgcolor="rgba(255,255,255,0.8)",
),
xaxis=dict(
title="Parameters (B)",
showgrid=True,
zeroline=False,
type="log",
dtick=x_dtick,
tickformat=".2f",
range=[x_min - 0.1, x_max + 0.1],
),
yaxis=dict(
title="Compression Ratio (%)",
showgrid=True,
zeroline=False,
type="log",
dtick=y_dtick,
tickformat=".2f",
range=[y_min - 0.1, y_max + 0.1],
autorange="reversed",
),
coloraxis=create_fit_delta_coloraxis(point_delta_values),
)
return fig
def create_category_scaling_plot(
data_manager: DataManager, period: str, selected_datasets: list, display_mode: str = "separate", use_pareto: bool = False
):
"""
为选中的数据集绘制 scaling law 拟合线
display_mode: "separate" - 每个数据集单独显示, "average" - 计算选中数据集的平均值
use_pareto: True - 只拟合帕累托前沿的点, False - 拟合所有数据点
"""
new_df = data_manager.query(
period=period,
metric_code="cr",
param_range=(0, 40),
model_groups=None,
visible_columns=None,
)
if len(new_df) == 0 or not selected_datasets:
fig = go.Figure()
fig.update_layout(
title={"text": "Scaling Law by Dataset", "x": 0.5},
width=SCALING_PLOT_WIDTH,
height=700,
margin=SCALING_PLOT_MARGIN,
)
return fig
# 颜色配色方案 - 使用高对比度、饱和度高的颜色
color_palette = [
"#1f77b4", # 蓝色
"#ff7f0e", # 橙色
"#2ca02c", # 绿色
"#d62728", # 红色
"#9467bd", # 紫色
"#8c564b", # 棕色
"#e377c2", # 粉色
"#17becf", # 青色
"#bcbd22", # 黄绿色
"#7f7f7f", # 灰色
]
fig = go.Figure()
# 用于计算全局坐标范围
all_x_values = []
all_y_values = []
all_delta_values = []
if display_mode == "average":
# 平均模式:计算选中数据集的平均值
x_values = new_df["Params (B)"].astype(float).tolist()
names = new_df["Name"].tolist()
# 计算每个模型在选中数据集上的平均值
avg_y_values = []
for i in range(len(new_df)):
valid_values = []
for dataset in selected_datasets:
if dataset in new_df.columns:
val = new_df[dataset].iloc[i]
if not np.isnan(val) and val > 0:
valid_values.append(val)
if valid_values:
avg_y_values.append(np.mean(valid_values))
else:
avg_y_values.append(np.nan)
# 过滤掉无效值
valid_data = [(x, y, n) for x, y, n in zip(x_values, avg_y_values, names) if x > 0 and not np.isnan(y) and y > 0]
if len(valid_data) >= 2:
x_vals, y_vals, name_vals = zip(*valid_data)
x_vals, y_vals, name_vals = list(x_vals), list(y_vals), list(name_vals)
all_x_values.extend(x_vals)
all_y_values.extend(y_vals)
color = "#39C5BB"
# 如果选择帕累托前沿,筛选数据点
if use_pareto:
fit_x_vals, fit_y_vals, fit_name_vals = filter_pareto_frontier(x_vals, y_vals, name_vals)
if len(fit_x_vals) < 2:
return fig
else:
fit_x_vals, fit_y_vals, fit_name_vals = x_vals, y_vals, name_vals
# 使用筛选后的数据进行拟合
params, raw_rmse, log_rmse, fit_x, fit_y = fit_power_law_with_offset(
fit_x_vals,
fit_y_vals,
extrapolate_max_b=SCALING_EXTRAPOLATE_MAX_B,
)
a, b, c = params
positive_fit_y = fit_y[fit_y > 0]
if positive_fit_y.size:
all_y_values.extend(positive_fit_y.tolist())
point_delta_values = calculate_fit_delta_percent(x_vals, y_vals, params)
all_delta_values.extend(point_delta_values)
# 构建数据集名称列表(用于hover显示)
datasets_label = f"Average of {len(selected_datasets)} datasets"
# Pareto 模式下,普通散点仅显示非 Pareto 点,避免同一点重复绘制
base_points = list(zip(x_vals, y_vals, name_vals, point_delta_values))
if use_pareto:
pareto_points = set(zip(fit_x_vals, fit_y_vals, fit_name_vals))
base_points = [p for p in base_points if (p[0], p[1], p[2]) not in pareto_points]
if base_points:
base_x_vals, base_y_vals, base_name_vals, base_delta_vals = zip(*base_points)
fig.add_trace(
go.Scatter(
x=list(base_x_vals),
y=list(base_y_vals),
mode="markers",
name=f"{datasets_label} (Non-Pareto)" if use_pareto else datasets_label,
marker=dict(size=12, color=create_fit_delta_color_values(base_delta_vals), coloraxis="coloraxis", opacity=0.85),
text=list(base_name_vals),
customdata=list(zip(base_x_vals, base_y_vals, base_delta_vals)),
hovertemplate=(
f"%{{text}}
{datasets_label}
"
+ "Params: %{customdata[0]:.2f}B
"
+ "CR: %{customdata[1]:.2f}%
"
+ "vs Fit: %{customdata[2]:+.2f}%
"
+ ""
),
)
)
# 如果使用帕累托前沿,高亮显示帕累托前沿的点
if use_pareto:
pareto_delta_vals = calculate_fit_delta_percent(fit_x_vals, fit_y_vals, params)
fig.add_trace(
go.Scatter(
x=fit_x_vals,
y=fit_y_vals,
mode="markers",
name="Pareto Frontier",
marker=dict(
size=14,
color=create_fit_delta_color_values(pareto_delta_vals),
coloraxis="coloraxis",
symbol="diamond",
opacity=1.0,
line=dict(color="#263238", width=1),
),
text=fit_name_vals,
customdata=list(zip(fit_x_vals, fit_y_vals, pareto_delta_vals)),
hovertemplate=(
f"%{{text}} (Pareto)
{datasets_label}
"
+ "Params: %{customdata[0]:.2f}B
"
+ "CR: %{customdata[1]:.2f}%
"
+ "vs Fit: %{customdata[2]:+.2f}%
"
+ ""
),
)
)
# 添加拟合曲线
fit_type = "Pareto Fit" if use_pareto else "Fit"
fit_label = f"{datasets_label} ({fit_type})"
fit_legend_name = fit_label
if len(selected_datasets) == 1:
fit_legend_name = build_fit_summary_legend_text(fit_label, a, b, c, raw_rmse, log_rmse)
fig.add_trace(
go.Scatter(
x=fit_x.tolist(),
y=fit_y.tolist(),
mode="lines",
name=fit_legend_name,
line=dict(color=FIT_LINE_COLOR, width=2, dash="dash"),
hovertemplate=build_fit_line_hovertemplate(fit_label, a, b, c, raw_rmse, log_rmse),
)
)
else:
# 单独显示模式:为每个数据集创建散点图和拟合线
fig.add_trace(
go.Scatter(
x=[None],
y=[None],
mode="markers",
name="Non-Pareto",
marker=dict(size=9, color="rgba(55,65,81,0.9)", symbol="circle"),
hoverinfo="skip",
)
)
if use_pareto:
fig.add_trace(
go.Scatter(
x=[None],
y=[None],
mode="markers",
name="Pareto",
marker=dict(
size=10,
color="rgba(255,255,255,0.95)",
symbol="diamond",
line=dict(color="#263238", width=1.1),
),
hoverinfo="skip",
)
)
for idx, dataset in enumerate(selected_datasets):
if dataset not in new_df.columns:
continue
# 提取该数据集的数据
x_values = new_df["Params (B)"].astype(float).tolist()
y_values = new_df[dataset].astype(float).tolist()
names = new_df["Name"].tolist()
# 过滤掉无效值
valid_data = [(x, y, n) for x, y, n in zip(x_values, y_values, names) if x > 0 and y > 0 and not np.isnan(x) and not np.isnan(y)]
if len(valid_data) < 2: # 至少需要2个点才能拟合
continue
x_vals, y_vals, name_vals = zip(*valid_data)
x_vals, y_vals, name_vals = list(x_vals), list(y_vals), list(name_vals)
all_x_values.extend(x_vals)
all_y_values.extend(y_vals)
color = color_palette[idx % len(color_palette)]
# 如果选择帕累托前沿,筛选数据点
if use_pareto:
fit_x_vals, fit_y_vals, fit_name_vals = filter_pareto_frontier(x_vals, y_vals, name_vals)
if len(fit_x_vals) < 2:
continue
else:
fit_x_vals, fit_y_vals, fit_name_vals = x_vals, y_vals, name_vals
# 使用筛选后的数据进行拟合
params, raw_rmse, log_rmse, fit_x, fit_y = fit_power_law_with_offset(
fit_x_vals,
fit_y_vals,
extrapolate_max_b=SCALING_EXTRAPOLATE_MAX_B,
)
a, b, c = params
positive_fit_y = fit_y[fit_y > 0]
if positive_fit_y.size:
all_y_values.extend(positive_fit_y.tolist())
point_delta_values = calculate_fit_delta_percent(x_vals, y_vals, params)
all_delta_values.extend(point_delta_values)
# Pareto 模式下,普通散点仅显示非 Pareto 点,避免同一点重复绘制
base_points = list(zip(x_vals, y_vals, name_vals, point_delta_values))
if use_pareto:
pareto_points = set(zip(fit_x_vals, fit_y_vals, fit_name_vals))
base_points = [p for p in base_points if (p[0], p[1], p[2]) not in pareto_points]
if base_points:
base_x_vals, base_y_vals, base_name_vals, base_delta_vals = zip(*base_points)
fig.add_trace(
go.Scatter(
x=list(base_x_vals),
y=list(base_y_vals),
mode="markers",
name=f"{dataset} (Non-Pareto)" if use_pareto else f"{dataset}",
marker=dict(size=10, color=create_fit_delta_color_values(base_delta_vals), coloraxis="coloraxis", opacity=0.8),
text=list(base_name_vals),
customdata=list(zip(base_x_vals, base_y_vals, base_delta_vals)),
hovertemplate=(
f"%{{text}}
{dataset}
"
+ "Params: %{customdata[0]:.2f}B
"
+ "CR: %{customdata[1]:.2f}%
"
+ "vs Fit: %{customdata[2]:+.2f}%
"
+ ""
),
legendgroup=dataset,
showlegend=False,
)
)
# 如果使用帕累托前沿,高亮显示帕累托前沿的点
if use_pareto:
pareto_delta_vals = calculate_fit_delta_percent(fit_x_vals, fit_y_vals, params)
fig.add_trace(
go.Scatter(
x=fit_x_vals,
y=fit_y_vals,
mode="markers",
name=f"{dataset} (Pareto)",
marker=dict(
size=12,
color=create_fit_delta_color_values(pareto_delta_vals),
coloraxis="coloraxis",
symbol="diamond",
opacity=1.0,
line=dict(color="#263238", width=1),
),
text=fit_name_vals,
customdata=list(zip(fit_x_vals, fit_y_vals, pareto_delta_vals)),
hovertemplate=(
f"%{{text}} (Pareto)
{dataset}
"
+ "Params: %{customdata[0]:.2f}B
"
+ "CR: %{customdata[1]:.2f}%
"
+ "vs Fit: %{customdata[2]:+.2f}%
"
+ ""
),
legendgroup=dataset,
showlegend=False,
)
)
# 添加拟合曲线
fit_type = "Pareto Fit" if use_pareto else "Fit"
fit_label = f"{dataset} ({fit_type})"
fit_legend_name = dataset
if len(selected_datasets) == 1:
fit_legend_name = build_fit_summary_legend_text(dataset, a, b, c, raw_rmse, log_rmse)
fig.add_trace(
go.Scatter(
x=fit_x.tolist(),
y=fit_y.tolist(),
mode="lines",
name=fit_legend_name,
line=dict(color=color, width=2, dash="dash"),
hovertemplate=build_fit_line_hovertemplate(fit_label, a, b, c, raw_rmse, log_rmse),
legendgroup=dataset,
showlegend=True,
)
)
if not all_x_values or not all_y_values:
fig = go.Figure()
fig.update_layout(
title={"text": "Scaling Law by Dataset - No Valid Data", "x": 0.5},
width=SCALING_PLOT_WIDTH,
height=700,
margin=SCALING_PLOT_MARGIN,
)
return fig
# 计算全局坐标范围
x_min_val = min(all_x_values)
x_max_val = max(all_x_values)
x_axis_max = x_max_val
x_min, x_max = np.log10(x_min_val), np.log10(x_axis_max)
y_min, y_max = np.log10(min(all_y_values)), np.log10(max(all_y_values))
x_dtick = (x_max - x_min) / 4
y_dtick = (y_max - y_min) / 4
coloraxis = create_fit_delta_coloraxis(all_delta_values)
coloraxis["colorbar"].update(
dict(
x=1.02,
xanchor="left",
y=0.5,
yanchor="middle",
len=0.72,
thickness=28,
tickfont=dict(size=9),
title=dict(text="vs fit", side="top"),
)
)
fig.update_layout(
title={"text": "Scaling Law by Dataset", "x": 0.5, "xanchor": "center", "yanchor": "top"},
width=SCALING_PLOT_WIDTH,
height=700,
showlegend=True,
legend=dict(
yanchor="top",
y=0.98,
xanchor="left",
x=0.02,
bgcolor="rgba(255,255,255,0.78)",
bordercolor="rgba(15,23,42,0.08)",
borderwidth=1,
font=dict(size=9),
tracegroupgap=2,
),
xaxis=dict(
title="Parameters (B)",
showgrid=True,
zeroline=False,
type="log",
dtick=x_dtick,
tickformat=".2f",
range=[x_min - 0.1, x_max + 0.1],
),
yaxis=dict(
title="Compression Ratio (%)",
showgrid=True,
zeroline=False,
type="log",
dtick=y_dtick,
tickformat=".2f",
range=[y_min - 0.1, y_max + 0.1],
autorange="reversed",
),
coloraxis=coloraxis,
margin=dict(l=70, r=80, t=70, b=65),
)
return fig
if __name__ == "__main__":
data_manager = DataManager("data")
time_list = data_manager.get_available_periods()
last_period = time_list[0]
# Long Context Data
lc_dm = LongContextDataManager("longctx_data")
lc_periods = lc_dm.get_available_periods()
default_lc_period = lc_periods[0]
MODE_ABS_AVG = "Absolute (Averaged by Model)"
MODE_ABS_SINGLE = "Absolute (By Dataset)"
MODE_REL_AVG = "Relative (Averaged by Model)"
MODE_REL_SINGLE = "Relative (By Dataset)"
lc_modes = [MODE_ABS_AVG, MODE_ABS_SINGLE, MODE_REL_AVG, MODE_REL_SINGLE]
default_lc_mode = MODE_ABS_AVG
# init_lc_choices = lc_dm.get_model_choices(default_lc_period)
init_lc_choices = lc_dm.get_model_choices(default_lc_period)
def get_default_model(choices):
"""获取默认模型,优先选择 Qwen3-8B-Base,否则返回第一个模型"""
if not choices:
return None
for display_name, model_name in choices:
if model_name == "Qwen3-8B-Base":
return model_name
return choices[0][1]
def create_initial_lc_plot():
if not init_lc_choices:
return None
default_model = get_default_model(init_lc_choices)
data_map = {}
paths = lc_dm.get_paths_for_model(default_lc_period, default_model)
data_map[default_model] = paths
return draw_long_context_plot(default_lc_mode, data_map, None, 0.1, 32, 32, [None, None], 0.02)
initial_lc_plot = create_initial_lc_plot()
initial_fig = create_scaling_plot(data_manager, last_period) if last_period else go.Figure()
initial_metric = metric_list[0]
initial_columns = data_manager.get_available_columns(last_period)
initial_colors = ["Average", "Individual Tests"]
initial_size_range = [0, 50]
# 默认不显示 ao3 nonenglish 列
default_visible_columns = [c for c in initial_columns if c != "ao3 nonenglish"]
initial_data = update_table(
data_manager, last_period, model_size_list, initial_metric, default_visible_columns, initial_colors, initial_size_range
)
theme = gr.themes.Default(
font=[
gr.themes.GoogleFont("Source Sans Pro"),
"ui-sans-serif",
"system-ui",
"sans-serif",
],
font_mono=[
"IBM Plex Mono",
"ui-monospace",
"Consolas",
"monospace",
],
)
with gr.Blocks(theme=theme, css=css) as demo:
gr.HTML(TITLE_HTML)
gr.HTML(SUBTITLE_HTML)
gr.HTML(LINKS_HTML)
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=last_period)
metric_selector = gr.Dropdown(label="Metric", choices=metric_list, value=initial_metric)
model_selector = gr.CheckboxGroup(label="Model Size", choices=model_size_list, value=model_size_list)
size_range_slider = RangeSlider(minimum=0, maximum=50, value=[0, 50], step=0.1, label="Model Size Range")
midpoint_slider = gr.Slider(minimum=0.1, maximum=0.9, value=0.5, step=0.01, label="Color Gradient Midpoint")
color_selector = gr.CheckboxGroup(label="Colored Columns", choices=["Average", "Individual Tests"], value=initial_colors)
with gr.Column():
# Data Source 分组定义
code_cols = ["github cpp", "github javascript", "github python", "github markdown", "github other"]
science_cols = ["arxiv math", "arxiv physics", "arxiv cs", "arxiv other", "biorxiv all"]
knowledge_cols = ["wikipedia english", "bbc news", "ao3 english"]
multilingual_cols = ["wikipedia nonenglish", "ao3 nonenglish"]
initial_code = [c for c in code_cols if c in initial_columns]
initial_science = [c for c in science_cols if c in initial_columns]
initial_knowledge = [c for c in knowledge_cols if c in initial_columns]
initial_multilingual = [c for c in multilingual_cols if c in initial_columns]
default_multilingual = [c for c in initial_multilingual if c != "ao3 nonenglish"]
with gr.Column(elem_classes=["data-source-box"]):
gr.Markdown("Data Sources")
# 代码 (Code)
with gr.Row():
toggle_code = gr.Checkbox(label="💻 Code", value=True, scale=0, min_width=150)
colfilter_code = gr.CheckboxGroup(
choices=initial_code, value=initial_code, show_label=False, scale=3, elem_classes=["aligned-checkboxes"]
)
# 科学 (Science)
with gr.Row():
toggle_science = gr.Checkbox(label="🔬 Science", value=True, scale=0, min_width=150)
colfilter_science = gr.CheckboxGroup(
choices=initial_science, value=initial_science, show_label=False, scale=3, elem_classes=["aligned-checkboxes"]
)
# 世界知识 (Knowledge)
with gr.Row():
toggle_knowledge = gr.Checkbox(label="📖 Knowledge", value=True, scale=0, min_width=150)
colfilter_knowledge = gr.CheckboxGroup(
choices=initial_knowledge, value=initial_knowledge, show_label=False, scale=3, elem_classes=["aligned-checkboxes"]
)
# 多语言 (Multilingual)
with gr.Row():
toggle_multilingual = gr.Checkbox(label="🌍 Multilingual", value=True, scale=0, min_width=150)
colfilter_multilingual = gr.CheckboxGroup(
choices=initial_multilingual,
value=default_multilingual,
show_label=False,
scale=3,
elem_classes=["aligned-checkboxes"],
)
table = gr.HTML(initial_data, elem_classes=["leaderboard-table"])
def update_table_wrapper(
period, models_size, metric, code_sel, science_sel, knowledge_sel, multilingual_sel, color_columns, size_range, midpoint
):
visible_columns = code_sel + science_sel + knowledge_sel + multilingual_sel
return update_table(data_manager, period, models_size, metric, visible_columns, color_columns, size_range, midpoint)
def update_column_choices(period, cur_code, cur_science, cur_knowledge, cur_multilingual):
if not period:
empty = gr.update(choices=[], value=[])
return empty, empty, empty, empty
columns = data_manager.get_available_columns(period)
new_code = [c for c in code_cols if c in columns]
new_science = [c for c in science_cols if c in columns]
new_knowledge = [c for c in knowledge_cols if c in columns]
new_multilingual = [c for c in multilingual_cols if c in columns]
sel_code = [c for c in cur_code if c in new_code] if cur_code else new_code
sel_science = [c for c in cur_science if c in new_science] if cur_science else new_science
sel_knowledge = [c for c in cur_knowledge if c in new_knowledge] if cur_knowledge else new_knowledge
sel_multilingual = [c for c in cur_multilingual if c in new_multilingual] if cur_multilingual else new_multilingual
if not sel_code:
sel_code = new_code
if not sel_science:
sel_science = new_science
if not sel_knowledge:
sel_knowledge = new_knowledge
if not sel_multilingual:
sel_multilingual = new_multilingual
return (
gr.update(choices=new_code, value=sel_code),
gr.update(choices=new_science, value=sel_science),
gr.update(choices=new_knowledge, value=sel_knowledge),
gr.update(choices=new_multilingual, value=sel_multilingual),
)
# 总开关功能
def toggle_group(enabled, group_cols, available_cols):
valid_cols = [c for c in group_cols if c in available_cols]
return valid_cols if enabled else []
toggle_code.change(lambda enabled: toggle_group(enabled, code_cols, initial_columns), inputs=[toggle_code], outputs=[colfilter_code])
toggle_science.change(
lambda enabled: toggle_group(enabled, science_cols, initial_columns), inputs=[toggle_science], outputs=[colfilter_science]
)
toggle_knowledge.change(
lambda enabled: toggle_group(enabled, knowledge_cols, initial_columns), inputs=[toggle_knowledge], outputs=[colfilter_knowledge]
)
toggle_multilingual.change(
lambda enabled: toggle_group(enabled, multilingual_cols, initial_columns),
inputs=[toggle_multilingual],
outputs=[colfilter_multilingual],
)
shared_inputs = [
period_selector,
model_selector,
metric_selector,
colfilter_code,
colfilter_science,
colfilter_knowledge,
colfilter_multilingual,
color_selector,
size_range_slider,
midpoint_slider,
]
period_selector.change(
update_column_choices,
inputs=[period_selector, colfilter_code, colfilter_science, colfilter_knowledge, colfilter_multilingual],
outputs=[colfilter_code, colfilter_science, colfilter_knowledge, colfilter_multilingual],
)
period_selector.change(update_table_wrapper, inputs=shared_inputs, outputs=table)
model_selector.change(update_table_wrapper, inputs=shared_inputs, outputs=table)
metric_selector.change(update_table_wrapper, inputs=shared_inputs, outputs=table)
colfilter_code.change(update_table_wrapper, inputs=shared_inputs, outputs=table)
colfilter_science.change(update_table_wrapper, inputs=shared_inputs, outputs=table)
colfilter_knowledge.change(update_table_wrapper, inputs=shared_inputs, outputs=table)
colfilter_multilingual.change(update_table_wrapper, inputs=shared_inputs, outputs=table)
color_selector.change(update_table_wrapper, inputs=shared_inputs, outputs=table)
size_range_slider.change(update_table_wrapper, inputs=shared_inputs, outputs=table)
midpoint_slider.change(update_table_wrapper, inputs=shared_inputs, outputs=table)
with gr.Tab("📚 Long Context"):
gr.Markdown(read_longctx_about_md())
with gr.Row():
with gr.Column(scale=1):
lc_period_dropdown = gr.Dropdown(label="Period", choices=lc_periods, value=default_lc_period)
lc_mode_radio = gr.Radio(label="Visualization Mode", choices=lc_modes, value=default_lc_mode)
gr.Markdown("### Model / Dataset Selection")
default_model = get_default_model(init_lc_choices)
default_selected_models = [default_model] if default_model else []
lc_select_abs = gr.Dropdown(
label="Select Models", choices=init_lc_choices, value=default_selected_models, multiselect=True, visible=True
)
lc_select_base = gr.Dropdown(
label="Baseline Model",
choices=init_lc_choices,
value=None,
multiselect=False,
visible=False,
)
lc_select_comp = gr.Dropdown(label="Comparison Models", choices=init_lc_choices, value=[], multiselect=True, visible=False)
# By Dataset mode selectors
init_dataset_choices = lc_dm.get_dataset_choices(default_lc_period) if default_lc_period else []
default_selected_datasets = [init_dataset_choices[0][1]] if init_dataset_choices else []
lc_select_datasets = gr.Dropdown(
label="Select Datasets", choices=init_dataset_choices, value=default_selected_datasets, multiselect=True, visible=False
)
lc_select_models_single = gr.Dropdown(
label="Select Models", choices=init_lc_choices, value=default_selected_models, multiselect=True, visible=False
)
lc_select_base_model_single = gr.Dropdown(
label="Baseline Model",
choices=init_lc_choices,
value=None,
multiselect=False,
visible=False,
)
lc_select_comp_models_single = gr.Dropdown(
label="Comparison Models", choices=init_lc_choices, value=[], multiselect=True, visible=False
)
with gr.Accordion("Advanced Settings", open=True):
lc_smooth = gr.Slider(1, 125, 32, step=1, label="Smooth Window")
lc_cutoff = gr.Slider(0.05, 1.0, 0.1, step=0.05, label="Cutoff Ratio")
lc_tail_drop = gr.Slider(0.0, 0.9, 0.02, step=0.01, label="Tail Drop Ratio")
lc_offset = gr.Number(32, label="Start Offset (Bytes)")
with gr.Row():
lc_ymin = gr.Textbox(label="Y Min", placeholder="Auto", value="")
lc_ymax = gr.Textbox(label="Y Max", placeholder="Auto", value="")
lc_btn_plot = gr.Button("Visualize", variant="primary")
with gr.Column(scale=3):
lc_plot_output = gr.Plot(label="Visualization Result", value=initial_lc_plot)
def update_lc_inputs(period, mode):
if not period:
return tuple([gr.update()] * 7)
is_model_agg = "Averaged by Model" in mode
is_single_dataset = "By Dataset" in mode
is_relative = "Relative" in mode
def get_default_model(choices):
"""获取默认模型,优先选择 Qwen3-8B-Base,否则返回第一个模型"""
if not choices:
return None
for display_name, model_name in choices:
if model_name == "Qwen3-8B-Base":
return model_name
return choices[0][1] if choices else None
if is_model_agg:
# Averaged by Model mode - use existing logic
choices = lc_dm.get_model_choices(period)
label_suffix = "Models"
if not is_relative:
# Absolute (Averaged by Model) - 默认选择 Qwen3-8B-Base
default_model = get_default_model(choices)
default_selected = [default_model] if default_model else []
return (
gr.update(visible=True, choices=choices, label=f"Select {label_suffix}", value=default_selected),
gr.update(visible=False, choices=choices, value=None),
gr.update(visible=False, choices=choices, value=[]),
gr.update(visible=False, value=[]),
gr.update(visible=False, value=[]),
gr.update(visible=False, value=None),
gr.update(visible=False, value=[]),
)
else:
default_baseline = get_default_model(choices)
return (
gr.update(visible=False, choices=choices, value=[]),
gr.update(visible=True, choices=choices, label=f"Baseline", value=default_baseline),
gr.update(visible=True, choices=choices, label=f"Comparison", value=[]),
gr.update(visible=False, value=[]),
gr.update(visible=False, value=[]),
gr.update(visible=False, value=None),
gr.update(visible=False, value=[]),
)
else:
# By Dataset mode
dataset_choices = lc_dm.get_dataset_choices(period)
model_choices = lc_dm.get_model_choices(period)
if not is_relative:
# Absolute By Dataset - 默认选择 Qwen3-8B-Base
default_model = get_default_model(model_choices)
default_selected = [default_model] if default_model else []
return (
gr.update(visible=False, value=[]),
gr.update(visible=False, value=None),
gr.update(visible=False, value=[]),
gr.update(visible=True, choices=dataset_choices, value=[]),
gr.update(visible=True, choices=model_choices, value=default_selected),
gr.update(visible=False, value=None),
gr.update(visible=False, value=[]),
)
else:
# Relative By Dataset - use same datasets for all models
default_baseline = get_default_model(model_choices)
return (
gr.update(visible=False, value=[]),
gr.update(visible=False, value=None),
gr.update(visible=False, value=[]),
gr.update(visible=True, choices=dataset_choices, value=[]),
gr.update(visible=False, value=[]),
gr.update(visible=True, choices=model_choices, value=default_baseline),
gr.update(visible=True, choices=model_choices, value=[]),
)
lc_period_dropdown.change(
fn=update_lc_inputs,
inputs=[lc_period_dropdown, lc_mode_radio],
outputs=[
lc_select_abs,
lc_select_base,
lc_select_comp,
lc_select_datasets,
lc_select_models_single,
lc_select_base_model_single,
lc_select_comp_models_single,
],
)
lc_mode_radio.change(
fn=update_lc_inputs,
inputs=[lc_period_dropdown, lc_mode_radio],
outputs=[
lc_select_abs,
lc_select_base,
lc_select_comp,
lc_select_datasets,
lc_select_models_single,
lc_select_base_model_single,
lc_select_comp_models_single,
],
)
def run_lc_plot(
mode,
period,
sel_abs,
sel_base,
sel_comp,
sel_datasets,
sel_models_single,
sel_base_model_single,
sel_comp_models_single,
smooth,
cutoff,
tail_drop,
offset,
ymin,
ymax,
):
data_map = {}
baseline_key = None
is_model_agg = "Averaged by Model" in mode
is_relative = "Relative" in mode
if is_model_agg:
# Averaged by Model mode - existing logic
if not is_relative:
selection = sel_abs
else:
if not sel_base:
return None
selection = [sel_base] + sel_comp
baseline_key = sel_base
if not selection:
return None
for item in selection:
paths = lc_dm.get_paths_for_model(period, item)
if paths:
data_map[item] = paths
else:
# By Dataset mode
if not is_relative:
# Absolute By Dataset
if not sel_datasets or not sel_models_single:
return None
for model_name in sel_models_single:
paths = lc_dm.get_paths_for_model_and_datasets(period, model_name, sel_datasets)
if paths:
data_map[model_name] = paths
else:
# Relative By Dataset - use same datasets for all models
if not sel_datasets or not sel_base_model_single:
return None
# Baseline model with selected datasets (averaged)
baseline_paths = lc_dm.get_paths_for_model_and_datasets(period, sel_base_model_single, sel_datasets)
if baseline_paths:
baseline_key = sel_base_model_single
data_map[baseline_key] = baseline_paths
# Comparison models with same datasets (averaged)
if sel_comp_models_single:
for model_name in sel_comp_models_single:
paths = lc_dm.get_paths_for_model_and_datasets(period, model_name, sel_datasets)
if paths:
data_map[model_name] = paths
if not data_map:
return None
def _to_float_or_none(val):
if val is None:
return None
s = str(val).strip()
if not s:
return None
try:
return float(s)
except ValueError:
return None
ymin = _to_float_or_none(ymin)
ymax = _to_float_or_none(ymax)
y_range = [ymin, ymax]
return draw_long_context_plot(mode, data_map, baseline_key, cutoff, smooth, int(offset), y_range, tail_drop)
lc_btn_plot.click(
fn=run_lc_plot,
inputs=[
lc_mode_radio,
lc_period_dropdown,
lc_select_abs,
lc_select_base,
lc_select_comp,
lc_select_datasets,
lc_select_models_single,
lc_select_base_model_single,
lc_select_comp_models_single,
lc_smooth,
lc_cutoff,
lc_tail_drop,
lc_offset,
lc_ymin,
lc_ymax,
],
outputs=lc_plot_output,
)
with gr.Tab("📈 Scaling Law"):
gr.Markdown("### Compression Ratio Scaling Law")
gr.Markdown("Explore how compression ratio scales with model parameters across different datasets.")
# 显示模式选择
MODE_OVERALL = "📊 Overall (Average)"
MODE_BY_DATASET = "📈 By Dataset"
scaling_modes = [MODE_OVERALL, MODE_BY_DATASET]
# 数据集列表
all_datasets = [
"github cpp",
"github javascript",
"github python",
"github markdown",
"github other",
"arxiv math",
"arxiv physics",
"arxiv cs",
"arxiv other",
"biorxiv all",
"wikipedia english",
"wikipedia nonenglish",
"bbc news",
"ao3 english",
"ao3 nonenglish",
]
initial_datasets = all_datasets[:4]
with gr.Row():
with gr.Column(scale=1):
scaling_period_selector = gr.Dropdown(label="Period", choices=time_list, value=last_period)
scaling_mode_radio = gr.Radio(label="Display Mode", choices=scaling_modes, value=MODE_OVERALL)
# 拟合方式选择
scaling_fit_mode = gr.Radio(
label="Fitting Method",
choices=[("Fit All Data", False), ("Fit Pareto Frontier", True)],
value=True,
info="Pareto Frontier: only fit points where no other model has both smaller parameters and lower compression ratio",
)
# 数据集选择器(初始隐藏)
scaling_dataset_selector = gr.CheckboxGroup(
label="Select Datasets", choices=all_datasets, value=initial_datasets, visible=False
)
# 数据集显示方式(初始隐藏)
scaling_dataset_display_mode = gr.Radio(
label="Dataset Display",
choices=[("Average Selected", "average"), ("Show Separately", "separate")],
value="separate",
visible=False,
)
with gr.Column(scale=5, min_width=1440):
with gr.Row(equal_height=False):
with gr.Column(scale=4, min_width=1040):
initial_scaling_fig = create_scaling_plot(data_manager, last_period, use_pareto=True) if last_period else go.Figure()
scaling_plot = gr.Plot(initial_scaling_fig, elem_classes=["scaling-plot"])
with gr.Column(scale=1, min_width=380):
initial_scaling_frontier_df = (
create_scaling_frontier_table(data_manager, last_period, MODE_OVERALL, initial_datasets, "separate", use_pareto=True)
if last_period
else pd.DataFrame(columns=FRONTIER_TABLE_COLUMNS)
)
scaling_frontier_table = gr.HTML(
value=render_scaling_frontier_table(initial_scaling_frontier_df),
elem_classes=["frontier-table"],
)
def update_scaling_mode_visibility(mode):
"""根据模式切换数据集选择器的可见性"""
is_by_dataset = mode == MODE_BY_DATASET
return gr.update(visible=is_by_dataset), gr.update(visible=is_by_dataset)
def update_scaling_plot_unified(period, mode, datasets, dataset_display_mode, use_pareto):
"""统一的绑图更新函数"""
if mode == MODE_OVERALL:
fig = create_scaling_plot(data_manager, period, use_pareto)
else: # MODE_BY_DATASET
fig = create_category_scaling_plot(data_manager, period, datasets, dataset_display_mode, use_pareto)
frontier_table = create_scaling_frontier_table(
data_manager,
period,
mode,
datasets,
dataset_display_mode,
use_pareto=use_pareto,
)
return fig, gr.update(value=render_scaling_frontier_table(frontier_table), visible=bool(use_pareto))
# 模式切换时更新可见性和图表
scaling_mode_radio.change(
fn=update_scaling_mode_visibility, inputs=[scaling_mode_radio], outputs=[scaling_dataset_selector, scaling_dataset_display_mode]
)
scaling_mode_radio.change(
fn=update_scaling_plot_unified,
inputs=[scaling_period_selector, scaling_mode_radio, scaling_dataset_selector, scaling_dataset_display_mode, scaling_fit_mode],
outputs=[scaling_plot, scaling_frontier_table],
)
# Period 改变时更新图表
scaling_period_selector.change(
fn=update_scaling_plot_unified,
inputs=[scaling_period_selector, scaling_mode_radio, scaling_dataset_selector, scaling_dataset_display_mode, scaling_fit_mode],
outputs=[scaling_plot, scaling_frontier_table],
)
# 数据集选择改变时更新图表
scaling_dataset_selector.change(
fn=update_scaling_plot_unified,
inputs=[scaling_period_selector, scaling_mode_radio, scaling_dataset_selector, scaling_dataset_display_mode, scaling_fit_mode],
outputs=[scaling_plot, scaling_frontier_table],
)
# 数据集显示模式改变时更新图表
scaling_dataset_display_mode.change(
fn=update_scaling_plot_unified,
inputs=[scaling_period_selector, scaling_mode_radio, scaling_dataset_selector, scaling_dataset_display_mode, scaling_fit_mode],
outputs=[scaling_plot, scaling_frontier_table],
)
# 拟合方式改变时更新图表
scaling_fit_mode.change(
fn=update_scaling_plot_unified,
inputs=[scaling_period_selector, scaling_mode_radio, scaling_dataset_selector, scaling_dataset_display_mode, scaling_fit_mode],
outputs=[scaling_plot, scaling_frontier_table],
)
with gr.Tab("ℹ️ About"):
gr.Markdown(read_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(share=False)