import logging import os os.makedirs("tmp", exist_ok=True) os.environ["TMP_DIR"] = "tmp" import subprocess import shutil import glob import gradio as gr import numpy as np from src.radial.radial import create_plot from apscheduler.schedulers.background import BackgroundScheduler from gradio_leaderboard import Leaderboard, SelectColumns from gradio_space_ci import enable_space_ci import json from io import BytesIO def handle_file_upload(file): file_path = file.name.split("/")[-1] if "/" in file.name else file.name logging.info("File uploaded: %s", file_path) with open(file.name, "r") as f: v = json.load(f) return v, file_path def submit_file(v, file_path, mn, profile: gr.OAuthProfile | None): """ Обрабатывает загрузку файлов с результатами пользовательских моделей Args: v: Загруженные данные результатов из JSON file_path: Путь к загруженному файлу mn: Имя модели, указанное пользователем profile: Профиль пользователя HuggingFace Returns: str: Сообщение об успехе или ошибке """ if profile is None: return "Hub Login Required: Войдите в HuggingFace, чтобы загрузить результаты" try: # Проверяем наличие обязательных полей required_fields = ["score", "math_score", "physics_score"] for field in required_fields: if field not in v: return f"Error: Отсутствует обязательное поле '{field}' в JSON файле" # Создаем новый объект для сохранения в базе данных new_file = { "model_name": profile.username + "/" + mn, "score": float(v["score"]), "math_score": float(v["math_score"]), "physics_score": float(v["physics_score"]), "total_tokens": int(v.get("total_tokens", 0)), "evaluation_time": float(v.get("evaluation_time", 0.0)), "system_prompt": v.get( "system_prompt", "Вы - полезный помощник по математике и физике. Ответьте на русском языке." ), } # Проверка значений на корректность if not ( 0 <= new_file["score"] <= 1 and 0 <= new_file["math_score"] <= 1 and 0 <= new_file["physics_score"] <= 1 ): return "Error: Все значения оценок должны быть в диапазоне от 0 до 1" # Создаем уникальное имя файла на основе username и имени модели safe_filename = profile.username + "_" + mn.replace("/", "_").replace(" ", "_") + ".json" # Создаем JSON в памяти и загружаем его в репозиторий buf = BytesIO() buf.write(json.dumps(new_file, ensure_ascii=False).encode("utf-8")) # Загружаем файл в репозиторий API.upload_file( path_or_fileobj=buf, path_in_repo="model_data/external/" + safe_filename, repo_id="Vikhrmodels/DeathMath-leaderboard-data", repo_type="dataset", ) # Устанавливаем флаг для обновления лидерборда os.environ[RESET_JUDGEMENT_ENV] = "1" # Логируем успешную загрузку logging.info(f"Successfully uploaded model results: {new_file['model_name']}") return f"Success! Результаты модели '{mn}' успешно отправлены в лидерборд DeathMath." except Exception as e: logging.error(f"Error submitting file: {e}") return f"Error: Произошла ошибка при отправке файла: {str(e)}" from src.display.about import INTRODUCTION_TEXT, TITLE, LLM_BENCHMARKS_TEXT from src.display.css_html_js import custom_css from src.display.utils import ( AutoEvalColumn, fields, ) from src.envs import API, H4_TOKEN, HF_HOME, REPO_ID, RESET_JUDGEMENT_ENV from src.leaderboard.build_leaderboard import build_leadearboard_df, download_openbench, download_dataset import huggingface_hub # huggingface_hub.login(token=H4_TOKEN) os.environ["GRADIO_ANALYTICS_ENABLED"] = "false" # Configure logging logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") # Start ephemeral Spaces on PRs (see config in README.md) enable_space_ci() # download_openbench() def restart_space(): API.restart_space(repo_id=REPO_ID) download_openbench() def update_plot(selected_models): return create_plot(selected_models) def build_demo(): """ Строит интерфейс лидерборда DeathMath """ # Загружаем данные для лидерборда download_openbench() # Создаем интерфейс с настройками темы demo = gr.Blocks( title="DeathMath Leaderboard", css=custom_css, theme=gr.themes.Default( primary_hue="indigo", secondary_hue="purple", ), ) # Получаем данные для лидерборда leaderboard_df = build_leadearboard_df() # Строим интерфейс with demo: # Заголовок и введение gr.HTML(TITLE) gr.Markdown(INTRODUCTION_TEXT, elem_classes="markdown-text") # Основные вкладки with gr.Tabs(elem_classes="tab-buttons"): # Вкладка лидерборда with gr.TabItem("🏅 Лидерборд", elem_id="llm-benchmark-tab-table", id=0): gr.Markdown("### Таблица результатов моделей DeathMath") # Таблица с результатами Leaderboard( value=leaderboard_df, datatype=[c.type for c in fields(AutoEvalColumn)], select_columns=SelectColumns( default_selection=[c.name for c in fields(AutoEvalColumn) if c.displayed_by_default], cant_deselect=[c.name for c in fields(AutoEvalColumn) if c.never_hidden or c.dummy], label="Выберите колонки для отображения:", ), search_columns=[ AutoEvalColumn.model.name, ], ) # Вкладка для отправки результатов with gr.TabItem("🚀 Отправить результаты", elem_id="submit-tab", id=1): with gr.Row(): gr.Markdown(LLM_BENCHMARKS_TEXT, elem_classes="markdown-text") with gr.Column(): model_name_textbox = gr.Textbox(label="Название модели") file_output = gr.File(label="Перетащите JSON файл с результатами сюда", type="filepath") uploaded_file = gr.State() file_path = gr.State() with gr.Row(): with gr.Column(): out = gr.Textbox("Здесь будет показан статус отправки") with gr.Column(): login_button = gr.LoginButton(elem_id="oauth-button") submit_button = gr.Button("Отправить результаты", elem_id="submit_button", variant="primary") # Обработчики событий file_output.upload(handle_file_upload, file_output, [uploaded_file, file_path]) submit_button.click(submit_file, [uploaded_file, file_path, model_name_textbox], [out]) # Вкладка с аналитикой with gr.TabItem("📊 Аналитика", elem_id="analytics-tab", id=2): with gr.Column(): if len(leaderboard_df) > 0: model_dropdown = gr.Dropdown( choices=leaderboard_df["model"].tolist(), label="Модели", value=leaderboard_df["model"].tolist()[:5] if len(leaderboard_df) >= 5 else leaderboard_df["model"].tolist(), multiselect=True, info="Выберите модели для сравнения", ) else: model_dropdown = gr.Dropdown( choices=["example/model-1", "example/model-2"], label="Модели", value=["example/model-1", "example/model-2"], multiselect=True, info="Выберите модели для сравнения", ) # Вкладки для разных типов визуализации with gr.Tabs(): with gr.TabItem("Столбчатая диаграмма"): bar_plot = gr.Plot(create_plot(model_dropdown.value)) with gr.TabItem("Радарная диаграмма"): from src.radial.radial import create_radar_plot radar_plot = gr.Plot(create_radar_plot(model_dropdown.value)) # Обновление графиков при изменении выбранных моделей model_dropdown.change(fn=create_plot, inputs=[model_dropdown], outputs=[bar_plot]) model_dropdown.change(fn=create_radar_plot, inputs=[model_dropdown], outputs=[radar_plot]) return demo # print(os.system('cd src/gen && ../../.venv/bin/python gen_judgment.py')) # print(os.system('cd src/gen/ && python show_result.py --output')) def update_board(): need_reset = os.environ.get(RESET_JUDGEMENT_ENV) logging.info("Updating the leaderboard: %s", need_reset) if need_reset != "1": return os.environ[RESET_JUDGEMENT_ENV] = "0" try: # Загружаем актуальные данные из репозитория download_dataset("Vikhrmodels/DeathMath-leaderboard-data", "m_data") logging.info("Successfully downloaded model evaluation data") # Собираем данные из всех файлов моделей data_list = [] seen_models = set() # Для отслеживания дубликатов for file in glob.glob("./m_data/model_data/external/*.json"): try: with open(file) as f: data = json.load(f) # Проверяем наличие необходимых полей model_name = data.get("model_name", "") if not model_name: logging.error(f"Failed to parse {file}: 'model_name' not found") continue # Предотвращаем дублирование моделей model_base_name = model_name.split("/")[-1].split("_v")[0] if model_base_name in seen_models: logging.info(f"Skipping duplicate model: {model_name}") continue seen_models.add(model_base_name) # Добавляем модель в список data_list.append( { "model_name": model_name, "score": float(data.get("score", 0.0)), "math_score": float(data.get("math_score", 0.0)), "physics_score": float(data.get("physics_score", 0.0)), "total_tokens": int(data.get("total_tokens", 0)), "evaluation_time": float(data.get("evaluation_time", 0.0)), "system_prompt": data.get( "system_prompt", "Вы - полезный помощник по математике и физике. Ответьте на русском языке.", ), } ) except Exception as e: logging.error(f"Failed to process file {file}: {e}") continue # Если есть данные, сортируем их по общему баллу и сохраняем if data_list: # Сортируем по общему баллу data_list.sort(key=lambda x: x["score"], reverse=True) # Сохраняем обновленный лидерборд with open("genned.json", "w", encoding="utf-8") as f: json.dump(data_list, f, ensure_ascii=False, indent=2) # Загружаем обновленный лидерборд в репозиторий API.upload_file( path_or_fileobj="genned.json", path_in_repo="leaderboard.json", repo_id="Vikhrmodels/DeathMath-leaderboard-metainfo", repo_type="dataset", ) logging.info(f"Updated leaderboard with {len(data_list)} models") # Генерируем README с таблицей лидерборда update_readme(data_list) except Exception as e: logging.error(f"Error updating leaderboard: {e}") def update_readme(data_list): """ Генерирует README.md с таблицей лидерборда """ try: import pandas as pd from datetime import datetime # Создаем DataFrame для удобного форматирования таблицы df = pd.DataFrame(data_list) # Форматируем числовые колонки for col in ["score", "math_score", "physics_score"]: if col in df.columns: df[col] = df[col].apply(lambda x: f"{x:.3f}") if "total_tokens" in df.columns: df["total_tokens"] = df["total_tokens"].apply(lambda x: f"{int(x):,}") if "evaluation_time" in df.columns: df["evaluation_time"] = df["evaluation_time"].apply(lambda x: f"{x:.1f}s") # Создаем содержимое README current_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S") readme_content = f"""--- title: DeathMath Leaderboard emoji: 🔢 colorFrom: indigo colorTo: purple sdk: gradio sdk_version: "4.20.0" app_file: app.py pinned: false hf_oauth: true --- # DeathMath Leaderboard DeathMath - это бенчмарк для оценки способности моделей решать сложные математические и физические задачи на русском языке. ## Текущий лидерборд Последнее обновление: {current_date} | Модель | Общий балл | Математика | Физика | Токены | Время оценки | |--------|------------|------------|---------|---------|--------------| """ # Добавляем строки таблицы for _, row in df.iterrows(): model_name = row.get("model_name", "") readme_content += f"| {model_name} | {row.get('score', 'N/A')} | {row.get('math_score', 'N/A')} | {row.get('physics_score', 'N/A')} | {row.get('total_tokens', 'N/A')} | {row.get('evaluation_time', 'N/A')} |\n" readme_content += """ ## Как принять участие в бенчмарке Для участия в бенчмарке DeathMath: 1. Клонируйте репозиторий и запустите тесты вашей модели 2. Загрузите результаты через [HuggingFace Space](https://huggingface.co/spaces/Vikhrmodels/DeathMath-leaderboard) 3. Дождитесь проверки и добавления результатов в лидерборд ## Формат результатов Результаты должны быть в формате JSON со следующей структурой: ```json { "score": 0.586, "math_score": 0.8, "physics_score": 0.373, "total_tokens": 1394299, "evaluation_time": 4533.2, "system_prompt": "Вы - полезный помощник по математике и физике. Ответьте на русском языке." } ``` ## Лицензия Бенчмарк распространяется под лицензией Apache 2.0 """ # Сохраняем README with open("README.md", "w", encoding="utf-8") as f: f.write(readme_content) # Загружаем README в репозиторий API.upload_file( path_or_fileobj="README.md", path_in_repo="README.md", repo_id="Vikhrmodels/DeathMath-leaderboard-metainfo", repo_type="dataset", ) logging.info("Updated README with leaderboard table") except Exception as e: logging.error(f"Failed to update README: {e}") def update_board_(): """ Инициализирующая версия функции обновления лидерборда. Удаляет все существующие данные перед первой загрузкой """ try: # Очищаем каталоги перед загрузкой данных if os.path.exists("m_data"): # Удаляем все JSON-файлы моделей из старой версии Small Shlepa for old_file in glob.glob("./m_data/model_data/external/*.json"): try: os.remove(old_file) logging.info(f"Removed old file: {old_file}") except Exception as e: logging.error(f"Error removing old file {old_file}: {e}") # Вызываем основную функцию обновления update_board() except Exception as e: logging.error(f"Error in update_board_(): {e}") if __name__ == "__main__": os.environ[RESET_JUDGEMENT_ENV] = "1" scheduler = BackgroundScheduler() update_board_() scheduler.add_job(update_board, "interval", minutes=10) scheduler.start() demo_app = build_demo() demo_app.launch()