"""Hugging Face Hub — direct HTML only (no `/api/` endpoints): the hub pages embed their data as JSON in `data-props` attributes.
Sources (tier 2 for hub metadata — model-card statements are the authors' own):
* listing pages https://huggingface.co/models?author=&sort=downloads&p= → `ModelList.initialValues.models` (30 per page:
id, downloads, likes, pipeline_tag, lastModified, gated, numParameters). The organization page itself only embeds
its 10 most recent models, so the author listing is used for discovery (configurable `models_per_org`, default 20).
* model pages https://huggingface.co// → `ModelHeader.model` (cardData, config, safetensors,
tags, license, gated, createdAt, lastModified, downloads, likes) + GGUF/safetensors file links.
* raw model card https://huggingface.co///raw/main/README.md → YAML front matter (license, base_model, datasets,
language, quantized_by…) and the card text for later LLM passes. Not fetched for gated repositories (401 without auth).
* daily papers https://huggingface.co/papers → `DailyPapers.dailyPapers` → paper entities (arXiv id).
Identity (ontology `aiatlas.ontology.models`): a repository is either the official checkpoint of a MODEL (`Qwen/Qwen3-8B` → model
"Qwen3-8B", family hint "Qwen3") or an ARTIFACT of one — quantisation (`bartowski/Qwen3.8-27B-GGUF`, `zai-org/GLM-5-FP8`), conversion
(`mlx-community/Kimi-K2.5-bf16`, ONNX/CoreML repacks) or packaging (a converter organisation re-uploading the same weights). Artifacts keep the
full repo id as name and point to their canonical model through `EntityRef.canonical` (the `base_model` tag when present, otherwise the
analysed base name with medium identity confidence); the writer materialises `canonical_id` / `artifact_of`. Effort labels, gating and
licences are canonical: `license` is the ontology key (raw slug in `license_raw`), gating is `access: gated|open` (not an openness value),
`weights_available: true` for every hub repository and `openness` derived from the ontology dimensions.
Discovery is bounded: `max_targets` (default 1500) and `models_per_org`; at 20 requests/minute a 1 100-target run takes ~55 minutes
(`aia run huggingface --max-targets 700` stays under 40 minutes).
"""
from __future__ import annotations
import math
import re
from datetime import UTC, datetime
from typing import Any
from urllib.parse import quote
from aiatlas.connectors._identity import family_ref
from aiatlas.ontology.licenses import normalize_license
from aiatlas.ontology.models import CONVERTER_ORGS, PRECISION_FORMATS, QUANT_FORMATS, NameAnalysis, analyze_model_name
from aiatlas.ontology.openness import derive_openness, openness_dimensions
from aiatlas.ontology.taxonomy import normalize_modalities
from aiatlas.registry import org_by_hf, org_ref, organizations
from aiatlas.sdk.connector import BaseConnector, Parsed, RunContext
from aiatlas.sdk.extract.dates import parse_datetime
from aiatlas.sdk.extract.numbers import parse_active_params, parse_param_count
from aiatlas.sdk.facts import EntityRef, Facts, Target
from aiatlas.sdk.fetch import FetchResult
HF = "https://huggingface.co"
PER_PAGE = 30
QUANT_TAGS = ("gguf", "mlx", "awq", "gptq", "fp8", "exl2", "exl3", "bitsandbytes", "onnx", "compressed-tensors", "quantized")
GGUF_QUANT = re.compile(r"[-_.](IQ\d+_[A-Z0-9_]+|Q\d+_[A-Z0-9_]+|Q\d+|BF16|F16|F32|MXFP4)\.gguf$", re.IGNORECASE)
ORIGINAL_MODEL = re.compile(r"(?:Original|Base|Source) model:?\s*\[?(?:https?://huggingface\.co/)?([\w.-]+/[\w.-]+)", re.IGNORECASE)
BASE_MODEL_TAG = re.compile(r"^base_model:(?:(finetune|quantized|merge|adapter):)?([\w.-]+/[\w.-]+)$")
RELATION_BY_KIND = {"finetune": "fine_tuned_from", "quantized": "quantized_from", "merge": "merged_from", "adapter": "derived_from", None: "derived_from"}
NOISE_TAGS = {"endpoints_compatible", "text-generation-inference", "eval-results", "autotrain_compatible", "has_space", "conversational"}
# Organizations that re-publish other labs' weights (quantizations, mirrors). Their repos are artifacts and keep the full `org/name` as
# entity name so that `unsloth/Llama-3.1-8B-Instruct` never merges by alias into Meta's `Llama-3.1-8B-Instruct`.
REDISTRIBUTORS = {"bartowski", "unsloth", "mlx-community", "thebloke", "lmstudio-community", "quantfactory", "mradermacher", "nvidia-community",
"ggml-org", "second-state", "turboderp", "casperhansen", "hugging-quants", "neuralmagic", "redhatai"} | CONVERTER_ORGS
# Model families and the hub organizations that publish them: `NousResearch/Meta-Llama-3.1-70B-Instruct` is a mirror of Meta's repo,
# not Meta's entity, so it keeps its full id as name; the canonical model of a `bartowski/Qwen3.8-27B-GGUF` artifact belongs to Qwen.
FAMILY_ORGS = {r"^(meta-)?llama": {"meta-llama"}, r"^qwen|^qwq|^qvq": {"qwen"}, r"^deepseek": {"deepseek-ai"}, r"^(mistral|mixtral|magistral|devstral|codestral|ministral|pixtral|voxtral)": {"mistralai"},
r"^gemma|^paligemma|^shieldgemma|^medgemma": {"google"}, r"^phi-": {"microsoft"}, r"^glm|^chatglm|^cogview|^cogvideo": {"zai-org", "thudm"}, r"^kimi": {"moonshotai"},
r"^minimax": {"minimaxai"}, r"^gpt-oss": {"openai"}, r"^whisper": {"openai"}, r"^granite": {"ibm-granite"}, r"^(nvidia-)?nemotron": {"nvidia"},
r"^smollm|^smolvlm": {"huggingfacetb"}, r"^olmo|^molmo|^tulu": {"allenai"}, r"^(c4ai-)?command|^aya": {"coherelabs", "cohereforai"}, r"^flux": {"black-forest-labs"},
r"^stable-diffusion|^sdxl|^sd3": {"stabilityai"}, r"^claude": {"anthropic"}, r"^grok": {"xai-org"}, r"^seed-|^bagel": {"bytedance-seed"}, r"^hunyuan": {"tencent"},
r"^ernie": {"baidu"}, r"^jamba": {"ai21labs"}, r"^lfm": {"liquidai"}, r"^dbrx": {"databricks"}, r"^hermes": {"nousresearch"}}
_QUANT_TOKEN = re.compile(r"^(w\d+a\d+|q\d(_[a-z0-9]+)*|iq\d(_[a-z0-9]+)*|\d-?bit|int\d|fp\d|nvfp\d|mxfp\d|ud-q\d.*|bnb-\d+bit)$", re.IGNORECASE)
# pipeline tags without a "-to-" arrow → (input modalities, output modalities); `-to-` tags are split on the arrow
PIPELINE_MODALITIES: dict[str, tuple[list[str], list[str]]] = {
"text-generation": (["text"], ["text"]), "text2text-generation": (["text"], ["text"]), "fill-mask": (["text"], ["text"]), "translation": (["text"], ["text"]),
"summarization": (["text"], ["text"]), "question-answering": (["text"], ["text"]), "conversational": (["text"], ["text"]),
"text-classification": (["text"], ["structured"]), "token-classification": (["text"], ["structured"]), "zero-shot-classification": (["text"], ["structured"]),
"feature-extraction": (["text"], ["embedding"]), "sentence-similarity": (["text"], ["embedding"]), "image-feature-extraction": (["image"], ["embedding"]),
"automatic-speech-recognition": (["audio"], ["text"]), "audio-classification": (["audio"], ["structured"]), "text-to-speech": (["text"], ["audio"]),
"text-to-audio": (["text"], ["audio"]), "audio-to-audio": (["audio"], ["audio"]), "voice-activity-detection": (["audio"], ["structured"]),
"image-classification": (["image"], ["structured"]), "object-detection": (["image"], ["structured"]), "image-segmentation": (["image"], ["image"]),
"zero-shot-image-classification": (["image", "text"], ["structured"]), "zero-shot-object-detection": (["image", "text"], ["structured"]),
"depth-estimation": (["image"], ["image"]), "image-to-image": (["image"], ["image"]), "image-to-text": (["image"], ["text"]), "text-to-image": (["text"], ["image"]),
"text-to-video": (["text"], ["video"]), "image-to-video": (["image"], ["video"]), "video-classification": (["video"], ["structured"]), "video-text-to-text": (["video", "text"], ["text"]),
"image-text-to-text": (["image", "text"], ["text"]), "audio-text-to-text": (["audio", "text"], ["text"]), "visual-question-answering": (["image", "text"], ["text"]),
"document-question-answering": (["document", "text"], ["text"]), "text-to-3d": (["text"], ["3d"]), "image-to-3d": (["image"], ["3d"]), "robotics": (["image", "text"], ["action"]),
"reinforcement-learning": ([], ["action"]), "any-to-any": ([], []), "text-ranking": (["text"], ["structured"]), "mask-generation": (["image"], ["image"]),
"keypoint-detection": (["image"], ["structured"]), "unconditional-image-generation": ([], ["image"]), "tabular-classification": (["structured"], ["structured"]),
"tabular-regression": (["structured"], ["structured"]), "time-series-forecasting": (["structured"], ["structured"]), "graph-ml": (["structured"], ["structured"]),
}
class HuggingFaceConnector(BaseConnector):
name = "huggingface"
label = "Hugging Face Hub — models of the registry organizations, model cards, daily papers"
description = "Hub listing pages, model pages (embedded JSON), raw model cards and the daily-papers page — direct HTML, no API."
source_key = "huggingface.co"
version = "2"
parser_version = "2"
interval_seconds = 6 * 3600
min_interval_seconds = 3 * 3600
max_interval_seconds = 2 * 86400
rate_per_min = 20 # huggingface.co answers 429 at 30/min with 3 parallel fetches (observed 2026-09-11); 20/min × 2 is clean
tier = 2
priority = 1
expected_min_records = 200
concurrency = 2
def __init__(self, config: dict[str, Any] | None = None):
super().__init__(config)
self.config.setdefault("max_targets", 1500)
self.config.setdefault("models_per_org", 20)
self.config.setdefault("listing_sort", "downloads")
self.config.setdefault("fetch_readme", True)
# ------------------------------------------------------------------------------------------ discovery
def hf_orgs(self) -> list[str]:
if self.config.get("orgs"):
return list(self.config["orgs"])
return sorted({o["hf_org"] for o in organizations().values() if o.get("hf_org")}, key=str.lower)
async def discover(self, ctx: RunContext) -> list[Target]:
targets = [Target(url=f"{HF}/papers", doc_type="listing", key="papers", min_bytes=5000, priority=1)]
per_org = int(self.config["models_per_org"])
pages = max(1, math.ceil(per_org / PER_PAGE))
sort = self.config["listing_sort"]
for org in self.hf_orgs():
for p in range(pages):
targets.append(Target(url=f"{HF}/models?author={quote(org)}&sort={sort}&p={p}", doc_type="listing", key=f"listing:{org}:{p}",
meta={"hf_org": org, "page": p}, min_bytes=5000, priority=1))
return targets
# ------------------------------------------------------------------------------------------ extraction
async def extract(self, ctx: RunContext, target: Target, res: FetchResult, parsed: Parsed) -> Facts:
facts = Facts()
key = target.key or ""
if key == "papers" and parsed.html:
self._papers(facts, parsed)
elif key.startswith("listing:") and parsed.html:
self._listing(facts, target, parsed)
elif target.doc_type == "model_page" and parsed.html:
self._model_page(facts, target, parsed)
elif target.doc_type == "model_card" and parsed.markdown:
self._model_card(facts, target, parsed)
return facts
# ------------------------------------------------------------------------------------------ listing
def _listing(self, facts: Facts, target: Target, parsed: Parsed) -> None:
html = parsed.html
assert html
props = html.embedded_json.get("data-props:ModelList") or {}
models = ((props.get("initialValues") or {}).get("models")) or []
hf_org = target.meta.get("hf_org") or ""
author_data = next((m.get("authorData") for m in models if m.get("authorData")), None)
org = self._org_ref(facts, hf_org, author_data)
per_org = int(self.config["models_per_org"])
page = int(target.meta.get("page") or 0)
budget = per_org - page * PER_PAGE
for m in models[: max(0, budget)]:
repo_id = m.get("id")
if not repo_id or m.get("private"):
continue
ref = self._model_ref(facts, repo_id, org)
facts.claim(ref, "hf_repo", repo_id)
facts.claim(ref, "model_card_url", f"{HF}/{repo_id}")
self._pipeline_claims(facts, ref, m.get("pipeline_tag"))
facts.claim(ref, "metric.downloads", m.get("downloads"))
facts.claim(ref, "metric.likes", m.get("likes"))
facts.claim(ref, "last_modified", _iso(m.get("lastModified")))
gated = m.get("gated")
self._access_claims(facts, ref, gated)
if isinstance(m.get("numParameters"), int) and m["numParameters"] > 0:
facts.claim(ref, "parameter_count", m["numParameters"])
facts.follow(f"{HF}/{repo_id}", doc_type="model_page", entity=ref, key=f"model:{repo_id}", min_bytes=5000,
meta={"hf_repo": repo_id, "hf_org": hf_org, "gated": bool(gated), "num_parameters": m.get("numParameters")})
facts.document_entity = org
facts.document_title = f"Hugging Face models — {hf_org}"
# ------------------------------------------------------------------------------------------ model page
def _model_page(self, facts: Facts, target: Target, parsed: Parsed) -> None:
html = parsed.html
assert html
header = html.embedded_json.get("data-props:ModelHeader") or {}
model = header.get("model") or (html.embedded_json.get("data-props:ModelTensorsParams") or {}).get("model") or {}
repo_id = model.get("id") or target.meta.get("hf_repo")
if not repo_id:
return
author = model.get("author") or repo_id.split("/")[0]
org = self._org_ref(facts, author, header.get("author"))
card = model.get("cardData") or {}
config = model.get("config") or {}
tags = [t for t in (model.get("tags") or []) if isinstance(t, str)]
gated = model.get("gated")
bases = self._base_models(card, tags)
quant_format = next((q for q in QUANT_TAGS if q in tags and q != "quantized"), None)
if not quant_format and repo_id.lower().endswith(("-gguf", "_gguf")):
quant_format = "gguf"
if not quant_format and author.lower() == "mlx-community":
quant_format = "mlx"
ref = self._model_ref(facts, repo_id, org, bases=bases, quant_format=quant_format)
is_artifact = ref.entity_type == "artifact"
facts.claim(ref, "hf_repo", repo_id)
facts.claim(ref, "model_card_url", f"{HF}/{repo_id}")
self._pipeline_claims(facts, ref, model.get("pipeline_tag") or card.get("pipeline_tag"))
facts.claim(ref, "library_name", model.get("library_name") or card.get("library_name"))
license_ = card.get("license") or next((t.split(":", 1)[1] for t in tags if t.startswith("license:")), None)
if isinstance(license_, list):
license_ = license_[0] if license_ else None
license_key = self._license_claims(facts, ref, license_, card.get("license_name"))
facts.claim(ref, "license_url", card.get("license_link"))
self._access_claims(facts, ref, gated, license_key=license_key)
facts.claim(ref, "release_date", _date(model.get("createdAt")))
facts.claim(ref, "last_modified", _iso(model.get("lastModified")))
facts.claim(ref, "metric.downloads", model.get("downloads"))
facts.claim(ref, "metric.downloads_all_time", model.get("downloadsAllTime"))
facts.claim(ref, "metric.likes", model.get("likes"))
# parameters: the packaged size on artifacts, the safetensors count on official checkpoints (never on the canonical model of an artifact)
st = model.get("safetensors") or {}
total = st.get("total") if isinstance(st, dict) else None
listed = target.meta.get("num_parameters") # the hub's own count from the listing (GGUF repos have no safetensors)
params = total if isinstance(total, int) and total > 0 else listed if isinstance(listed, int) and listed > 0 else parse_param_count(repo_id.split("/")[-1])
facts.claim(ref, "parameter_count", params)
facts.claim(ref, "active_parameter_count", parse_active_params(repo_id.split("/")[-1]))
if isinstance(st, dict) and st.get("parameters"):
facts.claim(ref, "weights_dtype", sorted(st["parameters"]))
if isinstance(st, dict) and isinstance(st.get("totalFileSize"), int):
facts.claim(ref, "file_size_gb", round(st["totalFileSize"] / 1e9, 2), unit="GB")
archs = config.get("architectures") if isinstance(config, dict) else None
facts.claim(ref, "architecture", archs[0] if isinstance(archs, list) and archs else None)
facts.claim(ref, "model_type", config.get("model_type") if isinstance(config, dict) else None)
facts.claim(ref, "languages", _listify(card.get("language")))
datasets = _listify(card.get("datasets"))
facts.claim(ref, "datasets", datasets)
for ds in datasets[:20]:
if "/" in ds or re.fullmatch(r"[\w.-]+", ds):
facts.relate(ref, "uses_dataset", facts.entity("dataset", ds, identifiers={"hf_dataset": ds}))
clean_tags = [t for t in tags if ":" not in t and t not in NOISE_TAGS][:30]
facts.claim(ref, "tags", clean_tags)
for t in tags:
if t.startswith("arxiv:"):
paper = facts.entity("paper", f"arXiv:{t[6:]}", identifiers={"arxiv": t[6:]})
facts.relate(ref, "described_by", paper)
# quantization
facts.claim(ref, "quant_format", quant_format)
quants = sorted({m.group(1).upper() for href, _ in html.links for m in [GGUF_QUANT.search(href)] if m and "/blob/main/" in href})
if not quants and quant_format == "mlx":
m = re.search(r"-(\d)bit\b", repo_id, re.IGNORECASE)
quants = [f"{m.group(1)}bit"] if m else []
facts.claim(ref, "quantization", quants)
if quant_format or (is_artifact and ref.artifact_kind == "quantization"):
facts.claim(ref, "is_quantized", True)
if is_artifact:
facts.claim(ref, "artifact_kind", ref.artifact_kind)
# base model relations (artifact → model, fine-tune → base…)
for base_id, kind in bases:
base = self._base_ref(facts, base_id)
facts.relate(ref, RELATION_BY_KIND.get(kind, "derived_from"), base, attributes={"base_model_relation": kind} if kind else {})
facts.claim(ref, "base_model", [b for b, _ in bases] or None)
facts.claim(ref, "quantized_by", card.get("quantized_by"))
providers = [p.get("provider") for p in (model.get("availableInferenceProviders") or []) if isinstance(p, dict) and p.get("provider")]
facts.claim(ref, "hf_inference_providers", sorted(set(providers)))
facts.document_entity = ref
facts.document_title = repo_id
if self.config.get("fetch_readme", True) and not gated and model.get("cardExists", True):
facts.follow(f"{HF}/{repo_id}/raw/main/README.md", doc_type="model_card", entity=ref, key=f"card:{repo_id}", priority=3,
meta={"hf_repo": repo_id, "hf_org": author, "content_type": "text/markdown", "llm_task": "model_passport"}, min_bytes=32)
# ------------------------------------------------------------------------------------------ raw model card
def _model_card(self, facts: Facts, target: Target, parsed: Parsed) -> None:
md = parsed.markdown
assert md
repo_id = target.meta.get("hf_repo")
if not repo_id:
m = re.search(r"huggingface\.co/([\w.-]+/[\w.-]+)/raw/", target.url)
repo_id = m.group(1) if m else None
if not repo_id:
return
fm = md.front_matter or {}
bases = [(b, None) for b in _listify(fm.get("base_model")) if "/" in b]
quantized = bool(fm.get("quantized_by")) or "gguf" in [t.lower() for t in _listify(fm.get("tags"))]
if not bases:
m = ORIGINAL_MODEL.search(md.body)
if m and m.group(1).lower() != repo_id.lower():
bases = [(m.group(1), "quantized")]
quantized = True
org = self._org_ref(facts, repo_id.split("/")[0], None)
ref = target.entity or self._model_ref(facts, repo_id, org, bases=bases, quant_format="gguf" if quantized else None)
if ref not in facts.entities:
facts.entities.append(ref)
lic = fm.get("license")
self._license_claims(facts, ref, lic[0] if isinstance(lic, list) and lic else (lic if isinstance(lic, str) else None), fm.get("license_name"))
self._pipeline_claims(facts, ref, fm.get("pipeline_tag"))
facts.claim(ref, "library_name", fm.get("library_name"))
facts.claim(ref, "languages", _listify(fm.get("language")))
facts.claim(ref, "datasets", _listify(fm.get("datasets")))
facts.claim(ref, "quantized_by", fm.get("quantized_by"))
for base_id, _kind in bases:
base = self._base_ref(facts, base_id)
# the typed relation (finetune/quantized/merge/adapter) comes from the model page tags; the card only proves quantization
if quantized:
facts.relate(ref, "quantized_from", base)
facts.claim(ref, "base_model", [b for b, _ in bases] or None)
h1 = next((t for lvl, t in md.headings if lvl == 1), None)
facts.document_title = h1 or repo_id
facts.document_entity = ref
# ------------------------------------------------------------------------------------------ daily papers
def _papers(self, facts: Facts, parsed: Parsed) -> None:
html = parsed.html
assert html
props = html.embedded_json.get("data-props:DailyPapers") or {}
for item in props.get("dailyPapers") or []:
paper = item.get("paper") or {}
arxiv_id = paper.get("id")
title = (paper.get("title") or item.get("title") or "").strip()
if not arxiv_id or not title:
continue
published = parse_datetime(paper.get("publishedAt") or item.get("publishedAt"))
ref = facts.entity("paper", title[:300], identifiers={"arxiv": arxiv_id}, first_seen_hint=published)
facts.claim(ref, "arxiv_id", arxiv_id)
facts.claim(ref, "authors", [a.get("name") for a in paper.get("authors") or [] if isinstance(a, dict) and a.get("name")][:50])
facts.claim(ref, "abstract", (paper.get("summary") or "").strip()[:4000] or None)
facts.claim(ref, "published_at", published.isoformat(timespec="seconds") if published else None)
facts.claim(ref, "pdf_url", f"https://arxiv.org/pdf/{arxiv_id}")
facts.claim(ref, "official_url", f"https://arxiv.org/abs/{arxiv_id}")
facts.claim(ref, "hf_paper_url", f"{HF}/papers/{arxiv_id}")
facts.claim(ref, "metric.upvotes", paper.get("upvotes"))
facts.claim(ref, "metric.hf_comments", item.get("numComments"))
facts.claim(ref, "github_repo", _github_repo(paper.get("githubRepo")))
if paper.get("githubStars") is not None:
facts.claim(ref, "metric.github_stars", paper.get("githubStars"))
facts.document_title = f"Hugging Face daily papers — {props.get('dateString') or ''}".strip()
# ------------------------------------------------------------------------------------------ helpers
def _org_ref(self, facts: Facts, hf_org: str, author_data: dict[str, Any] | None) -> EntityRef:
known = org_by_hf(hf_org)
if known:
ref = org_ref(known["key"])
else:
display = (author_data or {}).get("fullname") or hf_org
kind = "company" if (author_data or {}).get("type") == "org" else "organization"
ref = EntityRef(entity_type=kind, name=display, identifiers={"hf_org": hf_org}, aliases=[hf_org])
facts.claim(ref, "hf_org", hf_org)
for e in facts.entities:
if e.entity_type == ref.entity_type and e.identifiers and e.identifiers == ref.identifiers:
return e
facts.entities.append(ref)
return ref
def _model_ref(self, facts: Facts, repo_id: str, org: EntityRef, *, bases: list[tuple[str, str | None]] | None = None,
quant_format: str | None = None) -> EntityRef:
"""Model or artifact EntityRef for a repository (one per repo inside a Facts)."""
for e in facts.entities:
if e.entity_type in ("model", "artifact") and e.identifiers.get("hf_repo") == repo_id:
return e
org_slug, _, repo_name = repo_id.partition("/")
analysis = analyze_model_name(repo_id)
kind = artifact_kind(repo_id, analysis, bases=bases or [], quant_format=quant_format)
if kind:
canonical = self._canonical_ref(facts, repo_id, analysis, bases or [])
ref = facts.entity("artifact", repo_id, identifiers={"hf_repo": repo_id}, organization=org, artifact_kind=kind, canonical=canonical,
identity_confidence="high" if any("/" in b for b, _ in (bases or [])) else "medium")
facts.relate(ref, "published_by", org)
return ref
foreign = _foreign_family(org_slug, repo_name)
name = repo_id if foreign else (repo_name or repo_id)
# an artifact seen earlier in the document may already have created this model by name (`Qwen/Qwen3.8-27B-FP8` before
# `Qwen/Qwen3.8-27B`): upgrade that ref in place instead of creating a second one
existing = next((e for e in facts.entities if e.entity_type == "model" and not e.identifiers and e.name.lower() == name.lower()
and (e.organization is None or e.organization.identifiers == org.identifiers)), None)
if existing is not None and not foreign:
existing.identifiers["hf_repo"] = repo_id
existing.organization = org
existing.aliases = [a for a in dict.fromkeys([*existing.aliases, repo_id]) if a != name]
existing.family = existing.family or family_ref(repo_name, org)
existing.identity_confidence = "high"
facts.relate(org, "develops", existing)
return existing
ref = facts.entity("model", name, identifiers={"hf_repo": repo_id}, organization=org, aliases=[repo_id] if not foreign else [],
family=family_ref(repo_name, org), identity_confidence="medium" if foreign else "high")
facts.relate(org, "develops", ref)
return ref
def _base_ref(self, facts: Facts, base_id: str) -> EntityRef:
"""The model behind a `base_model` tag: an official repository → model ref (no `develops` here; its own page states it)."""
for e in facts.entities:
if e.entity_type in ("model", "artifact") and e.identifiers.get("hf_repo") == base_id:
return e
base_org_slug, _, base_name = base_id.partition("/")
base_org = self._org_ref(facts, base_org_slug, None)
analysis = analyze_model_name(base_id)
if artifact_kind(base_id, analysis, bases=[], quant_format=None):
# a quantisation of a quantisation: the base is itself an artifact of the analysed model
ref = facts.entity("artifact", base_id, identifiers={"hf_repo": base_id}, organization=base_org,
artifact_kind=artifact_kind(base_id, analysis, bases=[], quant_format=None),
canonical=self._canonical_ref(facts, base_id, analysis, []), identity_confidence="medium")
else:
foreign = _foreign_family(base_org_slug, base_name)
ref = facts.entity("model", base_id if foreign else base_name, identifiers={"hf_repo": base_id}, organization=base_org,
aliases=[base_id] if not foreign else [], family=family_ref(base_name, base_org), identity_confidence="high")
facts.claim(ref, "hf_repo", base_id)
return ref
def _canonical_ref(self, facts: Facts, repo_id: str, analysis: NameAnalysis, bases: list[tuple[str, str | None]]) -> EntityRef | None:
"""The model an artifact packages: the `base_model` repo when present, otherwise the analysed base name (medium confidence) —
`nvidia/Gemma-4-31B-IT-NVFP4` → same-org model "Gemma-4-31B-IT". None when the base cannot be named confidently (no token to strip
from the name, or a redistributor repo whose family is unknown): never an invented model."""
base = next((b for b, k in bases if k == "quantized" and "/" in b), None) or next((b for b, _ in bases if "/" in b), None)
if base and base.lower() != repo_id.lower():
return self._base_ref(facts, base)
org_slug, _, repo_name = repo_id.partition("/")
name = canonical_name(repo_name)
if not name or name.lower() == repo_name.lower() or not re.search(r"[a-z]", name, re.IGNORECASE):
return None
org: EntityRef | None = None
family_orgs = _family_orgs(repo_name)
if family_orgs and org_slug.lower() in family_orgs:
org = self._org_ref(facts, org_slug, None)
elif family_orgs:
hf_org = next((o for o in sorted(family_orgs) if org_by_hf(o)), None)
org = self._org_ref(facts, org_by_hf(hf_org)["hf_org"], None) if hf_org else None # type: ignore[index]
elif org_slug.lower() not in REDISTRIBUTORS:
org = self._org_ref(facts, org_slug, None) # the official developer's own quantisation / conversion → its own model
if org is None:
return None
for e in facts.entities:
if e.entity_type == "model" and e.name.lower() == name.lower() and (org is None or e.organization is None or e.organization.identifiers == org.identifiers):
return e
ref = facts.entity("model", name, organization=org, family=family_ref(name, org), identity_confidence="medium")
return ref
@staticmethod
def _pipeline_claims(facts: Facts, ref: EntityRef, tag: Any) -> None:
if not isinstance(tag, str) or not tag:
return
facts.claim(ref, "pipeline_tag", tag)
mods_in, mods_out = pipeline_modalities(tag)
facts.claim(ref, "modalities_input", mods_in)
facts.claim(ref, "modalities_output", mods_out)
facts.claim(ref, "modalities", sorted(set(mods_in) | set(mods_out)))
@staticmethod
def _license_claims(facts: Facts, ref: EntityRef, raw: str | None, license_name: Any = None) -> str | None:
if not raw:
return None
key = normalize_license(raw)
if (key is None or key == "Other") and isinstance(license_name, str) and license_name.strip():
key = normalize_license(license_name) or key
facts.claim(ref, "license_name", license_name.strip())
facts.claim(ref, "license_raw", raw)
facts.claim(ref, "license", key or raw)
return key
@staticmethod
def _access_claims(facts: Facts, ref: EntityRef, gated: Any, *, license_key: str | None = None) -> None:
facts.claim(ref, "access", "gated" if gated else "open")
if isinstance(gated, str) and gated:
facts.claim(ref, "gated_mode", gated)
facts.claim(ref, "weights_available", True)
if license_key is not None: # the category needs the licence terms; weights alone only prove `weights_available`
dims = openness_dimensions(weights_available=True, license_key=license_key)
facts.claim(ref, "openness", derive_openness(dims, license_key=license_key))
@staticmethod
def _base_models(card: dict[str, Any], tags: list[str]) -> list[tuple[str, str | None]]:
kinds: dict[str, str | None] = {}
for t in tags:
m = BASE_MODEL_TAG.match(t)
if m:
kinds[m.group(2)] = m.group(1) or kinds.get(m.group(2))
for b in _listify(card.get("base_model")):
if "/" in b:
kinds.setdefault(b, None)
return list(kinds.items())
# ---------------------------------------------------------------------------------------------- identity helpers (pure)
def artifact_kind(repo_id: str, analysis: NameAnalysis | None = None, *, bases: list[tuple[str, str | None]], quant_format: str | None) -> str | None:
"""quantization | conversion | packaging for a repository that is not the official checkpoint of a model, else None."""
a = analysis or analyze_model_name(repo_id)
org_slug = repo_id.split("/")[0].lower() if "/" in repo_id else ""
if a.is_quantized or quant_format or any(k == "quantized" for _, k in bases):
return "quantization"
if a.quant_formats or a.precision:
return "conversion"
if org_slug in REDISTRIBUTORS:
return "packaging"
return None
def canonical_name(repo_name: str) -> str:
"""Repository name without quantisation / precision / packaging tokens, original casing kept: 'Qwen3.8-27B-GGUF' → 'Qwen3.8-27B',
'Llama-3.1-8B-Instruct-bnb-4bit' → 'Llama-3.1-8B-Instruct', 'Kimi-K2.5-bf16' → 'Kimi-K2.5'."""
kept: list[str] = []
for tok in repo_name.split("-"):
low = tok.lower()
if not tok or low in QUANT_FORMATS or low in PRECISION_FORMATS or _QUANT_TOKEN.match(low) or low == "ud" or re.match(r"^i?q\d", low):
continue
kept.append(tok)
return "-".join(kept).strip("-") or repo_name
def pipeline_modalities(tag: str) -> tuple[list[str], list[str]]:
"""Hub pipeline tag → (input modalities, output modalities), canonical vocabulary."""
low = tag.strip().lower()
if low in PIPELINE_MODALITIES:
i, o = PIPELINE_MODALITIES[low]
return normalize_modalities(i), normalize_modalities(o)
if "-to-" in low:
left, _, right = low.partition("-to-")
return normalize_modalities(left.split("-")), normalize_modalities(right.split("-"))
return [], []
def _family_orgs(repo_name: str) -> set[str] | None:
low = repo_name.lower()
for pattern, orgs in FAMILY_ORGS.items():
if re.search(pattern, low):
return {o.lower() for o in orgs}
return None
def _foreign_family(org_slug: str, repo_name: str) -> bool:
"""True when the repo name belongs to a well-known model family published by another hub organization (mirror / re-upload)."""
orgs = _family_orgs(repo_name)
return org_slug.lower() not in orgs if orgs else False
def _listify(v: Any) -> list[str]:
if v is None:
return []
if isinstance(v, str):
return [v] if v.strip() else []
if isinstance(v, list):
return [str(x) for x in v if isinstance(x, (str, int, float)) and str(x).strip()]
return []
def _iso(v: Any) -> str | None:
dt = parse_datetime(v) if isinstance(v, str) else None
return dt.astimezone(UTC).isoformat(timespec="seconds") if dt else None
def _date(v: Any) -> str | None:
dt = parse_datetime(v) if isinstance(v, str) else None
return dt.date().isoformat() if dt else None
def _github_repo(v: Any) -> str | None:
if not isinstance(v, str):
return None
m = re.search(r"github\.com/([\w.-]+/[\w.-]+)", v)
return m.group(1).removesuffix(".git") if m else None
def _now() -> datetime:
return datetime.now(UTC)
CONNECTORS = [HuggingFaceConnector]