#!/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- 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: " bytes: data = handle.read(n) if len(data) != n: raise ValueError("truncated GGUF header") return data def read_string() -> bytes: size = struct.unpack(" 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(" 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: " bytes: data = handle.read(n) if len(data) != n: raise ValueError("truncated GGUF header") return data def read_string() -> bytes: size = struct.unpack(" 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(" 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 "" in low or "" 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 block and often never emits . # 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 . " "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: ./ 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: ./-NVFP4 in the current directory") quant.add_argument("--name", help="Ollama name to register with --install. Default: :nvfp4 or -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()