unsloth/studio/backend/hub/services/models/gguf_variants.py
oobabooga 2b319e8d3a
Studio: support separate-file MTP GGUF drafters (Gemma 4) (#6125)
* Studio: support separate-file MTP GGUF drafters (Gemma 4)

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: fix review findings for separate-file MTP drafters

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: pair local MTP drafters by name and include them in reload dedup

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: manage --model-draft in extras and reject MTP/ copies as models

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
2026-06-10 08:45:12 -07:00

648 lines
25 KiB
Python

# SPDX-License-Identifier: AGPL-3.0-only
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
"""GGUF variant resolution."""
from __future__ import annotations
import asyncio
import threading
import time
from collections import OrderedDict
from typing import NamedTuple, Optional
from fastapi import HTTPException
from loggers import get_logger
from hub.schemas.inventory import GgufVariantDetail, GgufVariantsResponse
from hub.utils import download_manifest
from hub.utils import download_registry
from hub.utils import inventory_scan as hf_cache_scan
from hub.utils.hf_errors import hf_error_status
from hub.utils.hf_cache_state import (
INCOMPLETE_SUFFIX,
iter_destructive_repo_cache_dirs,
)
from hub.utils.gguf import (
extract_quant_label,
iter_hf_cache_snapshots,
list_gguf_variants,
list_gguf_variants_from_hf_cache,
list_local_gguf_variants,
list_partial_gguf_variants_from_state,
pick_best_gguf,
)
from hub.utils.paths import (
is_local_path,
is_valid_repo_id as _is_valid_repo_id,
)
from hub.services.models.common import (
_is_mmproj_filename,
_is_mtp_drafter_path,
_iter_gguf_paths,
)
from hub.utils.gguf_plan import (
GgufVariantPlan as _GgufVariantRequirement,
build_gguf_variant_plans,
is_main_gguf_variant_path,
)
logger = get_logger(__name__)
_VARIANT_HASH_CACHE: "OrderedDict[tuple[str, str, str, bool], tuple[frozenset[str], float]]" = (
OrderedDict()
)
_VARIANT_REQUIREMENT_CACHE: "OrderedDict[tuple[str, str, str], tuple[_GgufVariantRequirement, float]]" = OrderedDict()
_VARIANT_REQUIREMENT_NEG_CACHE: "OrderedDict[tuple[str, str], float]" = OrderedDict()
_VARIANT_HASH_MAX = 512
# Blob hashes are derived from the same mutable remote revision metadata as
# variant requirements, so they must not outlive that freshness window.
_VARIANT_HASH_POS_TTL = 60.0
# Refresh resolved variant requirements so a moved repo revision is picked up
# within the session instead of being pinned for the backend's lifetime.
_VARIANT_REQUIREMENT_POS_TTL = 60.0
# Suppress retries on a metadata-fetch failure so a slow/flaky link doesn't
# re-hammer the API on every page refresh.
_VARIANT_REQUIREMENT_NEG_TTL = 60.0
# Fail fast on a slow link so the variant render isn't blocked for seconds.
_GGUF_METADATA_TIMEOUT_SECONDS = 5.0
_VARIANT_HASH_LOCK = threading.Lock()
class VariantIncompleteDeleteResult(NamedTuple):
deleted: int
unresolved: bool
def _variant_hash_cache_key(
repo_id: str, variant: str, hf_token: Optional[str]
) -> tuple[str, str, str]:
return (
repo_id.lower(),
variant.lower(),
hf_cache_scan.token_fingerprint(hf_token),
)
def _variant_blob_hash_cache_key(
repo_id: str, variant: str, hf_token: Optional[str], include_companions: bool
) -> tuple[str, str, str, bool]:
base = _variant_hash_cache_key(repo_id, variant, hf_token)
return (*base, include_companions)
def _variant_repo_cache_key(repo_id: str, hf_token: Optional[str]) -> tuple[str, str]:
return (repo_id.lower(), hf_cache_scan.token_fingerprint(hf_token))
def _variant_requirement_neg_cache_active(key: tuple[str, str]) -> bool:
with _VARIANT_HASH_LOCK:
cached_at = _VARIANT_REQUIREMENT_NEG_CACHE.get(key)
if cached_at is None:
return False
if (time.monotonic() - cached_at) < _VARIANT_REQUIREMENT_NEG_TTL:
_VARIANT_REQUIREMENT_NEG_CACHE.move_to_end(key)
return True
_VARIANT_REQUIREMENT_NEG_CACHE.pop(key, None)
return False
def _variant_requirement_neg_cache_set(key: tuple[str, str]) -> None:
with _VARIANT_HASH_LOCK:
_VARIANT_REQUIREMENT_NEG_CACHE[key] = time.monotonic()
_VARIANT_REQUIREMENT_NEG_CACHE.move_to_end(key)
while len(_VARIANT_REQUIREMENT_NEG_CACHE) > _VARIANT_HASH_MAX:
_VARIANT_REQUIREMENT_NEG_CACHE.popitem(last = False)
def _variant_requirement_neg_cache_clear(key: tuple[str, str]) -> None:
with _VARIANT_HASH_LOCK:
_VARIANT_REQUIREMENT_NEG_CACHE.pop(key, None)
def _variant_hash_cache_get(key: tuple[str, str, str, bool]) -> Optional[frozenset[str]]:
with _VARIANT_HASH_LOCK:
cached = _VARIANT_HASH_CACHE.get(key)
if cached is None:
return None
hashes, ts = cached
if (time.monotonic() - ts) >= _VARIANT_HASH_POS_TTL:
_VARIANT_HASH_CACHE.pop(key, None)
return None
_VARIANT_HASH_CACHE.move_to_end(key)
return hashes
def _variant_hash_cache_set(key: tuple[str, str, str, bool], hashes: frozenset[str]) -> None:
with _VARIANT_HASH_LOCK:
_VARIANT_HASH_CACHE[key] = (hashes, time.monotonic())
_VARIANT_HASH_CACHE.move_to_end(key)
while len(_VARIANT_HASH_CACHE) > _VARIANT_HASH_MAX:
_VARIANT_HASH_CACHE.popitem(last = False)
def _variant_requirement_cache_get(key: tuple[str, str, str]) -> Optional[_GgufVariantRequirement]:
with _VARIANT_HASH_LOCK:
cached = _VARIANT_REQUIREMENT_CACHE.get(key)
if cached is None:
return None
requirement, ts = cached
if (time.monotonic() - ts) >= _VARIANT_REQUIREMENT_POS_TTL:
_VARIANT_REQUIREMENT_CACHE.pop(key, None)
return None
_VARIANT_REQUIREMENT_CACHE.move_to_end(key)
return requirement
def _variant_requirement_cache_set_many(
repo_id: str, hf_token: Optional[str], requirements: dict[str, _GgufVariantRequirement]
) -> None:
with _VARIANT_HASH_LOCK:
now = time.monotonic()
for quant, requirement in requirements.items():
key = _variant_hash_cache_key(repo_id, quant, hf_token)
_VARIANT_REQUIREMENT_CACHE[key] = (requirement, now)
_VARIANT_REQUIREMENT_CACHE.move_to_end(key)
while len(_VARIANT_REQUIREMENT_CACHE) > _VARIANT_HASH_MAX:
_VARIANT_REQUIREMENT_CACHE.popitem(last = False)
def _build_gguf_variant_requirements(siblings: list) -> dict[str, _GgufVariantRequirement]:
return build_gguf_variant_plans(siblings)
def gguf_variant_requirements(
repo_id: str,
variant: str,
hf_token: Optional[str] = None,
) -> Optional[_GgufVariantRequirement]:
key = _variant_hash_cache_key(repo_id, variant, hf_token)
cached = _variant_requirement_cache_get(key)
if cached is not None:
return cached
requirements = _fetch_gguf_variant_requirements(repo_id, hf_token)
return requirements.get(variant.lower())
def _fetch_gguf_variant_requirements(
repo_id: str,
hf_token: Optional[str] = None,
*,
siblings: Optional[list] = None,
) -> dict[str, _GgufVariantRequirement]:
repo_key = _variant_repo_cache_key(repo_id, hf_token)
if siblings is None:
if _variant_requirement_neg_cache_active(repo_key):
return {}
try:
from huggingface_hub import HfApi
info = HfApi(token = hf_token).model_info(
repo_id,
files_metadata = True,
timeout = _GGUF_METADATA_TIMEOUT_SECONDS,
)
except Exception as e:
logger.warning(
"model_info failed resolving GGUF files for %s: %s",
repo_id,
download_registry.scrub_secrets(str(e), hf_token = hf_token),
)
_variant_requirement_neg_cache_set(repo_key)
return {}
siblings = list(info.siblings)
requirements = _build_gguf_variant_requirements(siblings)
if requirements:
_variant_requirement_cache_set_many(repo_id, hf_token, requirements)
_variant_requirement_neg_cache_clear(repo_key)
return requirements
def _gguf_all_variant_requirements(
repo_id: str,
hf_token: Optional[str] = None,
*,
siblings: Optional[list] = None,
) -> dict[str, _GgufVariantRequirement]:
return _fetch_gguf_variant_requirements(repo_id, hf_token, siblings = siblings)
def _manifest_variant_blob_hashes(
repo_id: str,
variant: str,
*,
include_companions: bool = True,
) -> frozenset[str]:
manifest = download_manifest.read_manifest("model", repo_id, variant)
if manifest is None:
return frozenset()
variant_key = variant.lower()
hashes: set[str] = set()
for expected in manifest.expected_files:
if not expected.sha256:
continue
if include_companions:
hashes.add(expected.sha256)
continue
if is_main_gguf_variant_path(expected.path, variant_key):
hashes.add(expected.sha256)
return frozenset(hashes)
def gguf_variant_blob_hashes(
repo_id: str,
variant: str,
hf_token: Optional[str] = None,
*,
include_companions: bool = True,
allow_remote: bool = True,
) -> frozenset[str]:
key = _variant_blob_hash_cache_key(
repo_id,
variant,
hf_token,
include_companions,
)
cached = _variant_hash_cache_get(key)
if cached is not None:
return cached
hashes = _manifest_variant_blob_hashes(
repo_id,
variant,
include_companions = include_companions,
)
if hashes:
_variant_hash_cache_set(key, hashes)
return hashes
requirement_key = _variant_hash_cache_key(repo_id, variant, hf_token)
requirement = _variant_requirement_cache_get(requirement_key)
if requirement is None and allow_remote:
requirement = gguf_variant_requirements(repo_id, variant, hf_token)
if requirement is not None:
hashes = requirement.required_hashes if include_companions else requirement.main_hashes
if hashes:
_variant_hash_cache_set(key, hashes)
return hashes
return frozenset()
def _partial_transport_for_variant(repo_id: str, variant: str) -> Optional[str]:
return hf_cache_scan.partial_transport_for("model", repo_id, variant)
def delete_variant_incomplete_blobs_result(
repo_id: str,
variant: str,
hf_token: Optional[str],
*,
extra_hashes: frozenset[str] = frozenset(),
companions: bool = True,
) -> VariantIncompleteDeleteResult:
# With a sibling still downloading, ``companions=False`` keeps a shared mmproj
# from being unlinked out from under it; the repo's last delete reclaims it.
target_hashes = (
gguf_variant_blob_hashes(repo_id, variant, hf_token, include_companions = companions)
| extra_hashes
)
if not target_hashes:
has_variant_partial_state = hf_cache_scan.is_variant_partial(
repo_id,
variant,
incomplete_blob_hashes = set(),
variant_blob_hashes = frozenset(),
)
has_repo_partials = bool(download_registry.incomplete_blob_hashes("model", repo_id))
return VariantIncompleteDeleteResult(
deleted = 0,
unresolved = has_variant_partial_state and has_repo_partials,
)
deleted = 0
# Destructive iterator: only the exact-case match (or abort if ambiguous),
# so a case-variant sibling repo's partials are never unlinked.
for entry in iter_destructive_repo_cache_dirs("model", repo_id):
blobs_dir = entry / "blobs"
if not blobs_dir.is_dir():
continue
for h in target_hashes:
incomplete = blobs_dir / f"{h}{INCOMPLETE_SUFFIX}"
if incomplete.exists():
try:
incomplete.unlink()
deleted += 1
except OSError as e:
logger.warning(f"Failed to unlink {incomplete}: {e}")
return VariantIncompleteDeleteResult(deleted = deleted, unresolved = False)
async def get_gguf_variants_response(
repo_id: str,
prefer_local_cache: bool = False,
offline: bool = False,
local_path: Optional[str] = None,
hf_token: Optional[str] = None,
):
"""
List available GGUF quantization variants for a HuggingFace repo
or a local directory (e.g. LM Studio model folder).
Returns all available quantization variants (Q4_K_M, Q8_0, BF16, etc.)
with file sizes, whether the model supports vision, and the recommended
default variant.
"""
def _compute() -> GgufVariantsResponse:
def _local_response(
response_repo_id: str, variants, has_vision: bool
) -> GgufVariantsResponse:
filenames = [v.filename for v in variants]
best = pick_best_gguf(filenames)
default_variant = extract_quant_label(best) if best else None
return GgufVariantsResponse(
repo_id = response_repo_id,
variants = [
GgufVariantDetail(
filename = v.filename,
quant = v.quant,
display_label = v.display_label,
size_bytes = v.size_bytes,
download_size_bytes = v.size_bytes,
downloaded = True,
)
for v in variants
],
has_vision = has_vision,
default_variant = default_variant,
)
def _partial_local_response(
response_repo_id: str, variants, has_vision: bool
) -> GgufVariantsResponse:
filenames = [v.filename for v in variants]
best = pick_best_gguf(filenames)
default_variant = extract_quant_label(best) if best else None
return GgufVariantsResponse(
repo_id = response_repo_id,
variants = [
GgufVariantDetail(
filename = v.filename,
quant = v.quant,
display_label = v.display_label,
size_bytes = v.size_bytes,
download_size_bytes = v.download_size_bytes or v.size_bytes,
downloaded = False,
partial = True,
partial_transport = _partial_transport_for_variant(
response_repo_id,
v.quant,
),
)
for v in variants
],
has_vision = has_vision,
default_variant = default_variant,
)
# Local directory path (e.g. LM Studio models) — scan filesystem
if is_local_path(repo_id):
variants, has_vision = list_local_gguf_variants(repo_id)
return _local_response(repo_id, variants, has_vision)
# Reject invalid remote repo_ids up front (like download/delete) so a
# malformed id returns 400 instead of a 500 from the HF client.
if not _is_valid_repo_id(repo_id):
raise HTTPException(status_code = 400, detail = f"Invalid repo_id: {repo_id!r}")
local_only = prefer_local_cache or offline
if local_only:
cached = list_gguf_variants_from_hf_cache(repo_id)
if cached is not None:
variants, has_vision = cached
return _local_response(repo_id, variants, has_vision)
if local_path and is_local_path(local_path):
variants, has_vision = list_local_gguf_variants(local_path)
if variants or has_vision:
return _local_response(repo_id, variants, has_vision)
partial = list_partial_gguf_variants_from_state(repo_id)
if partial is not None:
variants, has_vision = partial
return _partial_local_response(repo_id, variants, has_vision)
if local_path and offline:
return GgufVariantsResponse(
repo_id = repo_id,
variants = [],
has_vision = False,
default_variant = None,
)
if offline:
raise HTTPException(
status_code = 404,
detail = "No cached GGUF variants available while offline.",
)
try:
variants, has_vision, siblings = list_gguf_variants(repo_id, hf_token = hf_token)
except Exception:
cached = list_gguf_variants_from_hf_cache(repo_id)
if cached is not None:
variants, has_vision = cached
return _local_response(repo_id, variants, has_vision)
partial = list_partial_gguf_variants_from_state(repo_id)
if partial is not None:
variants, has_vision = partial
return _partial_local_response(repo_id, variants, has_vision)
raise
filenames = [v.filename for v in variants]
best = pick_best_gguf(filenames)
default_variant = extract_quant_label(best) if best else None
# Per-snapshot accounting: a variant counts as present only when one
# snapshot holds all its files (split GGUFs need every shard together),
# sizes are max across snapshots so shared blobs aren't double-counted,
# and keys are lowercased since cache dir casing can differ from repo_id.
cached_filenames_by_snapshot: list[dict[str, int]] = []
cached_quant_bytes_by_snapshot: list[dict[str, int]] = []
if _is_valid_repo_id(repo_id):
for snap in iter_hf_cache_snapshots(repo_id):
try:
gguf_paths = list(_iter_gguf_paths(snap))
except (OSError, RuntimeError, ValueError) as e:
logger.debug("Skipping GGUF cache snapshot %s: %s", snap, e)
continue
by_filename: dict[str, int] = {}
by_quant: dict[str, int] = {}
for f in gguf_paths:
try:
rel = f.relative_to(snap).as_posix()
size = f.stat().st_size
except (OSError, RuntimeError, ValueError) as e:
logger.debug("Skipping GGUF cache file %s: %s", f, e)
continue
key = rel.lower()
by_filename[key] = max(by_filename.get(key, 0), size)
if _is_mmproj_filename(f.name) or _is_mtp_drafter_path(rel):
continue
q = extract_quant_label(rel).lower()
by_quant[q] = by_quant.get(q, 0) + size
if by_filename:
cached_filenames_by_snapshot.append(by_filename)
if by_quant:
cached_quant_bytes_by_snapshot.append(by_quant)
requirements_by_quant = {
v.quant.lower(): _variant_requirement_cache_get(
_variant_hash_cache_key(repo_id, v.quant, hf_token)
)
for v in variants
}
if any(req is None for req in requirements_by_quant.values()):
fetched_requirements = _gguf_all_variant_requirements(
repo_id, hf_token, siblings = siblings
)
for v in variants:
key = v.quant.lower()
if requirements_by_quant.get(key) is None:
requirements_by_quant[key] = fetched_requirements.get(key)
def _filenames_cached(filenames: frozenset[str], expected_size: int) -> bool:
if not filenames:
return False
wanted = [name.lower() for name in filenames]
# All files must live in a single snapshot, not spread across several.
for by_filename in cached_filenames_by_snapshot:
cached = 0
for name in wanted:
size = by_filename.get(name)
if size is None:
break
cached += size
else:
return expected_size <= 0 or cached >= expected_size * 0.99
return False
def _any_mmproj_cached(filenames: frozenset[str]) -> bool:
return any(
by_filename.get(name.lower()) is not None
for by_filename in cached_filenames_by_snapshot
for name in filenames
)
def _is_fully_downloaded(variant) -> bool:
requirement = requirements_by_quant.get(variant.quant.lower())
if requirement is None:
if variant.size_bytes == 0:
return False
quant = variant.quant.lower()
# Allow small rounding tolerance (symlinks vs real sizes).
return any(
by_quant.get(quant, 0) >= variant.size_bytes * 0.99
for by_quant in cached_quant_bytes_by_snapshot
)
if not _filenames_cached(
requirement.main_filenames,
requirement.main_size_bytes,
):
return False
# Vision repos ship an mmproj adapter per variant. Any mmproj
# precision on disk suffices (the loader picks whichever is present);
# requiring the API-preferred one would falsely demote variants.
if requirement.mmproj_filenames and not _any_mmproj_cached(
requirement.mmproj_filenames,
):
return False
return True
partial_quants: set[str] = set()
partial_quant_transports: dict[str, Optional[str]] = {}
try:
incomplete_hashes = download_registry.incomplete_blob_hashes("model", repo_id)
except Exception as e:
logger.warning(f"Failed to compute partial GGUF variants for {repo_id}: {e}")
incomplete_hashes = set()
scan_snapshot_dir = hf_cache_scan.resolve_snapshot_dir_for_scan("model", repo_id)
# Manifest + marker + main incomplete-blob check: catches variants whose
# download was cancelled or whose expected shards are missing/undersized.
for variant in variants:
try:
requirement = requirements_by_quant.get(variant.quant.lower())
variant_hashes = requirement.main_hashes if requirement is not None else None
if variant_hashes is None and incomplete_hashes:
variant_hashes = gguf_variant_blob_hashes(
repo_id,
variant.quant,
hf_token,
include_companions = False,
)
if hf_cache_scan.is_variant_partial(
repo_id,
variant.quant,
scan_snapshot_dir,
incomplete_blob_hashes = incomplete_hashes,
variant_blob_hashes = variant_hashes,
):
partial_quants.add(variant.quant)
partial_quant_transports[variant.quant] = _partial_transport_for_variant(
repo_id,
variant.quant,
)
except Exception as e:
logger.warning(
f"Manifest-based partial check failed for " f"{repo_id}/{variant.quant}: {e}"
)
if incomplete_hashes:
for variant in variants:
requirement = requirements_by_quant.get(variant.quant.lower())
if requirement is None:
continue
# companion_hashes adds the MTP drafter (mmproj_hashes covers
# every mmproj precision in the repo, not just the planned one).
if (
(requirement.mmproj_hashes | requirement.companion_hashes) & incomplete_hashes
) and _filenames_cached(
requirement.main_filenames,
requirement.main_size_bytes,
):
partial_quants.add(variant.quant)
partial_quant_transports.setdefault(
variant.quant,
_partial_transport_for_variant(repo_id, variant.quant),
)
def _variant_detail(v) -> GgufVariantDetail:
is_partial = v.quant in partial_quants
requirement = requirements_by_quant.get(v.quant.lower())
return GgufVariantDetail(
filename = v.filename,
quant = v.quant,
display_label = v.display_label,
size_bytes = v.size_bytes,
download_size_bytes = (
requirement.download_size_bytes if requirement is not None else v.size_bytes
),
downloaded = _is_fully_downloaded(v) and not is_partial,
partial = is_partial,
partial_transport = (partial_quant_transports.get(v.quant) if is_partial else None),
)
return GgufVariantsResponse(
repo_id = repo_id,
variants = [_variant_detail(v) for v in variants],
has_vision = has_vision,
default_variant = default_variant,
)
try:
return await asyncio.to_thread(_compute)
except HTTPException:
raise
except Exception as e:
scrubbed = download_registry.scrub_secrets(str(e), hf_token = hf_token)
# Client-side HF error (missing repo, gated, bad token): pass the status through.
status = hf_error_status(e)
if status is not None:
raise HTTPException(status_code = status, detail = scrubbed)
logger.error("Error listing GGUF variants for %s: %s", repo_id, scrubbed)
raise HTTPException(
status_code = 500,
detail = "Failed to list GGUF variants: " + scrubbed,
)