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"""
Pareto Frontier Models
{rows_html}
paramsmodelratio%vs fit
""" 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)