| """Minimal GGUF reader and llama.cpp iMatrix loader. |
| |
| The V2 quantizer works on float tensors, so GGUF ingestion is deliberately a |
| thin, read-only layer: |
| |
| * parse the GGUF header, metadata and tensor directory; |
| * dequantize the common source types (F32, F16, BF16, Q8_0, Q4_0 and Q2_K); |
| * read llama.cpp ``.gguf``/legacy ``.dat`` importance matrices and normalize |
| each vector to mean 1.0, matching ``imatrix_reader.h``. |
| |
| No GGUF writer is provided here; the quantizer emits Q2_K arrays that can be |
| packed by an external GGUF writer. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import re |
| import struct |
| import time |
| from dataclasses import dataclass |
| from pathlib import Path |
|
|
| import numpy as np |
|
|
| from .q2k import QK_K, blocks_to_bytes, dequantize_params_batch, unpack_codes |
| from .q3_k import Q3_K_BLOCK_BYTES, QK3_K, dequantize_q3_k_bytes, quantize_tensor_q3_k |
| from .q4_0 import Q4_0_BLOCK_BYTES, QK4_0, quantize_tensor_q4_0 |
| from .q1_0 import QK1_0, Q1_0_BLOCK_BYTES, quantize_tensor_q1_0 |
|
|
| GGUF_MAGIC = 0x46554747 |
| DEFAULT_ALIGNMENT = 32 |
|
|
| GGML_TYPE_F32 = 0 |
| GGML_TYPE_F16 = 1 |
| GGML_TYPE_Q4_0 = 2 |
| GGML_TYPE_Q4_1 = 3 |
| GGML_TYPE_Q5_0 = 6 |
| GGML_TYPE_Q5_1 = 7 |
| GGML_TYPE_Q8_0 = 8 |
| GGML_TYPE_Q2_K = 10 |
| GGML_TYPE_Q3_K = 11 |
| GGML_TYPE_BF16 = 30 |
| GGML_TYPE_Q1_0 = 41 |
|
|
| TYPE_NAME = { |
| GGML_TYPE_F32: "F32", |
| GGML_TYPE_F16: "F16", |
| GGML_TYPE_Q4_0: "Q4_0", |
| GGML_TYPE_Q4_1: "Q4_1", |
| GGML_TYPE_Q5_0: "Q5_0", |
| GGML_TYPE_Q5_1: "Q5_1", |
| GGML_TYPE_Q8_0: "Q8_0", |
| GGML_TYPE_Q2_K: "Q2_K", |
| GGML_TYPE_Q3_K: "Q3_K", |
| GGML_TYPE_BF16: "BF16", |
| GGML_TYPE_Q1_0: "Q1_0", |
| } |
|
|
| BLOCK_SIZE = { |
| GGML_TYPE_F32: 1, |
| GGML_TYPE_F16: 1, |
| GGML_TYPE_BF16: 1, |
| GGML_TYPE_Q4_0: 32, |
| GGML_TYPE_Q4_1: 32, |
| GGML_TYPE_Q5_0: 32, |
| GGML_TYPE_Q5_1: 32, |
| GGML_TYPE_Q8_0: 32, |
| GGML_TYPE_Q2_K: QK_K, |
| GGML_TYPE_Q3_K: QK3_K, |
| GGML_TYPE_Q1_0: QK1_0, |
| } |
| BLOCK_BYTES = { |
| GGML_TYPE_F32: 4, |
| GGML_TYPE_F16: 2, |
| GGML_TYPE_BF16: 2, |
| GGML_TYPE_Q4_0: 18, |
| GGML_TYPE_Q4_1: 20, |
| GGML_TYPE_Q5_0: 22, |
| GGML_TYPE_Q5_1: 24, |
| GGML_TYPE_Q8_0: 34, |
| GGML_TYPE_Q2_K: 84, |
| GGML_TYPE_Q3_K: Q3_K_BLOCK_BYTES, |
| GGML_TYPE_Q1_0: Q1_0_BLOCK_BYTES, |
| } |
| TYPE_BPW = { |
| GGML_TYPE_F32: 32.0, |
| GGML_TYPE_F16: 16.0, |
| GGML_TYPE_BF16: 16.0, |
| GGML_TYPE_Q8_0: 8.5, |
| GGML_TYPE_Q5_1: 6.0, |
| GGML_TYPE_Q5_0: 5.5, |
| GGML_TYPE_Q4_1: 5.0, |
| GGML_TYPE_Q4_0: 4.5, |
| GGML_TYPE_Q2_K: 2.625, |
| GGML_TYPE_Q3_K: 3.4375, |
| GGML_TYPE_Q1_0: 1.125, |
| 12: 4.5, |
| 13: 5.5, |
| 14: 6.5625, |
| 15: 8.5, |
| 16: 2.0625, |
| 17: 2.3125, |
| 18: 3.0625, |
| 19: 1.5625, |
| 20: 4.5, |
| 21: 3.4375, |
| 22: 2.5625, |
| 23: 4.25, |
| } |
|
|
|
|
| def source_type_bpw(ttype: int) -> float: |
| if ttype in TYPE_BPW: |
| return TYPE_BPW[ttype] |
| if ttype in BLOCK_SIZE and ttype in BLOCK_BYTES: |
| return (BLOCK_BYTES[ttype] * 8.0) / BLOCK_SIZE[ttype] |
| return 16.0 |
| SUPPORTED_READ_TYPES = { |
| GGML_TYPE_F32, |
| GGML_TYPE_F16, |
| GGML_TYPE_BF16, |
| GGML_TYPE_Q4_0, |
| GGML_TYPE_Q8_0, |
| GGML_TYPE_Q2_K, |
| GGML_TYPE_Q3_K, |
| GGML_TYPE_Q1_0, |
| } |
|
|
|
|
| @dataclass(slots=True) |
| class TensorInfo: |
| name: str |
| dims: tuple[int, ...] |
| type: int |
| offset: int |
|
|
| @property |
| def n_elements(self) -> int: |
| n = 1 |
| for d in self.dims: |
| n *= int(d) |
| return n |
|
|
| @property |
| def row_width(self) -> int: |
| return int(self.dims[0]) if self.dims else 1 |
|
|
| @property |
| def type_name(self) -> str: |
| return TYPE_NAME.get(self.type, f"type_{self.type}") |
|
|
| @property |
| def n_bytes(self) -> int: |
| if self.type not in BLOCK_SIZE: |
| raise ValueError(f"unsupported GGUF type {self.type} ({self.type_name})") |
| block = BLOCK_SIZE[self.type] |
| if self.n_elements % block != 0: |
| raise ValueError(f"{self.name}: element count is not block-aligned") |
| return (self.n_elements // block) * BLOCK_BYTES[self.type] |
|
|
|
|
| def _align(value: int, alignment: int = DEFAULT_ALIGNMENT) -> int: |
| return (value + alignment - 1) & ~(alignment - 1) |
|
|
|
|
| def _read_string(f) -> str: |
| n = struct.unpack("<Q", f.read(8))[0] |
| return f.read(n).decode("utf-8", errors="replace") |
|
|
|
|
| def _read_value(f, vtype: int): |
| if vtype == 0: |
| return struct.unpack("<B", f.read(1))[0] |
| if vtype == 1: |
| return struct.unpack("<b", f.read(1))[0] |
| if vtype == 2: |
| return struct.unpack("<H", f.read(2))[0] |
| if vtype == 3: |
| return struct.unpack("<h", f.read(2))[0] |
| if vtype == 4: |
| return struct.unpack("<I", f.read(4))[0] |
| if vtype == 5: |
| return struct.unpack("<i", f.read(4))[0] |
| if vtype == 6: |
| return struct.unpack("<f", f.read(4))[0] |
| if vtype == 7: |
| return bool(struct.unpack("<B", f.read(1))[0]) |
| if vtype == 8: |
| return _read_string(f) |
| if vtype == 9: |
| subtype = struct.unpack("<I", f.read(4))[0] |
| length = struct.unpack("<Q", f.read(8))[0] |
| if subtype == 8: |
| for _ in range(length): |
| _read_string(f) |
| else: |
| sizes = {0: 1, 1: 1, 2: 2, 3: 2, 4: 4, 5: 4, 6: 4, 7: 1, 10: 8, 11: 8, 12: 8} |
| f.read(sizes.get(subtype, 4) * length) |
| return None |
| if vtype == 10: |
| return struct.unpack("<Q", f.read(8))[0] |
| if vtype == 11: |
| return struct.unpack("<q", f.read(8))[0] |
| if vtype == 12: |
| return struct.unpack("<d", f.read(8))[0] |
| raise ValueError(f"unknown GGUF value type {vtype}") |
|
|
|
|
| class GGUFReader: |
| def __init__(self, path: str | Path): |
| self.path = Path(path) |
| self.metadata: dict[str, object] = {} |
| self.kv_pairs: list[tuple[str, int, bytes]] = [] |
| self.tensors: list[TensorInfo] = [] |
| with self.path.open("rb") as f: |
| magic, version = struct.unpack("<II", f.read(8)) |
| if magic != GGUF_MAGIC: |
| raise ValueError(f"{self.path} is not a GGUF file") |
| self.version = version |
| self.kv_start = f.tell() |
| n_tensors, n_kv = struct.unpack("<QQ", f.read(16)) |
| for _ in range(n_kv): |
| key = _read_string(f) |
| vtype = struct.unpack("<I", f.read(4))[0] |
| value_start = f.tell() |
| self.metadata[key] = _read_value(f, vtype) |
| value_end = f.tell() |
| f.seek(value_start) |
| raw = f.read(value_end - value_start) |
| f.seek(value_end) |
| self.kv_pairs.append((key, vtype, raw)) |
| self.kv_end = f.tell() |
| for _ in range(n_tensors): |
| name = _read_string(f) |
| n_dims = struct.unpack("<I", f.read(4))[0] |
| dims = tuple(struct.unpack("<Q", f.read(8))[0] for _ in range(n_dims)) |
| ttype = struct.unpack("<I", f.read(4))[0] |
| offset = struct.unpack("<Q", f.read(8))[0] |
| self.tensors.append(TensorInfo(name, dims, ttype, offset)) |
| self.info_end = f.tell() |
| self.alignment = int(self.metadata.get("general.alignment", DEFAULT_ALIGNMENT)) |
| self.data_start = _align(self.info_end, self.alignment) |
| self._by_name = {ti.name: ti for ti in self.tensors} |
|
|
| def info(self, name: str) -> TensorInfo: |
| try: |
| return self._by_name[name] |
| except KeyError: |
| raise KeyError(f"tensor '{name}' not found in {self.path}") from None |
|
|
| def list_tensors(self) -> list[TensorInfo]: |
| return list(self.tensors) |
|
|
| def read_tensor_raw(self, name: str) -> bytes: |
| ti = self.info(name) |
| with self.path.open("rb") as f: |
| f.seek(self.data_start + ti.offset) |
| raw = f.read(ti.n_bytes) |
| if len(raw) != ti.n_bytes: |
| raise ValueError(f"{name}: truncated tensor data") |
| return raw |
|
|
| def copy_tensor_raw_to(self, name: str, out, chunk: int = 1 << 20) -> None: |
| ti = self.info(name) |
| remaining = ti.n_bytes |
| with self.path.open("rb") as f: |
| f.seek(self.data_start + ti.offset) |
| while remaining > 0: |
| data = f.read(min(chunk, remaining)) |
| if not data: |
| raise ValueError(f"{name}: truncated tensor data") |
| out.write(data) |
| remaining -= len(data) |
|
|
| def read_tensor(self, name: str) -> tuple[np.ndarray, int]: |
| raw = self.read_tensor_raw(name) |
| return _dequantize(raw, self.info(name)), self.info(name).row_width |
|
|
| def iter_tensor_row_chunks(self, name: str, max_elements: int = 8_388_608): |
| """Stream a tensor in row-aligned float32 chunks to bound peak RAM.""" |
| ti = self.info(name) |
| row = ti.row_width |
| blk = BLOCK_SIZE[ti.type] |
| if row <= 0 or row % blk != 0 or ti.n_elements % row != 0: |
| x, r = self.read_tensor(name) |
| yield x, r, 0, ti.n_elements // max(r, 1) |
| return |
| n_rows = ti.n_elements // row |
| rows_per_chunk = max(1, max_elements // row) |
| bytes_per_row = (row // blk) * BLOCK_BYTES[ti.type] |
| with self.path.open("rb") as f: |
| for r0 in range(0, n_rows, rows_per_chunk): |
| r1 = min(r0 + rows_per_chunk, n_rows) |
| chunk_elems = (r1 - r0) * row |
| chunk_bytes = (r1 - r0) * bytes_per_row |
| f.seek(self.data_start + ti.offset + r0 * bytes_per_row) |
| raw = f.read(chunk_bytes) |
| if len(raw) != chunk_bytes: |
| raise ValueError(f"{name}: truncated tensor data") |
| sub_ti = TensorInfo(ti.name, (chunk_elems,), ti.type, 0) |
| yield _dequantize(raw, sub_ti), row, r0, r1 |
|
|
|
|
| def _dequantize(raw: bytes, ti: TensorInfo) -> np.ndarray: |
| t = ti.type |
| if t == GGML_TYPE_F32: |
| return np.frombuffer(raw, dtype="<f4").astype(np.float32) |
| if t == GGML_TYPE_F16: |
| return np.frombuffer(raw, dtype="<f2").astype(np.float32) |
| if t == GGML_TYPE_BF16: |
| bits = np.frombuffer(raw, dtype="<u2").astype(np.uint32) << 16 |
| return bits.view(np.float32) |
| if t == GGML_TYPE_Q8_0: |
| blocks = np.frombuffer(raw, dtype=np.uint8).reshape(-1, 34) |
| d = blocks[:, :2].copy().view("<f2").reshape(-1).astype(np.float32) |
| q = blocks[:, 2:].view(np.int8).astype(np.float32) |
| return (d[:, None] * q).reshape(-1) |
| if t == GGML_TYPE_Q4_0: |
| blocks = np.frombuffer(raw, dtype=np.uint8).reshape(-1, 18) |
| d = blocks[:, :2].copy().view("<f2").reshape(-1).astype(np.float32) |
| qs = blocks[:, 2:] |
| low = (qs & 0x0F).astype(np.float32) - 8.0 |
| high = (qs >> 4).astype(np.float32) - 8.0 |
| q = np.concatenate([low, high], axis=1) |
| return (d[:, None] * q).reshape(-1) |
| if t == GGML_TYPE_Q1_0: |
| blocks = np.frombuffer(raw, dtype=np.uint8).reshape(-1, Q1_0_BLOCK_BYTES) |
| d = blocks[:, :2].copy().view("<f2").reshape(-1, 1).astype(np.float32) |
| bits = np.unpackbits(blocks[:, 2:], axis=1, bitorder="little")[:, :QK1_0].astype(np.float32) |
| return (d * (bits * 2.0 - 1.0)).reshape(-1) |
| if t == GGML_TYPE_Q3_K: |
| n_blocks = len(raw) // Q3_K_BLOCK_BYTES |
| return dequantize_q3_k_bytes(raw, n_blocks) |
| if t == GGML_TYPE_Q2_K: |
| blocks = np.frombuffer(raw, dtype=np.uint8).reshape(-1, 84) |
| n = blocks.shape[0] |
| scales = blocks[:, :16] |
| qs = blocks[:, 16:80] |
| d = blocks[:, 80:82].copy().view("<f2").reshape(-1).astype(np.float32) |
| dmin = blocks[:, 82:84].copy().view("<f2").reshape(-1).astype(np.float32) |
| codes = np.empty((n, QK_K), dtype=np.uint8) |
| for base in (0, 128): |
| q = qs[:, base // 4 : base // 4 + 32] |
| for s_lvl in range(4): |
| lo = base + 32 * s_lvl |
| codes[:, lo : lo + 32] = (q >> np.uint8(2 * s_lvl)) & np.uint8(3) |
| return dequantize_params_batch(d, dmin, scales & 0x0F, scales >> 4, codes).reshape(-1) |
| raise ValueError(f"unsupported source GGUF type {ti.type_name} for {ti.name}") |
|
|
|
|
| def read_imatrix(path: str | Path) -> dict[str, np.ndarray]: |
| """Read a llama.cpp imatrix (GGUF) or legacy .dat file. |
| |
| Returns ``name -> float32 vector`` normalized to mean 1.0, exactly the |
| per-input-column ``E[a^2]`` weight used by the C engine. |
| """ |
| path = Path(path) |
| imat: dict[str, np.ndarray] = {} |
| with path.open("rb") as f: |
| magic = struct.unpack("<I", f.read(4))[0] |
| if magic == GGUF_MAGIC: |
| f.seek(0) |
| reader = GGUFReader(path) |
| sums: dict[str, np.ndarray] = {} |
| counts: dict[str, np.ndarray] = {} |
| with reader.path.open("rb") as fh: |
| for ti in reader.tensors: |
| if not (ti.name.endswith(".in_sum2") or ti.name.endswith(".counts")): |
| continue |
| fh.seek(reader.data_start + ti.offset) |
| raw = fh.read(ti.n_bytes) |
| data = np.frombuffer(raw, dtype="<f4").astype(np.float32) |
| if ti.name.endswith(".in_sum2"): |
| sums[ti.name[: -len(".in_sum2")]] = data |
| else: |
| counts[ti.name[: -len(".counts")]] = data |
| for base, sum2 in sums.items(): |
| c = counts.get(base) |
| if c is not None and c.size and float(c.flat[0]) > 0.0: |
| values = sum2 / float(c.flat[0]) |
| else: |
| values = np.ones_like(sum2) |
| mean = float(values.mean()) |
| imat[base] = values / mean if mean > 1.0e-30 else np.ones_like(values) |
| else: |
| f.seek(0) |
| n_entries = struct.unpack("<i", f.read(4))[0] |
| for _ in range(n_entries): |
| name_len = struct.unpack("<i", f.read(4))[0] |
| name = f.read(name_len).decode("utf-8", errors="replace") |
| n_values = struct.unpack("<i", f.read(4))[0] |
| struct.unpack("<i", f.read(4))[0] |
| values = np.frombuffer(f.read(n_values * 4), dtype="<f4").astype(np.float32) |
| mean = float(values.mean()) |
| imat[name] = values / mean if mean > 1.0e-30 else np.ones_like(values) |
| return imat |
|
|
|
|
| def lookup_imatrix(imat: dict[str, np.ndarray], name: str) -> np.ndarray | None: |
| if name in imat: |
| return imat[name] |
| for suffix in (".weight", ".in_sum2", ".counts"): |
| alt = name[: -len(suffix)] if name.endswith(suffix) else name + suffix |
| if alt in imat: |
| return imat[alt] |
| return None |
|
|
|
|
| def expand_importance( |
| values: np.ndarray | None, n_elements: int, row_width: int |
| ) -> np.ndarray | None: |
| if values is None: |
| return None |
| v = np.asarray(values, dtype=np.float32).ravel() |
| if v.size == n_elements: |
| return v |
| if row_width > 0 and v.size == row_width and n_elements % row_width == 0: |
| return np.tile(v, n_elements // row_width) |
| raise ValueError( |
| f"importance length {v.size} does not match {n_elements} elements " |
| f"(row width {row_width})" |
| ) |
|
|
|
|
| _BLOCK_PREFIX_RE = re.compile(r"^blk\.\d+\.") |
|
|
|
|
| def tensor_type_key(name: str) -> str: |
| """Return the layer-independent tensor type (e.g. ``blk.4.ffn_up.weight`` -> ``ffn_up``).""" |
| key = _BLOCK_PREFIX_RE.sub("", name) |
| for suffix in (".weight", ".bias"): |
| if key.endswith(suffix): |
| return key[: -len(suffix)] |
| return key |
|
|
|
|
| def _is_skipped_metadata_tensor(name: str, dims: tuple[int, ...]) -> bool: |
| if len(dims) < 2: |
| return True |
| if "norm" in name or ".bias" in name: |
| return True |
| if "ffn_gate_inp" in name: |
| return True |
| if name.startswith("mm.") or name.startswith("a.") or name.startswith("v."): |
| return True |
| if "altup" in name or "laurel" in name or "layer_output_scale" in name: |
| return True |
| if "ssm_a" in name.split(".") or "A_log" in name: |
| return True |
| if "ssm_dt" in name or "dt_bias" in name: |
| return True |
| if "ssm_conv1d" in name or "conv1d.weight" in name: |
| return True |
| return False |
|
|
|
|
| def _is_embd_or_head(name: str) -> bool: |
| return "token_embd" in name or name == "output.weight" or name.endswith(".output.weight") |
|
|
|
|
| def _is_attention_projection(name: str) -> bool: |
| """True for attention and recurrent/SSM projections (Q/K/V/O, QKV, gates, SSM state).""" |
| if "norm" in name or ".bias" in name: |
| return False |
| return any( |
| tok in name |
| for tok in ( |
| ".attn_", |
| ".ssm_", |
| "self_attn.", |
| "linear_attn.", |
| "attention.w", |
| "attn.q", |
| "attn.k", |
| "attn.v", |
| "attn.o", |
| "attn.wq", |
| "attn.wk", |
| "attn.wv", |
| "attn.wo", |
| ) |
| ) |
|
|
|
|
| def _is_tiny_ssm_gate(name: str, dims: tuple[int, ...]) -> bool: |
| """True for low-rank recurrent step-size/decay gates (e.g. ssm_alpha, ssm_beta, ssm_f_a/b).""" |
| if len(dims) < 2 or max(int(d) for d in dims) < 1024: |
| return False |
| if any( |
| k in name |
| for k in ("ssm_alpha", "ssm_beta", "ssm_f_a", "ssm_f_b", "ssm_g_a", "ssm_g_b") |
| ): |
| return True |
| if ".ssm_" in name and min(int(dims[0]), int(dims[1])) <= 128: |
| return True |
| return False |
|
|
|
|
| def is_q2k_eligible(name: str, dims: tuple[int, ...]) -> bool: |
| if _is_skipped_metadata_tensor(name, dims): |
| return False |
| if _is_embd_or_head(name): |
| return False |
| if dims[0] % QK_K != 0: |
| return False |
| n_elements = 1 |
| for d in dims: |
| n_elements *= int(d) |
| return n_elements >= QK_K and n_elements % QK_K == 0 |
|
|
|
|
| def is_q3_k_eligible(name: str, dims: tuple[int, ...]) -> bool: |
| """True for 256-aligned 2D/3D weight tensors eligible for Q3_K.""" |
| if _is_skipped_metadata_tensor(name, dims): |
| return False |
| if _is_embd_or_head(name): |
| return False |
| if dims[0] % QK3_K != 0: |
| return False |
| n_elements = 1 |
| for d in dims: |
| n_elements *= int(d) |
| return n_elements >= QK3_K and n_elements % QK3_K == 0 |
|
|
|
|
| def is_q4_0_eligible( |
| name: str, |
| dims: tuple[int, ...], |
| include_embd: bool = True, |
| promote_attn: bool = False, |
| allow_q2k_tensors: bool = False, |
| ) -> bool: |
| """True for 32-aligned 2D/3D weight tensors eligible for Q4_0.""" |
| if _is_skipped_metadata_tensor(name, dims): |
| return False |
| is_embd = _is_embd_or_head(name) |
| if is_embd and not include_embd: |
| return False |
| if not allow_q2k_tensors and not is_embd and not (promote_attn and _is_attention_projection(name)): |
| if is_q2k_eligible(name, dims): |
| return False |
| if dims[0] % QK4_0 != 0: |
| return False |
| n_elements = 1 |
| for d in dims: |
| n_elements *= int(d) |
| return n_elements >= QK4_0 and n_elements % QK4_0 == 0 |
|
|
|
|
| def is_q1_0_eligible( |
| name: str, |
| dims: tuple[int, ...], |
| include_embd: bool = True, |
| ) -> bool: |
| """True for 128-aligned 2D/3D weight tensors eligible for Q1_0.""" |
| if _is_skipped_metadata_tensor(name, dims): |
| return False |
| if _is_tiny_ssm_gate(name, dims): |
| return False |
| if _is_embd_or_head(name) and not include_embd: |
| return False |
| if len(dims) < 2 or dims[0] % QK1_0 != 0: |
| return False |
| n_elements = 1 |
| for d in dims: |
| n_elements *= int(d) |
| return n_elements >= QK1_0 and n_elements % QK1_0 == 0 |
|
|
|
|
| def target_quant_type( |
| name: str, |
| dims: tuple[int, ...], |
| enable_q4_fallback: bool = True, |
| quantize_embd_q4: bool = True, |
| promote_attn_q4: bool = False, |
| quant_format: str = "mixed", |
| source_type: int | None = None, |
| custom_quant_map: dict[str, str] | None = None, |
| ) -> str | None: |
| """Determine target quantization format ('Q1_0', 'Q2_K', 'Q4_0', or None for Copy). |
| |
| Modes: |
| * 'mixed' (default): all Q2_K-eligible tensors -> Q2_K; non-Q2_K 32-aligned |
| tensors (and embeddings/head if quantize_embd_q4) -> Q4_0. |
| * 'mixed_q1': FFN/expert 128-aligned tensors -> Q1_0; attention/SSM projections, |
| embeddings/head, and non-128-aligned 32-aligned tensors -> Q4_0. |
| * 'custom': per-tensor-type overrides from ``custom_quant_map``; unlisted |
| tensor types fall back to the 'mixed' plan. |
| * 'q4_0': all 32-aligned eligible weight tensors -> Q4_0. |
| * 'q2_k': only Q2_K-eligible tensors -> Q2_K; everything else copied. |
| * 'q3_k': only Q3_K-eligible tensors -> Q3_K; everything else copied. |
| * 'q1_0': all 128-aligned eligible weight tensors -> Q1_0. |
| |
| ``custom_quant_map`` maps a layer-independent tensor type (see |
| :func:`tensor_type_key`) to 'Q1_0', 'Q2_K', 'Q3_K', 'Q4_0', 'Copy', or 'Auto'. |
| Non-Auto entries override the selected mode for that tensor type. A target |
| is applied only when the tensor is actually eligible for it; explicit |
| 'Copy' (and ineligible targets) leave the tensor at its source precision. |
| |
| Never promotes a lower-bit source tensor to a higher-bit target format |
| (e.g., a Q2_K source tensor is never re-quantized to Q4_0). |
| """ |
| fmt = "mixed" if quant_format == "custom" else quant_format |
| tgt: str | None = None |
| custom_choice = None |
| if custom_quant_map: |
| custom_choice = custom_quant_map.get(tensor_type_key(name)) |
| if custom_choice is not None and str(custom_choice).strip().lower() not in ("", "auto"): |
| choice = str(custom_choice).strip().upper() |
| if choice == "Q1_0": |
| if is_q1_0_eligible(name, dims, include_embd=True): |
| tgt = "Q1_0" |
| elif choice == "Q2_K": |
| if is_q2k_eligible(name, dims): |
| tgt = "Q2_K" |
| elif choice == "Q3_K": |
| if is_q3_k_eligible(name, dims): |
| tgt = "Q3_K" |
| elif choice == "Q4_0": |
| if is_q4_0_eligible(name, dims, include_embd=True, allow_q2k_tensors=True): |
| tgt = "Q4_0" |
| |
| elif fmt == "q1_0": |
| if promote_attn_q4 and _is_attention_projection(name) and is_q4_0_eligible( |
| name, dims, include_embd=quantize_embd_q4, promote_attn=True, allow_q2k_tensors=True |
| ): |
| tgt = "Q4_0" |
| elif _is_embd_or_head(name) and quantize_embd_q4 and enable_q4_fallback and is_q4_0_eligible( |
| name, dims, include_embd=True, allow_q2k_tensors=True |
| ): |
| tgt = "Q4_0" |
| elif is_q1_0_eligible(name, dims, include_embd=not quantize_embd_q4): |
| tgt = "Q1_0" |
| elif enable_q4_fallback and is_q4_0_eligible( |
| name, dims, include_embd=quantize_embd_q4, promote_attn=True, allow_q2k_tensors=True |
| ): |
| tgt = "Q4_0" |
| elif fmt == "mixed_q1": |
| if (_is_attention_projection(name) or _is_tiny_ssm_gate(name, dims)) and is_q4_0_eligible( |
| name, dims, include_embd=quantize_embd_q4, promote_attn=True, allow_q2k_tensors=True |
| ): |
| tgt = "Q4_0" |
| elif _is_embd_or_head(name): |
| if quantize_embd_q4 and is_q4_0_eligible(name, dims, include_embd=True): |
| tgt = "Q4_0" |
| elif is_q1_0_eligible(name, dims, include_embd=False): |
| tgt = "Q1_0" |
| elif enable_q4_fallback and is_q4_0_eligible( |
| name, dims, include_embd=quantize_embd_q4, promote_attn=True, allow_q2k_tensors=True |
| ): |
| tgt = "Q4_0" |
| elif fmt == "q4_0": |
| if is_q4_0_eligible( |
| name, dims, include_embd=quantize_embd_q4, allow_q2k_tensors=True |
| ): |
| tgt = "Q4_0" |
| elif fmt == "q2_k": |
| if is_q2k_eligible(name, dims): |
| tgt = "Q2_K" |
| elif fmt == "q3_k": |
| if is_q3_k_eligible(name, dims): |
| tgt = "Q3_K" |
| else: |
| if promote_attn_q4 and _is_attention_projection(name) and is_q4_0_eligible( |
| name, dims, include_embd=quantize_embd_q4, promote_attn=True |
| ): |
| tgt = "Q4_0" |
| elif is_q2k_eligible(name, dims): |
| tgt = "Q2_K" |
| elif enable_q4_fallback and is_q4_0_eligible( |
| name, dims, include_embd=quantize_embd_q4, promote_attn=False |
| ): |
| tgt = "Q4_0" |
|
|
| if tgt is not None and source_type is not None: |
| src_bpw = source_type_bpw(int(source_type)) |
| tgt_bpw = {"Q2_K": 2.625, "Q3_K": 3.4375, "Q4_0": 4.5, "Q1_0": 1.125}[tgt] |
| if src_bpw <= tgt_bpw + 1.0e-6: |
| return None |
| return tgt |
|
|
|
|
| def _write_string(f, value: str) -> None: |
| data = value.encode("utf-8") |
| f.write(struct.pack("<Q", len(data))) |
| f.write(data) |
|
|
|
|
| def write_quantized_gguf( |
| input_path: str | Path, |
| output_path: str | Path, |
| quantizer, |
| imat: dict[str, np.ndarray] | None = None, |
| quantize_all: bool = True, |
| tensor_names: set[str] | None = None, |
| limit: int = 0, |
| verbose: bool = True, |
| progress_callback=None, |
| stage_callback=None, |
| stop_event=None, |
| ) -> dict[str, float]: |
| """Quantize a source GGUF with the V2 engine and emit a valid GGUF. |
| |
| Eligible tensors become Q2_K; everything else is copied byte-for-byte, |
| including the metadata/tokenizer section. Returns aggregate statistics. |
| """ |
| reader = GGUFReader(input_path) |
| alignment = reader.alignment |
| names = set(tensor_names) if tensor_names else None |
| if names is not None: |
| missing = names - set(reader._by_name) |
| if missing: |
| raise KeyError(f"tensors not found: {sorted(missing)}") |
|
|
| cfg = getattr(quantizer, "config", None) |
| enable_q4 = bool(getattr(cfg, "enable_q4_fallback", True)) |
| embd_q4 = bool(getattr(cfg, "quantize_embd_q4", True)) |
| attn_q4 = bool(getattr(cfg, "promote_attn_q4", False)) |
| q_format = str(getattr(cfg, "quant_format", "mixed")) |
| custom_map = getattr(cfg, "custom_quant_map", None) or None |
|
|
| plan: list[str | None] = [] |
| quantized_count = 0 |
| for ti in reader.tensors: |
| target_fmt: str | None = None |
| if names is not None: |
| if ti.name in names: |
| target_fmt = target_quant_type( |
| ti.name, |
| ti.dims, |
| enable_q4_fallback=enable_q4, |
| quantize_embd_q4=embd_q4, |
| promote_attn_q4=attn_q4, |
| quant_format=q_format, |
| source_type=ti.type, |
| custom_quant_map=custom_map, |
| ) |
| custom_choice = None |
| if custom_map: |
| custom_choice = custom_map.get(tensor_type_key(ti.name)) |
| explicit_custom = ( |
| custom_choice is not None |
| and str(custom_choice).strip().lower() not in ("", "auto") |
| ) |
| if target_fmt is None and not explicit_custom and q_format != "q3_k": |
| src_bpw = source_type_bpw(ti.type) |
| if q_format == "q1_0" and len(ti.dims) >= 2 and ti.dims[0] % QK1_0 == 0 and src_bpw > 1.125: |
| target_fmt = "Q1_0" |
| elif q_format == "q4_0" and len(ti.dims) >= 2 and ti.dims[0] % QK4_0 == 0 and src_bpw > 4.5: |
| target_fmt = "Q4_0" |
| elif len(ti.dims) >= 2 and ti.dims[0] % QK_K == 0 and ti.n_elements % QK_K == 0 and src_bpw > 2.625: |
| target_fmt = "Q2_K" |
| elif len(ti.dims) >= 2 and ti.dims[0] % QK4_0 == 0 and ti.n_elements % QK4_0 == 0 and src_bpw > 4.5: |
| target_fmt = "Q4_0" |
| elif verbose: |
| print(f" copy (already <= target precision or unaligned): {ti.name}") |
| elif quantize_all: |
| target_fmt = target_quant_type( |
| ti.name, |
| ti.dims, |
| enable_q4_fallback=enable_q4, |
| quantize_embd_q4=embd_q4, |
| promote_attn_q4=attn_q4, |
| quant_format=q_format, |
| source_type=ti.type, |
| custom_quant_map=custom_map, |
| ) |
| if target_fmt and ti.type not in SUPPORTED_READ_TYPES: |
| if verbose: |
| print(f" copy (unsupported source {ti.type_name}): {ti.name}") |
| target_fmt = None |
| if target_fmt and limit and quantized_count >= limit: |
| target_fmt = None |
| if target_fmt: |
| quantized_count += 1 |
| plan.append(target_fmt) |
|
|
| out_infos = [] |
| data_offset = 0 |
| total_planned_blocks = 0 |
| for ti, fmt in zip(reader.tensors, plan): |
| if fmt == "Q2_K": |
| n_blocks = ti.n_elements // QK_K |
| out_type = GGML_TYPE_Q2_K |
| out_size = n_blocks * 84 |
| total_planned_blocks += n_blocks |
| elif fmt == "Q3_K": |
| n_blocks = ti.n_elements // QK3_K |
| out_type = GGML_TYPE_Q3_K |
| out_size = n_blocks * Q3_K_BLOCK_BYTES |
| total_planned_blocks += max(1, n_blocks // 16) |
| elif fmt == "Q4_0": |
| n_blocks = ti.n_elements // QK4_0 |
| out_type = GGML_TYPE_Q4_0 |
| out_size = n_blocks * Q4_0_BLOCK_BYTES |
| total_planned_blocks += max(1, n_blocks // 8) |
| elif fmt == "Q1_0": |
| n_blocks = ti.n_elements // QK1_0 |
| out_type = GGML_TYPE_Q1_0 |
| out_size = n_blocks * Q1_0_BLOCK_BYTES |
| total_planned_blocks += max(1, n_blocks // 32) |
| else: |
| out_type = ti.type |
| out_size = ti.n_bytes |
| out_infos.append((ti, out_type, out_size, data_offset)) |
| data_offset = _align(data_offset + out_size, alignment) |
|
|
| has_q2k = any(fmt == "Q2_K" for fmt in plan) |
| has_q3k = any(fmt == "Q3_K" for fmt in plan) |
| has_q1_0 = any(fmt == "Q1_0" for fmt in plan) |
| |
| ftype_code = 11 if has_q3k else (10 if has_q2k else (40 if has_q1_0 else 2)) |
| kv_pairs = [] |
| for key, vtype, raw in reader.kv_pairs: |
| if quantized_count and key == "general.file_type" and vtype == 4: |
| raw = struct.pack("<I", ftype_code) |
| elif quantized_count and key == "general.quantization_version" and vtype == 4: |
| raw = struct.pack("<I", 2) |
| kv_pairs.append((key, vtype, raw)) |
|
|
| output_path = Path(output_path) |
| output_path.parent.mkdir(parents=True, exist_ok=True) |
| total_rmse = 0.0 |
| total_quantized = 0 |
| total_q2k = 0 |
| total_q3k = 0 |
| total_q4_0 = 0 |
| total_q1_0 = 0 |
| total_elements = 0 |
| completed_planned_blocks = 0 |
| model_t0 = time.time() |
| with output_path.open("wb") as fout: |
| fout.write(struct.pack("<IIQQ", GGUF_MAGIC, reader.version, len(out_infos), len(kv_pairs))) |
| for key, vtype, raw in kv_pairs: |
| _write_string(fout, key) |
| fout.write(struct.pack("<I", vtype)) |
| fout.write(raw) |
| for ti, out_type, out_size, offset in out_infos: |
| _write_string(fout, ti.name) |
| fout.write(struct.pack("<I", len(ti.dims))) |
| for d in ti.dims: |
| fout.write(struct.pack("<Q", int(d))) |
| fout.write(struct.pack("<I", out_type)) |
| fout.write(struct.pack("<Q", offset)) |
|
|
| data_start = _align(fout.tell(), alignment) |
| fout.write(b"\x00" * (data_start - fout.tell())) |
|
|
| for idx, (ti, out_type, out_size, offset) in enumerate(out_infos): |
| if stop_event is not None and stop_event.is_set(): |
| raise InterruptedError("Quantization cancelled by user") |
| target = data_start + offset |
| if target > fout.tell(): |
| fout.write(b"\x00" * (target - fout.tell())) |
| fmt = plan[idx] |
| if fmt == "Q2_K": |
| t_blocks = ti.n_elements // QK_K |
| has_imat = imat is not None and lookup_imatrix(imat, ti.name) is not None |
|
|
| def _emit_stage(ev: dict, _i=idx + 1, _ti=ti, _base=completed_planned_blocks, _tb=t_blocks): |
| if stage_callback is None: |
| return |
| phase_name = ev.get("phase", "") |
| done_b = int(ev.get("done", 0)) |
| tot_b = max(int(ev.get("total", _tb)), 1) |
| if phase_name.startswith("Phase 2b") or "Phase 1–2b" in phase_name: |
| frac = 0.80 * (done_b / tot_b) |
| elif phase_name.startswith("Phase 3"): |
| frac = 0.82 |
| elif "Phase 4" in phase_name and not phase_name.startswith("Phase 4.8"): |
| frac = 0.82 + 0.17 * (done_b / tot_b) |
| elif phase_name.startswith("Phase 4.8") or phase_name.startswith("Phase 5"): |
| frac = 1.0 |
| else: |
| frac = 0.0 |
| eff_blocks = _base + frac * _tb |
| elapsed_model = max(time.time() - model_t0, 1.0e-6) |
| model_rate = eff_blocks / elapsed_model |
| rem_blocks = max(total_planned_blocks - eff_blocks, 0.0) |
| model_eta = rem_blocks / model_rate if model_rate > 1.0e-6 else 0.0 |
| model_pct = 100.0 * eff_blocks / max(total_planned_blocks, 1) |
| stage_callback( |
| _i, |
| len(out_infos), |
| _ti, |
| { |
| **ev, |
| "model_pct": model_pct, |
| "model_eta": model_eta, |
| "model_done_blocks": int(eff_blocks), |
| "model_total_blocks": total_planned_blocks, |
| }, |
| ) |
|
|
| _emit_stage( |
| { |
| "phase": f"Reading & Dequantizing Source ({ti.type_name})", |
| "done": 0, |
| "total": t_blocks, |
| "detail": f"dims={ti.dims} elements={ti.n_elements:,} ({ti.n_bytes / 1048576:.2f} MB) imatrix={'yes' if has_imat else 'no'}", |
| } |
| ) |
| x, row = reader.read_tensor(ti.name) |
| imp = None |
| if imat is not None: |
| raw_imp = lookup_imatrix(imat, ti.name) |
| if raw_imp is not None: |
| imp = expand_importance(raw_imp, x.size, row) |
| res = quantizer.quantize( |
| x, |
| importance=imp, |
| row_width=row, |
| status_callback=_emit_stage if stage_callback is not None else None, |
| stop_event=stop_event, |
| ) |
| completed_planned_blocks += t_blocks |
| payload = blocks_to_bytes(res.scales, res.qs, res.dm) |
| if len(payload) != out_size: |
| raise RuntimeError(f"{ti.name}: Q2_K payload size mismatch") |
| fout.write(payload) |
| total_rmse += res.stats.rmse * res.stats.rmse * x.size |
| total_elements += x.size |
| total_quantized += 1 |
| total_q2k += 1 |
| if verbose: |
| print( |
| f" Q2_K {ti.name:48s} row={row:6d} " |
| f"rmse={res.stats.rmse:.6e} imatrix={'yes' if imp is not None else 'no'}" |
| ) |
| if progress_callback is not None: |
| progress_callback( |
| idx + 1, |
| len(out_infos), |
| ti.name, |
| "Q2_K", |
| float(res.stats.rmse), |
| imp is not None, |
| ) |
| elif fmt == "Q3_K": |
| t_blocks_q3 = ti.n_elements // QK3_K |
| eff_tb = max(1, t_blocks_q3 // 16) |
| raw_imp = lookup_imatrix(imat, ti.name) if imat is not None else None |
| has_imat = raw_imp is not None and raw_imp.size == ti.row_width |
| written_bytes = 0 |
| sse_q3 = 0.0 |
| done_q3_blks = 0 |
| t_q3_start = time.time() |
|
|
| for x_chunk, row, _r0, _r1 in reader.iter_tensor_row_chunks(ti.name, max_elements=8_388_608): |
| if stop_event is not None and stop_event.is_set(): |
| raise InterruptedError("Quantization cancelled by user") |
| imp_chunk = expand_importance(raw_imp, x_chunk.size, row) if has_imat else None |
| q3_bytes, nb_chunk, chunk_rmse = quantize_tensor_q3_k( |
| x_chunk, importance=imp_chunk, cfg=cfg, chunk_blocks=8192 |
| ) |
| fout.write(q3_bytes) |
| written_bytes += len(q3_bytes) |
| sse_q3 += (chunk_rmse * chunk_rmse) * x_chunk.size |
| done_q3_blks += nb_chunk |
| if stage_callback is not None: |
| frac = done_q3_blks / max(t_blocks_q3, 1) |
| eff_blocks = completed_planned_blocks + frac * eff_tb |
| elapsed_model = max(time.time() - model_t0, 1.0e-6) |
| model_rate = eff_blocks / elapsed_model |
| model_eta = max(total_planned_blocks - eff_blocks, 0.0) / max(model_rate, 1.0e-6) |
| el_q3 = max(time.time() - t_q3_start, 1.0e-6) |
| r_q3 = done_q3_blks / el_q3 |
| eta_q3 = (t_blocks_q3 - done_q3_blks) / max(r_q3, 1.0e-6) |
| stage_callback( |
| idx + 1, |
| len(out_infos), |
| ti, |
| { |
| "phase": "Q3_K: Analog + WLS + Sieve/Viterbi + FP16-ULP Quantization", |
| "done": done_q3_blks, |
| "total": t_blocks_q3, |
| "detail": f"blk {done_q3_blks:,}/{t_blocks_q3:,} ({r_q3:.0f} blk/s) rmse={chunk_rmse:.6e}", |
| "rate": r_q3, |
| "eta": eta_q3, |
| "rmse": chunk_rmse, |
| "sigma": float(np.sqrt(np.mean(x_chunk.astype(np.float64) ** 2))), |
| "model_pct": 100.0 * eff_blocks / max(total_planned_blocks, 1), |
| "model_eta": model_eta, |
| "model_done_blocks": int(eff_blocks), |
| "model_total_blocks": total_planned_blocks, |
| }, |
| ) |
|
|
| if written_bytes != out_size: |
| raise RuntimeError(f"{ti.name}: Q3_K payload size mismatch") |
| completed_planned_blocks += eff_tb |
| q3_rmse = float(np.sqrt(sse_q3 / ti.n_elements)) |
| total_rmse += sse_q3 |
| total_elements += ti.n_elements |
| total_quantized += 1 |
| total_q3k += 1 |
| if verbose: |
| print( |
| f" Q3_K {ti.name:48s} row={ti.row_width:6d} " |
| f"rmse={q3_rmse:.6e} imatrix={'yes' if has_imat else 'no'}" |
| ) |
| if progress_callback is not None: |
| progress_callback( |
| idx + 1, |
| len(out_infos), |
| ti.name, |
| "Q3_K", |
| q3_rmse, |
| has_imat, |
| ) |
| elif fmt == "Q4_0": |
| t_blocks_q4 = ti.n_elements // QK4_0 |
| eff_tb = max(1, t_blocks_q4 // 8) |
| raw_imp = lookup_imatrix(imat, ti.name) if imat is not None else None |
| has_imat = raw_imp is not None and raw_imp.size == ti.row_width |
| written_bytes = 0 |
| sse_q4 = 0.0 |
| done_q4_blks = 0 |
| t_q4_start = time.time() |
|
|
| for x_chunk, row, _r0, _r1 in reader.iter_tensor_row_chunks(ti.name, max_elements=8_388_608): |
| if stop_event is not None and stop_event.is_set(): |
| raise InterruptedError("Quantization cancelled by user") |
| imp_chunk = expand_importance(raw_imp, x_chunk.size, row) if has_imat else None |
| q4_bytes, nb_chunk, chunk_rmse = quantize_tensor_q4_0( |
| x_chunk, importance=imp_chunk, cfg=cfg, chunk_blocks=16384 |
| ) |
| fout.write(q4_bytes) |
| written_bytes += len(q4_bytes) |
| sse_q4 += (chunk_rmse * chunk_rmse) * x_chunk.size |
| done_q4_blks += nb_chunk |
| if stage_callback is not None: |
| frac = done_q4_blks / max(t_blocks_q4, 1) |
| eff_blocks = completed_planned_blocks + frac * eff_tb |
| elapsed_model = max(time.time() - model_t0, 1.0e-6) |
| model_rate = eff_blocks / elapsed_model |
| model_eta = max(total_planned_blocks - eff_blocks, 0.0) / max(model_rate, 1.0e-6) |
| el_q4 = max(time.time() - t_q4_start, 1.0e-6) |
| r_q4 = done_q4_blks / el_q4 |
| eta_q4 = (t_blocks_q4 - done_q4_blks) / max(r_q4, 1.0e-6) |
| stage_callback( |
| idx + 1, |
| len(out_infos), |
| ti, |
| { |
| "phase": "Q4_0: Analog + WLS + FP16-ULP Quantization", |
| "done": done_q4_blks, |
| "total": t_blocks_q4, |
| "detail": f"blk {done_q4_blks:,}/{t_blocks_q4:,} ({r_q4:.0f} blk/s) rmse={chunk_rmse:.6e}", |
| "rate": r_q4, |
| "eta": eta_q4, |
| "rmse": chunk_rmse, |
| "sigma": float(np.sqrt(np.mean(x_chunk.astype(np.float64) ** 2))), |
| "model_pct": 100.0 * eff_blocks / max(total_planned_blocks, 1), |
| "model_eta": model_eta, |
| "model_done_blocks": int(eff_blocks), |
| "model_total_blocks": total_planned_blocks, |
| }, |
| ) |
|
|
| if written_bytes != out_size: |
| raise RuntimeError(f"{ti.name}: Q4_0 payload size mismatch") |
| completed_planned_blocks += eff_tb |
| q4_rmse = float(np.sqrt(sse_q4 / ti.n_elements)) |
| total_rmse += sse_q4 |
| total_elements += ti.n_elements |
| total_quantized += 1 |
| total_q4_0 += 1 |
| if verbose: |
| print( |
| f" Q4_0 {ti.name:48s} row={ti.row_width:6d} " |
| f"rmse={q4_rmse:.6e} imatrix={'yes' if has_imat else 'no'}" |
| ) |
| if progress_callback is not None: |
| progress_callback( |
| idx + 1, |
| len(out_infos), |
| ti.name, |
| "Q4_0", |
| q4_rmse, |
| has_imat, |
| ) |
| elif fmt == "Q1_0": |
| t_blocks_q1 = ti.n_elements // QK1_0 |
| eff_tb = max(1, t_blocks_q1 // 32) |
| raw_imp = lookup_imatrix(imat, ti.name) if imat is not None else None |
| has_imat = raw_imp is not None and raw_imp.size == ti.row_width |
| written_bytes = 0 |
| sse_q1 = 0.0 |
| done_q1_blks = 0 |
| t_q1_start = time.time() |
|
|
| for x_chunk, row, _r0, _r1 in reader.iter_tensor_row_chunks(ti.name, max_elements=8_388_608): |
| if stop_event is not None and stop_event.is_set(): |
| raise InterruptedError("Quantization cancelled by user") |
| imp_chunk = expand_importance(raw_imp, x_chunk.size, row) if has_imat else None |
| q1_bytes, nb_chunk, chunk_rmse = quantize_tensor_q1_0( |
| x_chunk, importance=imp_chunk, cfg=cfg |
| ) |
| fout.write(q1_bytes) |
| written_bytes += len(q1_bytes) |
| sse_q1 += (chunk_rmse * chunk_rmse) * x_chunk.size |
| done_q1_blks += nb_chunk |
| if stage_callback is not None: |
| frac = done_q1_blks / max(t_blocks_q1, 1) |
| eff_blocks = completed_planned_blocks + frac * eff_tb |
| elapsed_model = max(time.time() - model_t0, 1.0e-6) |
| model_rate = eff_blocks / elapsed_model |
| model_eta = max(total_planned_blocks - eff_blocks, 0.0) / max(model_rate, 1.0e-6) |
| el_q1 = max(time.time() - t_q1_start, 1.0e-6) |
| r_q1 = done_q1_blks / el_q1 |
| eta_q1 = (t_blocks_q1 - done_q1_blks) / max(r_q1, 1.0e-6) |
| stage_callback( |
| idx + 1, |
| len(out_infos), |
| ti, |
| { |
| "phase": "Q1_0: Analog + WLS + Sieve/Viterbi + FP16-ULP Quantization", |
| "done": done_q1_blks, |
| "total": t_blocks_q1, |
| "detail": f"blk {done_q1_blks:,}/{t_blocks_q1:,} ({r_q1:.0f} blk/s) rmse={chunk_rmse:.6e}", |
| "rate": r_q1, |
| "eta": eta_q1, |
| "rmse": chunk_rmse, |
| "sigma": float(np.sqrt(np.mean(x_chunk.astype(np.float64) ** 2))), |
| "model_pct": 100.0 * eff_blocks / max(total_planned_blocks, 1), |
| "model_eta": model_eta, |
| "model_done_blocks": int(eff_blocks), |
| "model_total_blocks": total_planned_blocks, |
| }, |
| ) |
|
|
| if written_bytes != out_size: |
| raise RuntimeError(f"{ti.name}: Q1_0 payload size mismatch") |
| completed_planned_blocks += eff_tb |
| q1_rmse = float(np.sqrt(sse_q1 / ti.n_elements)) |
| total_rmse += sse_q1 |
| total_elements += ti.n_elements |
| total_quantized += 1 |
| total_q1_0 += 1 |
| if verbose: |
| print( |
| f" Q1_0 {ti.name:48s} row={ti.row_width:6d} " |
| f"rmse={q1_rmse:.6e} imatrix={'yes' if has_imat else 'no'}" |
| ) |
| if progress_callback is not None: |
| progress_callback( |
| idx + 1, |
| len(out_infos), |
| ti.name, |
| "Q1_0", |
| q1_rmse, |
| has_imat, |
| ) |
| else: |
| if stage_callback is not None: |
| stage_callback( |
| idx + 1, |
| len(out_infos), |
| ti, |
| { |
| "phase": f"Passthrough Copy ({ti.type_name})", |
| "done": 1, |
| "total": 1, |
| "detail": f"dims={ti.dims} ({ti.n_bytes / 1048576:.2f} MB)", |
| }, |
| ) |
| reader.copy_tensor_raw_to(ti.name, fout) |
| if verbose and idx % 25 == 0: |
| print(f" copy {ti.name}") |
| if progress_callback is not None: |
| progress_callback(idx + 1, len(out_infos), ti.name, "copy", None, False) |
|
|
| rmse = float(np.sqrt(total_rmse / total_elements)) if total_elements else 0.0 |
| return { |
| "tensors": float(len(out_infos)), |
| "quantized": float(total_quantized), |
| "q2_k": float(total_q2k), |
| "q3_k": float(total_q3k), |
| "q4_0": float(total_q4_0), |
| "q1_0": float(total_q1_0), |
| "elements": float(total_elements), |
| "rmse": rmse, |
| } |
|
|
