| """Exact parameter accounting for candidate models -> MODEL_MANIFEST.md + reports/param_counts.json. |
| |
| Two independent counts per model: |
| stored sum of every tensor in the repo's safetensors headers (read via HTTP range requests, |
| no weight download), split into text / vision / audio / MTP heads by tensor name. |
| Tied input/output embeddings are counted once. |
| instantiated parameters of the module transformers builds (meta device, no weights) with |
| AutoModelForCausalLM, falling back to AutoModelForImageTextToText. |
| |
| LoRA/adapter repos are counted as base model + adapter tensors (shared-task rule). |
| Usage: python scripts/count_params.py [--models id1 id2 ...] |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import re |
| import sys |
| from pathlib import Path |
|
|
| sys.path.insert(0, str(Path(__file__).resolve().parents[1])) |
| from eval.common import ROOT, load_env |
|
|
| LIMIT = 8_000_000_000 |
| CUTOFF = "2026-10-01" |
| MANIFEST = ROOT / "MODEL_MANIFEST.md" |
| BEGIN, END = "<!-- AUTO:PARAMS:BEGIN -->", "<!-- AUTO:PARAMS:END -->" |
|
|
| PATTERNS = [ |
| ("vision", re.compile(r"vision|visual|image|siglip|patch_embed|multi_modal_projector|mm_projector|embed_vision")), |
| ("audio", re.compile(r"audio|embed_audio|speech")), |
| ("mtp", re.compile(r"(^|\.)mtp[._]")), |
| ] |
|
|
|
|
| def tensor_kind(name: str) -> str: |
| for kind, pat in PATTERNS: |
| if pat.search(name): |
| return kind |
| return "text" |
|
|
|
|
| def _tied(cfg: dict) -> bool: |
| for c in (cfg, cfg.get("text_config") or {}): |
| if "tie_word_embeddings" in c: |
| return bool(c["tie_word_embeddings"]) |
| return False |
|
|
|
|
| def stored_counts(api, repo: str, sha: str) -> dict: |
| from huggingface_hub import hf_hub_download |
|
|
| meta = api.get_safetensors_metadata(repo, revision=sha) |
| counts = {"text": 0, "vision": 0, "audio": 0, "mtp": 0} |
| names = {} |
| for fm in meta.files_metadata.values(): |
| for name, t in fm.tensors.items(): |
| n = 1 |
| for d in t.shape: |
| n *= d |
| names[name] = n |
| counts[tensor_kind(name)] += n |
| cfg = json.load(open(hf_hub_download(repo, "config.json", revision=sha))) |
| tied = _tied(cfg) |
| lm_head = [k for k in names if k.endswith("lm_head.weight")] |
| deduped = 0 |
| if tied and lm_head: |
| deduped = sum(names[k] for k in lm_head) |
| counts["text"] -= deduped |
| emb = sum(v for k, v in names.items() if re.search(r"embed_tokens|wte|lm_head|embeddings\.word", k) |
| and tensor_kind(k) == "text") - deduped |
| return {**counts, "embeddings": emb, "tied_embeddings": tied, "total": sum(counts.values()), |
| "model_type": cfg.get("model_type"), "config": cfg} |
|
|
|
|
| def instantiated_count(repo: str, sha: str) -> tuple[int | None, str]: |
| import torch |
| from accelerate import init_empty_weights |
| from transformers import AutoConfig, AutoModelForCausalLM |
|
|
| try: |
| cfg = AutoConfig.from_pretrained(repo, revision=sha) |
| except Exception as e: |
| return None, f"config: {type(e).__name__}" |
| for cls_name in ("AutoModelForCausalLM", "AutoModelForImageTextToText"): |
| try: |
| import transformers |
|
|
| cls = getattr(transformers, cls_name) |
| with init_empty_weights(): |
| m = cls.from_config(cfg) |
| m.tie_weights() |
| return sum(p.numel() for p in m.parameters()), type(m).__name__ |
| except Exception as e: |
| err = f"{cls_name}: {type(e).__name__}: {str(e)[:80]}" |
| return None, err |
|
|
|
|
| def count_repo(api, repo: str) -> dict: |
| info = api.model_info(repo) |
| sha = info.sha |
| row = {"id": repo, "sha": sha, "created": str(info.created_at)[:10], |
| "last_modified": str(info.last_modified)[:10], |
| "license": (info.card_data or {}).get("license") if info.card_data else None, |
| "gated": info.gated} |
| files = {s.rfilename for s in info.siblings} |
| if "adapter_config.json" in files: |
| from huggingface_hub import hf_hub_download |
|
|
| acfg = json.load(open(hf_hub_download(repo, "adapter_config.json", revision=sha))) |
| base = acfg["base_model_name_or_path"] |
| fm = api.parse_safetensors_file_metadata(repo, "adapter_model.safetensors", revision=sha) |
| adapter = sum(_numel(t.shape) for t in fm.tensors.values()) |
| b = count_repo(api, base) |
| row.update(kind="adapter", base=base, adapter_params=adapter, |
| modules_to_save=acfg.get("modules_to_save"), |
| text=b.get("text"), vision=b.get("vision"), audio=b.get("audio"), mtp=b.get("mtp"), |
| embeddings=b.get("embeddings"), tied_embeddings=b.get("tied_embeddings"), |
| total=(b.get("total") or 0) + adapter, text_loaded=(b.get("text_loaded") or 0) + adapter, |
| instantiated=None, inst_class=f"base {b.get('total')} + adapter") |
| return row |
| s = stored_counts(api, repo, sha) |
| s.pop("config") |
| inst, cls = instantiated_count(repo, sha) |
| row.update(kind="model", **s, text_loaded=s["text"], instantiated=inst, inst_class=cls) |
| return row |
|
|
|
|
| def _numel(shape) -> int: |
| n = 1 |
| for d in shape: |
| n *= d |
| return n |
|
|
|
|
| def verdict(r: dict) -> str: |
| if r.get("error"): |
| return "ERROR" |
| flags = [] |
| if r["created"] >= CUTOFF: |
| flags.append("RELEASED TOO LATE") |
| if r["total"] <= LIMIT: |
| flags.append("OK") |
| elif r["text_loaded"] <= LIMIT: |
| flags.append("OK only text-only (strip vision/audio/MTP)") |
| else: |
| flags.append("OVER 8B") |
| return "; ".join(flags) |
|
|
|
|
| def b(n) -> str: |
| return "-" if n in (None, 0) else f"{n / 1e9:.3f}B" |
|
|
|
|
| def write_manifest(rows: list[dict]) -> None: |
| lines = [BEGIN, "", |
| "| Model | Verdict | Total (all stored) | Text model | Vision | Audio | MTP | Embeddings | Tied | " |
| "Instantiated (class) | Repo created | License | Commit |", |
| "|---|---|---:|---:|---:|---:|---:|---:|---|---|---|---|---|"] |
| for r in rows: |
| if r.get("error"): |
| lines.append(f"| `{r['id']}` | ERROR: {r['error']} |" + " |" * 11) |
| continue |
| inst = f"{b(r['instantiated'])} ({r['inst_class']})" if r.get("instantiated") else r.get("inst_class", "-") |
| name = f"`{r['id']}`" + (f"<br>adapter on `{r['base']}` (+{b(r['adapter_params'])})" if r["kind"] == "adapter" else "") |
| lines.append(f"| {name} | {verdict(r)} | {b(r['total'])} | {b(r['text_loaded'])} | {b(r['vision'])} | " |
| f"{b(r['audio'])} | {b(r['mtp'])} | {b(r['embeddings'])} | {'yes' if r['tied_embeddings'] else 'no'} | " |
| f"{inst} | {r['created']} | {r['license'] or '?'} | `{r['sha'][:10]}` |") |
| lines += ["", END] |
| text = MANIFEST.read_text() |
| start, stop = text.index(BEGIN), text.index(END) + len(END) |
| MANIFEST.write_text(text[:start] + "\n".join(lines) + text[stop:]) |
|
|
|
|
| def main() -> None: |
| ap = argparse.ArgumentParser() |
| ap.add_argument("--models", nargs="*") |
| args = ap.parse_args() |
| load_env() |
| import yaml |
| from huggingface_hub import HfApi |
|
|
| repos = args.models or [c["id"] for c in yaml.safe_load(open(ROOT / "configs" / "models.yaml"))["candidates"]] |
| api = HfApi() |
| rows = [] |
| for repo in repos: |
| try: |
| r = count_repo(api, repo) |
| except Exception as e: |
| r = {"id": repo, "error": f"{type(e).__name__}: {str(e)[:100]}"} |
| rows.append(r) |
| print(f"{repo:42s} {verdict(r):45s} total={b(r.get('total'))} text={b(r.get('text_loaded'))} " |
| f"inst={b(r.get('instantiated'))}") |
| out = ROOT / "reports" / "param_counts.json" |
| out.parent.mkdir(exist_ok=True) |
| if args.models and out.exists(): |
| fresh = {r["id"]: r for r in rows} |
| old = json.loads(out.read_text()) |
| rows = [fresh.pop(r["id"], r) for r in old] + list(fresh.values()) |
| out.write_text(json.dumps(rows, indent=2, default=str)) |
| write_manifest(rows) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
