Upload slimshard.py

slimshard.py

@@ -2681,6 +2681,520 @@ def slimshard_info(path: str | Path) -> dict:
         }
 
 
+# ---------------------------------------------------------------------------
+# Universal SafeTensors -> GGUF Exporter (supports FP4/FP8/BF16/F16/F32 + Q2_K/Q4_0)
+# ---------------------------------------------------------------------------
+_GGUF_MAGIC = 0x46554747
+_GGUF_ALIGNMENT = 32
+_GGML_F32 = 0
+_GGML_F16 = 1
+_GGML_Q4_0 = 2
+_GGML_Q2_K = 10
+_GGML_BF16 = 30
+
+_LAYER_SUFFIX_TO_GGUF = {
+    "input_layernorm.weight": "attn_norm.weight",
+    "attn_norm.weight": "attn_norm.weight",
+    "ln_1.weight": "attn_norm.weight",
+    "post_attention_layernorm.weight": "ffn_norm.weight",
+    "ffn_norm.weight": "ffn_norm.weight",
+    "ln_2.weight": "ffn_norm.weight",
+    "self_attn.q_proj.weight": "attn_q.weight",
+    "attn.q_proj.weight": "attn_q.weight",
+    "attention.wq.weight": "attn_q.weight",
+    "self_attn.k_proj.weight": "attn_k.weight",
+    "attn.k_proj.weight": "attn_k.weight",
+    "attention.wk.weight": "attn_k.weight",
+    "self_attn.v_proj.weight": "attn_v.weight",
+    "attn.v_proj.weight": "attn_v.weight",
+    "attention.wv.weight": "attn_v.weight",
+    "self_attn.o_proj.weight": "attn_output.weight",
+    "attn.o_proj.weight": "attn_output.weight",
+    "attention.wo.weight": "attn_output.weight",
+    "attn.wo_b.weight": "attn_output.weight",
+    "attn.wo_a.weight": "attn_o_a.weight",
+    "self_attn.q_a_proj.weight": "attn_q_a.weight",
+    "attn.wq_a.weight": "attn_q_a.weight",
+    "self_attn.q_a_layernorm.weight": "attn_q_a_norm.weight",
+    "attn.q_norm.weight": "attn_q_a_norm.weight",
+    "self_attn.q_b_proj.weight": "attn_q_b.weight",
+    "attn.wq_b.weight": "attn_q_b.weight",
+    "self_attn.kv_a_proj_with_mqa.weight": "attn_kv_a_mqa.weight",
+    "attn.wkv.weight": "attn_kv_a_mqa.weight",
+    "self_attn.kv_a_layernorm.weight": "attn_kv_a_norm.weight",
+    "attn.kv_norm.weight": "attn_kv_a_norm.weight",
+    "self_attn.kv_b_proj.weight": "attn_kv_b.weight",
+    "self_attn.g_proj.weight": "attn_gate.weight",
+    "mlp.gate_proj.weight": "ffn_gate.weight",
+    "feed_forward.w1.weight": "ffn_gate.weight",
+    "mlp.up_proj.weight": "ffn_up.weight",
+    "feed_forward.w3.weight": "ffn_up.weight",
+    "mlp.down_proj.weight": "ffn_down.weight",
+    "feed_forward.w2.weight": "ffn_down.weight",
+    "moe.gate.weight": "ffn_gate_inp.weight",
+    "mlp.gate.weight": "ffn_gate_inp.weight",
+    "ffn.gate.weight": "ffn_gate_inp.weight",
+    "block_sparse_moe.gate.weight": "ffn_gate_inp.weight",
+    "moe.gate.e_score_correction_bias": "exp_probs_b.bias",
+    "ffn.gate.bias": "exp_probs_b.bias",
+    "moe.shared_experts.gate_proj.weight": "ffn_gate_shexp.weight",
+    "mlp.shared_experts.gate_proj.weight": "ffn_gate_shexp.weight",
+    "ffn.shared_experts.w1.weight": "ffn_gate_shexp.weight",
+    "moe.shared_experts.up_proj.weight": "ffn_up_shexp.weight",
+    "mlp.shared_experts.up_proj.weight": "ffn_up_shexp.weight",
+    "ffn.shared_experts.w3.weight": "ffn_up_shexp.weight",
+    "moe.shared_experts.down_proj.weight": "ffn_down_shexp.weight",
+    "mlp.shared_experts.down_proj.weight": "ffn_down_shexp.weight",
+    "ffn.shared_experts.w2.weight": "ffn_down_shexp.weight",
+}
+
+_EXPERT_PROJ_TO_GGUF = {
+    "gate_proj": "ffn_gate_exps.weight",
+    "w1": "ffn_gate_exps.weight",
+    "fc1": "ffn_gate_exps.weight",
+    "up_proj": "ffn_up_exps.weight",
+    "w3": "ffn_up_exps.weight",
+    "down_proj": "ffn_down_exps.weight",
+    "w2": "ffn_down_exps.weight",
+    "fc2": "ffn_down_exps.weight",
+}
+
+
+def _gguf_align(val: int, alignment: int = _GGUF_ALIGNMENT) -> int:
+    return (val + alignment - 1) & ~(alignment - 1)
+
+
+def _pack_gguf_str(s: str) -> bytes:
+    b = s.encode("utf-8")
+    return struct.pack("<Q", len(b)) + b
+
+
+def _build_gguf_kv_pairs(
+    arch: dict,
+    tokenizer_path: Path | None,
+    chat_template_path: Path | None,
+    quant_format: str,
+    model_name: str,
+) -> list[tuple[str, int, bytes]]:
+    raw_mtype = str(arch.get("model_type", "llama")).lower()
+    if "deepseek" in raw_mtype:
+        g_arch = "deepseek2"
+    elif "qwen" in raw_mtype:
+        g_arch = "qwen2moe" if int(arch.get("num_experts") or 0) > 0 else "qwen2"
+    else:
+        g_arch = raw_mtype
+
+    ftype_map = {"bf16": 32, "q4_0": 2, "q2_k": 10, "mixed": 10}
+    kvs: list[tuple[str, int, bytes]] = [
+        ("general.architecture", 8, _pack_gguf_str(g_arch)),
+        ("general.name", 8, _pack_gguf_str(model_name)),
+        ("general.file_type", 4, struct.pack("<I", ftype_map.get(quant_format, 10))),
+        ("general.quantization_version", 4, struct.pack("<I", 2)),
+    ]
+
+    def add_u32(k: str, v: int) -> None:
+        if v is not None and int(v) > 0:
+            kvs.append((k, 4, struct.pack("<I", int(v))))
+
+    def add_f32(k: str, v: float) -> None:
+        if v is not None:
+            kvs.append((k, 6, struct.pack("<f", float(v))))
+
+    add_u32(f"{g_arch}.block_count", int(arch.get("num_hidden_layers", 0)))
+    add_u32(f"{g_arch}.embedding_length", int(arch.get("hidden_size", 0)))
+    add_u32(
+        f"{g_arch}.feed_forward_length",
+        int(arch.get("intermediate_size") or arch.get("moe_intermediate_size") or 0),
+    )
+    add_u32(f"{g_arch}.expert_feed_forward_length", int(arch.get("moe_intermediate_size") or 0))
+    add_u32(f"{g_arch}.attention.head_count", int(arch.get("num_attention_heads", 0)))
+    add_u32(
+        f"{g_arch}.attention.head_count_kv",
+        int(arch.get("num_key_value_heads") or arch.get("num_attention_heads") or 0),
+    )
+    add_f32(
+        f"{g_arch}.attention.layer_norm_rms_epsilon",
+        float(arch.get("rms_norm_eps", 1.0e-5)),
+    )
+    add_u32(f"{g_arch}.expert_count", int(arch.get("num_experts") or 0))
+    add_u32(
+        f"{g_arch}.expert_used_count",
+        int(arch.get("num_experts_per_token") or arch.get("num_experts_per_tok") or 0),
+    )
+    add_u32(f"{g_arch}.context_length", int(arch.get("max_position_embeddings", 32768)))
+    add_u32(f"{g_arch}.vocab_size", int(arch.get("vocab_size", 0)))
+    add_u32(f"{g_arch}.attention.q_lora_rank", int(arch.get("q_lora_rank") or 0))
+    add_u32(f"{g_arch}.attention.kv_lora_rank", int(arch.get("kv_lora_rank") or 0))
+
+    vocab_sz = int(arch.get("vocab_size", 0))
+    if tokenizer_path is not None and tokenizer_path.exists():
+        try:
+            tok_data = json.loads(tokenizer_path.read_text())
+            vmap: dict[str, int] = tok_data.get("model", {}).get("vocab", {})
+            added = tok_data.get("added_tokens", [])
+            max_id = max(
+                [vocab_sz - 1]
+                + list(vmap.values())
+                + [int(a.get("id", -1)) for a in added]
+            )
+            n_tok = max(max_id + 1, 1)
+            tokens = [f"<|tok_{i}|>" for i in range(n_tok)]
+            ttypes = np.ones(n_tok, dtype="<i4")
+            for s, tid in vmap.items():
+                if 0 <= tid < n_tok:
+                    tokens[tid] = s
+            for a in added:
+                tid = int(a.get("id", -1))
+                if 0 <= tid < n_tok:
+                    tokens[tid] = str(a.get("content", tokens[tid]))
+                    if a.get("special", False):
+                        ttypes[tid] = 3
+
+            kvs.append(("tokenizer.ggml.model", 8, _pack_gguf_str("gpt2")))
+            kvs.append(("tokenizer.ggml.pre", 8, _pack_gguf_str("default")))
+            tok_buf = bytearray(struct.pack("<IQ", 8, len(tokens)))
+            for s in tokens:
+                tok_buf.extend(_pack_gguf_str(s))
+            kvs.append(("tokenizer.ggml.tokens", 9, bytes(tok_buf)))
+            kvs.append(
+                (
+                    "tokenizer.ggml.token_type",
+                    9,
+                    struct.pack("<IQ", 5, len(tokens)) + ttypes.tobytes(),
+                )
+            )
+            merges = tok_data.get("model", {}).get("merges", [])
+            if merges:
+                m_buf = bytearray(struct.pack("<IQ", 8, len(merges)))
+                for m in merges:
+                    ms = " ".join(m) if isinstance(m, list) else str(m)
+                    m_buf.extend(_pack_gguf_str(ms))
+                kvs.append(("tokenizer.ggml.merges", 9, bytes(m_buf)))
+        except Exception:
+            pass
+
+    if "bos_token_id" in arch and arch["bos_token_id"] is not None:
+        kvs.append(("tokenizer.ggml.bos_token_id", 4, struct.pack("<I", int(arch["bos_token_id"]))))
+    if "eos_token_id" in arch and arch["eos_token_id"] is not None:
+        kvs.append(("tokenizer.ggml.eos_token_id", 4, struct.pack("<I", int(arch["eos_token_id"]))))
+    if "pad_token_id" in arch and arch["pad_token_id"] is not None:
+        kvs.append(("tokenizer.ggml.padding_token_id", 4, struct.pack("<I", int(arch["pad_token_id"]))))
+    if chat_template_path is not None and chat_template_path.exists():
+        kvs.append(("tokenizer.chat_template", 8, _pack_gguf_str(chat_template_path.read_text())))
+
+    return kvs
+
+
+def export_safetensors_to_gguf(
+    safetensors_path: str | Path,
+    output_gguf: str | Path,
+    config: str | Path | None = None,
+    quant_format: str = "mixed",
+    device: str = "auto",
+    verbose: bool = True,
+    progress_cb=None,
+    stop_event=None,
+) -> dict:
+    """Convert a SafeTensors model (including FP4/FP8 DeepSeek-V4.1, Raon, Llama, Qwen, Mixtral) to GGUF.
+
+    * Unpacks FP4 (``I8`` + ``.scale``) and FP8 (``F8_E4M3`` + ``.scale`` / ``.weight_scale_inv``)
+      tensor pairs on the fly.
+    * Stacks routed MoE experts ``0..E-1`` per layer into standard 3D GGUF expert tensors
+      (``blk.{l}.ffn_{gate,up,down}_exps.weight``).
+    * Quantizes eligible weights using HPC-Quantize V2 (``mixed``, ``q2_k``, ``q4_0``, or ``bf16``)
+      with cross-layer payload caching for identical CheckFree anchor layers.
+    """
+    if safe_open is None or torch is None:
+        raise RuntimeError("safetensors + torch are required for GGUF export")
+
+    st_path = Path(safetensors_path)
+    out_path = Path(output_gguf)
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+    cfg_path = Path(config) if config else st_path.parent / "config.json"
+    raw_cfg = json.loads(cfg_path.read_text()) if cfg_path.exists() else {}
+    arch = _normalize_arch_config(raw_cfg)
+    tok_path = cfg_path.parent / "tokenizer.json"
+    if not tok_path.exists():
+        tok_path = st_path.parent / "tokenizer.json"
+    chat_path = cfg_path.parent / "chat_template.jinja"
+    if not chat_path.exists():
+        chat_path = st_path.parent / "chat_template.jinja"
+
+    use_cuda = (device in ("auto", "cuda")) and torch.cuda.is_available()
+    dev = torch.device("cuda:0" if use_cuda else "cpu")
+    lut = torch.tensor(_FP4_LUT_CPU, dtype=torch.float32, device=dev)
+
+    hpc_cfg = HPCConfig(
+        device="cuda" if use_cuda else "cpu",
+        quant_format=quant_format if quant_format in ("mixed", "q2_k", "q4_0") else "mixed",
+        verbose=False,
+    ).validate()
+    quantizer = Q2KQuantizer(hpc_cfg)
+
+    with safe_open(str(st_path), framework="pt", device="cpu") as f:
+        keys = sorted(f.keys())
+        keys_set = set(keys)
+
+        # Group routed experts by (layer_idx, proj) -> sorted [(expert_idx, weight_key, scale_key)]
+        expert_groups: dict[tuple[int, str], list[tuple[int, str, str | None]]] = {}
+        consumed_keys: set[str] = set()
+
+        for k in keys:
+            if k.endswith((".scale", ".weight_scale_inv", ".scales", ".weight_scale")):
+                base_w = k.rsplit(".", 1)[0] + ".weight"
+                if base_w in keys_set:
+                    consumed_keys.add(k)
+                    continue
+            em = EXPERT_RE.match(k)
+            if em and em.group(6) == "weight":
+                lidx = int(em.group(2))
+                eidx = int(em.group(4))
+                proj = em.group(5)
+                sc_k = None
+                for sfx in (".scale", ".weight_scale_inv", ".scales", ".weight_scale"):
+                    cand_s = k[:-7] + sfx
+                    if cand_s in keys_set:
+                        sc_k = cand_s
+                        consumed_keys.add(cand_s)
+                        break
+                expert_groups.setdefault((lidx, proj), []).append((eidx, k, sc_k))
+                consumed_keys.add(k)
+
+        for gk in expert_groups:
+            expert_groups[gk].sort(key=lambda x: x[0])
+
+        # Build work items: ("single", src_key, scale_key, gguf_name) or ("experts", (lidx, proj), items, gguf_name)
+        work_items: list[tuple] = []
+        for k in keys:
+            if k in consumed_keys:
+                continue
+            sc_k = None
+            if k.endswith(".weight"):
+                for sfx in (".scale", ".weight_scale_inv", ".scales", ".weight_scale"):
+                    cand_s = k[:-7] + sfx
+                    if cand_s in keys_set:
+                        sc_k = cand_s
+                        break
+            if _is_embed_key(k):
+                gname = "token_embd.weight"
+            elif _is_head_key(k):
+                gname = "output.weight"
+            elif k in _NORM_NAMES:
+                gname = "output_norm.weight"
+            else:
+                lm = LAYER_RE.match(k)
+                if lm:
+                    lidx = int(lm.group(2))
+                    suf = lm.group(3)
+                    mapped = _LAYER_SUFFIX_TO_GGUF.get(suf, suf)
+                    gname = f"blk.{lidx}.{mapped}"
+                else:
+                    gname = k
+            work_items.append(("single", k, sc_k, gname))
+
+        for (lidx, proj), members in sorted(expert_groups.items()):
+            mapped_proj = _EXPERT_PROJ_TO_GGUF.get(proj, f"ffn_{proj}_exps.weight")
+            gname = f"blk.{lidx}.{mapped_proj}"
+            work_items.append(("experts", (lidx, proj), members, gname))
+
+        # Stream quantized/converted tensor payloads into a temporary data file and record GGUF directory
+        tmp_data = tempfile.NamedTemporaryFile(delete=False, dir=str(out_path.parent))
+        gguf_tensors: list[tuple[str, tuple[int, ...], int, int]] = []
+        data_offset = 0
+        payload_cache: dict[tuple, tuple[int, tuple[int, ...], int, int]] = {}
+        stats = {"tensors": 0, "q2_k": 0, "q4_0": 0, "bf16": 0, "f32": 0}
+        total_items = len(work_items)
+
+        def _dequant_single_to_f32_torch(w_key: str, s_key: str | None) -> "torch.Tensor":
+            wt = f.get_tensor(w_key).to(dev)
+            dt_s = _safetensors_dtype(f.get_slice(w_key).get_dtype())
+            if s_key is not None and s_key in keys_set:
+                st = f.get_tensor(s_key).to(dev)
+                if dt_s == "I8":
+                    return _ds4_dequant_fp4_1x32(wt, st, lut)
+                if dt_s.startswith("F8"):
+                    if st.ndim == 2 and wt.shape[0] // st.shape[0] == 32 and wt.shape[1] // st.shape[1] == 32:
+                        return _ds4_dequant_fp8_32x32(wt, st)
+                    return _dequant_mxfp8_torch(wt, st)
+            return wt.float()
+
+        def _choose_gguf_target(gname: str, torch_shape: tuple[int, ...]) -> str:
+            if len(torch_shape) < 2 or "norm" in gname or gname.endswith(".bias") or "ffn_gate_inp" in gname or "hc_" in gname or "attn_sink" in gname:
+                return "F32" if len(torch_shape) == 1 else "BF16"
+            if quant_format == "bf16":
+                return "BF16"
+            row_w = int(torch_shape[-1])
+            n_el = int(np.prod(torch_shape))
+            is_emb = gname in ("token_embd.weight", "output.weight")
+            if quant_format == "q4_0":
+                return "Q4_0" if (row_w % QK4_0 == 0 and n_el % QK4_0 == 0) else "BF16"
+            if quant_format == "q2_k":
+                if not is_emb and row_w % QK_K == 0 and n_el % QK_K == 0:
+                    return "Q2_K"
+                return "BF16"
+            # mixed
+            if is_emb:
+                return "Q4_0" if (row_w % QK4_0 == 0 and n_el % QK4_0 == 0) else "BF16"
+            if row_w % QK_K == 0 and n_el % QK_K == 0:
+                return "Q2_K"
+            if row_w % QK4_0 == 0 and n_el % QK4_0 == 0:
+                return "Q4_0"
+            return "BF16"
+
+        def _encode_tensor_f32(t_f32: "torch.Tensor", tgt: str) -> tuple[int, bytes]:
+            row_w = int(t_f32.shape[-1])
+            if tgt == "F32":
+                return _GGML_F32, t_f32.detach().cpu().numpy().astype("<f4").tobytes()
+            if tgt == "BF16":
+                b16 = t_f32.to(torch.bfloat16).view(torch.uint16).detach().cpu().numpy().astype("<u2").tobytes()
+                return _GGML_BF16, b16
+            if tgt == "Q4_0":
+                total_r = int(np.prod(t_f32.shape[:-1]))
+                t2d = t_f32.reshape(total_r, row_w)
+                chunk_r = max(1, (8 * 1024 * 1024) // max(row_w, 1))
+                out_chunks = []
+                for r0 in range(0, total_r, chunk_r):
+                    sub = t2d[r0 : min(r0 + chunk_r, total_r)].detach().cpu().numpy().astype(np.float32)
+                    raw_q4, _, _ = quantize_tensor_q4_0(sub, cfg=hpc_cfg)
+                    out_chunks.append(raw_q4)
+                return _GGML_Q4_0, b"".join(out_chunks)
+            if tgt == "Q2_K":
+                flat_np = t_f32.detach().cpu().numpy().astype(np.float32).reshape(-1)
+                res = quantizer.quantize(flat_np, row_width=row_w)
+                return _GGML_Q2_K, blocks_to_bytes(res.scales, res.qs, res.dm)
+            raise ValueError(tgt)
+
+        for idx_w, item in enumerate(work_items, 1):
+            if stop_event is not None and stop_event.is_set():
+                tmp_data.close()
+                os.unlink(tmp_data.name)
+                raise InterruptedError("GGUF export cancelled by user.")
+
+            kind = item[0]
+            if kind == "single":
+                _, w_key, s_key, gname = item
+                sl = f.get_slice(w_key)
+                t_shape = tuple(sl.get_shape())
+                g_dims = tuple(reversed(t_shape)) if t_shape else (1,)
+                tgt = _choose_gguf_target(gname, t_shape or (1,))
+
+                # Fast signature for identical anchor layer reuse
+                lm = LAYER_RE.match(w_key)
+                cache_key = None
+                if lm:
+                    suf = lm.group(3)
+                    w_Sample = _payload_bytes(f.get_tensor(w_key).flatten()[:256])
+                    cache_key = ("single", suf, t_shape, tgt, w_Sample)
+                if cache_key is not None and cache_key in payload_cache:
+                    out_type, g_dims, src_off, src_len = payload_cache[cache_key]
+                    raw_bytes = os.pread(tmp_data.fileno(), src_len, src_off)
+                else:
+                    t_f32 = _dequant_single_to_f32_torch(w_key, s_key)
+                    if not t_shape:
+                        t_f32 = t_f32.reshape(1)
+                    out_type, raw_bytes = _encode_tensor_f32(t_f32, tgt)
+                    del t_f32
+            else:
+                _, (lidx, proj), members, gname = item
+                first_w = members[0][1]
+                e_shape = tuple(f.get_slice(first_w).get_shape())
+                # If packed FP4 I8, logical in_dim is 2 * stored in_dim
+                dt_e = _safetensors_dtype(f.get_slice(first_w).get_dtype())
+                if dt_e == "I8" and members[0][2] is not None:
+                    e_shape = (e_shape[0], e_shape[1] * 2)
+                full_shape = (len(members), e_shape[0], e_shape[1])
+                g_dims = (e_shape[1], e_shape[0], len(members))
+                tgt = _choose_gguf_target(gname, full_shape)
+
+                w_sample = _payload_bytes(f.get_tensor(first_w).flatten()[:256])
+                cache_key = ("experts", proj, full_shape, tgt, w_sample)
+                if cache_key in payload_cache:
+                    out_type, g_dims, src_off, src_len = payload_cache[cache_key]
+                    raw_bytes = os.pread(tmp_data.fileno(), src_len, src_off)
+                else:
+                    chunks = []
+                    for _, wk, sk in members:
+                        ef32 = _dequant_single_to_f32_torch(wk, sk)
+                        _, eb = _encode_tensor_f32(ef32, tgt)
+                        chunks.append(eb)
+                        del ef32
+                    raw_bytes = b"".join(chunks)
+                    out_type = (
+                        _GGML_Q2_K
+                        if tgt == "Q2_K"
+                        else (_GGML_Q4_0 if tgt == "Q4_0" else (_GGML_BF16 if tgt == "BF16" else _GGML_F32))
+                    )
+
+            pad = _gguf_align(data_offset) - data_offset
+            if pad > 0:
+                tmp_data.write(b"\x00" * pad)
+                data_offset += pad
+            cur_off = data_offset
+            tmp_data.write(raw_bytes)
+            tmp_data.flush()
+            data_offset += len(raw_bytes)
+
+            if cache_key is not None and cache_key not in payload_cache:
+                payload_cache[cache_key] = (out_type, g_dims, cur_off, len(raw_bytes))
+
+            gguf_tensors.append((gname, g_dims, out_type, cur_off))
+            stats["tensors"] += 1
+            if out_type == _GGML_Q2_K:
+                stats["q2_k"] += 1
+            elif out_type == _GGML_Q4_0:
+                stats["q4_0"] += 1
+            elif out_type == _GGML_BF16:
+                stats["bf16"] += 1
+            else:
+                stats["f32"] += 1
+
+            if progress_cb is not None and (idx_w % 8 == 0 or idx_w == total_items):
+                progress_cb(idx_w, total_items, gname, tgt, stats)
+
+    # Update effective expert count in arch metadata if stacked experts were pruned
+    if expert_groups:
+        first_group = next(iter(expert_groups.values()))
+        arch["num_experts"] = len(first_group)
+
+    kv_pairs = _build_gguf_kv_pairs(
+        arch,
+        tok_path if tok_path.exists() else None,
+        chat_path if chat_path.exists() else None,
+        quant_format,
+        st_path.stem,
+    )
+
+    with out_path.open("wb") as fout:
+        fout.write(struct.pack("<IIQQ", _GGUF_MAGIC, 3, len(gguf_tensors), len(kv_pairs)))
+        for k, vt, raw_v in kv_pairs:
+            fout.write(_pack_gguf_str(k))
+            fout.write(struct.pack("<I", vt))
+            fout.write(raw_v)
+        for gname, g_dims, out_type, off in gguf_tensors:
+            fout.write(_pack_gguf_str(gname))
+            fout.write(struct.pack("<I", len(g_dims)))
+            for d in g_dims:
+                fout.write(struct.pack("<Q", int(d)))
+            fout.write(struct.pack("<I", int(out_type)))
+            fout.write(struct.pack("<Q", int(off)))
+        data_start = _gguf_align(fout.tell())
+        if data_start > fout.tell():
+            fout.write(b"\x00" * (data_start - fout.tell()))
+        tmp_data.seek(0)
+        shutil.copyfileobj(tmp_data, fout, length=16 * 1024 * 1024)
+
+    tmp_data.close()
+    os.unlink(tmp_data.name)
+    stats["output_bytes"] = out_path.stat().st_size
+    if verbose:
+        print(
+            f"Exported GGUF -> {out_path} ({stats['output_bytes'] / 1e9:.2f} GB, "
+            f"{stats['tensors']} tensors: Q2_K={stats['q2_k']}, Q4_0={stats['q4_0']}, "
+            f"BF16={stats['bf16']}, F32={stats['f32']})"
+        )
+    return stats
+
+
 def main() -> None:
     ap = argparse.ArgumentParser(description="HPC SlimShard / CheckFree single-shard LLM recovery")
     sub = ap.add_subparsers(dest="cmd", required=False)