1839 lines
74 KiB
Python
1839 lines
74 KiB
Python
#!/usr/bin/env python3
|
|
"""List Ollama's sha256 model store in readable form, and quantize a model to NVFP4.
|
|
|
|
Ollama keeps weights in blobs/sha256-<digest> and names them through JSON manifests.
|
|
`list` and `show` resolve those hashes. `list` prints a short capability
|
|
code for each model. `list --full-scan` also reads GGUF headers, which is
|
|
slower, and uses them to find encoders and capabilities that the manifest
|
|
does not name. `list --ctx-scan` reads each parameter blob and names the
|
|
models whose Modelfile sets num_ctx. `--info MODEL` runs gguf_info.py on
|
|
that model's sha256 weight blob. Arguments after the model name are passed
|
|
through, so `--info MODEL --metadata` is gguf_info.py on the blob with
|
|
`--metadata`. `export` writes the original GGUF and a Modelfile into a
|
|
directory. `quant` does the same after fitting NVFP4, and can
|
|
register the result as a new Ollama model. The Modelfile keeps the source
|
|
template, stop list, system prompt, renderer, and parser, which is what
|
|
Ollama uses for thinking and for stopping generation.
|
|
|
|
Most Ollama blobs are already quantized (Q4_K_M, Q8_0, and so on). NVFP4 is then fit
|
|
to those dequantized weights, so it cannot recover detail the source quant discarded.
|
|
A float16 or BF16 blob is the better source.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import shutil
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
log = logging.getLogger("ollama_nvfp4")
|
|
|
|
MODEL_MEDIA = "application/vnd.ollama.image.model"
|
|
CONFIG_MEDIA = "application/vnd.docker.container.image.v1+json"
|
|
TEMPLATE_MEDIA = "application/vnd.ollama.image.template"
|
|
MESSAGES_MEDIA = "application/vnd.ollama.image.messages"
|
|
ADAPTER_MEDIA = "application/vnd.ollama.image.adapter"
|
|
PROJECTOR_MEDIA = "application/vnd.ollama.image.projector"
|
|
EMBED_MEDIA = "application/vnd.ollama.image.embed"
|
|
PARAMS_MEDIA = "application/vnd.ollama.image.params"
|
|
LAYER_KIND = {
|
|
TEMPLATE_MEDIA: "TEMPLATE",
|
|
"application/vnd.ollama.image.prompt": "TEMPLATE",
|
|
"application/vnd.ollama.image.system": "SYSTEM",
|
|
"application/vnd.ollama.image.license": "LICENSE",
|
|
PARAMS_MEDIA: "PARAMETER",
|
|
MESSAGES_MEDIA: "MESSAGE",
|
|
PROJECTOR_MEDIA: "PROJECTOR",
|
|
ADAPTER_MEDIA: "ADAPTER",
|
|
}
|
|
FLOAT_TYPES = {"F32", "F16", "BF16"}
|
|
|
|
|
|
def default_models_dir() -> Path:
|
|
env = os.environ.get("OLLAMA_MODELS")
|
|
if env:
|
|
return Path(env)
|
|
system = Path("/models/ollama")
|
|
if (system / "blobs").is_dir():
|
|
return system
|
|
return Path.home() / ".ollama" / "models"
|
|
|
|
|
|
def digest_to_name(digest: str) -> str:
|
|
return digest.replace(":", "-") if digest.startswith("sha256:") else f"sha256-{digest}"
|
|
|
|
|
|
def blob_path(root: Path, digest: str) -> Path:
|
|
return root / "blobs" / digest_to_name(digest)
|
|
|
|
|
|
def format_size(num_bytes: int) -> str:
|
|
size = float(num_bytes)
|
|
for unit in ("B", "KiB", "MiB", "GiB", "TiB"):
|
|
if size < 1024 or unit == "TiB":
|
|
return f"{int(size)} {unit}" if unit == "B" else f"{size:.1f} {unit}"
|
|
size /= 1024
|
|
return f"{size:.1f} TiB"
|
|
|
|
|
|
def sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def read_json(path: Path) -> dict:
|
|
with path.open("r", encoding="utf-8") as handle:
|
|
data = json.load(handle)
|
|
if not isinstance(data, dict):
|
|
raise ValueError(f"{path} is not a JSON object")
|
|
return data
|
|
|
|
|
|
@dataclass
|
|
class OllamaModel:
|
|
name: str
|
|
manifest_path: Path
|
|
store: str
|
|
config_digest: str
|
|
config: dict
|
|
layers: list[dict]
|
|
model_digest: str
|
|
model_size: int
|
|
gguf_path: Path
|
|
metadata: dict = field(default_factory=dict)
|
|
runner: str = ""
|
|
manifest_format: str = ""
|
|
|
|
@property
|
|
def family(self) -> str:
|
|
families = self.config.get("model_families") or []
|
|
if families:
|
|
return str(families[0])
|
|
return str(self.config.get("model_family") or "")
|
|
|
|
@property
|
|
def params(self) -> str:
|
|
return str(self.config.get("model_type") or self.metadata.get("general.size_label") or "")
|
|
|
|
@property
|
|
def quant(self) -> str:
|
|
if self.config.get("file_type"):
|
|
return str(self.config["file_type"])
|
|
file_type = self.metadata.get("general.file_type")
|
|
return "" if file_type is None else str(file_type)
|
|
|
|
@property
|
|
def context(self) -> str:
|
|
found = _context_from_metadata(self.metadata)
|
|
if found:
|
|
return found
|
|
# Ollama's metadata cache is often mode 600 and owned by root. The GGUF
|
|
# header is readable and is what /api/show uses.
|
|
return _context_from_gguf(self.gguf_path)
|
|
|
|
|
|
def _is_model_context_key(name: str) -> bool:
|
|
return name.endswith(".context_length") and ".rope." not in name
|
|
|
|
|
|
def _scalar_context(value) -> int | None:
|
|
if isinstance(value, bool) or isinstance(value, (list, dict)):
|
|
return None
|
|
try:
|
|
length = int(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
if length <= 0:
|
|
return None
|
|
return length
|
|
|
|
|
|
def _context_from_metadata(metadata: dict) -> str:
|
|
for key, value in metadata.items():
|
|
if not _is_model_context_key(key):
|
|
continue
|
|
length = _scalar_context(value)
|
|
if length is not None:
|
|
return str(length)
|
|
return ""
|
|
|
|
|
|
def _context_from_gguf(path: Path) -> str:
|
|
"""Read only the GGUF header. GGUFReader maps the whole weight file."""
|
|
if not path.is_file():
|
|
return ""
|
|
scalar = {0: 1, 1: 1, 2: 2, 3: 2, 4: 4, 5: 4, 6: 4, 7: 1, 10: 8, 11: 8, 12: 8}
|
|
integer = {4: "<I", 5: "<i", 10: "<Q", 11: "<q"}
|
|
|
|
def read(n: int) -> bytes:
|
|
data = handle.read(n)
|
|
if len(data) != n:
|
|
raise ValueError("truncated GGUF header")
|
|
return data
|
|
|
|
def read_string() -> bytes:
|
|
size = struct.unpack("<Q", read(8))[0]
|
|
if size > 16 * 1024 * 1024:
|
|
raise ValueError("GGUF string is too long")
|
|
return read(size)
|
|
|
|
def skip(value_type: int) -> None:
|
|
if value_type in scalar:
|
|
read(scalar[value_type])
|
|
return
|
|
if value_type == 8:
|
|
read_string()
|
|
return
|
|
if value_type != 9:
|
|
raise ValueError("unknown GGUF value type")
|
|
subtype = struct.unpack("<I", read(4))[0]
|
|
count = struct.unpack("<Q", read(8))[0]
|
|
if subtype in scalar:
|
|
handle.seek(scalar[subtype] * count, os.SEEK_CUR)
|
|
return
|
|
if subtype == 8:
|
|
for _ in range(count):
|
|
read_string()
|
|
return
|
|
raise ValueError("unsupported GGUF array")
|
|
|
|
try:
|
|
with path.open("rb") as handle:
|
|
magic, version = struct.unpack("<II", read(8))
|
|
if magic != 0x46554747 or version < 2:
|
|
return ""
|
|
_tensor_count, kv_count = struct.unpack("<QQ", read(16))
|
|
for _ in range(kv_count):
|
|
key = read_string().decode("utf-8", "replace")
|
|
value_type = struct.unpack("<I", read(4))[0]
|
|
if _is_model_context_key(key) and value_type in integer:
|
|
length = _scalar_context(struct.unpack(integer[value_type], read(scalar[value_type]))[0])
|
|
if length is not None:
|
|
return str(length)
|
|
skip(value_type)
|
|
except (OSError, ValueError, struct.error, UnicodeError):
|
|
return ""
|
|
return ""
|
|
|
|
|
|
def manifest_name(root: Path, manifest: Path) -> tuple[str, str]:
|
|
for store in ("manifests", "manifests-v2"):
|
|
base = root / store
|
|
try:
|
|
parts = manifest.relative_to(base).parts
|
|
except ValueError:
|
|
continue
|
|
if len(parts) < 3:
|
|
break
|
|
tag = parts[-1]
|
|
model = parts[-2]
|
|
namespace = parts[-3]
|
|
name = f"{model}:{tag}" if namespace == "library" else f"{namespace}/{model}:{tag}"
|
|
return name, store
|
|
return manifest.name, "loose"
|
|
|
|
|
|
def iter_manifests(root: Path):
|
|
for store in ("manifests", "manifests-v2"):
|
|
base = root / store
|
|
if not base.is_dir():
|
|
continue
|
|
for path in sorted(base.rglob("*")):
|
|
if path.is_file():
|
|
yield path
|
|
|
|
|
|
def load_models(root: Path) -> list[OllamaModel]:
|
|
models: list[OllamaModel] = []
|
|
for manifest_path in iter_manifests(root):
|
|
try:
|
|
manifest = read_json(manifest_path)
|
|
except (OSError, json.JSONDecodeError, ValueError):
|
|
continue
|
|
layers = manifest.get("layers") or []
|
|
model_layer = next((layer for layer in layers if layer.get("mediaType") == MODEL_MEDIA), None)
|
|
config_info = manifest.get("config") or {}
|
|
if model_layer is None or "digest" not in config_info:
|
|
continue
|
|
config_path = blob_path(root, config_info["digest"])
|
|
try:
|
|
config = read_json(config_path) if config_path.is_file() else {}
|
|
except (OSError, json.JSONDecodeError, ValueError):
|
|
config = {}
|
|
meta_path = root / "metadata" / digest_to_name(model_layer["digest"])
|
|
meta_file = meta_path.with_suffix(".json")
|
|
metadata = {}
|
|
if meta_file.is_file():
|
|
try:
|
|
metadata = read_json(meta_file).get("kv") or {}
|
|
except (OSError, json.JSONDecodeError, ValueError):
|
|
metadata = {}
|
|
name, store = manifest_name(root, manifest_path)
|
|
models.append(OllamaModel(
|
|
name=name,
|
|
manifest_path=manifest_path,
|
|
store=store,
|
|
config_digest=config_info["digest"],
|
|
config=config,
|
|
layers=layers,
|
|
model_digest=model_layer["digest"],
|
|
model_size=int(model_layer.get("size") or 0),
|
|
gguf_path=blob_path(root, model_layer["digest"]),
|
|
metadata=metadata,
|
|
runner=str(manifest.get("runner") or ""),
|
|
manifest_format=str(manifest.get("format") or ""),
|
|
))
|
|
models.sort(key=lambda item: item.name.lower())
|
|
return models
|
|
|
|
|
|
def referenced_digests(root: Path) -> set[str]:
|
|
found: set[str] = set()
|
|
for manifest_path in iter_manifests(root):
|
|
try:
|
|
manifest = read_json(manifest_path)
|
|
except (OSError, json.JSONDecodeError, ValueError):
|
|
continue
|
|
config = manifest.get("config") or {}
|
|
if isinstance(config.get("digest"), str):
|
|
found.add(config["digest"])
|
|
for layer in manifest.get("layers") or []:
|
|
if isinstance(layer.get("digest"), str):
|
|
found.add(layer["digest"])
|
|
return found
|
|
|
|
|
|
def resolve_model(models: list[OllamaModel], query: str) -> OllamaModel:
|
|
exact = [model for model in models if model.name == query]
|
|
if len(exact) == 1:
|
|
return exact[0]
|
|
if len(exact) > 1:
|
|
raise SystemExit("More than one model is named " + query + ":\n" + "\n".join(model.manifest_path.as_posix() for model in exact))
|
|
folded = query.lower()
|
|
partial = [model for model in models if folded in model.name.lower() or folded in model.model_digest.lower()]
|
|
if len(partial) == 1:
|
|
return partial[0]
|
|
if not partial:
|
|
raise SystemExit(f"No Ollama model matches {query!r}. Run list to see names.")
|
|
preview = "\n".join(f" {model.name}" for model in partial[:20])
|
|
raise SystemExit(f"{query!r} matches {len(partial)} models:\n{preview}")
|
|
|
|
|
|
# Letters match Ollama's capability set. X is text because T is tool calls.
|
|
CAPABILITY_LETTERS = (
|
|
("completion", "X", "text"),
|
|
("tools", "T", "tool calls"),
|
|
("vision", "V", "vision (images and video)"),
|
|
("audio", "A", "audio input"),
|
|
("image", "I", "image generation"),
|
|
("embedding", "E", "embeddings"),
|
|
("thinking", "R", "thinking"),
|
|
("insert", "M", "infill (fill-in-the-middle)"),
|
|
("decision", "D", "decision (classification, not chat)"),
|
|
("speech", "S", "speech output"),
|
|
)
|
|
# Families and architectures that always need a vision encoder. Gemma 3 is
|
|
# not here: its 270M and 1B builds are text, and the larger ones are not.
|
|
_VISION_IDS = {
|
|
"qwen2vl", "qwen25vl", "qwen2.5vl", "qwen3vl", "qwen3vlmoe",
|
|
"llama4", "mllama", "llava", "bakllava", "moondream",
|
|
"minicpm-v", "minicpmv", "mistral3", "pixtral", "internvl",
|
|
"cogvlm", "deepseek2-ocr", "glmocr", "glm-ocr", "gemma3n",
|
|
"hunyuan_vl", "paddleocr", "sensenova_u1", "llama3.2-vision",
|
|
}
|
|
_EMBED_IDS = {
|
|
"nomic-bert", "nomic-bert-moe", "modern-bert", "neo-bert",
|
|
"jina-bert-v2", "jina-bert-v3", "eurobert",
|
|
"gemma-embedding", "gemma-embedding2", "llama-embed", "pangu-embedded",
|
|
}
|
|
_SPEECH_ARCH = {"qwen3tts", "pockettts", "talkie", "wavtokenizer-dec"}
|
|
_THINKING_PARSERS = {
|
|
"qwen3-thinking", "qwen3-vl-thinking", "deepseek3", "gemma4",
|
|
"lfm2-thinking", "olmo3-think", "harmony",
|
|
}
|
|
_TOOL_PARSERS = {"qwen3-coder", "functiongemma", "harmony"}
|
|
_AUDIO_PROJECTORS = {
|
|
"qwen2a", "qwen3a", "glma", "voxtral", "voxtral_rt", "meralion",
|
|
"lfm2a", "musicflamingo", "granite_speech", "mimo_audio", "parakeet",
|
|
"dots3note_a", "d1omni_a", "muse-spark-audio", "ultravox",
|
|
"qwen2.5o", "nemotron_omni",
|
|
}
|
|
_SPEECH_PROJECTORS = {
|
|
"qwen3tts_spkenc", "qwen3tts_gen", "pockettts_spkenc", "pockettts_gen", "soprano",
|
|
}
|
|
_VISION_PROJECTORS = {
|
|
"gemma3", "gemma3nv", "gemma3na", "gemma4v", "gemma4a", "gemma4uv", "gemma4ua",
|
|
"phi4", "idefics3", "pixtral", "llama4", "qwen2vl_merger", "qwen2.5vl_merger",
|
|
"exaone4_5", "qwen3vl_merger", "ling3vl", "step3vl", "internvl", "lfm2",
|
|
"lfm3vl", "kimivl", "paddleocr", "kimik25", "lightonocr", "cogvlm",
|
|
"janus_pro", "dots_ocr", "dots3note_v", "deepseekocr", "deepseekocr2",
|
|
"deepseek4v", "deepseek41v", "glm4v", "glm5v", "youtuvl", "nemotron_v2_vl",
|
|
"nemotron_omni", "qwen2.5o", "hunyuanvl", "minimax_m3", "minicpmv4_6",
|
|
"minicpmv4_7", "minicpm_ocr_1_0", "mimovl", "granite4_vision", "muse-glimmer",
|
|
"cohere2v", "d1omni_v",
|
|
}
|
|
_TEMPLATE_MEDIAS = {
|
|
TEMPLATE_MEDIA,
|
|
"application/vnd.ollama.image.prompt",
|
|
}
|
|
_GEMMA3_VISION_SIZE = re.compile(r"(^|[^0-9.])(4(?:\.3)?|12|27)\s*b\b")
|
|
# Shared chat templates often contain an image or audio branch even for a text
|
|
# checkpoint. A real requirement comes from the name, the architecture, or a
|
|
# projector the manifest actually lists.
|
|
_NAME_VISION = re.compile(r"(?:^|[^a-z0-9])(?:vl|vision|ocr)(?:[^a-z0-9]|$)")
|
|
_NAME_AUDIO = re.compile(r"(?:^|[^a-z0-9])audio(?:[^a-z0-9]|$)")
|
|
_GGUF_CACHE: dict[tuple[str, int, int], "GGUFProfile"] = {}
|
|
|
|
|
|
@dataclass
|
|
class GGUFProfile:
|
|
context: str = ""
|
|
architecture: str = ""
|
|
general_name: str = ""
|
|
general_type: str = ""
|
|
pooling: bool = False
|
|
decision: bool = False
|
|
vision_blocks: bool = False
|
|
audio_blocks: bool = False
|
|
has_vision_encoder: bool | None = None
|
|
has_audio_encoder: bool | None = None
|
|
projector_type: str = ""
|
|
image_tokens: bool = False
|
|
video_tokens: bool = False
|
|
audio_tokens: bool = False
|
|
templates: list[str] = field(default_factory=list)
|
|
vision_tensors: bool = False
|
|
audio_tensors: bool = False
|
|
speech_tensors: bool = False
|
|
readable: bool = False
|
|
|
|
|
|
@dataclass
|
|
class CapabilityReport:
|
|
marks: list[tuple[str, str, bool]] = field(default_factory=list)
|
|
notes: list[str] = field(default_factory=list)
|
|
projector: str = "-"
|
|
context: str = ""
|
|
|
|
def letters(self) -> str:
|
|
return " ".join(letter if ok else f"{letter}!" for letter, _label, ok in self.marks)
|
|
|
|
def column(self) -> str:
|
|
text = self.letters()
|
|
if self.notes:
|
|
return f"{text} {'; '.join(self.notes)}" if text else "; ".join(self.notes)
|
|
return text
|
|
|
|
def labels(self) -> str:
|
|
rendered = []
|
|
for _letter, label, ok in self.marks:
|
|
rendered.append(label if ok else f"{label}, weights missing")
|
|
return ", ".join(rendered)
|
|
|
|
|
|
def _gguf_cache_key(path: Path) -> tuple[str, int, int] | None:
|
|
try:
|
|
stat = path.stat()
|
|
except OSError:
|
|
return None
|
|
return (str(path), stat.st_mtime_ns, stat.st_size)
|
|
|
|
|
|
def _note_gguf_key(profile: GGUFProfile, key: str, value) -> None:
|
|
low = key.lower()
|
|
if low == "general.architecture" and isinstance(value, str):
|
|
profile.architecture = value
|
|
elif low == "general.name" and isinstance(value, str):
|
|
profile.general_name = value
|
|
elif low == "general.type" and isinstance(value, str):
|
|
profile.general_type = value
|
|
elif low.endswith("pooling_type"):
|
|
profile.pooling = True
|
|
elif low == "decision.type" or low.endswith(".decision.type"):
|
|
profile.decision = isinstance(value, str) and bool(value.strip())
|
|
elif low.endswith("vision.block_count"):
|
|
profile.vision_blocks = True
|
|
elif low.endswith("audio.block_count"):
|
|
profile.audio_blocks = True
|
|
elif low.endswith("has_vision_encoder") and isinstance(value, int):
|
|
profile.has_vision_encoder = value != 0
|
|
elif low.endswith("has_audio_encoder") and isinstance(value, int):
|
|
profile.has_audio_encoder = value != 0
|
|
elif low.endswith("projector_type") and isinstance(value, str):
|
|
profile.projector_type = value
|
|
elif low.endswith("image_token_id"):
|
|
profile.image_tokens = True
|
|
elif low.endswith("video_token_id"):
|
|
profile.video_tokens = True
|
|
elif low.endswith("audio_token_id"):
|
|
profile.audio_tokens = True
|
|
elif "chat_template" in low and isinstance(value, str) and value:
|
|
profile.templates.append(value[:500_000])
|
|
elif _is_model_context_key(key):
|
|
length = _scalar_context(value)
|
|
if length is not None and not profile.context:
|
|
profile.context = str(length)
|
|
|
|
|
|
def _note_tensor_name(profile: GGUFProfile, name: str) -> None:
|
|
low = name.lower()
|
|
head, _, rest = low.partition(".")
|
|
if head in {"v", "vis", "vision", "mm", "mmproj"}:
|
|
profile.vision_tensors = True
|
|
elif head in {"audio", "gen_audio"} or low.startswith("a.blk"):
|
|
profile.audio_tensors = True
|
|
elif head in {"tts", "spk", "speaker", "codec"} or rest.startswith("spk"):
|
|
profile.speech_tensors = True
|
|
|
|
|
|
def _scan_gguf(path: Path) -> GGUFProfile:
|
|
"""Read a GGUF header for capability metadata. Does not map the weights."""
|
|
key = _gguf_cache_key(path)
|
|
if key is None:
|
|
return GGUFProfile()
|
|
cached = _GGUF_CACHE.get(key)
|
|
if cached is not None:
|
|
return cached
|
|
profile = _scan_gguf_uncached(path)
|
|
_GGUF_CACHE[key] = profile
|
|
return profile
|
|
|
|
|
|
def _scan_gguf_uncached(path: Path) -> GGUFProfile:
|
|
profile = GGUFProfile()
|
|
scalar = {0: 1, 1: 1, 2: 2, 3: 2, 4: 4, 5: 4, 6: 4, 7: 1, 10: 8, 11: 8, 12: 8}
|
|
integer = {0: "<B", 1: "<b", 2: "<H", 3: "<h", 4: "<I", 5: "<i", 7: "<?", 10: "<Q", 11: "<q"}
|
|
|
|
def read(n: int) -> bytes:
|
|
data = handle.read(n)
|
|
if len(data) != n:
|
|
raise ValueError("truncated GGUF header")
|
|
return data
|
|
|
|
def read_string() -> bytes:
|
|
size = struct.unpack("<Q", read(8))[0]
|
|
if size > 16 * 1024 * 1024:
|
|
raise ValueError("GGUF string is too long")
|
|
return read(size)
|
|
|
|
def read_scalar(value_type: int):
|
|
if value_type == 8:
|
|
return read_string().decode("utf-8", "replace")
|
|
if value_type not in scalar:
|
|
raise ValueError("unknown GGUF value type")
|
|
raw = read(scalar[value_type])
|
|
if value_type in integer:
|
|
return struct.unpack(integer[value_type], raw)[0]
|
|
return None
|
|
|
|
def skip_array(subtype: int, count: int, keep_strings: bool) -> None:
|
|
if subtype in scalar:
|
|
handle.seek(scalar[subtype] * count, os.SEEK_CUR)
|
|
return
|
|
if subtype != 8:
|
|
raise ValueError("unsupported GGUF array")
|
|
kept = 0
|
|
for _ in range(count):
|
|
text = read_string()
|
|
if keep_strings and kept < 4:
|
|
profile.templates.append(text.decode("utf-8", "replace")[:500_000])
|
|
kept += 1
|
|
|
|
try:
|
|
with path.open("rb") as handle:
|
|
magic, version = struct.unpack("<II", read(8))
|
|
if magic != 0x46554747 or version < 2:
|
|
return profile
|
|
tensor_count, kv_count = struct.unpack("<QQ", read(16))
|
|
for _ in range(kv_count):
|
|
key = read_string().decode("utf-8", "replace")
|
|
value_type = struct.unpack("<I", read(4))[0]
|
|
if value_type == 9:
|
|
subtype = struct.unpack("<I", read(4))[0]
|
|
count = struct.unpack("<Q", read(8))[0]
|
|
skip_array(subtype, count, "chat_template" in key.lower())
|
|
continue
|
|
_note_gguf_key(profile, key, read_scalar(value_type))
|
|
for _ in range(tensor_count):
|
|
name = read_string().decode("utf-8", "replace")
|
|
n_dims = struct.unpack("<I", read(4))[0]
|
|
if n_dims:
|
|
read(8 * n_dims)
|
|
read(4)
|
|
read(8)
|
|
_note_tensor_name(profile, name)
|
|
profile.readable = True
|
|
except (OSError, ValueError, struct.error, UnicodeError):
|
|
return profile
|
|
return profile
|
|
|
|
|
|
def _config_capabilities(config: dict) -> set[str]:
|
|
raw = config.get("capabilities") or []
|
|
if isinstance(raw, str):
|
|
raw = [raw]
|
|
if not isinstance(raw, list):
|
|
return set()
|
|
return {str(item).strip().lower() for item in raw if str(item).strip()}
|
|
|
|
|
|
def _model_ids(model: OllamaModel, main: GGUFProfile) -> set[str]:
|
|
found = set()
|
|
if model.family:
|
|
found.add(model.family.lower())
|
|
for item in model.config.get("model_families") or []:
|
|
found.add(str(item).lower())
|
|
if main.architecture:
|
|
found.add(main.architecture.lower())
|
|
return found
|
|
|
|
|
|
def _template_capabilities(text: str) -> set[str]:
|
|
if not text:
|
|
return set()
|
|
low = text.lower()
|
|
found: set[str] = set()
|
|
if "tools" in low or "tool_call" in low or ".tools" in low:
|
|
found.add("tools")
|
|
if "<think>" in low or "</think>" in low or "enable_thinking" in low or "<|think|>" in low:
|
|
found.add("thinking")
|
|
if ".suffix" in low:
|
|
found.add("insert")
|
|
return found
|
|
|
|
|
|
def _name_modalities(text: str) -> set[str]:
|
|
"""Vision or audio claimed by the model name, not by a shared chat template."""
|
|
low = text.lower()
|
|
found: set[str] = set()
|
|
if _NAME_VISION.search(low):
|
|
found.add("vision")
|
|
if _NAME_AUDIO.search(low):
|
|
found.add("audio")
|
|
return found
|
|
|
|
|
|
def _blob_text(path: Path, limit: int = 2_000_000) -> str:
|
|
if not path.is_file():
|
|
return ""
|
|
try:
|
|
with path.open("r", encoding="utf-8", errors="replace") as handle:
|
|
return handle.read(limit)
|
|
except OSError:
|
|
return ""
|
|
|
|
|
|
def _looks_like_embedding(model: OllamaModel) -> bool:
|
|
if _model_ids(model, GGUFProfile()) & _EMBED_IDS:
|
|
return True
|
|
return "embed" in f"{model.name} {model.family}".lower()
|
|
|
|
|
|
def _gemma3_expects_vision(model: OllamaModel, main: GGUFProfile, ids: set[str]) -> bool:
|
|
if "gemma3n" in ids:
|
|
return False
|
|
if "gemma3" not in ids and "gemma3" not in model.name.lower():
|
|
return False
|
|
text = " ".join((model.params, model.name, main.general_name))
|
|
return _GEMMA3_VISION_SIZE.search(text.lower()) is not None
|
|
|
|
|
|
def _projector_modes(profile: GGUFProfile) -> tuple[bool, bool, bool]:
|
|
"""Return vision, audio, speech for one projector header."""
|
|
ptype = (profile.projector_type or "").lower()
|
|
speech = ptype in _SPEECH_PROJECTORS or profile.speech_tensors
|
|
audio = bool(
|
|
profile.has_audio_encoder
|
|
or profile.audio_blocks
|
|
or profile.audio_tensors
|
|
or profile.audio_tokens
|
|
or ptype in _AUDIO_PROJECTORS
|
|
)
|
|
if ptype == "gemma3nv":
|
|
audio = False
|
|
if profile.has_vision_encoder is True:
|
|
vision = True
|
|
elif profile.has_vision_encoder is False:
|
|
vision = False
|
|
elif profile.vision_blocks or profile.vision_tensors or profile.image_tokens or profile.video_tokens:
|
|
vision = True
|
|
elif ptype in _VISION_PROJECTORS:
|
|
vision = True
|
|
elif audio or speech:
|
|
vision = False
|
|
else:
|
|
vision = True
|
|
return vision, audio, speech
|
|
|
|
|
|
def assess_capabilities(
|
|
model: OllamaModel,
|
|
*,
|
|
full: bool,
|
|
weights_present: bool,
|
|
main: GGUFProfile,
|
|
projectors: list[tuple[bool, GGUFProfile | None]],
|
|
templates: list[str],
|
|
) -> CapabilityReport:
|
|
"""Decide which capabilities a model has, and which are missing their weights.
|
|
|
|
`full` is the GGUF header scan. Without it, only the manifest, config,
|
|
parser, and small template blobs are used, so a packed-in encoder stays
|
|
invisible and a missing sidecar is reported only when the manifest names it.
|
|
"""
|
|
explicit = _config_capabilities(model.config)
|
|
ids = _model_ids(model, main if full else GGUFProfile())
|
|
scanned = full and main.readable
|
|
present_projectors = [(ok, profile) for ok, profile in projectors if ok and profile is not None]
|
|
missing_projector = any(not ok for ok, _profile in projectors)
|
|
readable_modes = [_projector_modes(profile) for _ok, profile in present_projectors if profile.readable]
|
|
# A projector file that is on disk but unreadable still counts as present.
|
|
opaque_projector = any(ok and (profile is None or not profile.readable) for ok, profile in projectors)
|
|
|
|
wanted: set[str] = set(explicit)
|
|
template_found: set[str] = set()
|
|
for text in templates:
|
|
template_found |= _template_capabilities(text)
|
|
if scanned:
|
|
for text in main.templates:
|
|
template_found |= _template_capabilities(text)
|
|
wanted |= template_found
|
|
wanted |= _name_modalities(model.name)
|
|
if scanned:
|
|
wanted |= _name_modalities(main.general_name)
|
|
|
|
parser = str(model.config.get("parser") or "").lower()
|
|
renderer = str(model.config.get("renderer") or "").lower()
|
|
if parser in _TOOL_PARSERS or renderer in _TOOL_PARSERS:
|
|
wanted.add("tools")
|
|
if parser in _THINKING_PARSERS or renderer in _THINKING_PARSERS or ids & {"gptoss", "gpt-oss"}:
|
|
wanted.add("thinking")
|
|
|
|
embed = "embedding" in explicit or (scanned and main.pooling) or (not scanned and _looks_like_embedding(model))
|
|
decision = "decision" in explicit or (scanned and main.decision)
|
|
if embed:
|
|
wanted.add("embedding")
|
|
if decision:
|
|
wanted.add("decision")
|
|
if "completion" in explicit or (not embed and not decision):
|
|
wanted.add("completion")
|
|
|
|
vision_inline = scanned and main.vision_tensors
|
|
audio_inline = scanned and main.audio_tensors
|
|
speech_inline = scanned and main.speech_tensors
|
|
projector_vision = any(vision for vision, _audio, _speech in readable_modes) or opaque_projector
|
|
projector_audio = any(audio for _vision, audio, _speech in readable_modes)
|
|
projector_speech = any(speech for _vision, _audio, speech in readable_modes)
|
|
|
|
if projector_vision or missing_projector:
|
|
wanted.add("vision")
|
|
if scanned and (main.vision_blocks or main.image_tokens or main.video_tokens or vision_inline):
|
|
wanted.add("vision")
|
|
if scanned and (ids & _VISION_IDS or _gemma3_expects_vision(model, main, ids)):
|
|
wanted.add("vision")
|
|
if projector_audio or (scanned and (main.audio_blocks or main.audio_tokens or audio_inline)):
|
|
wanted.add("audio")
|
|
if "image" in explicit or (scanned and main.general_type.lower() in {"image", "flux", "sdxl"}):
|
|
wanted.add("image")
|
|
if projector_speech or speech_inline or (scanned and main.architecture.lower() in _SPEECH_ARCH):
|
|
wanted.add("speech")
|
|
if "speech" in explicit:
|
|
wanted.add("speech")
|
|
|
|
if decision:
|
|
wanted.difference_update({"completion", "tools", "thinking", "insert"})
|
|
|
|
# Audio sidecars are separate files. A name that says audio, with no projector
|
|
# on disk, is missing its encoder even before the GGUF header is read.
|
|
# Vision is often packed into the model file, so a vision name stays
|
|
# unmarked until --full-scan looks for those tensors.
|
|
name_claims_audio = "audio" in _name_modalities(model.name) or (
|
|
scanned and "audio" in _name_modalities(main.general_name)
|
|
)
|
|
|
|
have = set(wanted)
|
|
notes: list[str] = []
|
|
if not weights_present:
|
|
have.clear()
|
|
notes.append("model blob missing")
|
|
else:
|
|
# Without a header scan, declared vision can be packed into the model
|
|
# file, so it is left unmarked. A projector blob the manifest names and
|
|
# cannot find is broken either way. A model whose name says audio is
|
|
# broken when that sidecar is absent, because the audio encoder is not
|
|
# packed into the language file.
|
|
if "vision" in wanted and not (vision_inline or projector_vision):
|
|
if scanned or missing_projector:
|
|
have.discard("vision")
|
|
if missing_projector:
|
|
notes.append("mmproj blob missing")
|
|
elif main.vision_blocks and not main.vision_tensors:
|
|
notes.append("vision encoder missing from the model file")
|
|
else:
|
|
notes.append("no mmproj")
|
|
if "audio" in wanted and not (audio_inline or projector_audio):
|
|
if scanned or missing_projector or name_claims_audio:
|
|
have.discard("audio")
|
|
if missing_projector and "mmproj blob missing" not in notes:
|
|
notes.append("mmproj blob missing")
|
|
elif scanned and main.audio_blocks and not main.audio_tensors:
|
|
notes.append("audio encoder missing from the model file")
|
|
elif "no mmproj" not in notes and "mmproj blob missing" not in notes:
|
|
notes.append("no audio mmproj")
|
|
if "speech" in wanted and not (speech_inline or projector_speech or opaque_projector):
|
|
needs_sidecar = scanned and main.architecture.lower() in _SPEECH_ARCH and not speech_inline
|
|
if needs_sidecar or missing_projector:
|
|
have.discard("speech")
|
|
if missing_projector and "mmproj blob missing" not in notes:
|
|
notes.append("mmproj blob missing")
|
|
elif "no mmproj" not in notes and "mmproj blob missing" not in notes:
|
|
notes.append("no speech mmproj")
|
|
|
|
broken = [name for name in ("vision", "audio", "speech") if name in wanted and name not in have]
|
|
if missing_projector and not broken and "mmproj blob missing" not in notes:
|
|
notes.append("mmproj blob missing")
|
|
|
|
present_count = sum(1 for ok, _profile in projectors if ok)
|
|
missing_count = len(projectors) - present_count
|
|
inline = vision_inline or audio_inline or speech_inline
|
|
if missing_count and present_count:
|
|
projector = f"yes, {missing_count} MISSING"
|
|
elif missing_count:
|
|
projector = "MISSING"
|
|
elif present_count > 1:
|
|
projector = f"yes x{present_count}"
|
|
elif present_count == 1:
|
|
projector = "yes+inline" if inline else "yes"
|
|
elif inline:
|
|
projector = "inline"
|
|
else:
|
|
projector = "-"
|
|
|
|
marks = []
|
|
for name, letter, label in CAPABILITY_LETTERS:
|
|
if name not in wanted:
|
|
continue
|
|
marks.append((letter, label, name in have and weights_present))
|
|
return CapabilityReport(marks=marks, notes=notes, projector=projector, context=main.context)
|
|
|
|
|
|
def capability_report(root: Path, model: OllamaModel, full: bool) -> CapabilityReport:
|
|
weights_present = model.gguf_path.is_file()
|
|
projectors: list[tuple[bool, GGUFProfile | None]] = []
|
|
templates: list[str] = []
|
|
try:
|
|
for layer in model.layers:
|
|
media = str(layer.get("mediaType") or "")
|
|
digest = str(layer.get("digest") or "")
|
|
if not digest:
|
|
continue
|
|
path = blob_path(root, digest)
|
|
if media == PROJECTOR_MEDIA:
|
|
profile = _scan_gguf(path) if full and path.is_file() else None
|
|
projectors.append((path.is_file(), profile))
|
|
elif media in _TEMPLATE_MEDIAS:
|
|
templates.append(_blob_text(path))
|
|
main = _scan_gguf(model.gguf_path) if full and weights_present else GGUFProfile()
|
|
return assess_capabilities(
|
|
model,
|
|
full=full,
|
|
weights_present=weights_present,
|
|
main=main,
|
|
projectors=projectors,
|
|
templates=templates,
|
|
)
|
|
except (OSError, ValueError):
|
|
return CapabilityReport(notes=["could not read capabilities"])
|
|
|
|
|
|
def _report_context(model: OllamaModel, report: CapabilityReport, full: bool) -> str:
|
|
found = _context_from_metadata(model.metadata)
|
|
if found:
|
|
return found
|
|
if full and report.context:
|
|
return report.context
|
|
return _context_from_gguf(model.gguf_path)
|
|
|
|
|
|
def print_capability_legend(full: bool) -> None:
|
|
print()
|
|
print("CAPS " + " ".join(f"{letter} {label}" for _name, letter, label in CAPABILITY_LETTERS))
|
|
print("A letter with ! has that capability, but the file that implements it is absent.")
|
|
print("MMPROJ yes = projector blob, inline = encoder is inside the model file, MISSING = the manifest names a blob that is absent, - = no projector file.")
|
|
if full:
|
|
print("Each GGUF header was read. An encoder packed into the model file shows up as inline. A capability whose encoder is absent is marked !.")
|
|
else:
|
|
print("Caps here come from the manifest, config, template blobs, and the model name.")
|
|
print("A model with audio in its name is marked A! when it has no projector. Vision from the name is left unmarked until --full-scan, because that encoder is sometimes inside the model file.")
|
|
print("--full-scan reads each GGUF header, which walks the vocabulary, and can add tools, thinking, and encoders stored only in that file.")
|
|
|
|
|
|
def modelfile_num_ctx(root: Path, model: OllamaModel) -> tuple[str, str]:
|
|
"""Read PARAMETER num_ctx from the model's parameter blob.
|
|
|
|
The first item is the value to print: the number, '-' when it is unset,
|
|
MISSING when the blob is gone, or '?' when the blob is not JSON.
|
|
The second item is '' unless this model needs the num_ctx removed.
|
|
"""
|
|
saw_params = False
|
|
value = None
|
|
for layer in model.layers:
|
|
if layer.get("mediaType") != PARAMS_MEDIA:
|
|
continue
|
|
digest = str(layer.get("digest") or "")
|
|
if not digest:
|
|
continue
|
|
saw_params = True
|
|
path = blob_path(root, digest)
|
|
if not path.is_file():
|
|
return "MISSING", ""
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, UnicodeError, json.JSONDecodeError):
|
|
return "?", ""
|
|
if isinstance(data, dict) and "num_ctx" in data:
|
|
value = data["num_ctx"]
|
|
if not saw_params or value is None:
|
|
return "-", ""
|
|
rendered = value if isinstance(value, str) else json.dumps(value, separators=(",", ":"))
|
|
return rendered, rendered
|
|
|
|
|
|
def cmd_list(root: Path, full_scan: bool = False, ctx_scan: bool = False) -> None:
|
|
models = load_models(root)
|
|
if not models:
|
|
print(f"No manifests in {root}")
|
|
return
|
|
rows = []
|
|
needs_fix: list[tuple[str, str]] = []
|
|
for model in models:
|
|
if full_scan:
|
|
print(f"Reading {model.name}...", file=sys.stderr)
|
|
report = capability_report(root, model, full_scan)
|
|
present = "MISSING" if not model.gguf_path.is_file() else format_size(model.model_size)
|
|
num_ctx = "-"
|
|
if ctx_scan:
|
|
num_ctx, fix = modelfile_num_ctx(root, model)
|
|
if fix:
|
|
needs_fix.append((model.name, fix))
|
|
rows.append((model, present, _report_context(model, report, full_scan), report, num_ctx))
|
|
name_width = max(len(model.name) for model, _present, _context, _report, _num_ctx in rows)
|
|
ctx_width = max([len("CTX")] + [len(context) for _model, _present, context, _report, _num_ctx in rows])
|
|
proj_width = max([len("MMPROJ")] + [len(report.projector) for _model, _present, _context, report, _num_ctx in rows])
|
|
num_width = max([len("NUM_CTX")] + [len(num_ctx) for *_rest, num_ctx in rows]) if ctx_scan else 0
|
|
header = (
|
|
f"{'NAME':<{name_width}} {'PARAMS':<8} {'QUANT':<12} {'FAMILY':<14} "
|
|
f"{'SIZE':>10} {'CTX':<{ctx_width}} {'MMPROJ':<{proj_width}} "
|
|
)
|
|
if ctx_scan:
|
|
header += f"{'NUM_CTX':<{num_width}} "
|
|
header += "CAPS"
|
|
print(header)
|
|
print("-" * len(header))
|
|
for model, present, context, report, num_ctx in rows:
|
|
line = (
|
|
f"{model.name:<{name_width}} {model.params:<8} {model.quant:<12} "
|
|
f"{model.family:<14} {present:>10} {context:<{ctx_width}} "
|
|
f"{report.projector:<{proj_width}} "
|
|
)
|
|
if ctx_scan:
|
|
line += f"{num_ctx:<{num_width}} "
|
|
line += report.column()
|
|
print(line)
|
|
blobs = list((root / "blobs").glob("sha256-*")) if (root / "blobs").is_dir() else []
|
|
referenced = referenced_digests(root)
|
|
orphan = [path for path in blobs if "sha256:" + path.name.removeprefix("sha256-") not in referenced]
|
|
orphan_bytes = sum(path.stat().st_size for path in orphan)
|
|
print()
|
|
print(f"{len(models)} models, {len(blobs)} blobs, {len(orphan)} blobs not referenced by a manifest ({format_size(orphan_bytes)})")
|
|
if ctx_scan:
|
|
print()
|
|
if needs_fix:
|
|
noun = "model" if len(needs_fix) == 1 else "models"
|
|
print(f"{len(needs_fix)} {noun} set num_ctx in the Modelfile and need reinstalling without it:")
|
|
for name, value in needs_fix:
|
|
print(f" {name} {value}")
|
|
else:
|
|
print("No Modelfile sets num_ctx.")
|
|
print_capability_legend(full_scan)
|
|
|
|
|
|
def cmd_show(root: Path, query: str) -> None:
|
|
model = resolve_model(load_models(root), query)
|
|
report = capability_report(root, model, True)
|
|
print(model.name)
|
|
print(f" store {model.store}")
|
|
print(f" manifest {model.manifest_path}")
|
|
print(f" family {model.family}")
|
|
print(f" params {model.params}")
|
|
print(f" quant {model.quant}")
|
|
print(f" context {_report_context(model, report, True) or '-'}")
|
|
print(f" caps {report.letters() or '-'}")
|
|
if report.labels():
|
|
print(f" cap detail {report.labels()}")
|
|
if report.notes:
|
|
print(f" broken {'; '.join(report.notes)}")
|
|
print(f" projector {report.projector}")
|
|
if model.config.get("renderer"):
|
|
print(f" renderer {model.config['renderer']}")
|
|
if model.config.get("parser"):
|
|
print(f" parser {model.config['parser']}")
|
|
if model.config.get("requires"):
|
|
print(f" requires {model.config['requires']}")
|
|
architecture = model.metadata.get("general.architecture", "")
|
|
if architecture:
|
|
print(f" arch {architecture}")
|
|
for suffix in ("block_count", "embedding_length", "expert_count", "vocab_size"):
|
|
key = f"{architecture}.{suffix}"
|
|
if key in model.metadata and not isinstance(model.metadata[key], list):
|
|
print(f" {suffix:<18} {model.metadata[key]}")
|
|
print(f" weights {model.model_digest}")
|
|
print(f" file {model.gguf_path} ({format_size(model.model_size)})")
|
|
print(" layers")
|
|
for layer in model.layers:
|
|
kind = str(layer.get("mediaType", "")).removeprefix("application/vnd.ollama.image.")
|
|
digest = str(layer.get("digest", ""))
|
|
short = digest[:19] + "…" if len(digest) > 19 else digest
|
|
path = blob_path(root, digest)
|
|
state = "ok" if path.is_file() else "MISSING"
|
|
print(f" {kind:<16} {format_size(int(layer.get('size') or 0)):>10} {short} {state}")
|
|
|
|
|
|
def _pythonize(value):
|
|
import numpy as np
|
|
if isinstance(value, np.generic):
|
|
return value.item()
|
|
if isinstance(value, list):
|
|
return [_pythonize(item) for item in value]
|
|
return value
|
|
|
|
|
|
def _logical_shape(tensor) -> tuple[int, ...]:
|
|
return tuple(int(dim) for dim in tensor.shape[::-1])
|
|
|
|
|
|
def _to_float(tensor, gguf_module):
|
|
import numpy as np
|
|
qtype = tensor.tensor_type
|
|
if qtype == gguf_module.GGMLQuantizationType.F32:
|
|
return np.asarray(tensor.data, dtype=np.float32)
|
|
if qtype == gguf_module.GGMLQuantizationType.F16:
|
|
return np.asarray(tensor.data, dtype=np.float32)
|
|
return np.asarray(gguf_module.quants.dequantize(tensor.data, qtype), dtype=np.float32)
|
|
|
|
|
|
def _checkpoint_file(directory: Path, name: str) -> Path:
|
|
digest = hashlib.sha1(name.encode("utf-8")).hexdigest()[:16]
|
|
safe = "".join(ch if ch.isalnum() or ch in "._-" else "_" for ch in name)[:80]
|
|
return directory / f"{safe}.{digest}.npz"
|
|
|
|
|
|
def _load_checkpoint(directory: Path, name: str, shape: tuple[int, ...]):
|
|
import numpy as np
|
|
path = _checkpoint_file(directory, name)
|
|
if not path.is_file():
|
|
return None
|
|
with np.load(path) as blob:
|
|
if tuple(int(v) for v in blob["shape"]) != tuple(shape):
|
|
return None
|
|
raw = np.array(blob["raw"], copy=True)
|
|
scale = np.array(blob["scale"], copy=True)
|
|
return raw, None if scale.size == 0 else scale
|
|
|
|
|
|
def _save_checkpoint(directory: Path, name: str, shape: tuple[int, ...], raw, scale) -> None:
|
|
import numpy as np
|
|
path = _checkpoint_file(directory, name)
|
|
temporary = path.with_name(path.name + ".tmp")
|
|
with temporary.open("wb") as handle:
|
|
np.savez_compressed(
|
|
handle,
|
|
shape=np.asarray(shape, dtype=np.int64),
|
|
raw=raw,
|
|
scale=np.zeros((0,), np.float32) if scale is None else np.asarray(scale, dtype=np.float32),
|
|
)
|
|
temporary.replace(path)
|
|
|
|
|
|
def _copy_metadata(reader, writer, gguf_module, context_length: int | None = None) -> None:
|
|
from nvfp4gguf.bundle import is_model_context_key, normalize_chat_template, require_context_length, scalar_context_length
|
|
requested = require_context_length(context_length)
|
|
source_context = None
|
|
wrote_context = False
|
|
for field in reader.fields.values():
|
|
if field.name.startswith("GGUF.") or field.name == "general.architecture" or not field.types:
|
|
continue
|
|
value = _pythonize(field.contents())
|
|
if value is None:
|
|
continue
|
|
value_type = field.types[0]
|
|
if value_type == gguf_module.GGUFValueType.ARRAY and not value:
|
|
continue
|
|
if field.name == "general.file_type":
|
|
value = int(gguf_module.LlamaFileType.MOSTLY_NVFP4)
|
|
if field.name == "general.alignment":
|
|
writer.data_alignment = int(value)
|
|
subtype = field.types[-1] if value_type == gguf_module.GGUFValueType.ARRAY else None
|
|
if field.name == "tokenizer.chat_template" or field.name.startswith("tokenizer.chat_template."):
|
|
value = normalize_chat_template(value)
|
|
if is_model_context_key(field.name):
|
|
source_context = scalar_context_length(value)
|
|
if requested is not None:
|
|
value = requested
|
|
value_type = gguf_module.GGUFValueType.UINT32
|
|
subtype = None
|
|
wrote_context = True
|
|
writer.add_key_value(field.name, value, value_type, sub_type=subtype)
|
|
if requested is not None and not wrote_context:
|
|
writer.add_context_length(requested)
|
|
if requested is not None:
|
|
log.info("context length %d", requested)
|
|
elif source_context is not None:
|
|
log.info("context length %d from the source GGUF", source_context)
|
|
|
|
|
|
def _source_description(tensor) -> str:
|
|
return tensor.tensor_type.name
|
|
|
|
|
|
def quantize_gguf(args, source: Path, output: Path, quant_args) -> None:
|
|
from nvfp4gguf import (
|
|
TensorQuantizer,
|
|
bind_exclude,
|
|
eligible_weight,
|
|
ensure_cuda_interpreter,
|
|
import_gguf,
|
|
load_learned_converter,
|
|
must_bake_nvfp4_scale,
|
|
mtp_layer_tensor,
|
|
preserve_scale_tensor,
|
|
quantize_and_pack,
|
|
report_reconstruction,
|
|
require_quant_settings,
|
|
scale_bake_notice,
|
|
)
|
|
gguf = import_gguf(args.llama_cpp)
|
|
|
|
ensure_cuda_interpreter(quant_args.device)
|
|
if output.exists() and not _wants_overwrite(args):
|
|
raise SystemExit(f"{output} already exists. Pass --overwrite to replace it.")
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
reader = gguf.GGUFReader(source)
|
|
if reader.byte_order != "I":
|
|
raise SystemExit(f"{source} is not little-endian GGUF")
|
|
architecture = reader.fields["general.architecture"].contents()
|
|
if isinstance(architecture, bytes):
|
|
architecture = architecture.decode("utf-8", "replace")
|
|
else:
|
|
architecture = str(architecture)
|
|
if not hasattr(quant_args, "exclude"):
|
|
quant_args.exclude = None
|
|
bind_exclude(quant_args, architecture)
|
|
notice = scale_bake_notice(architecture)
|
|
if notice:
|
|
log.warning(notice)
|
|
file_type = reader.fields.get("general.file_type")
|
|
file_type_name = ""
|
|
if file_type is not None:
|
|
try:
|
|
file_type_name = gguf.LlamaFileType(int(file_type.contents())).name
|
|
except ValueError:
|
|
file_type_name = str(file_type.contents())
|
|
float_source = file_type_name in {"ALL_F32", "MOSTLY_F16", "MOSTLY_BF16"}
|
|
if not float_source:
|
|
message = (
|
|
f"Source file type is {file_type_name or 'quantized'}. "
|
|
"NVFP4 will be fit to the dequantized weights, so detail already discarded by the source quant stays discarded."
|
|
)
|
|
if args.float_only:
|
|
raise SystemExit(message)
|
|
log.warning(message)
|
|
|
|
names = {tensor.name for tensor in reader.tensors}
|
|
|
|
def field_int(key: str) -> int:
|
|
field = reader.fields.get(key)
|
|
if field is None:
|
|
return 0
|
|
try:
|
|
return int(field.contents())
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
|
|
n_blocks = field_int(f"{architecture}.block_count")
|
|
n_nextn = field_int(f"{architecture}.nextn_predict_layers")
|
|
quant_names = []
|
|
for tensor in reader.tensors:
|
|
logical = _logical_shape(tensor)
|
|
if eligible_weight(tensor.name, logical, quant_args, gguf):
|
|
quant_names.append(tensor.name)
|
|
quant_name_set = set(quant_names)
|
|
log.info("%d matrices queued for NVFP4, %d tensors copied", len(quant_names), len(reader.tensors) - len(quant_names))
|
|
|
|
work = output.with_name(output.name + ".nvfp4-work")
|
|
if args.fresh and work.exists():
|
|
shutil.rmtree(work)
|
|
work.mkdir(parents=True, exist_ok=True)
|
|
require_quant_settings(work, quant_args)
|
|
_converter_cls, unpack_uint4 = load_learned_converter(quant_args.convert_to_quant)
|
|
del _converter_cls
|
|
quantizer = TensorQuantizer(quant_args, gguf, unpack_uint4)
|
|
writer = gguf.GGUFWriter(output, architecture, use_temp_file=not args.no_temp_file)
|
|
_copy_metadata(reader, writer, gguf, getattr(args, "context_length", None))
|
|
|
|
done = 0
|
|
for tensor in reader.tensors:
|
|
scale_for = ""
|
|
if tensor.name.endswith(".scale") or tensor.name.endswith(".input_scale"):
|
|
stem = tensor.name.rsplit(".", 1)[0]
|
|
scale_for = stem + ".weight"
|
|
if scale_for in quant_name_set and not (
|
|
tensor.name.endswith(".scale") and preserve_scale_tensor(architecture, scale_for)
|
|
):
|
|
continue
|
|
if tensor.name not in quant_name_set:
|
|
writer.add_tensor(tensor.name, tensor.data, raw_dtype=tensor.tensor_type)
|
|
continue
|
|
done += 1
|
|
logical = _logical_shape(tensor)
|
|
bias_name = tensor.name.replace(".weight", ".bias")
|
|
bake = must_bake_nvfp4_scale(
|
|
architecture,
|
|
tensor.name,
|
|
(".ffn_" in tensor.name or tensor.name.startswith("ffn_")) and bias_name in names,
|
|
) or mtp_layer_tensor(tensor.name, n_blocks, n_nextn)
|
|
cached = _load_checkpoint(work, tensor.name, logical)
|
|
if cached is not None and (cached[1] is None) != bake:
|
|
log.info(" checkpoint scale layout does not match %s; recomputing", tensor.name)
|
|
cached = None
|
|
log.info(
|
|
"NVFP4 [%d/%d] %s %s from %s%s",
|
|
done, len(quant_names), tensor.name, logical, _source_description(tensor),
|
|
" (checkpoint)" if cached else "",
|
|
)
|
|
if cached is None:
|
|
weight = _to_float(tensor, gguf)
|
|
raw, scale = quantize_and_pack(quantizer, weight, bake, tensor.name)
|
|
report_reconstruction(quantizer, weight, raw, scale)
|
|
_save_checkpoint(work, tensor.name, logical, raw, scale)
|
|
del weight
|
|
else:
|
|
raw, scale = cached
|
|
writer.add_tensor(tensor.name, raw, raw_dtype=gguf.GGMLQuantizationType.NVFP4)
|
|
if scale is not None:
|
|
writer.add_tensor(tensor.name[: -len(".weight")] + ".scale", scale)
|
|
del raw
|
|
|
|
log.info("Writing %s", output)
|
|
from nvfp4gguf.bundle import warn_missing_gguf_metadata
|
|
warn_missing_gguf_metadata(writer)
|
|
writer.write_header_to_file()
|
|
writer.write_kv_data_to_file()
|
|
writer.write_tensors_to_file(progress=True)
|
|
writer.close()
|
|
if not args.keep_checkpoints:
|
|
shutil.rmtree(work, ignore_errors=True)
|
|
log.info("Wrote %s", output)
|
|
|
|
|
|
def _split_name(name: str) -> tuple[str, str, str]:
|
|
if ":" not in name:
|
|
name = name + ":latest"
|
|
model, tag = name.rsplit(":", 1)
|
|
if "/" in model:
|
|
namespace, model = model.split("/", 1)
|
|
else:
|
|
namespace = "library"
|
|
return namespace, model, tag
|
|
|
|
|
|
def _read_blob(root: Path, digest: str) -> str:
|
|
path = blob_path(root, digest)
|
|
if not path.is_file():
|
|
raise FileNotFoundError(f"Missing blob {path}")
|
|
return path.read_text(encoding="utf-8")
|
|
|
|
|
|
def _quote_modelfile_value(text: str) -> str:
|
|
"""Quote a Modelfile argument the way Ollama's parser reads it back."""
|
|
if text == "" or "\n" in text or text[:1].isspace() or text[-1:].isspace() or '"' in text:
|
|
if '"""' in text:
|
|
raise ValueError("Modelfile value contains a triple quote, which a Modelfile cannot hold")
|
|
if "\n" in text or '"' in text:
|
|
return '"""' + text + '"""'
|
|
return '"' + text + '"'
|
|
return text
|
|
|
|
|
|
def _format_parameter(value) -> str:
|
|
if isinstance(value, bool):
|
|
return "true" if value else "false"
|
|
if isinstance(value, (int, float)) and not isinstance(value, bool):
|
|
return str(value)
|
|
if isinstance(value, (list, dict)):
|
|
return _quote_modelfile_value(json.dumps(value, separators=(",", ":")))
|
|
return _quote_modelfile_value(str(value))
|
|
|
|
|
|
def _config_lines(config: dict) -> list[str]:
|
|
"""Renderer, parser, requires, and explicit capabilities.
|
|
|
|
Ollama uses the renderer and parser for thinking and tool calls. Gemma 4,
|
|
Qwen3.5, and LFM2 thinking models stop and split thinking only when these
|
|
are present. A copied template alone does not turn thinking on.
|
|
"""
|
|
lines: list[str] = []
|
|
if config.get("requires"):
|
|
lines.append(f"REQUIRES {_quote_modelfile_value(str(config['requires']))}")
|
|
if config.get("renderer"):
|
|
lines.append(f"RENDERER {_quote_modelfile_value(str(config['renderer']))}")
|
|
if config.get("parser"):
|
|
lines.append(f"PARSER {_quote_modelfile_value(str(config['parser']))}")
|
|
capabilities = config.get("capabilities") or []
|
|
if isinstance(capabilities, str):
|
|
capabilities = [capabilities]
|
|
for capability in capabilities:
|
|
lines.append(f"CAPABILITY {_quote_modelfile_value(str(capability))}")
|
|
return lines
|
|
|
|
|
|
def _quote_block(command: str, text: str) -> str:
|
|
body = text.replace("\r\n", "\n")
|
|
return f"{command} {_quote_modelfile_value(body)}\n"
|
|
|
|
|
|
def _message_lines(text: str) -> list[str]:
|
|
try:
|
|
messages = json.loads(text)
|
|
except json.JSONDecodeError:
|
|
return ["# Messages blob was not JSON and was left out."]
|
|
if not isinstance(messages, list):
|
|
return ["# Messages blob was not a list and was left out."]
|
|
lines: list[str] = []
|
|
for item in messages:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
role = str(item.get("role") or "user")
|
|
content = item.get("content") if isinstance(item.get("content"), str) else ""
|
|
if not content and item.get("tool_calls"):
|
|
content = json.dumps(item["tool_calls"], separators=(",", ":"))
|
|
lines.append(f"MESSAGE {role} {_quote_modelfile_value(content)}")
|
|
return lines
|
|
|
|
|
|
# This fine-tune drafts inside a <think> block and often never emits <end_of_turn>.
|
|
# temperature 0.3, repeat_penalty 1.2, and repeat_last_n 128 got that story prompt
|
|
# to close the thought and stop. The chat template stays the source template.
|
|
_GEMMA_THINK_SAMPLER = (
|
|
("temperature", 0.3),
|
|
("repeat_penalty", 1.2),
|
|
("repeat_last_n", 128),
|
|
)
|
|
|
|
|
|
def _gemma_thinking_sampler(source: OllamaModel | None) -> bool:
|
|
if source is None or source.family.lower() not in {"gemma3", "gemma3n"}:
|
|
return False
|
|
if source.config.get("renderer") or source.config.get("parser"):
|
|
return False
|
|
text = " ".join((
|
|
source.name,
|
|
str(source.metadata.get("general.name") or ""),
|
|
str(source.metadata.get("general.finetune") or ""),
|
|
)).lower()
|
|
return "thinking" in text
|
|
|
|
|
|
def _extra_sampler(source: OllamaModel | None, existing: dict) -> list[tuple[str, object]]:
|
|
if not _gemma_thinking_sampler(source):
|
|
return []
|
|
missing = [(key, value) for key, value in _GEMMA_THINK_SAMPLER if key not in existing]
|
|
if missing:
|
|
log.warning(
|
|
"This Gemma thinking fine-tune drafts without writing <end_of_turn>. "
|
|
"Adding temperature 0.3, repeat_penalty 1.2, and repeat_last_n 128 so the reply reaches the stop token."
|
|
)
|
|
return missing
|
|
|
|
|
|
def render_modelfile(root: Path, source: OllamaModel | None, from_path: str) -> str:
|
|
"""Rebuild the source Modelfile without changing its prompt or sampling.
|
|
|
|
Renderer and parser are what turn thinking on for Gemma 4, Qwen, and LFM2.
|
|
The chat template and stop list are copied as stored, because rewriting them
|
|
is what made Gemma keep generating.
|
|
"""
|
|
lines = [f"FROM {from_path}", ""]
|
|
if source is None:
|
|
return "\n".join(lines)
|
|
if any(layer.get("mediaType") == PROJECTOR_MEDIA for layer in source.layers):
|
|
lines.append("FROM ./mmproj.gguf")
|
|
lines.append("")
|
|
config_lines = _config_lines(source.config)
|
|
if config_lines:
|
|
lines.extend(config_lines)
|
|
lines.append("")
|
|
notes: list[str] = []
|
|
for layer in source.layers:
|
|
media = str(layer.get("mediaType") or "")
|
|
kind = LAYER_KIND.get(media)
|
|
if media in (PROJECTOR_MEDIA, ADAPTER_MEDIA) or kind in (None, "PROJECTOR", "ADAPTER"):
|
|
if media == ADAPTER_MEDIA:
|
|
notes.append("# Ollama 0.40 no longer loads LoRA adapters, so the adapter layer was not written as ADAPTER.")
|
|
elif kind is None and media and media != MODEL_MEDIA:
|
|
notes.append(f"# Layer {media} is kept by --install and has no Modelfile command.")
|
|
continue
|
|
try:
|
|
text = _read_blob(root, layer["digest"])
|
|
except (OSError, KeyError) as exc:
|
|
notes.append(f"# Could not read {kind.lower()} blob: {exc}")
|
|
continue
|
|
if kind == "PARAMETER":
|
|
try:
|
|
params = json.loads(text)
|
|
except json.JSONDecodeError:
|
|
notes.append("# Parameters blob was not JSON and was left out.")
|
|
continue
|
|
if isinstance(params, dict):
|
|
if "num_ctx" in params:
|
|
log.info("Leaving num_ctx out of the Modelfile so OLLAMA_CONTEXT_LENGTH applies.")
|
|
for key, value in params.items():
|
|
if key == "num_ctx":
|
|
continue
|
|
values = value if isinstance(value, list) else [value]
|
|
for item in values:
|
|
lines.append(f"PARAMETER {key} {_format_parameter(item)}")
|
|
for key, value in _extra_sampler(source, params):
|
|
lines.append(f"PARAMETER {key} {_format_parameter(value)}")
|
|
lines.append("")
|
|
continue
|
|
if kind == "MESSAGE":
|
|
lines.extend(_message_lines(text))
|
|
lines.append("")
|
|
continue
|
|
lines.append(_quote_block(kind, text))
|
|
if notes:
|
|
lines.extend(notes)
|
|
lines.append("")
|
|
return "\n".join(lines).rstrip() + "\n"
|
|
|
|
|
|
def export_bundle(root: Path, export_dir: Path, gguf_file: Path, source: OllamaModel | None, overwrite: bool, copy_weights: bool = True) -> Path:
|
|
export_dir.mkdir(parents=True, exist_ok=True)
|
|
modelfile_path = export_dir / "Modelfile"
|
|
if modelfile_path.exists() and not overwrite:
|
|
raise SystemExit(f"{modelfile_path} already exists. Pass --overwrite to replace it.")
|
|
try:
|
|
relative = Path(os.path.relpath(gguf_file, export_dir))
|
|
from_path = relative.as_posix() if relative.is_absolute() else relative.as_posix()
|
|
except ValueError:
|
|
from_path = gguf_file.as_posix()
|
|
if not from_path.startswith(("/", "./", "../")):
|
|
from_path = "./" + from_path
|
|
modelfile_path.write_text(render_modelfile(root, source, from_path), encoding="utf-8")
|
|
if copy_weights and source is not None:
|
|
for layer in source.layers:
|
|
if layer.get("mediaType") != PROJECTOR_MEDIA:
|
|
continue
|
|
dest = export_dir / "mmproj.gguf"
|
|
src = blob_path(root, layer["digest"])
|
|
if not src.is_file():
|
|
log.warning("Projector blob is missing: %s", src)
|
|
continue
|
|
if dest.exists() and not overwrite:
|
|
raise SystemExit(f"{dest} already exists. Pass --overwrite to replace it.")
|
|
if dest.exists():
|
|
dest.unlink()
|
|
try:
|
|
os.link(src, dest)
|
|
except OSError:
|
|
shutil.copy2(src, dest)
|
|
log.info("Copied vision projector to %s", dest)
|
|
log.info("Wrote %s", modelfile_path)
|
|
return modelfile_path
|
|
|
|
|
|
def install_model(root: Path, gguf_file: Path, source: OllamaModel, new_name: str) -> Path:
|
|
digest = sha256_file(gguf_file)
|
|
blob = root / "blobs" / f"sha256-{digest}"
|
|
blob.parent.mkdir(parents=True, exist_ok=True)
|
|
if not blob.exists():
|
|
try:
|
|
os.link(gguf_file, blob)
|
|
except OSError:
|
|
shutil.copy2(gguf_file, blob)
|
|
config = {key: value for key, value in source.config.items() if key != "rootfs"}
|
|
config["model_format"] = source.config.get("model_format") or "gguf"
|
|
config["model_family"] = source.family
|
|
config["model_families"] = source.config.get("model_families") or ([source.family] if source.family else [])
|
|
config["model_type"] = source.params
|
|
config["file_type"] = "NVFP4"
|
|
config.setdefault("architecture", "amd64")
|
|
config.setdefault("os", "linux")
|
|
config["rootfs"] = {"type": "layers", "diff_ids": []}
|
|
layers = []
|
|
diff_ids = []
|
|
for layer in source.layers:
|
|
copied = dict(layer)
|
|
copied.pop("from", None)
|
|
if layer.get("mediaType") == MODEL_MEDIA:
|
|
copied = {
|
|
"mediaType": MODEL_MEDIA,
|
|
"digest": f"sha256:{digest}",
|
|
"size": gguf_file.stat().st_size,
|
|
}
|
|
elif layer.get("mediaType") == "application/vnd.ollama.image.params":
|
|
try:
|
|
current = json.loads(_read_blob(root, layer["digest"]))
|
|
except (OSError, KeyError, json.JSONDecodeError):
|
|
current = None
|
|
extra = _extra_sampler(source, current if isinstance(current, dict) else {})
|
|
if isinstance(current, dict) and (extra or "num_ctx" in current):
|
|
current.pop("num_ctx", None)
|
|
for key, value in extra:
|
|
current[key] = value
|
|
raw = json.dumps(current, separators=(",", ":")).encode("utf-8")
|
|
params_digest = hashlib.sha256(raw).hexdigest()
|
|
params_blob = root / "blobs" / f"sha256-{params_digest}"
|
|
params_blob.parent.mkdir(parents=True, exist_ok=True)
|
|
if not params_blob.exists():
|
|
params_blob.write_bytes(raw)
|
|
copied = {
|
|
"mediaType": layer["mediaType"],
|
|
"digest": f"sha256:{params_digest}",
|
|
"size": len(raw),
|
|
}
|
|
layers.append(copied)
|
|
diff_ids.append(copied["digest"])
|
|
config["rootfs"]["diff_ids"] = diff_ids
|
|
config_bytes = json.dumps(config, separators=(",", ":")).encode("utf-8")
|
|
config_digest = hashlib.sha256(config_bytes).hexdigest()
|
|
config_blob = root / "blobs" / f"sha256-{config_digest}"
|
|
config_blob.write_bytes(config_bytes)
|
|
namespace, model, tag = _split_name(new_name)
|
|
store = "manifests-v2" if source.store == "manifests-v2" else "manifests"
|
|
host = "ollama.com" if store == "manifests-v2" else "registry.ollama.ai"
|
|
manifest_path = root / store / host / namespace / model / tag
|
|
manifest_path.parent.mkdir(parents=True, exist_ok=True)
|
|
manifest = {
|
|
"schemaVersion": 2,
|
|
"mediaType": "application/vnd.docker.distribution.manifest.v2+json",
|
|
"config": {"mediaType": CONFIG_MEDIA, "digest": f"sha256:{config_digest}", "size": len(config_bytes)},
|
|
"layers": layers,
|
|
}
|
|
if source.runner:
|
|
manifest["runner"] = source.runner
|
|
elif store == "manifests-v2":
|
|
manifest["runner"] = "llamacpp"
|
|
if source.manifest_format:
|
|
manifest["format"] = source.manifest_format
|
|
elif store == "manifests-v2":
|
|
manifest["format"] = "gguf"
|
|
manifest_path.write_text(json.dumps(manifest, separators=(",", ":")), encoding="utf-8")
|
|
return manifest_path
|
|
|
|
|
|
def _resolve_source(args) -> tuple[Path, OllamaModel | None, str]:
|
|
query = args.model
|
|
if Path(query).is_file():
|
|
source = Path(query).resolve()
|
|
return source, None, source.stem
|
|
root = Path(args.models).resolve()
|
|
source_model = resolve_model(load_models(root), query)
|
|
if not source_model.gguf_path.is_file():
|
|
raise SystemExit(f"Weight blob is missing: {source_model.gguf_path}")
|
|
stem = source_model.name.replace("/", "-").replace(":", "-")
|
|
return source_model.gguf_path, source_model, stem
|
|
|
|
|
|
def _output_dir(args, stem: str) -> Path:
|
|
if args.output:
|
|
return Path(args.output).resolve()
|
|
return Path.cwd() / stem
|
|
|
|
|
|
def _wants_overwrite(args) -> bool:
|
|
return bool(getattr(args, "overwrite", False) or getattr(args, "force", False))
|
|
|
|
|
|
def _place_gguf(src: Path, dest: Path, overwrite: bool) -> None:
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
if dest.exists() and not overwrite:
|
|
raise SystemExit(f"{dest} already exists. Pass --overwrite to replace it.")
|
|
if dest.exists():
|
|
dest.unlink()
|
|
try:
|
|
os.link(src, dest)
|
|
except OSError:
|
|
shutil.copy2(src, dest)
|
|
|
|
|
|
def cmd_export(args) -> None:
|
|
"""Write the original GGUF and a Modelfile into a directory. No quantization."""
|
|
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
|
source, source_model, stem = _resolve_source(args)
|
|
out_dir = _output_dir(args, stem)
|
|
gguf_file = out_dir / f"{out_dir.name}.gguf"
|
|
overwrite = _wants_overwrite(args)
|
|
if not args.modelfile_only:
|
|
_place_gguf(source, gguf_file, overwrite)
|
|
else:
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
log.info("Skipping GGUF copy; writing Modelfile only")
|
|
export_bundle(Path(args.models).resolve(), out_dir, gguf_file, source_model, overwrite, copy_weights=not args.modelfile_only)
|
|
log.info("Exported %s", out_dir)
|
|
|
|
|
|
def cmd_quant(args) -> None:
|
|
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
|
source, source_model, stem = _resolve_source(args)
|
|
out_dir = _output_dir(args, f"{stem}-NVFP4")
|
|
output = out_dir / f"{out_dir.name}.gguf"
|
|
modelfile_path = out_dir / "Modelfile"
|
|
overwrite = _wants_overwrite(args)
|
|
if modelfile_path.exists() and not overwrite:
|
|
raise SystemExit(f"{modelfile_path} already exists. Pass --overwrite to replace it.")
|
|
quant_args = argparse.Namespace(
|
|
device=args.device,
|
|
optimizer=args.optimizer,
|
|
num_iter=args.num_iter,
|
|
lr=args.lr,
|
|
lr_min=args.lr_min,
|
|
top_p=args.top_p,
|
|
min_k=args.min_k,
|
|
max_k=args.max_k,
|
|
full_matrix=args.full_matrix,
|
|
simple=args.simple,
|
|
lr_schedule=args.lr_schedule,
|
|
early_stop_loss=args.early_stop_loss,
|
|
early_stop_stall=args.early_stop_stall,
|
|
early_stop_floor_stall=args.early_stop_floor_stall,
|
|
target_loss=args.target_loss,
|
|
scale_refinement_rounds=args.scale_refinement_rounds,
|
|
scale_optimization=args.scale_optimization,
|
|
convert_to_quant=args.convert_to_quant,
|
|
quantize_output=args.quantize_output,
|
|
exclude=args.exclude,
|
|
exclude_re=None,
|
|
calib_samples=args.calib_samples,
|
|
seed=args.seed,
|
|
no_bias_correction=args.no_bias_correction,
|
|
)
|
|
if args.modelfile_only:
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
log.info("Skipping quantization and GGUF creation; writing Modelfile only")
|
|
else:
|
|
quantize_gguf(args, source, output, quant_args)
|
|
export_bundle(Path(args.models).resolve(), out_dir, output, source_model, overwrite, copy_weights=not args.modelfile_only)
|
|
log.info("Quantized bundle %s" if not args.modelfile_only else "Wrote Modelfile %s", out_dir)
|
|
if args.install:
|
|
if source_model is None:
|
|
raise SystemExit("--install needs an Ollama model name, not a bare GGUF path")
|
|
if not output.is_file():
|
|
raise SystemExit(f"{output} is not there, so --install has no GGUF. Run quant without --modelfile-only first.")
|
|
root = Path(args.models).resolve()
|
|
new_name = args.name or _nvfp4_name(source_model.name)
|
|
manifest = install_model(root, output, source_model, new_name)
|
|
log.info("Registered %s", new_name)
|
|
log.info("Manifest %s", manifest)
|
|
|
|
|
|
_INFO_USAGE = "usage: ollama_nvfp4.py [--models DIR] --info MODEL [arguments for gguf_info.py...]"
|
|
|
|
|
|
def _without_models_dir(argv: list[str]) -> tuple[list[str], Path]:
|
|
"""Pull --models out of an --info command so it is not forwarded to gguf_info.py."""
|
|
kept: list[str] = []
|
|
models: Path | None = None
|
|
index = 0
|
|
while index < len(argv):
|
|
arg = argv[index]
|
|
if arg == "--models" or arg.startswith("--models="):
|
|
if arg == "--models":
|
|
if index + 1 >= len(argv):
|
|
raise SystemExit(f"{_INFO_USAGE}\n--models needs a directory")
|
|
value = argv[index + 1]
|
|
index += 2
|
|
else:
|
|
value = arg.split("=", 1)[1]
|
|
index += 1
|
|
models = Path(value)
|
|
continue
|
|
kept.append(arg)
|
|
index += 1
|
|
return kept, models if models is not None else default_models_dir()
|
|
|
|
|
|
def cmd_info(argv: list[str]) -> None:
|
|
"""Resolve MODEL to its sha256 weight blob and run gguf_info.py on that file.
|
|
|
|
Arguments after the model name are forwarded unchanged. --models is kept
|
|
here and may appear before or after --info.
|
|
"""
|
|
kept, models_dir = _without_models_dir(argv)
|
|
try:
|
|
info_at = kept.index("--info")
|
|
except ValueError:
|
|
raise SystemExit(_INFO_USAGE) from None
|
|
head, tail = kept[:info_at], kept[info_at + 1:]
|
|
if head or not tail or tail[0].startswith("-"):
|
|
extra = f"\nunrecognized: {' '.join(head)}" if head else ""
|
|
raise SystemExit(_INFO_USAGE + extra)
|
|
model_name, forwarded = tail[0], tail[1:]
|
|
root = Path(models_dir).resolve()
|
|
model = resolve_model(load_models(root), model_name)
|
|
if not model.gguf_path.is_file():
|
|
raise SystemExit(f"Weight blob is missing: {model.gguf_path}")
|
|
script = Path(__file__).resolve().parent / "gguf_info.py"
|
|
if not script.is_file():
|
|
raise SystemExit(f"gguf_info.py is not next to this script: {script}")
|
|
command = [sys.executable, str(script), str(model.gguf_path), *forwarded]
|
|
completed = subprocess.run(command)
|
|
raise SystemExit(completed.returncode)
|
|
|
|
|
|
def _nvfp4_name(name: str) -> str:
|
|
if ":" not in name:
|
|
return name + ":nvfp4"
|
|
model, tag = name.rsplit(":", 1)
|
|
if tag == "latest":
|
|
return model + ":nvfp4"
|
|
return f"{model}:{tag}-nvfp4"
|
|
|
|
|
|
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
|
from nvfp4gguf import (
|
|
DEFAULT_EARLY_STOP_FLOOR_STALL,
|
|
DEFAULT_TARGET_LOSS,
|
|
DEFAULT_EARLY_STOP_STALL,
|
|
DEFAULT_SCALE_OPTIMIZATION,
|
|
DEFAULT_SCALE_REFINEMENT_ROUNDS,
|
|
)
|
|
parser = argparse.ArgumentParser(
|
|
description=(
|
|
"Read an Ollama sha256 model store and quantize a model to NVFP4 GGUF. "
|
|
"--info MODEL runs gguf_info.py on that model's sha256 weight blob. "
|
|
"Arguments after MODEL are passed through, for example --metadata or --tensors."
|
|
),
|
|
)
|
|
parser.add_argument("--models", type=Path, default=default_models_dir(), help="Ollama models directory (blobs/ and manifests/)")
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
listed = sub.add_parser("list", help="Show every manifest as a readable model")
|
|
listed.add_argument(
|
|
"--full-scan",
|
|
action="store_true",
|
|
help="Read each GGUF header for capabilities and missing encoders. Slower on large models.",
|
|
)
|
|
listed.add_argument(
|
|
"--ctx-scan",
|
|
action="store_true",
|
|
help="Read parameter blobs and list models whose Modelfile sets num_ctx.",
|
|
)
|
|
|
|
show = sub.add_parser("show", help="Show one model, its blobs, and its GGUF metadata")
|
|
show.add_argument("model", help="Name from list, or a unique substring")
|
|
|
|
export = sub.add_parser("export", help="Write the original GGUF and a Modelfile into a directory")
|
|
export.add_argument("model", help="Ollama model name, or a path to a GGUF file")
|
|
export.add_argument("-o", "--output", type=Path, help="Output directory. Default: ./<model> in the current directory")
|
|
export.add_argument("--modelfile-only", action="store_true", help="Write the Modelfile and skip copying the GGUF and projector")
|
|
export.add_argument("--overwrite", action="store_true", help="Replace any output file that already exists")
|
|
export.add_argument("--force", action="store_true", help="Same as --overwrite")
|
|
|
|
quant = sub.add_parser("quant", help="Quantize into a directory containing the NVFP4 GGUF and a Modelfile")
|
|
quant.add_argument("model", help="Ollama model name, or a path to a GGUF file")
|
|
quant.add_argument("-o", "--output", type=Path, help="Output directory. Default: ./<model>-NVFP4 in the current directory")
|
|
quant.add_argument("--name", help="Ollama name to register with --install. Default: <model>:nvfp4 or <tag>-nvfp4")
|
|
quant.add_argument("--install", action="store_true", help="Register the NVFP4 GGUF in the Ollama store")
|
|
quant.add_argument("--modelfile-only", action="store_true", help="Write the Modelfile and skip quantization and GGUF creation")
|
|
quant.add_argument("--overwrite", action="store_true", help="Replace any output file that already exists")
|
|
quant.add_argument("--force", action="store_true", help="Same as --overwrite")
|
|
quant.add_argument("--float-only", action="store_true", help="Refuse to requantize a model that is already quantized")
|
|
quant.add_argument("--context-length", type=int, default=None,
|
|
help="Max context stored in the GGUF. Replaces a source value, or sets one when the source has no single context length.")
|
|
quant.add_argument("--llama-cpp", type=Path, default=Path.home() / "llama.cpp",
|
|
help="llama.cpp checkout used for the GGUF reader and writer (default: ~/llama.cpp)")
|
|
quant.add_argument("--convert-to-quant", type=Path, default=None)
|
|
quant.add_argument("--device", default="auto")
|
|
quant.add_argument("--simple", action="store_true", help="Round onto the E2M1 grid without learned rounding")
|
|
quant.add_argument("--optimizer", default="auto", choices=("auto", "prodigy", "adamw", "radam", "original"))
|
|
quant.add_argument("--num-iter", type=int, default=4000)
|
|
quant.add_argument("--lr", type=float, default=1.0, help="Initial learning rate")
|
|
quant.add_argument("--lr-min", type=float, default=1e-8, help="Lowest learning rate. The flat-loss check starts here (default: 1e-8).")
|
|
quant.add_argument("--lr-schedule", default="plateau", choices=("plateau", "exponential", "adaptive"))
|
|
quant.add_argument(
|
|
"--scale-optimization", default=DEFAULT_SCALE_OPTIMIZATION, choices=("fixed", "iterative", "joint"),
|
|
help="fixed keeps the initial scale. iterative refreshes it on the original optimizer. "
|
|
"joint trains the block scales with the codes (default: joint)",
|
|
)
|
|
quant.add_argument(
|
|
"--scale-refinement-rounds", type=int, default=DEFAULT_SCALE_REFINEMENT_ROUNDS,
|
|
help="Scale refreshes when --scale-optimization is iterative (default: 2). Ignored for fixed and joint.",
|
|
)
|
|
quant.add_argument("--full-matrix", action="store_true")
|
|
quant.add_argument("--top-p", type=float, default=0.2)
|
|
quant.add_argument("--min-k", type=int, default=128)
|
|
quant.add_argument("--max-k", type=int, default=1280)
|
|
quant.add_argument("--early-stop-loss", type=float, default=5e-9)
|
|
quant.add_argument(
|
|
"--early-stop-stall", type=int, default=DEFAULT_EARLY_STOP_STALL,
|
|
help="Stop after this many steps without a new best, at any learning rate (default: 2000)",
|
|
)
|
|
quant.add_argument(
|
|
"--target-loss", type=float, default=DEFAULT_TARGET_LOSS,
|
|
help="Stop when the best loss reaches this value, at any learning rate (default: 1e-5). 0 disables it.",
|
|
)
|
|
quant.add_argument(
|
|
"--early-stop-floor-stall", type=int, default=DEFAULT_EARLY_STOP_FLOOR_STALL,
|
|
help="If the loss never reaches --target-loss, stop after this many steps without a new best once it has otherwise flattened. 0 disables it (default: 64)",
|
|
)
|
|
quant.add_argument("--calib-samples", type=int, default=3072)
|
|
quant.add_argument("--seed", type=int, default=42)
|
|
quant.add_argument("--no-bias-correction", action="store_true")
|
|
quant.add_argument("--quantize-output", action="store_true")
|
|
quant.add_argument("--exclude", default=None, help="Regex of GGUF tensor names to leave unchanged")
|
|
quant.add_argument("--fresh", action="store_true")
|
|
quant.add_argument("--keep-checkpoints", action="store_true")
|
|
quant.add_argument("--no-temp-file", action="store_true")
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> None:
|
|
if argv is None:
|
|
argv = sys.argv[1:]
|
|
if "--info" in argv:
|
|
cmd_info(argv)
|
|
return
|
|
args = parse_args(argv)
|
|
root = Path(args.models).resolve()
|
|
if args.command == "list":
|
|
cmd_list(root, full_scan=args.full_scan, ctx_scan=args.ctx_scan)
|
|
return
|
|
if args.command == "show":
|
|
cmd_show(root, args.model)
|
|
return
|
|
if args.command == "export":
|
|
cmd_export(args)
|
|
return
|
|
if args.command == "quant":
|
|
cmd_quant(args)
|
|
return
|
|
raise SystemExit(f"Unknown command {args.command}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|