Spaces:
Running
Running
| """Ranking: for each model, the best quant for your hardware, then the models themselves. | |
| Every quant of every repo holding the model is fitted to the hardware; the best is the | |
| one that runs most comfortably (all on GPU, then experts in RAM, then more layers on | |
| GPU, then CPU), and among equally comfortable ones the one that matches the priority. | |
| Models are then ranked on that fit, popularity, recency and community reports. | |
| Nothing here looks at who uploaded a quant beyond its download count. | |
| """ | |
| from __future__ import annotations | |
| import math | |
| from dataclasses import dataclass, field | |
| from datetime import datetime, timezone | |
| from .catalogue import Group | |
| from .detail import ModelDetail | |
| from .fit import Fit, fit, llama_cpp_command, ollama_command | |
| from .geometry import estimate_without_header | |
| from .hardware import Hardware | |
| from .parsing import QUANT_BPW, QuantFile, min_model_bytes, quant_bits | |
| PRIORITIES = ["balanced", "quality", "small"] | |
| SORTS = ["best", "popular", "recent", "community"] | |
| WEIGHTS = {"fit": 0.35, "capacity": 0.20, "popularity": 0.25, "recency": 0.10, "community": 0.10} | |
| class Query: | |
| hardware: Hardware | None | |
| context: int = 8192 | |
| kv_type: str = "f16" | |
| priority: str = "balanced" | |
| min_bits: int = 0 # skip quants below this bit level (Q4 and up: 4) | |
| vram_only: bool = False # discrete GPUs: nothing may spill into system RAM | |
| mmproj: str = "gpu" # vision projector: "gpu" (llama-server default), "ram" or "off" | |
| class Option: | |
| repo: str | |
| repo_downloads: int | |
| quant: QuantFile | |
| fit: Fit | None | |
| estimated: bool | |
| projector_bytes: int | None = None | |
| def to_dict(self, q: Query) -> dict: | |
| d = {"repo": self.repo, "repo_downloads": self.repo_downloads, **self.quant.to_dict(), | |
| "estimated": self.estimated, "fit": self.fit.to_dict() if self.fit else None, | |
| "projector_size": self.projector_bytes, | |
| "url": f"https://huggingface.co/{self.repo}"} | |
| if self.fit: | |
| proj = q.mmproj if self.projector_bytes and q.mmproj != "gpu" else None | |
| d["commands"] = {"llama.cpp": llama_cpp_command(self.repo, self.quant.quant, self.fit, q.context, q.kv_type, proj), | |
| "ollama": ollama_command(self.repo, self.quant.quant)} | |
| else: | |
| d["commands"] = {"llama.cpp": f"llama-server -hf {self.repo}:{self.quant.quant} -c {q.context}", | |
| "ollama": ollama_command(self.repo, self.quant.quant)} | |
| return d | |
| def preference(opt: Option, priority: str) -> float: | |
| b = opt.quant.bpw or 4.85 | |
| if priority == "quality": | |
| return min(b, 8.5) | |
| if priority == "small": | |
| return -(opt.quant.size or 0) / 1e9 | |
| return -abs(b - 5.0) if b >= 3.5 else -3.0 - (3.5 - b) # Q4_K_M .. Q5_K_M | |
| def below_floor(qf: QuantFile, params: int | None, min_bits: int) -> bool: | |
| if not min_bits: | |
| return False | |
| bits = qf.bits | |
| if bits is None and params and qf.size: | |
| bits = quant_bits("", qf.size * 8 / params) | |
| return bits is not None and bits < min_bits | |
| PLACEHOLDER = {0: "Q4_K_M", 1: "Q4_K_M", 2: "Q4_K_M", 3: "Q4_K_M", 4: "Q4_K_M", 5: "Q5_K_M", 6: "Q6_K", 8: "Q8_0"} | |
| def options_for(group: Group, detail: ModelDetail | None, q: Query) -> list[Option]: | |
| opts: list[Option] = [] | |
| geo = detail.geometry if detail else None | |
| n_vocab = detail.n_vocab if detail else None | |
| downloads = {r.id: r.downloads for r in group.repos} | |
| if detail: | |
| for repo, rf in detail.repos.items(): | |
| for qf in rf.quants: | |
| if not qf.size or qf.size < min_model_bytes(group.params): | |
| continue # a draft model, head or encoder that survived the name checks | |
| if below_floor(qf, group.params, q.min_bits): | |
| continue | |
| if geo is not None: | |
| g = geo.scaled(qf.size, group.params) | |
| est = False | |
| elif group.params: | |
| g = estimate_without_header(group.params, qf.bpw or 4.85, qf.size, group.context_length, group.moe) | |
| est = True | |
| else: | |
| continue | |
| f = fit(g, q.hardware, q.context, q.kv_type, n_vocab, rf.mmproj_bytes or 0, q.vram_only, q.mmproj) if q.hardware else None | |
| opts.append(Option(repo, downloads.get(repo, 0), qf, f, est, rf.mmproj_bytes)) | |
| # A size guessed from the parameter count, only while the files are unknown. Once they are | |
| # listed and none is a usable quant (a drafter, a projector), there is nothing to guess about. | |
| files_known = bool(detail and detail.repos and all(rf.error is None for rf in detail.repos.values())) | |
| if not opts and group.params and group.repos and not files_known: | |
| lead = group.repos[0] | |
| name = PLACEHOLDER.get(q.min_bits, "Q8_0") | |
| qf = QuantFile(name, [], int(group.params * QUANT_BPW[name] / 8), QUANT_BPW[name]) | |
| g = (geo.scaled(qf.size, group.params) if geo else | |
| estimate_without_header(group.params, qf.bpw, qf.size, group.context_length, group.moe)) | |
| f = fit(g, q.hardware, q.context, q.kv_type, n_vocab, vram_only=q.vram_only) if q.hardware else None | |
| opts.append(Option(lead.id, lead.downloads, qf, f, geo is None)) | |
| return opts | |
| def best_option(opts: list[Option], priority: str) -> Option | None: | |
| if not opts: | |
| return None | |
| return max(opts, key=lambda o: ((o.fit.score if o.fit else 0.0), preference(o, priority), o.repo_downloads)) | |
| class Scored: | |
| group: Group | |
| best: Option | None | |
| options: list[Option] | |
| score: float | |
| components: dict | |
| reasons: list[str] = field(default_factory=list) | |
| detail_loaded: bool = False | |
| detail_repos: dict | None = None | |
| def to_dict(self, q: Query, with_options: bool = False) -> dict: | |
| d = {**self.group.summary(), "score": round(self.score, 1), "components": self.components, | |
| "reasons": self.reasons, "detail_loaded": self.detail_loaded, | |
| "best": self.best.to_dict(q) if self.best else None} | |
| if with_options: | |
| d["options"] = [o.to_dict(q) for o in self.options] | |
| return d | |
| def _age_days(ts: str | None) -> float | None: | |
| if not ts: | |
| return None | |
| try: | |
| t = datetime.fromisoformat(ts.replace("Z", "+00:00")) | |
| return max((datetime.now(timezone.utc) - t).total_seconds() / 86400, 0) | |
| except ValueError: | |
| return None | |
| class Engine: | |
| version = "fit-v1" | |
| def _capacity(g: Group, best: Option | None, q: Query) -> float: | |
| """Bigger models that still run well rank higher: a 27B that fits beats a 1B that fits. | |
| Scaled to what the hardware could hold at ~4.5 bits; neutral without hardware.""" | |
| if q.hardware is None or not g.params or best is None or best.fit is None or best.fit.tier in ("no", "no_vram"): | |
| return 0.5 if q.hardware is None else 0.0 | |
| mem = q.hardware.usable_vram_gb() if q.hardware.kind != "cpu" else q.hardware.ram_gb * 0.8 | |
| if q.hardware.kind == "gpu": | |
| mem += q.hardware.ram_gb * 0.25 # experts or layers can spill into RAM | |
| reference = max(mem * 1.6e9, 1e9) # params that fit in `mem` at ~5 bits | |
| return round(min(math.log1p(g.params / 1e9) / math.log1p(reference / 1e9), 1.0), 3) | |
| def score(self, groups: list[Group], q: Query, details: dict[str, ModelDetail], community: dict) -> list[Scored]: | |
| max_dl = max((g.downloads for g in groups), default=1) or 1 | |
| out = [] | |
| for g in groups: | |
| det = details.get(g.key) | |
| opts = options_for(g, det, q) | |
| best = best_option(opts, q.priority) | |
| comp, reasons = {}, [] | |
| if q.hardware is None or best is None or best.fit is None: | |
| comp["fit"] = 0.5 | |
| if q.hardware is None: | |
| reasons.append("Pick your hardware to see what fits.") | |
| else: | |
| comp["fit"] = round(best.fit.score / 5, 3) | |
| where = f"{best.quant.quant} from {best.repo.split('/')[0]}" | |
| reasons.append(f"{best.fit.label}: {where} ({(best.quant.size or 0) / 1e9:.1f} GB)" | |
| + (" — estimated from the parameter count" if best.estimated else "")) | |
| comp["capacity"] = self._capacity(g, best, q) | |
| comp["popularity"] = round(math.log1p(g.downloads) / math.log1p(max_dl), 3) | |
| age = _age_days(g.created) or _age_days(g.modified) | |
| comp["recency"] = round(math.exp(-age / 120), 3) if age is not None else 0.3 | |
| c = community.get(g.key) | |
| if c and (c["feedback"] or c["benchmarks"]): | |
| share = c["positive_share"] if c["positive_share"] is not None else 0.6 | |
| comp["community"] = round(min(1.0, (c["feedback"] + c["benchmarks"]) / 5) * share, 3) | |
| reasons.append(f"{c['feedback']} report(s), {c['benchmarks']} benchmark(s) from the community") | |
| else: | |
| comp["community"] = 0.0 | |
| total = 100 * sum(WEIGHTS[k] * v for k, v in comp.items()) | |
| out.append(Scored(g, best, opts, total, comp, reasons, bool(det and det.repos), det.repos if det else None)) | |
| return out | |
| def sort_results(scored: list[Scored], sort: str, community: dict | None = None) -> list[Scored]: | |
| if sort == "popular": | |
| return sorted(scored, key=lambda s: -s.group.downloads) | |
| if sort == "recent": | |
| return sorted(scored, key=lambda s: s.group.created or s.group.modified or "", reverse=True) | |
| if sort == "community": | |
| c = community or {} | |
| return sorted(scored, key=lambda s: (-(c.get(s.group.key, {}).get("feedback", 0) | |
| + c.get(s.group.key, {}).get("benchmarks", 0)), -s.score)) | |
| return sorted(scored, key=lambda s: (-s.score, -s.group.downloads)) | |