#!/usr/bin/env python3 """Build an HTML Open LLM Distribution Leaderboard from Hugging Face Hub metadata.""" from __future__ import annotations import argparse import csv import html import json import os import re import shutil import tempfile import time import urllib.error import urllib.parse import urllib.request from collections import Counter, defaultdict from datetime import datetime, timezone from pathlib import Path from typing import Any HF_MODELS_API = "https://huggingface.co/api/models" HF_ROUTER_MODELS_API = "https://router.huggingface.co/v1/models" USER_AGENT = "leaderboard/0.1" DEFAULT_MIN_DOWNLOADS = 100_000 DEFAULT_TOP_N = 250 DEFAULT_MAX_AGE_DAYS = 365 DEFAULT_BUCKET_DIR = Path(os.environ.get("OPEN_MODEL_BUCKET_DIR", "/data")) DEFAULT_LOCAL_DATA_DIR = Path(os.environ.get("OPEN_MODEL_LOCAL_DATA_DIR", "data")) DEFAULT_CREATORS = [ "CohereLabs", "EssentialAI", "MiniMaxAI", "Qwen", "Sao10K", "aisingapore", "allenai", "alpindale", "baidu", "deepcogito", "deepseek-ai", "google", "inclusionAI", "meta-llama", "moonshotai", "nvidia", "openai", "pearl-ai", "stepfun-ai", "swiss-ai", "utter-project", "zai-org", ] EXPAND_FIELDS = [ "author", "cardData", "createdAt", "downloads", "downloadsAllTime", "evalResults", "gated", "gguf", "inferenceProviderMapping", "lastModified", "library_name", "likes", "model-index", "pipeline_tag", "private", "safetensors", "sha", "siblings", "tags", "transformersInfo", "trendingScore", ] MODEL_CSV_FIELDS = [ "id", "author", "downloads", "downloads_all_time", "likes", "trending_score", "parameter_count", "parameter_count_b", "capacity_class", "pipeline_tag", "library_name", "license", "base_model_relation", "gated", "private", "created_at", "last_modified", "sha", "base_models", "has_safetensors", "has_gguf", "has_tokenizer", "has_chat_template", "siblings_count", "safetensors_files", "gguf_files", "bin_files", "eval_results_count", "inference_provider_count", "inference_providers", "live_inference_providers", "supports_tools_any", "supports_structured_output_any", "max_context_length", "router_served", "router_provider_count", "router_providers", "router_supports_tools_any", "router_supports_structured_output_any", "router_max_context_length", "tags", "group_key", "candidate_class", "candidate", "candidate_reason", ] GROUP_CSV_FIELDS = [ "rank", "group_key", "representative_id", "author", "downloads_per_month", "group_downloads_per_month", "representative_downloads_per_month", "downloads", "group_downloads", "representative_downloads", "downloads_all_time", "group_downloads_all_time", "representative_downloads_all_time", "likes_per_month", "group_likes_per_month", "representative_likes_per_month", "likes", "group_likes", "representative_likes", "parameter_count_b", "capacity_class", "candidate_class", "pipeline_tag", "license", "router_served", "supports_tools", "supports_structured_output", "artifact_count", "artifacts", "artifact_signals", "created_at", "last_modified", ] SECONDS_PER_DAY = 86_400 DAYS_PER_MONTH = 365.2425 / 12 MIN_DOWNLOAD_RATE_AGE_DAYS = DAYS_PER_MONTH MIN_LIKE_RATE_AGE_DAYS = DAYS_PER_MONTH ARTIFACT_REVISION_SUFFIX_RE = re.compile(r"-(?:gguf|awq|gptq|fp8|fp4|mxfp8|nvfp4)-v[0-9]+$") ARTIFACT_SUFFIX_RE = re.compile( r"-(?:" r"gguf|awq|gptq|fp8|fp4|mxfp8|nvfp4|bf16|fp16|int2|int3|int4|int8|" r"q[2-8](?:-(?:0|1|k|m|s|l))*|" r"4bit|8bit|bnb|bitsandbytes|mlx|exl2|onnx|openvino|tensorrt|" r"modelopt|qat|quantized|quant|compressed|safetensors|hf" r")$" ) QAT_SUFFIX_RE = re.compile(r"-qat-w[0-9]+a[0-9]+-ct$") LLM_PIPELINES = { "text-generation", "image-text-to-text", "video-text-to-text", "audio-text-to-text", "any-to-any", "text-to-audio", } EXCLUDED_PIPELINES = { "feature-extraction", "sentence-similarity", "text-ranking", "text-classification", "token-classification", "fill-mask", "automatic-speech-recognition", "text-to-speech", "voice-activity-detection", "image-classification", "image-segmentation", "object-detection", "zero-shot-image-classification", "zero-shot-object-detection", "image-text-to-image", "text-to-image", "image-to-image", "image-to-video", "text-to-video", "unconditional-image-generation", "image-to-3d", "text-to-3d", "robotics", "time-series-forecasting", "graph-ml", } TEXT_MODEL_TERMS = ( "instruct", "chat", "assistant", "reasoning", "thinking", "coder", "code", "llm", "vlm", "omni", "gpt-oss", "kimi", "glm", "llama", "qwen", "gemma", "deepseek", "nemotron", "olmo", "apertus", "eurollm", ) CODING_MODEL_NAME_RE = re.compile( r"(^|[-_/])(?:coder|coding|code)(?:$|[-_/])|" r"(^|[-_/])(?:codellama|codeqwen|opencoder|starcoder)[a-z0-9.]*($|[-_/])" ) PACKAGING_RELATION_TERMS = { "adapter", "compressed", "converted", "gguf", "quantized", } PACKAGING_SIGNAL_TERMS = { "2bit", "3bit", "4bit", "8bit", "awq", "bf16", "bnb", "compressed", "exl2", "fp4", "fp8", "gguf", "gptq", "int2", "int3", "int4", "int8", "llama.cpp", "mlx", "mxfp8", "nvfp4", "qat", "q2", "q3", "q4", "q5", "q6", "q8", "quant", "quantized", } DERIVATIVE_STRONG_TOKENS = { "adapter", "awq", "draft", "fp4", "fp8", "gguf", "gptq", "lora", "mlp", "mlx", "mxfp8", "nvfp4", "qat", "quant", "quantized", "speculative", "unquantized", } DERIVATIVE_AMBIGUOUS_TOKENS = { "assistant", } DERIVATIVE_COMPONENT_PREFIX_TOKENS = { "adapter", "assistant", "draft", "lora", "mlp", "speculative", } DERIVATIVE_SAFE_SUFFIX_TOKENS = DERIVATIVE_STRONG_TOKENS | DERIVATIVE_AMBIGUOUS_TOKENS | { "block", "dynamic", "static", "turbo", } PEER_VARIANT_TOKENS = { "base", "chat", "coder", "coding", "flash", "instruct", "math", "pro", "reasoning", "thinking", "vision", "vl", } MODEL_FAMILY_TOKENS = { "apertus", "deepseek", "ernie", "gemma", "glm", "gpt", "kimi", "llama", "minimax", "mistral", "mixtral", "nemotron", "olmo", "phi", "qwen", "qwen2", "qwen3", "step", } GLOBAL_ARTIFACT_SEARCH_QUERIES = ( "gguf", "awq", "gptq", "fp8", "fp4", "nvfp4", "quantized", "mlx", "llama.cpp", ) RELATED_ARTIFACT_SUFFIXES = ( "gguf", "awq", "gptq", "fp8", "fp4", "nvfp4", "mlx", ) RELATED_VARIANT_TOKENS = { "base", "chat", "flash", "instruct", "it", "pro", } DEFAULT_DISCOVERY_MAX_RESULTS_PER_QUERY = 500 DEFAULT_RELATED_DISCOVERY_SEED_LIMIT = 12 EXCLUDED_TERMS = ( "embedding", "embed", "reranker", "rerank", "whisper", "wav2vec", "speaker", "stable-diffusion", "diffusers", "flux", "sdxl", "clip", "siglip", "depth-anything", "ocr", "optical-character-recognition", "document-parse", "document-parser", "document-understanding", "pdf-parser", "receipt", ) EXCLUDED_TOKEN_TERMS = { "asr", "encoder", } EXCLUDED_AUDIO_TOKEN_TERMS = { "speech", } SPECIAL_ARCHITECTURE_TERMS = ( "dllm", "diffusion-language", "diffusion_language", "diffusion language", "text-diffusion", "text_diffusion", "language-diffusion", "language_diffusion", ) def now_utc() -> datetime: return datetime.now(timezone.utc) def stamp(dt: datetime | None = None) -> str: return (dt or now_utc()).strftime("%Y%m%dT%H%M%SZ") def iso(dt: datetime | None = None) -> str: return (dt or now_utc()).isoformat(timespec="seconds").replace("+00:00", "Z") def env_creators() -> list[str]: raw = os.environ.get("OPEN_MODEL_CREATORS", "") extra = [item.strip() for item in raw.replace("\n", ",").split(",") if item.strip()] return unique_preserve_order(DEFAULT_CREATORS + extra) def unique_preserve_order(values: list[str]) -> list[str]: seen: set[str] = set() out: list[str] = [] for value in values: if value in seen: continue seen.add(value) out.append(value) return out def str_list(value: Any) -> list[str]: if value is None: return [] if isinstance(value, str): return [value] if isinstance(value, list): return [item for item in value if isinstance(item, str)] return [] def request_json(url: str, token: str | None, retries: int = 4) -> tuple[Any, dict[str, str]]: headers = {"User-Agent": USER_AGENT} if token: headers["Authorization"] = f"Bearer {token}" request = urllib.request.Request(url, headers=headers) for attempt in range(retries + 1): try: with urllib.request.urlopen(request, timeout=90) as response: payload = json.loads(response.read().decode("utf-8")) return payload, dict(response.headers.items()) except urllib.error.HTTPError as err: if err.code in {429, 500, 502, 503, 504} and attempt < retries: time.sleep(min(2**attempt, 10)) continue body = err.read().decode("utf-8", errors="replace") raise RuntimeError(f"HTTP {err.code} for {url}: {body[:500]}") from err except urllib.error.URLError as err: if attempt < retries: time.sleep(min(2**attempt, 10)) continue raise RuntimeError(f"Request failed for {url}: {err}") from err raise RuntimeError(f"Request failed after retries: {url}") def parse_next_link(link_header: str | None) -> str | None: if not link_header: return None for part in link_header.split(","): match = re.search(r'<([^>]+)>;\s*rel="next"', part) if match: return match.group(1) return None def fetch_router_models(token: str | None) -> list[dict[str, Any]]: payload, _ = request_json(HF_ROUTER_MODELS_API, token) records = payload.get("data", []) if isinstance(payload, dict) else [] return [item for item in records if isinstance(item, dict) and item.get("id")] def router_creators(router_records: list[dict[str, Any]]) -> list[str]: creators = [] for record in router_records: owner = record.get("owned_by") if isinstance(owner, str) and owner: creators.append(owner) continue model_id = record.get("id") if isinstance(model_id, str) and "/" in model_id: creators.append(model_id.split("/", 1)[0]) return unique_preserve_order(creators) def list_creator_models( creator: str, *, page_size: int, sort: str, direction: int, token: str | None, limit: int | None, sleep_seconds: float, ) -> list[dict[str, Any]]: params: list[tuple[str, str | int]] = [ ("author", creator), ("limit", page_size), ("sort", sort), ("direction", direction), ] params.extend(("expand", field) for field in EXPAND_FIELDS) url = f"{HF_MODELS_API}?{urllib.parse.urlencode(params, doseq=True)}" models: list[dict[str, Any]] = [] while url: payload, headers = request_json(url, token) if not isinstance(payload, list): raise RuntimeError(f"Expected list payload for {creator}, got {type(payload).__name__}") for item in payload: if isinstance(item, dict): models.append(item) if limit is not None and len(models) >= limit: return models url = parse_next_link(headers.get("Link") or headers.get("link")) if url and sleep_seconds > 0: time.sleep(sleep_seconds) return models def list_search_models( search: str, *, page_size: int, min_downloads: int, max_results: int, token: str | None, sleep_seconds: float, ) -> list[dict[str, Any]]: params: list[tuple[str, str | int]] = [ ("search", search), ("limit", page_size), ("sort", "downloads"), ("direction", -1), ] params.extend(("expand", field) for field in EXPAND_FIELDS) url = f"{HF_MODELS_API}?{urllib.parse.urlencode(params, doseq=True)}" models: list[dict[str, Any]] = [] while url and len(models) < max_results: payload, headers = request_json(url, token) if not isinstance(payload, list): raise RuntimeError(f"Expected list payload for search {search}, got {type(payload).__name__}") stop = False for item in payload: if not isinstance(item, dict): continue downloads = int_metric(item.get("downloads")) if downloads <= min_downloads: stop = True break models.append(item) if len(models) >= max_results: break if stop or len(models) >= max_results: break url = parse_next_link(headers.get("Link") or headers.get("link")) if url and sleep_seconds > 0: time.sleep(sleep_seconds) return models def shortened_variant_slug(slug: str) -> str: tokens = slug_tokens(slug) while len(tokens) > 2 and tokens[-1] in RELATED_VARIANT_TOKENS: tokens.pop() return "-".join(tokens) def discovery_queries(seed_rows: list[dict[str, Any]], seed_limit: int = DEFAULT_RELATED_DISCOVERY_SEED_LIMIT) -> list[str]: queries = list(GLOBAL_ARTIFACT_SEARCH_QUERIES) ordered = sorted( seed_rows, key=lambda row: (int_metric(row.get("downloads")), int_metric(row.get("downloads_all_time"))), reverse=True, ) for row in ordered[:seed_limit]: slug = slug_dedupe_key(str(row.get("id") or "")) if not has_model_shape(slug): continue candidate_slugs = unique_preserve_order([slug, shortened_variant_slug(slug)]) for candidate_slug in candidate_slugs: if not candidate_slug or not has_model_shape(candidate_slug): continue for suffix in RELATED_ARTIFACT_SUFFIXES: queries.append(f"{candidate_slug}-{suffix}") return unique_preserve_order(queries) def discover_additional_models( *, seed_rows: list[dict[str, Any]], existing_ids: set[str], min_downloads: int, page_size: int, token: str | None, sleep_seconds: float, max_results_per_query: int, seed_limit: int, ) -> tuple[list[dict[str, Any]], list[dict[str, str]], dict[str, Any]]: discovered: dict[str, dict[str, Any]] = {} errors: list[dict[str, str]] = [] queries = discovery_queries(seed_rows, seed_limit=seed_limit) query_counts: dict[str, int] = {} for query in queries: try: rows = list_search_models( query, page_size=page_size, min_downloads=min_downloads, max_results=max_results_per_query, token=token, sleep_seconds=sleep_seconds, ) except Exception as err: # noqa: BLE001 errors.append({"discovery_query": query, "error": str(err)}) continue query_counts[query] = len(rows) for row in rows: model_id = row.get("id") or row.get("modelId") if not isinstance(model_id, str) or not model_id or model_id in existing_ids: continue discovered[model_id] = row summary = { "enabled": True, "queries": len(queries), "query_counts": query_counts, "records": len(discovered), } return list(discovered.values()), errors, summary def find_license(record: dict[str, Any]) -> str: card = record.get("cardData") if isinstance(card, dict) and isinstance(card.get("license"), str): return card["license"] for tag in str_list(record.get("tags")): if tag.startswith("license:"): return tag.split(":", 1)[1] return "" def find_base_models(record: dict[str, Any]) -> list[str]: values: list[str] = [] card = record.get("cardData") if isinstance(card, dict): values.extend(str_list(card.get("base_model"))) for tag in str_list(record.get("tags")): if tag.startswith("base_model:"): values.append(tag.split(":", 1)[1]) return unique_preserve_order(values) def find_base_model_relation(record: dict[str, Any]) -> str: card = record.get("cardData") if isinstance(card, dict) and isinstance(card.get("base_model_relation"), str): return str(card["base_model_relation"]) return "" def file_counts(record: dict[str, Any]) -> Counter[str]: counts: Counter[str] = Counter() for sibling in record.get("siblings") or []: if not isinstance(sibling, dict): continue name = sibling.get("rfilename") if not isinstance(name, str): continue lower = name.lower() if lower.endswith(".safetensors"): counts["safetensors"] += 1 elif lower.endswith(".gguf"): counts["gguf"] += 1 elif lower.endswith(".bin"): counts["bin"] += 1 if lower.endswith("tokenizer.json") or lower.endswith("tokenizer_config.json"): counts["tokenizer"] += 1 if lower.endswith("chat_template.jinja"): counts["chat_template"] += 1 return counts def extract_parameter_count(record: dict[str, Any]) -> int | None: safetensors = record.get("safetensors") if isinstance(safetensors, dict): total = safetensors.get("total") if isinstance(total, int): return total params = safetensors.get("parameters") if isinstance(params, dict): values = [value for value in params.values() if isinstance(value, int)] if values: return max(values) gguf = record.get("gguf") if isinstance(gguf, dict): for key in ("total", "parameters", "parameter_count", "parameterCount"): value = gguf.get(key) if isinstance(value, int): return value return None def capacity_class(parameter_count: int | None) -> str: if parameter_count is None: return "unknown" billion = parameter_count / 1_000_000_000 if billion <= 1: return "tiny" if billion <= 20: return "small" if billion <= 50: return "medium" return "large" def provider_summary(mapping: Any) -> dict[str, Any]: if not isinstance(mapping, list): mapping = [] providers: list[str] = [] live: list[str] = [] supports_tools = False supports_structured = False context_lengths: list[int] = [] for item in mapping: if not isinstance(item, dict): continue provider = item.get("provider") if isinstance(provider, str): providers.append(provider) if item.get("status") == "live": live.append(provider) features = item.get("features") if isinstance(features, dict): supports_tools = supports_tools or features.get("toolCalling") is True supports_structured = supports_structured or features.get("structuredOutput") is True details = item.get("providerDetails") if isinstance(details, dict): context = details.get("context_length") if isinstance(context, int): context_lengths.append(context) return { "providers": sorted(set(providers)), "live": sorted(set(live)), "supports_tools_any": supports_tools, "supports_structured_output_any": supports_structured, "max_context_length": max(context_lengths) if context_lengths else "", } def router_summary(router_entry: dict[str, Any] | None) -> dict[str, Any]: if not router_entry: return { "router_served": False, "router_provider_count": 0, "router_providers": [], "router_supports_tools_any": False, "router_supports_structured_output_any": False, "router_max_context_length": "", } providers = router_entry.get("providers") if not isinstance(providers, list): providers = [] provider_names: list[str] = [] context_lengths: list[int] = [] supports_tools = False supports_structured = False for provider in providers: if not isinstance(provider, dict): continue name = provider.get("provider") if isinstance(name, str): provider_names.append(name) supports_tools = supports_tools or provider.get("supports_tools") is True supports_structured = supports_structured or provider.get("supports_structured_output") is True context = provider.get("context_length") if isinstance(context, int): context_lengths.append(context) return { "router_served": True, "router_provider_count": len(provider_names), "router_providers": sorted(set(provider_names)), "router_supports_tools_any": supports_tools, "router_supports_structured_output_any": supports_structured, "router_max_context_length": max(context_lengths) if context_lengths else "", } def normalize_record(record: dict[str, Any], router_entry: dict[str, Any] | None) -> dict[str, Any]: counts = file_counts(record) parameter_count = extract_parameter_count(record) provider = provider_summary(record.get("inferenceProviderMapping")) router = router_summary(router_entry) tags = str_list(record.get("tags")) model_id = record.get("id") or record.get("modelId") or "" normalized = { "id": model_id, "author": record.get("author") or (str(model_id).split("/", 1)[0] if "/" in str(model_id) else ""), "downloads": record.get("downloads") if isinstance(record.get("downloads"), int) else 0, "downloads_all_time": record.get("downloadsAllTime") if isinstance(record.get("downloadsAllTime"), int) else "", "likes": record.get("likes") if isinstance(record.get("likes"), int) else 0, "trending_score": record.get("trendingScore") if isinstance(record.get("trendingScore"), int) else 0, "parameter_count": parameter_count or "", "parameter_count_b": round(parameter_count / 1_000_000_000, 3) if parameter_count else "", "capacity_class": capacity_class(parameter_count), "pipeline_tag": record.get("pipeline_tag") or "", "library_name": record.get("library_name") or "", "license": find_license(record), "base_model_relation": find_base_model_relation(record), "gated": record.get("gated"), "private": record.get("private"), "created_at": record.get("createdAt") or "", "last_modified": record.get("lastModified") or "", "sha": record.get("sha") or "", "base_models": ",".join(find_base_models(record)), "has_safetensors": bool(record.get("safetensors")) or counts["safetensors"] > 0, "has_gguf": bool(record.get("gguf")) or counts["gguf"] > 0, "has_tokenizer": counts["tokenizer"] > 0, "has_chat_template": counts["chat_template"] > 0, "siblings_count": len(record.get("siblings") or []), "safetensors_files": counts["safetensors"], "gguf_files": counts["gguf"], "bin_files": counts["bin"], "eval_results_count": len(record.get("evalResults") or []), "inference_provider_count": len(provider["providers"]), "inference_providers": ",".join(provider["providers"]), "live_inference_providers": ",".join(provider["live"]), "supports_tools_any": provider["supports_tools_any"], "supports_structured_output_any": provider["supports_structured_output_any"], "max_context_length": provider["max_context_length"], "router_served": router["router_served"], "router_provider_count": router["router_provider_count"], "router_providers": ",".join(router["router_providers"]), "router_supports_tools_any": router["router_supports_tools_any"], "router_supports_structured_output_any": router["router_supports_structured_output_any"], "router_max_context_length": router["router_max_context_length"], "tags": ",".join(tags), } normalized["candidate_class"] = candidate_class(normalized) normalized["candidate"], normalized["candidate_reason"] = candidate_decision(normalized) return normalized def normalized_repo_slug(model_id: str) -> str: slug = model_id.rsplit("/", 1)[-1].lower() slug = re.sub(r"[^a-z0-9]+", "-", slug) return slug.strip("-") def slug_dedupe_key(model_id: str) -> str: slug = normalized_repo_slug(model_id) previous = None while slug and slug != previous: previous = slug slug = QAT_SUFFIX_RE.sub("", slug) slug = ARTIFACT_REVISION_SUFFIX_RE.sub("", slug) slug = ARTIFACT_SUFFIX_RE.sub("", slug) return slug or normalized_repo_slug(model_id) def is_artifact_slug(model_id: str) -> bool: return normalized_repo_slug(model_id) != slug_dedupe_key(model_id) def quantized_base_refs(row: dict[str, Any]) -> list[str]: refs: list[str] = [] for source in [str(row.get("base_models") or ""), str(row.get("tags") or "")]: refs.extend(re.findall(r"(?:^|,)(?:base_model:)?quantized:([^,]+)", source)) return unique_preserve_order(refs) def base_model_refs(row: dict[str, Any]) -> list[tuple[str, str]]: refs: list[tuple[str, str]] = [] for value in str(row.get("base_models") or "").split(","): value = value.strip() if not value: continue relation = "" ref = value if ":" in value: prefix, rest = value.split(":", 1) if prefix in PACKAGING_RELATION_TERMS or prefix in {"finetune", "merge"}: relation = prefix ref = rest if ref: refs.append((relation, ref)) return refs def packaging_signals(row: dict[str, Any]) -> list[str]: text = f"{row.get('id', '')} {row.get('tags', '')} {row.get('library_name', '')}".lower() signals: list[str] = [] for term in sorted(PACKAGING_SIGNAL_TERMS, key=len, reverse=True): if term in text: signals.append(term) if row.get("has_gguf") is True: signals.append("gguf") return unique_preserve_order(signals) def packaging_base_refs(row: dict[str, Any]) -> list[str]: refs = list(quantized_base_refs(row)) relation = str(row.get("base_model_relation") or "").lower() if relation and relation not in PACKAGING_RELATION_TERMS: return unique_preserve_order(refs) for ref_relation, ref in base_model_refs(row): if ref_relation: if ref_relation in PACKAGING_RELATION_TERMS: refs.append(ref) continue if relation in PACKAGING_RELATION_TERMS: refs.append(ref) return unique_preserve_order(refs) def artifact_signals(row: dict[str, Any]) -> list[str]: text = f"{row.get('id', '')} {row.get('tags', '')} {row.get('library_name', '')}".lower() signals: list[str] = [] for name in ("gguf", "awq", "gptq", "fp8", "fp4", "nvfp4", "mxfp8", "bf16", "mlx", "qat"): if name in text: signals.append(name) if row.get("has_safetensors") is True: signals.append("safetensors") return unique_preserve_order(signals) def has_special_language_architecture(row: dict[str, Any]) -> bool: text = f"{row.get('id', '')} {row.get('pipeline_tag', '')} {row.get('tags', '')}".lower() if any(term in text for term in SPECIAL_ARCHITECTURE_TERMS): return True if "diffusion" in text and ( row.get("pipeline_tag") in LLM_PIPELINES or "text-generation" in text or "image-text-to-text" in text or "conversational" in text or "llm" in text ): return True return False def text_tokens(text: str) -> set[str]: return {token for token in re.split(r"[^a-z0-9]+", text.lower()) if token} def has_excluded_term(row: dict[str, Any]) -> bool: text = f"{row.get('id', '')} {row.get('pipeline_tag', '')} {row.get('tags', '')}".lower() if any(term in text for term in EXCLUDED_TERMS): return True tokens = text_tokens(text) if tokens & EXCLUDED_TOKEN_TERMS: return True pipeline = str(row.get("pipeline_tag") or "") if pipeline not in LLM_PIPELINES and tokens & EXCLUDED_AUDIO_TOKEN_TERMS: return True return False def is_coding_model_name(model_id: str) -> bool: return bool(CODING_MODEL_NAME_RE.search(model_id.lower())) def candidate_class(row: dict[str, Any]) -> str: model_id = str(row.get("id") or "").lower() text = f"{row.get('id', '')} {row.get('pipeline_tag', '')} {row.get('tags', '')}".lower() if has_special_language_architecture(row): return "special" if ( row.get("pipeline_tag") in {"image-text-to-text", "video-text-to-text", "audio-text-to-text", "any-to-any"} or "vision-language" in text or "multimodal" in text or "omni" in text or bool(re.search(r"(^|[-_/])vl($|[-_/])", model_id)) ): return "multimodal" if is_coding_model_name(model_id): return "coding" return "text" def has_text_model_signal(model_id: str) -> bool: if is_coding_model_name(model_id): return True boundary_terms = {"assistant", "chat", "code", "coder", "coding", "instruct", "reasoning", "thinking"} for term in TEXT_MODEL_TERMS: if term in boundary_terms: if re.search(rf"(^|[-_/]){re.escape(term)}($|[-_/])", model_id): return True continue if term in model_id: return True return False def candidate_decision(row: dict[str, Any]) -> tuple[bool, str]: model_id = str(row.get("id") or "").lower() text = f"{row.get('id', '')} {row.get('pipeline_tag', '')} {row.get('tags', '')}".lower() pipeline = str(row.get("pipeline_tag") or "") if row.get("private") is True: return False, "private" if pipeline in EXCLUDED_PIPELINES: return False, f"excluded pipeline: {pipeline}" if has_excluded_term(row): return False, "excluded narrow/non-LLM term" if pipeline in LLM_PIPELINES: return True, f"LLM pipeline: {pipeline}" if row.get("has_chat_template") is True: return True, "chat template" if row.get("supports_tools_any") is True or row.get("router_supports_tools_any") is True: return True, "tool support" if row.get("supports_structured_output_any") is True or row.get("router_supports_structured_output_any") is True: return True, "structured output support" if "conversational" in text: return True, "conversational tag" if has_text_model_signal(model_id): return True, "model-name signal" return False, "no LLM signal" def parse_hf_datetime(value: str) -> datetime | None: if not value: return None try: return datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: return None def int_metric(value: Any) -> int: if isinstance(value, bool): return int(value) if isinstance(value, int): return value if isinstance(value, float): return int(value) if isinstance(value, str): stripped = value.strip() if stripped: try: return int(float(stripped)) except ValueError: return 0 return 0 def slug_tokens(slug: str) -> list[str]: return [token for token in slug.split("-") if token] def has_model_shape(slug: str) -> bool: tokens = slug_tokens(slug) family = any( token in MODEL_FAMILY_TOKENS or any(token.startswith(f"{family}") for family in MODEL_FAMILY_TOKENS) for token in tokens[:3] ) sized_or_versioned = any( bool(re.fullmatch(r"[a-z]?[0-9]+(?:[a-z]*[0-9]*)?[bm]?", token)) and any(char.isdigit() for char in token) for token in tokens ) return family and sized_or_versioned and len(tokens) >= 2 def token_sequence_match(haystack: str, needle: str) -> tuple[list[str], list[str]] | None: haystack_tokens = slug_tokens(haystack) needle_tokens = slug_tokens(needle) if len(needle_tokens) < 2 or len(needle_tokens) >= len(haystack_tokens): return None for index in range(len(haystack_tokens) - len(needle_tokens) + 1): if haystack_tokens[index : index + len(needle_tokens)] == needle_tokens: return haystack_tokens[:index], haystack_tokens[index + len(needle_tokens) :] return None def component_scale(row: dict[str, Any], target: dict[str, Any] | None) -> bool: if target is None: return False try: row_params = float(row.get("parameter_count_b") or 0) target_params = float(target.get("parameter_count_b") or 0) except (TypeError, ValueError): return False return row_params > 0 and target_params > 0 and row_params < target_params * 0.5 def derivative_safe_suffix_token(token: str) -> bool: return ( token in DERIVATIVE_SAFE_SUFFIX_TOKENS or bool(re.fullmatch(r"q[2-8]|[0-9]+bit|[0-9]+", token)) or token in {"k", "m", "s", "l"} ) def derivative_tokens_allow_grouping( prefix_tokens: list[str], suffix_tokens: list[str], row: dict[str, Any], target: dict[str, Any] | None, ) -> bool: extra = set(prefix_tokens + suffix_tokens) if any(token in MODEL_FAMILY_TOKENS for token in extra): return False if extra & PEER_VARIANT_TOKENS: return False if prefix_tokens: prefix = set(prefix_tokens) if not (prefix & DERIVATIVE_COMPONENT_PREFIX_TOKENS): return False if any(not derivative_safe_suffix_token(token) for token in suffix_tokens): return False if extra & DERIVATIVE_STRONG_TOKENS: return True if extra & DERIVATIVE_AMBIGUOUS_TOKENS: return bool(packaging_signals(row)) or component_scale(row, target) return False def contained_lineage_key( slug: str, known_keys: set[str], row: dict[str, Any], key_representatives: dict[str, dict[str, Any]], ) -> str | None: matches: list[tuple[int, str, list[str], list[str]]] = [] for key in known_keys: if key == slug or not has_model_shape(key): continue match = token_sequence_match(slug, key) if match is None: continue prefix, suffix = match target = key_representatives.get(key) if derivative_tokens_allow_grouping(prefix, suffix, row, target): matches.append((len(slug_tokens(key)), key, prefix, suffix)) if not matches: return None matches.sort(reverse=True) return matches[0][1] def resolve_lineage_key( key: str, row: dict[str, Any], known_keys: set[str], key_representatives: dict[str, dict[str, Any]], ) -> str: return contained_lineage_key(key, known_keys, row, key_representatives) or key def representative_for_rows(rows: list[dict[str, Any]]) -> dict[str, Any]: return sorted( rows, key=lambda row: ( int_metric(row.get("downloads_all_time")), int_metric(row.get("downloads")), int_metric(row.get("likes")), str(row.get("id") or ""), ), reverse=True, )[0] def group_models(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: by_id = {str(row["id"]): row for row in rows} initial_key_map: dict[str, list[dict[str, Any]]] = defaultdict(list) for row in rows: initial_key_map[slug_dedupe_key(str(row["id"]))].append(row) known_keys = set(initial_key_map) key_representatives = {key: representative_for_rows(values) for key, values in initial_key_map.items()} group_map: dict[str, list[dict[str, Any]]] = defaultdict(list) for row in rows: key = slug_dedupe_key(str(row["id"])) for base in packaging_base_refs(row): if base in by_id: key = resolve_lineage_key(slug_dedupe_key(base), row, known_keys, key_representatives) break else: candidate_keys = known_keys - {key} contained_key = contained_lineage_key( normalized_repo_slug(str(row["id"])), candidate_keys, row, key_representatives, ) if contained_key: key = contained_key group_map[key].append(row) for key in list(group_map): target = f"{key}-it" if key.startswith("gemma-") and target in group_map and all(is_artifact_slug(str(row["id"])) for row in group_map[key]): group_map[target].extend(group_map.pop(key)) groups: list[dict[str, Any]] = [] for key, artifacts in group_map.items(): ordered_artifacts = sorted( artifacts, key=lambda row: ( int_metric(row.get("downloads_all_time")), int_metric(row.get("downloads")), int_metric(row.get("likes")), str(row.get("id") or ""), ), reverse=True, ) representative = ordered_artifacts[0] group_downloads = sum(int_metric(row.get("downloads")) for row in ordered_artifacts) group_downloads_all_time = sum(int_metric(row.get("downloads_all_time")) for row in ordered_artifacts) group_likes = sum(int_metric(row.get("likes")) for row in ordered_artifacts) signals = unique_preserve_order([signal for row in ordered_artifacts for signal in artifact_signals(row)]) router_served = any(row.get("router_served") is True for row in ordered_artifacts) supports_tools = any( row.get("supports_tools_any") is True or row.get("router_supports_tools_any") is True for row in ordered_artifacts ) supports_structured = any( row.get("supports_structured_output_any") is True or row.get("router_supports_structured_output_any") is True for row in ordered_artifacts ) group = { "group_key": key, "representative_id": representative["id"], "author": representative["author"], "downloads": group_downloads, "group_downloads": group_downloads, "representative_downloads": int_metric(representative.get("downloads")), "downloads_all_time": group_downloads_all_time, "group_downloads_all_time": group_downloads_all_time, "representative_downloads_all_time": int_metric(representative.get("downloads_all_time")), "likes": group_likes, "group_likes": group_likes, "representative_likes": int_metric(representative.get("likes")), "parameter_count_b": representative["parameter_count_b"], "capacity_class": representative["capacity_class"], "candidate_class": representative["candidate_class"], "pipeline_tag": representative["pipeline_tag"], "license": representative["license"] or "unknown", "router_served": router_served, "supports_tools": supports_tools, "supports_structured_output": supports_structured, "artifact_count": len(ordered_artifacts), "artifacts": [row["id"] for row in ordered_artifacts], "artifact_signals": signals, "created_at": representative["created_at"], "last_modified": representative["last_modified"], } groups.append(group) for row in ordered_artifacts: row["group_key"] = key groups.sort( key=lambda row: ( int_metric(row.get("downloads_all_time")), int_metric(row.get("downloads")), int_metric(row.get("likes")), str(row.get("group_key") or ""), ), reverse=True, ) for index, group in enumerate(groups, start=1): group["rank"] = index return groups def age_days(created_at: Any, as_of: datetime) -> float: created = parse_hf_datetime(str(created_at or "")) if created is None: return 0.0 return max((as_of - created).total_seconds() / SECONDS_PER_DAY, 0.0) def rate_metrics(group: dict[str, Any], as_of: datetime) -> dict[str, float]: age = age_days(group.get("created_at"), as_of) download_age = max(age, MIN_DOWNLOAD_RATE_AGE_DAYS) like_age = max(age, MIN_LIKE_RATE_AGE_DAYS) group_downloads_all_time = int_metric(group.get("group_downloads_all_time", group.get("downloads_all_time"))) representative_downloads_all_time = int_metric( group.get("representative_downloads_all_time", group_downloads_all_time) ) group_likes = int_metric(group.get("group_likes", group.get("likes"))) representative_likes = int_metric(group.get("representative_likes", group_likes)) group_downloads_per_month = round(group_downloads_all_time * DAYS_PER_MONTH / download_age, 4) representative_downloads_per_month = round(representative_downloads_all_time * DAYS_PER_MONTH / download_age, 4) group_likes_per_month = round(group_likes * DAYS_PER_MONTH / like_age, 4) representative_likes_per_month = round(representative_likes * DAYS_PER_MONTH / like_age, 4) return { "age_days": round(age, 4), "downloads_per_month": group_downloads_per_month, "group_downloads_per_month": group_downloads_per_month, "representative_downloads_per_month": representative_downloads_per_month, "likes_per_month": group_likes_per_month, "group_likes_per_month": group_likes_per_month, "representative_likes_per_month": representative_likes_per_month, } def add_rate_metrics(groups: list[dict[str, Any]], as_of: datetime) -> None: for group in groups: group.update(rate_metrics(group, as_of)) groups.sort( key=lambda row: ( float(row.get("downloads_per_month") or 0), int_metric(row.get("downloads_all_time")), int_metric(row.get("downloads")), float(row.get("likes_per_month") or 0), str(row.get("group_key") or ""), ), reverse=True, ) for index, group in enumerate(groups, start=1): group["rank"] = index def compare_group_snapshots(before: list[dict[str, Any]], after: list[dict[str, Any]]) -> dict[str, Any]: before_by_key = {str(group.get("group_key")): group for group in before} after_by_key = {str(group.get("group_key")): group for group in after} before_keys = set(before_by_key) after_keys = set(after_by_key) changed_artifacts: list[dict[str, Any]] = [] changed_metrics: list[dict[str, Any]] = [] for key in sorted(before_keys & after_keys): before_group = before_by_key[key] after_group = after_by_key[key] before_artifacts = list(before_group.get("artifacts") or []) after_artifacts = list(after_group.get("artifacts") or []) if before_artifacts != after_artifacts: changed_artifacts.append( { "group_key": key, "before_artifacts": before_artifacts, "after_artifacts": after_artifacts, } ) metric_delta: dict[str, Any] = {"group_key": key} changed = False for field in ("rank", "downloads", "downloads_all_time", "downloads_per_month", "likes_per_month"): before_value = before_group.get(field) after_value = after_group.get(field) if before_value != after_value: metric_delta[field] = {"before": before_value, "after": after_value} changed = True if changed: changed_metrics.append(metric_delta) return { "before_groups": len(before), "after_groups": len(after), "groups_added": sorted(after_keys - before_keys), "groups_removed": sorted(before_keys - after_keys), "artifact_changes": changed_artifacts, "metric_changes": changed_metrics, } def collect_metadata( *, creators: list[str], min_downloads: int, page_size: int, limit_per_creator: int | None, token: str | None, sleep_seconds: float, enable_discovery: bool, discovery_max_results_per_query: int, related_discovery_seed_limit: int, ) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, str]], list[dict[str, Any]], dict[str, Any]]: router_records = fetch_router_models(token) creators = unique_preserve_order(creators + router_creators(router_records)) router_by_id = {record["id"]: record for record in router_records} raw_rows: list[dict[str, Any]] = [] errors: list[dict[str, str]] = [] for creator in creators: try: raw_rows.extend( list_creator_models( creator, page_size=page_size, sort="downloads", direction=-1, token=token, limit=limit_per_creator, sleep_seconds=sleep_seconds, ) ) except Exception as err: # noqa: BLE001 errors.append({"creator": creator, "error": str(err)}) deduped_raw: dict[str, dict[str, Any]] = {} for row in raw_rows: model_id = row.get("id") or row.get("modelId") if isinstance(model_id, str) and model_id: deduped_raw[model_id] = row for record in router_records: model_id = record.get("id") if isinstance(model_id, str) and model_id not in deduped_raw: try: quoted = urllib.parse.quote(model_id, safe="/") detail, _ = request_json(f"{HF_MODELS_API}/{quoted}?{urllib.parse.urlencode([('expand', field) for field in EXPAND_FIELDS], doseq=True)}", token) if isinstance(detail, dict): deduped_raw[model_id] = detail except Exception as err: # noqa: BLE001 errors.append({"model": model_id, "error": str(err)}) discovery_summary: dict[str, Any] = {"enabled": False, "queries": 0, "query_counts": {}, "records": 0} if enable_discovery: initial_normalized = [ normalize_record(row, router_by_id.get(str(row.get("id") or row.get("modelId") or ""))) for row in deduped_raw.values() ] seed_rows = [ row for row in initial_normalized if int_metric(row.get("downloads")) > min_downloads and row.get("candidate") is True ] discovered_rows, discovery_errors, discovery_summary = discover_additional_models( seed_rows=seed_rows, existing_ids=set(deduped_raw), min_downloads=min_downloads, page_size=page_size, token=token, sleep_seconds=sleep_seconds, max_results_per_query=discovery_max_results_per_query, seed_limit=related_discovery_seed_limit, ) errors.extend(discovery_errors) for row in discovered_rows: model_id = row.get("id") or row.get("modelId") if isinstance(model_id, str) and model_id: deduped_raw[model_id] = row normalized = [normalize_record(row, router_by_id.get(str(row.get("id") or row.get("modelId") or ""))) for row in deduped_raw.values()] normalized.sort(key=lambda row: (int(row.get("downloads") or 0), int(row.get("likes") or 0), str(row.get("id"))), reverse=True) popular = [row for row in normalized if int(row.get("downloads") or 0) > min_downloads] return normalized, popular, errors, router_records, discovery_summary def build_payload( *, min_downloads: int = DEFAULT_MIN_DOWNLOADS, top_n: int = DEFAULT_TOP_N, max_age_days: int = DEFAULT_MAX_AGE_DAYS, creators: list[str] | None = None, page_size: int = 100, limit_per_creator: int | None = None, token: str | None = None, sleep_seconds: float = 0.05, enable_discovery: bool | None = None, discovery_max_results_per_query: int | None = None, related_discovery_seed_limit: int | None = None, ) -> dict[str, Any]: started = iso() if enable_discovery is None: enable_discovery = os.environ.get("OPEN_MODEL_ENABLE_DISCOVERY", "1").lower() not in {"0", "false", "no", "off"} if discovery_max_results_per_query is None: discovery_max_results_per_query = int( os.environ.get("OPEN_MODEL_DISCOVERY_MAX_RESULTS_PER_QUERY", DEFAULT_DISCOVERY_MAX_RESULTS_PER_QUERY) ) if related_discovery_seed_limit is None: related_discovery_seed_limit = int( os.environ.get("OPEN_MODEL_RELATED_DISCOVERY_SEED_LIMIT", DEFAULT_RELATED_DISCOVERY_SEED_LIMIT) ) normalized, popular, errors, router_records, discovery_summary = collect_metadata( creators=creators or env_creators(), min_downloads=min_downloads, page_size=page_size, limit_per_creator=limit_per_creator, token=token, sleep_seconds=sleep_seconds, enable_discovery=enable_discovery, discovery_max_results_per_query=discovery_max_results_per_query, related_discovery_seed_limit=related_discovery_seed_limit, ) candidates = [row for row in popular if row.get("candidate") is True] groups = group_models(candidates) generated_at_dt = now_utc() add_rate_metrics(groups, generated_at_dt) class_counts = Counter(group["candidate_class"] for group in groups) creator_counts = Counter(group["author"] for group in groups) summary = { "generated_at": iso(generated_at_dt), "started_at": started, "source_model_records": len(normalized), "popular_model_records": len(popular), "candidate_artifacts": len(candidates), "leaderboard_groups": len(groups), "multi_artifact_groups": sum(1 for group in groups if int(group["artifact_count"]) > 1), "router_records": len(router_records), "discovered_model_records": int(discovery_summary.get("records") or 0), "discovery_queries": int(discovery_summary.get("queries") or 0), "tool_supported_groups": sum(1 for group in groups if group["supports_tools"]), "structured_output_groups": sum(1 for group in groups if group["supports_structured_output"]), "class_counts": dict(sorted(class_counts.items())), "top_creators": [{"creator": name, "groups": count} for name, count in creator_counts.most_common(12)], "errors": errors, "discovery": discovery_summary, } return { "schema_version": 1, "title": "Open LLM Distribution Leaderboard", "generated_at": summary["generated_at"], "filters": { "min_downloads": min_downloads, "default_age_filter_days": max_age_days, "age_filter_options_days": ["off", 183, 365, 548, 730], "top_n": top_n, "metric_scope_default": "group", "candidate_filter": "Auto-filtered LLM-ish text, coding, multimodal, and special-architecture model repos; excludes embeddings, rerankers, audio utilities, image-only generation, and narrow support models.", "grouping": "Artifacts are grouped by normalized slug and explicit quantized base-model metadata when available.", }, "summary": summary, "groups": groups, "all_group_count": len(groups), "models": candidates, "popular_models": popular, "router_models": router_records, } def format_compact_number(value: Any) -> str: try: number = float(value) except (TypeError, ValueError): return "0" sign = "-" if number < 0 else "" number = abs(number) if 0 < number < 0.1: return f"{sign}<0.1" units = [(1_000_000_000, "B"), (1_000_000, "M"), (1_000, "K")] for index, (threshold, suffix) in enumerate(units): if number >= threshold: compact = number / threshold if round(compact, 1) >= 1000 and index > 0: threshold, suffix = units[index - 1] compact = number / threshold text = f"{compact:.1f}".removesuffix(".0") return f"{sign}{text}{suffix}" text = f"{number:.1f}".removesuffix(".0") return f"{sign}{text}" def format_params(value: Any) -> str: if value in {"", None}: return "unknown" try: number = float(value) except (TypeError, ValueError): return str(value) if number >= 100: return f"{number:.0f}B" if number >= 10: return f"{number:.1f}B" return f"{number:.2f}B" def format_date(value: Any) -> str: parsed = parse_hf_datetime(str(value or "")) if parsed is None: return "unknown" return parsed.date().isoformat() def format_age(created_at: Any, as_of: datetime) -> str: if parse_hf_datetime(str(created_at or "")) is None: return "" age = age_days(created_at, as_of) if age < 1: return "today" years = int(age // 365.2425) remaining_days = age - (years * 365.2425) months = int(remaining_days // DAYS_PER_MONTH) days = int(age) parts: list[str] = [] if years: parts.append(f"{years}yr") if months: parts.append(f"{months}mo") return " ".join(parts) if months: return f"{months}mo" return f"{days}d" def format_created(value: Any, as_of: datetime) -> str: date = format_date(value) age = format_age(value, as_of) if date == "unknown" or not age: return date return f"{date} ({age} ago)" def render_created(value: Any, as_of: datetime) -> str: date = format_date(value) age = format_age(value, as_of) if date == "unknown" or not age: return f'{html.escape(date)}' return ( f'{html.escape(date)}' f'{html.escape(age)} ago' ) def model_url(model_id: str) -> str: return f"https://huggingface.co/{urllib.parse.quote(model_id, safe='/')}" def pill(label: str, tone: str = "neutral") -> str: return f'{html.escape(label)}' def render_html(payload: dict[str, Any]) -> str: summary = payload["summary"] groups = payload["groups"] filters = payload["filters"] generated_js = json.dumps(str(payload["generated_at"])) default_age_days = str(filters.get("default_age_filter_days") or filters.get("max_age_days") or DEFAULT_MAX_AGE_DAYS) top_n_js = json.dumps(filters.get("top_n") or DEFAULT_TOP_N) min_downloads_label = format_compact_number(filters.get("min_downloads") or DEFAULT_MIN_DOWNLOADS) generated_at_dt = parse_hf_datetime(str(payload["generated_at"])) or now_utc() generated_at_label = generated_at_dt.date().isoformat() def age_selected(value: str) -> str: return " selected" if value == default_age_days else "" def attr(value: Any) -> str: return html.escape(str(value), quote=True) def repo_count_label(count: int) -> str: return f"{count} repo" if count == 1 else f"{count} repos" def more_variants_label(count: int) -> str: return f"{count} more variant" if count == 1 else f"{count} more variants" def artifact_link(model_id: str) -> str: return ( f'' f"{html.escape(model_id)}" ) table_rows = [] for group in groups: if ( "group_downloads_per_month" not in group or "representative_downloads_per_month" not in group or "group_likes_per_month" not in group or "representative_likes_per_month" not in group ): group.update(rate_metrics(group, generated_at_dt)) artifact_count = int_metric(group.get("artifact_count")) or len(group.get("artifacts") or []) group_downloads_per_month = float( group.get("group_downloads_per_month", group.get("downloads_per_month", group.get("downloads"))) or 0 ) representative_downloads_per_month = float( group.get("representative_downloads_per_month", group_downloads_per_month) or 0 ) group_likes_per_month = float(group.get("group_likes_per_month", group.get("likes_per_month")) or 0) representative_likes_per_month = float(group.get("representative_likes_per_month", group_likes_per_month) or 0) group_downloads = int_metric(group.get("group_downloads", group.get("downloads"))) representative_downloads = int_metric(group.get("representative_downloads", group_downloads)) group_downloads_all_time = int_metric(group.get("group_downloads_all_time", group.get("downloads_all_time"))) representative_downloads_all_time = int_metric( group.get("representative_downloads_all_time", group_downloads_all_time) ) group_likes = int_metric(group.get("group_likes", group.get("likes"))) representative_likes = int_metric(group.get("representative_likes", group_likes)) repo_label = repo_count_label(artifact_count) group_downloads_detail_primary = format_compact_number(group_downloads_all_time) + " all-time across " + repo_label group_downloads_detail_secondary = format_compact_number(group_downloads) + " recent HF downloads" representative_downloads_detail_primary = ( format_compact_number(representative_downloads_all_time) + " all-time for representative" ) representative_downloads_detail_secondary = ( format_compact_number(representative_downloads) + " recent HF downloads" ) group_likes_detail_primary = format_compact_number(group_likes) + " likes across " + repo_label group_likes_detail_secondary = "grouped repos" representative_likes_detail_primary = format_compact_number(representative_likes) + " likes for representative" representative_likes_detail_secondary = "main repo only" group_downloads_label = format_compact_number(group_downloads_per_month) + "/mo" representative_downloads_label = format_compact_number(representative_downloads_per_month) + "/mo" group_likes_label = format_compact_number(group_likes_per_month) + "/mo" representative_likes_label = format_compact_number(representative_likes_per_month) + "/mo" group_downloads_tooltip = "\n".join( [group_downloads_label, group_downloads_detail_primary, group_downloads_detail_secondary] ) representative_downloads_tooltip = "\n".join( [ representative_downloads_label, representative_downloads_detail_primary, representative_downloads_detail_secondary, ] ) group_likes_tooltip = "\n".join([group_likes_label, group_likes_detail_primary, group_likes_detail_secondary]) representative_likes_tooltip = "\n".join( [representative_likes_label, representative_likes_detail_primary, representative_likes_detail_secondary] ) artifacts = [str(item) for item in group.get("artifacts") or []] visible_artifact = artifact_link(artifacts[0]) if artifacts else "" more_count = max(len(artifacts) - 1, 0) extra_artifacts = "".join(artifact_link(item) for item in artifacts[1:]) more_label = more_variants_label(more_count) artifact_toggle = ( f'' if more_count else "" ) artifact_more = f'{artifact_toggle}' if more_count else "" artifact_extra = f'' if more_count else "" row_attrs = { "data-class": group["candidate_class"], "data-creator": group["author"], "data-created": group.get("created_at") or "", "data-group-downloads-per-month": group_downloads_per_month, "data-representative-downloads-per-month": representative_downloads_per_month, "data-group-likes-per-month": group_likes_per_month, "data-representative-likes-per-month": representative_likes_per_month, "data-group-downloads-per-month-label": group_downloads_label, "data-representative-downloads-per-month-label": representative_downloads_label, "data-group-likes-per-month-label": group_likes_label, "data-representative-likes-per-month-label": representative_likes_label, "data-group-downloads-tooltip": group_downloads_tooltip, "data-representative-downloads-tooltip": representative_downloads_tooltip, "data-group-likes-tooltip": group_likes_tooltip, "data-representative-likes-tooltip": representative_likes_tooltip, } row_attr_text = " ".join(f'{name}="{attr(value)}"' for name, value in row_attrs.items()) table_rows.append( f""" {group['rank']} {html.escape(str(group['group_key']))} {group_downloads_label} {group_likes_label} {format_params(group['parameter_count_b'])} {render_created(group.get('created_at'), generated_at_dt)} {visible_artifact}{artifact_more}{artifact_extra} """ ) return f""" Open LLM Distribution Leaderboard

Open LLM Distribution Leaderboard

This is work in progress, there might be inaccuracies. Last updated: {html.escape(generated_at_label)}.

{''.join(table_rows)}
Rank Model Avg Downloads / Month Avg Likes / month Params Created Related repos

How This Is Calculated

What counts Public Hugging Face model repos above the minimum download threshold are scanned, then filtered for LLM-ish text, coding, multimodal, and special architecture signals.
Models Related repos are grouped by normalized model name and explicit base-model metadata when available. The main row uses the most-downloaded repo in that group.
Metric scope Grouped metrics sum downloads, all-time downloads, and likes across related lineage repos, including package, quantized, and component repos when metadata or slug evidence links them safely. Representative metrics use only the main repo shown in the Model column.
Default sort Rows are sorted by average downloads per month unless you choose likes per month. Sorting follows the selected metric scope, and visible rank is recalculated after filters.
Rates avg downloads/month = all-time downloads / max(model age, 1 month)
avg likes/month = total likes / max(model age, 1 month)
Coverage Discovery scans configured creators, router models, and popular artifact searches. Valid popular artifacts without a safe grouping target stay as standalone rows. The snapshot omits repos below the configured download floor, currently {min_downloads_label}.
Age filter The default view hides models older than one year. Use Off to include older models from the generated snapshot.
Snapshots The leaderboard updates daily. Each refresh writes a new latest view and keeps timestamped historical snapshots in the bucket. Ages are calculated against the snapshot generation time.
Download counts Hugging Face download counts are public, deduplicated counts over tracked model files, meant to approximate model pulls rather than raw requests for every split model file. The recent downloads field is a recent-window count; downloadsAllTime is the same style of count over all time. This leaderboard uses downloadsAllTime divided by model age for Avg Downloads / Month, while recent downloads appear only in metric tooltips.
Sources Space source Data bucket
""" def write_json(path: Path, payload: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2, sort_keys=True, ensure_ascii=False) + "\n", encoding="utf-8") def write_jsonl(path: Path, rows: list[dict[str, Any]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8") as handle: for row in rows: handle.write(json.dumps(row, sort_keys=True, ensure_ascii=False) + "\n") def write_csv(path: Path, rows: list[dict[str, Any]], fieldnames: list[str]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames, lineterminator="\n") writer.writeheader() for row in rows: writer.writerow({field: row.get(field, "") for field in fieldnames}) def copy_tree_contents(source: Path, dest: Path) -> None: dest.mkdir(parents=True, exist_ok=True) for path in source.iterdir(): target = dest / path.name if path.is_dir(): if target.exists(): shutil.rmtree(target) shutil.copytree(path, target) else: shutil.copy2(path, target) def write_outputs(payload: dict[str, Any], output_dir: Path) -> Path: run_stamp = stamp() with tempfile.TemporaryDirectory(prefix="open-llm-leaderboard-") as tmp: tmp_path = Path(tmp) latest = tmp_path / "latest" snapshot = tmp_path / "snapshots" / run_stamp html_text = render_html(payload) for base in (latest, snapshot): base.mkdir(parents=True, exist_ok=True) (base / "index.html").write_text(html_text, encoding="utf-8") write_json(base / "leaderboard.json", payload) write_json(base / "summary.json", payload["summary"]) write_json(base / "groups.json", payload["groups"]) write_jsonl(base / "models.jsonl", payload["models"]) write_csv(base / "groups.csv", payload["groups"], GROUP_CSV_FIELDS) write_csv(base / "models.csv", payload["models"], MODEL_CSV_FIELDS) output_dir.mkdir(parents=True, exist_ok=True) copy_tree_contents(tmp_path / "latest", output_dir / "latest") copy_tree_contents(tmp_path / "snapshots", output_dir / "snapshots") (output_dir / "README.md").write_text( "# Open LLM Distribution Leaderboard Data\n\n" "Mutable bucket-backed data for the Open LLM Distribution Leaderboard Space.\n\n" "- `latest/index.html`: rendered leaderboard HTML.\n" "- `latest/leaderboard.json`: full machine-readable latest payload.\n" "- `latest/groups.csv`: grouped leaderboard table.\n" "- `latest/models.csv`: candidate artifact table.\n" "- `snapshots/`: timestamped historical runs.\n", encoding="utf-8", ) return output_dir / "latest" / "index.html" def refresh( *, output_dir: Path, min_downloads: int = DEFAULT_MIN_DOWNLOADS, top_n: int = DEFAULT_TOP_N, max_age_days: int = DEFAULT_MAX_AGE_DAYS, limit_per_creator: int | None = None, token: str | None = None, enable_discovery: bool | None = None, ) -> dict[str, Any]: payload = build_payload( min_downloads=min_downloads, top_n=top_n, max_age_days=max_age_days, limit_per_creator=limit_per_creator, token=token, enable_discovery=enable_discovery, ) html_path = write_outputs(payload, output_dir) payload["html_path"] = str(html_path) return payload def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--output-dir", type=Path, default=DEFAULT_LOCAL_DATA_DIR) parser.add_argument("--min-downloads", type=int, default=int(os.environ.get("OPEN_MODEL_MIN_DOWNLOADS", DEFAULT_MIN_DOWNLOADS))) parser.add_argument("--top-n", type=int, default=int(os.environ.get("OPEN_MODEL_TOP_N", DEFAULT_TOP_N))) parser.add_argument("--max-age-days", type=int, default=int(os.environ.get("OPEN_MODEL_MAX_AGE_DAYS", DEFAULT_MAX_AGE_DAYS))) parser.add_argument("--limit-per-creator", type=int, default=None) parser.add_argument("--disable-discovery", action="store_true") return parser.parse_args() def main() -> int: args = parse_args() token = os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN") payload = refresh( output_dir=args.output_dir, min_downloads=args.min_downloads, top_n=args.top_n, max_age_days=args.max_age_days, limit_per_creator=args.limit_per_creator, token=token, enable_discovery=False if args.disable_discovery else None, ) print( json.dumps( { "generated_at": payload["generated_at"], "groups": payload["summary"]["leaderboard_groups"], "artifacts": payload["summary"]["candidate_artifacts"], "html_path": payload["html_path"], }, sort_keys=True, ) ) return 0 if __name__ == "__main__": raise SystemExit(main())