""" Leaderboard storage backed by HuggingFace Dataset. Stores evaluation results persistently on HuggingFace Hub, allowing the leaderboard to survive Space restarts and be shared across deployments. """ import json import os from datetime import datetime from typing import Optional # Dataset configuration LEADERBOARD_DATASET_ID = "Sefaria/Rabbinic-Embedding-Leaderboard" LEADERBOARD_FILENAME = "leaderboard.json" # Cache for loaded leaderboard _leaderboard_cache: Optional[list[dict]] = None _cache_time: Optional[datetime] = None CACHE_TTL_SECONDS = 300 # Refresh cache every 5 minutes def _get_hf_token() -> Optional[str]: """Get HuggingFace token from environment.""" return os.environ.get("HF_TOKEN") or os.environ.get("HUGGINGFACE_TOKEN") def load_leaderboard(force_refresh: bool = False) -> list[dict]: """ Load leaderboard from HuggingFace Hub. Uses caching to avoid repeated downloads. Falls back to empty list if dataset doesn't exist or can't be accessed. Args: force_refresh: If True, bypass cache and reload from Hub Returns: List of leaderboard entries, sorted by MRR descending """ global _leaderboard_cache, _cache_time # Check cache if not force_refresh and _leaderboard_cache is not None: if _cache_time and (datetime.now() - _cache_time).seconds < CACHE_TTL_SECONDS: return _leaderboard_cache try: from huggingface_hub import hf_hub_download # Download the leaderboard file local_path = hf_hub_download( repo_id=LEADERBOARD_DATASET_ID, filename=LEADERBOARD_FILENAME, repo_type="dataset", token=_get_hf_token(), ) with open(local_path, "r", encoding="utf-8") as f: _leaderboard_cache = json.load(f) _cache_time = datetime.now() # Ensure sorted by MRR _leaderboard_cache.sort(key=lambda x: x.get("mrr", 0), reverse=True) return _leaderboard_cache except Exception as e: print(f"Could not load leaderboard from Hub: {e}") # Return cached data if available, otherwise empty list if _leaderboard_cache is not None: return _leaderboard_cache return [] def save_leaderboard(leaderboard: list[dict]) -> bool: """ Save leaderboard to HuggingFace Hub. Requires HF_TOKEN environment variable with write access to the dataset. Args: leaderboard: List of leaderboard entries Returns: True if saved successfully, False otherwise """ global _leaderboard_cache, _cache_time token = _get_hf_token() if not token: print("No HF_TOKEN found - leaderboard changes won't be persisted to Hub") # Still update local cache _leaderboard_cache = leaderboard _cache_time = datetime.now() return False try: from huggingface_hub import HfApi import tempfile api = HfApi(token=token) # Ensure dataset exists try: api.create_repo( repo_id=LEADERBOARD_DATASET_ID, repo_type="dataset", exist_ok=True, ) except Exception as e: print(f"Note: Could not create/verify repo: {e}") # Write to temp file and upload with tempfile.NamedTemporaryFile( mode="w", suffix=".json", delete=False, encoding="utf-8", ) as f: json.dump(leaderboard, f, indent=2, ensure_ascii=False) temp_path = f.name api.upload_file( path_or_fileobj=temp_path, path_in_repo=LEADERBOARD_FILENAME, repo_id=LEADERBOARD_DATASET_ID, repo_type="dataset", commit_message=f"Update leaderboard ({len(leaderboard)} entries)", ) # Update cache _leaderboard_cache = leaderboard _cache_time = datetime.now() print(f"Leaderboard saved to {LEADERBOARD_DATASET_ID}") return True except Exception as e: print(f"Failed to save leaderboard to Hub: {e}") # Still update local cache so current session sees changes _leaderboard_cache = leaderboard _cache_time = datetime.now() return False def add_result(entry: dict) -> bool: """ Add or update a result in the leaderboard. If a result for the same model_id exists, it will be replaced. Args: entry: Result dict with model_id, mrr, recall_at_1, etc. Returns: True if saved to Hub successfully, False otherwise """ # Load current leaderboard leaderboard = load_leaderboard(force_refresh=True) # Add timestamp if not present if "timestamp" not in entry: entry["timestamp"] = datetime.now().isoformat() # Remove existing entry for same model model_id = entry.get("model_id") leaderboard = [e for e in leaderboard if e.get("model_id") != model_id] # Add new entry leaderboard.append(entry) # Sort by MRR descending leaderboard.sort(key=lambda x: x.get("mrr", 0), reverse=True) # Save return save_leaderboard(leaderboard) def clear_cache(): """Clear the leaderboard cache to force a fresh load.""" global _leaderboard_cache, _cache_time _leaderboard_cache = None _cache_time = None # For local development/testing - create initial dataset def create_leaderboard_dataset(): """ Create the leaderboard dataset on HuggingFace Hub. Run this once to initialize the dataset: python -c "from leaderboard import create_leaderboard_dataset; create_leaderboard_dataset()" """ token = _get_hf_token() if not token: print("Error: HF_TOKEN environment variable required") return False try: from huggingface_hub import HfApi import tempfile api = HfApi(token=token) # Create the dataset repo api.create_repo( repo_id=LEADERBOARD_DATASET_ID, repo_type="dataset", exist_ok=True, ) print(f"Created/verified dataset: {LEADERBOARD_DATASET_ID}") # Create initial empty leaderboard with tempfile.NamedTemporaryFile( mode="w", suffix=".json", delete=False, encoding="utf-8", ) as f: json.dump([], f) temp_path = f.name api.upload_file( path_or_fileobj=temp_path, path_in_repo=LEADERBOARD_FILENAME, repo_id=LEADERBOARD_DATASET_ID, repo_type="dataset", commit_message="Initialize empty leaderboard", ) # Create README readme_content = """--- tags: - benchmark - leaderboard - embedding - hebrew - rabbinic license: mit --- # Rabbinic Embedding Benchmark Leaderboard This dataset stores the leaderboard results for the [Rabbinic Hebrew/Aramaic Embedding Benchmark](https://huggingface.co/spaces/Sefaria/Rabbinic-Embedding-Benchmark). ## Structure The `leaderboard.json` file contains an array of evaluation results: ```json [ { "model_id": "model-org/model-name", "model_name": "Model Display Name", "mrr": 0.85, "recall_at_1": 0.75, "recall_at_5": 0.90, "recall_at_10": 0.95, "bitext_accuracy": 0.92, "avg_true_pair_similarity": 0.85, "avg_random_pair_similarity": 0.35, "num_pairs": 3708, "timestamp": "2024-01-15T12:00:00", "categories": {"Talmud": 480, "Mishnah": 789, ...} } ] ``` ## Related - [Benchmark Dataset](https://huggingface.co/datasets/Sefaria/Rabbinic-Hebrew-English-Pairs) - [Benchmark Space](https://huggingface.co/spaces/Sefaria/Rabbinic-Embedding-Benchmark) """ with tempfile.NamedTemporaryFile( mode="w", suffix=".md", delete=False, encoding="utf-8", ) as f: f.write(readme_content) readme_path = f.name api.upload_file( path_or_fileobj=readme_path, path_in_repo="README.md", repo_id=LEADERBOARD_DATASET_ID, repo_type="dataset", commit_message="Add README", ) print(f"Leaderboard dataset created: https://huggingface.co/datasets/{LEADERBOARD_DATASET_ID}") return True except Exception as e: print(f"Failed to create leaderboard dataset: {e}") return False if __name__ == "__main__": # When run directly, create the leaderboard dataset create_leaderboard_dataset()