Spaces:
Running
Running
| 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"<b>{label}</b><br>" | |
| f"{equation}<br>" | |
| f"Raw RMSE: {raw_rmse:.2f}<br>" | |
| f"Log-RMSE: {log_rmse:.3f}<br>" | |
| "Params: %{x:.2f}B<br>" | |
| "Predicted CR: %{y:.2f}%<extra></extra>" | |
| ) | |
| 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 "<br>".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'<span class="{" ".join(classes)}"{tooltip_attr}>{safe_text}</span>' | |
| 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'<span class="cell-tooltip-trigger model-name-trigger" data-tooltip="{safe_value}">' | |
| f'<span class="model-name-cell">{safe_display_value}</span>' | |
| "</span>" | |
| ) | |
| 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 ( | |
| '<div class="leaderboard-table-scroll" style="width: 100%; max-width: 100%; overflow-x: auto; padding-bottom: 4px;">' | |
| f"{table_html}" | |
| "</div>" | |
| ) | |
| 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 = '<tr><td class="empty" colspan="4">No Pareto frontier models</td></tr>' | |
| 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'<tr style="--row-bg: {row_bg}; --row-hover-bg: {row_hover_bg};">' | |
| f'<td class="params">{params}</td>' | |
| f'<td class="model">{model}</td>' | |
| f'<td class="ratio">{ratio}</td>' | |
| f'<td class="delta">{fit_delta_text}</td>' | |
| f"</tr>" | |
| ) | |
| rows_html = "\n".join(rows) | |
| return f""" | |
| <div class="frontier-table-card"> | |
| <div class="frontier-table-title">Pareto Frontier Models</div> | |
| <div class="frontier-table-scroll"> | |
| <table class="frontier-table-inner"> | |
| <colgroup> | |
| <col class="params-col"> | |
| <col class="model-col"> | |
| <col class="ratio-col"> | |
| <col class="delta-col"> | |
| </colgroup> | |
| <thead> | |
| <tr><th>params</th><th>model</th><th>ratio%</th><th>vs fit</th></tr> | |
| </thead> | |
| <tbody> | |
| {rows_html} | |
| </tbody> | |
| </table> | |
| </div> | |
| </div> | |
| """ | |
| 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=( | |
| "<b>%{text}</b><br>" | |
| + "Params: %{customdata[0]:.2f}B<br>" | |
| + "Compression Ratio: %{customdata[1]:.2f}%<br>" | |
| + "vs Fit: %{customdata[2]:+.2f}%<br>" | |
| + "<extra></extra>" | |
| ), | |
| ) | |
| ) | |
| # 如果使用帕累托前沿,高亮显示帕累托前沿的点 | |
| 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=( | |
| "<b>%{text}</b> (Pareto)<br>" | |
| + "Params: %{customdata[0]:.2f}B<br>" | |
| + "Compression Ratio: %{customdata[1]:.2f}%<br>" | |
| + "vs Fit: %{customdata[2]:+.2f}%<br>" | |
| + "<extra></extra>" | |
| ), | |
| ) | |
| ) | |
| # 添加拟合曲线 | |
| 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"<b>%{{text}}</b><br>{datasets_label}<br>" | |
| + "Params: %{customdata[0]:.2f}B<br>" | |
| + "CR: %{customdata[1]:.2f}%<br>" | |
| + "vs Fit: %{customdata[2]:+.2f}%<br>" | |
| + "<extra></extra>" | |
| ), | |
| ) | |
| ) | |
| # 如果使用帕累托前沿,高亮显示帕累托前沿的点 | |
| 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"<b>%{{text}}</b> (Pareto)<br>{datasets_label}<br>" | |
| + "Params: %{customdata[0]:.2f}B<br>" | |
| + "CR: %{customdata[1]:.2f}%<br>" | |
| + "vs Fit: %{customdata[2]:+.2f}%<br>" | |
| + "<extra></extra>" | |
| ), | |
| ) | |
| ) | |
| # 添加拟合曲线 | |
| 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"<b>%{{text}}</b><br>{dataset}<br>" | |
| + "Params: %{customdata[0]:.2f}B<br>" | |
| + "CR: %{customdata[1]:.2f}%<br>" | |
| + "vs Fit: %{customdata[2]:+.2f}%<br>" | |
| + "<extra></extra>" | |
| ), | |
| 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"<b>%{{text}}</b> (Pareto)<br>{dataset}<br>" | |
| + "Params: %{customdata[0]:.2f}B<br>" | |
| + "CR: %{customdata[1]:.2f}%<br>" | |
| + "vs Fit: %{customdata[2]:+.2f}%<br>" | |
| + "<extra></extra>" | |
| ), | |
| 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) | |