#!/usr/bin/env python3 """合成迷你 GGUF 回环测试: 复刻真实 HauhauCS GGUF 的全部布局约定 （2D 数据 [out,in] + 元数据 [in,out]、v-head value-major 置换、范数增量 -1、ssm_a 存 -A）， 验证 gguf_qwen35 加载器的映射/反量化/置换/变换正确，含数值断言（cos）。 量化类型用 Q8_0（gguf 库支持量化；Q4_K 反量化已由真实文件 426/426 逐张量验证）。 用法: python test_gguf_loader.py [--keep]（--keep 保留临时文件供 inspect_gguf 查看） """ from __future__ import annotations import argparse import os import sys import tempfile import numpy as np import torch HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) N_EMBD = 2560 FFN = 9216 N_HEAD, N_HEAD_KV, HEAD_DIM = 16, 4, 256 L_KEY_HEADS, L_KEY_HD, L_V_HEADS, L_V_HD = 16, 128, 32, 128 CONV_K = 4 GGUF_TO_HF = np.array([(i % 16) * 2 + (i // 16) for i in range(32)], dtype=np.int64) def hf_to_gguf(x): return x[GGUF_TO_HF] def w2d(rng, out, inn): return rng.standard_normal((out, inn)).astype(np.float32) def add2d(w, name, data, qtype=None): from gguf.quants import quantize from gguf import GGML_QUANT_SIZES out, inn = int(data.shape[0]), int(data.shape[1]) if qtype is not None: data = quantize(np.ascontiguousarray(data), qtype) block, tsize = GGML_QUANT_SIZES[qtype] w.add_tensor(name, data, raw_shape=(out, inn * tsize // block), raw_dtype=qtype) else: w.add_tensor(name, data, raw_shape=(out, inn)) def add1d(w, name, data): w.add_tensor(name, data) def add_layer(w, rng, i: int, is_full: bool, expect: dict): base = f"blk.{i}" pref = f"language_model.layers.{i}." norm_in = w2d(rng, N_EMBD, 1).ravel() + 0.5 norm_out = w2d(rng, N_EMBD, 1).ravel() + 0.5 add1d(w, f"{base}.attn_norm.weight", norm_in) add1d(w, f"{base}.post_attention_norm.weight", norm_out) expect[f"{pref}input_layernorm.weight"] = norm_in - 1.0 expect[f"{pref}post_attention_layernorm.weight"] = norm_out - 1.0 gate = w2d(rng, FFN, N_EMBD) up = w2d(rng, FFN, N_EMBD) down = w2d(rng, N_EMBD, FFN) add2d(w, f"{base}.ffn_gate.weight", gate, qtype=Q8) add2d(w, f"{base}.ffn_up.weight", up, qtype=Q8) add2d(w, f"{base}.ffn_down.weight", down, qtype=Q8) expect[f"{pref}mlp.gate_proj.weight"] = gate expect[f"{pref}mlp.up_proj.weight"] = up expect[f"{pref}mlp.down_proj.weight"] = down if is_full: q = w2d(rng, 2 * N_HEAD * HEAD_DIM, N_EMBD) k = w2d(rng, N_HEAD_KV * HEAD_DIM, N_EMBD) v = w2d(rng, N_HEAD_KV * HEAD_DIM, N_EMBD) o = w2d(rng, N_EMBD, N_HEAD * HEAD_DIM) qn = w2d(rng, HEAD_DIM, 1).ravel() + 0.5 kn = w2d(rng, HEAD_DIM, 1).ravel() + 0.5 add2d(w, f"{base}.attn_q.weight", q, qtype=Q8) add2d(w, f"{base}.attn_k.weight", k, qtype=Q8) add2d(w, f"{base}.attn_v.weight", v, qtype=Q8) add2d(w, f"{base}.attn_output.weight", o, qtype=Q8) add1d(w, f"{base}.attn_q_norm.weight", qn) add1d(w, f"{base}.attn_k_norm.weight", kn) expect[f"{pref}self_attn.q_proj.weight"] = q expect[f"{pref}self_attn.k_proj.weight"] = k expect[f"{pref}self_attn.v_proj.weight"] = v expect[f"{pref}self_attn.o_proj.weight"] = o expect[f"{pref}self_attn.q_norm.weight"] = qn - 1.0 expect[f"{pref}self_attn.k_norm.weight"] = kn - 1.0 else: k_dim, v_dim = L_KEY_HEADS * L_KEY_HD, L_V_HEADS * L_V_HD conv_dim = k_dim * 2 + v_dim qkv = w2d(rng, conv_dim, N_EMBD) qkv_hf = qkv.copy() qkv[v_dim:] = hf_to_gguf(qkv[v_dim:].reshape(32, 128, -1)).reshape(v_dim, -1) z = w2d(rng, v_dim, N_EMBD) z_hf = z.copy() z = hf_to_gguf(z.reshape(32, 128, -1)).reshape(v_dim, -1) conv = w2d(rng, conv_dim, CONV_K) conv_hf = conv.copy() conv[v_dim:] = hf_to_gguf(conv[v_dim:].reshape(32, 128, -1)).reshape(v_dim, -1) dt_hf = rng.standard_normal(L_V_HEADS).astype(np.float32) dt = hf_to_gguf(dt_hf) a_log = rng.standard_normal(L_V_HEADS).astype(np.float32) beta_hf = w2d(rng, L_V_HEADS, N_EMBD) beta = hf_to_gguf(beta_hf) alpha_hf = w2d(rng, L_V_HEADS, N_EMBD) alpha = hf_to_gguf(alpha_hf) s_norm = w2d(rng, L_V_HD, 1).ravel() + 0.5 out_proj = w2d(rng, N_EMBD, v_dim) out_proj_gguf = out_proj.reshape(N_EMBD, 32, 128)[:, GGUF_TO_HF, :].reshape(N_EMBD, v_dim) add2d(w, f"{base}.attn_qkv.weight", qkv, qtype=Q8) add2d(w, f"{base}.attn_gate.weight", z, qtype=Q8) w.add_tensor(f"{base}.ssm_conv1d.weight", conv, raw_shape=(conv_dim, CONV_K)) add1d(w, f"{base}.ssm_dt.bias", dt) add1d(w, f"{base}.ssm_a", hf_to_gguf(-np.exp(a_log)).astype(np.float32)) add2d(w, f"{base}.ssm_beta.weight", beta, qtype=Q8) add2d(w, f"{base}.ssm_alpha.weight", alpha, qtype=Q8) add1d(w, f"{base}.ssm_norm.weight", s_norm) add2d(w, f"{base}.ssm_out.weight", out_proj_gguf, qtype=Q8) expect[f"{pref}linear_attn.in_proj_qkv.weight"] = qkv_hf expect[f"{pref}linear_attn.in_proj_z.weight"] = z_hf expect[f"{pref}linear_attn.conv1d.weight"] = conv_hf.reshape(conv_dim, 1, CONV_K) expect[f"{pref}linear_attn.dt_bias"] = dt_hf expect[f"{pref}linear_attn.A_log"] = a_log expect[f"{pref}linear_attn.in_proj_b.weight"] = beta_hf expect[f"{pref}linear_attn.in_proj_a.weight"] = alpha_hf expect[f"{pref}linear_attn.norm.weight"] = s_norm expect[f"{pref}linear_attn.out_proj.weight"] = out_proj def main() -> None: global Q8 from gguf import GGUFWriter, GGMLQuantizationType Q8 = GGMLQuantizationType.Q8_0 ap = argparse.ArgumentParser() ap.add_argument("--keep", action="store_true") args = ap.parse_args() tmpdir = tempfile.mkdtemp(prefix="gguf_test_") path = os.path.join(tmpdir, "toy_qwen35.gguf") print(f"写入合成 GGUF -> {path}") rng = np.random.default_rng(42) expect: dict[str, np.ndarray] = {} w = GGUFWriter(path, "qwen35") emb = w2d(rng, 248320, N_EMBD) add2d(w, "token_embd.weight", emb) expect["language_model.embed_tokens.weight"] = emb out_norm = w2d(rng, N_EMBD, 1).ravel() + 0.5 add1d(w, "output_norm.weight", out_norm) expect["language_model.norm.weight"] = out_norm - 1.0 add_layer(w, rng, 0, is_full=False, expect=expect) add_layer(w, rng, 3, is_full=True, expect=expect) w.write_header_to_file() w.write_kv_data_to_file() w.write_tensors_to_file() w.close() print(f"GGUF 写入完成 ({os.path.getsize(path) / 1e6:.0f} MB)") from gguf_qwen35 import load_gguf_state_dict, build_hf_model sd = load_gguf_state_dict(path) print(f"映射出 {len(sd)} 个键") fails = 0 for key, want in sorted(expect.items()): if key not in sd: print(f" [FAIL] {key} MISSING") fails += 1 continue got = sd[key] ws = torch.from_numpy(want).float() gs = got.float() if tuple(gs.shape) != tuple(ws.shape): print(f" [FAIL] {key} shape {tuple(gs.shape)} 期望 {tuple(ws.shape)}") fails += 1 continue dot = torch.zeros((), dtype=torch.float64) na = torch.zeros((), dtype=torch.float64) nb = torch.zeros((), dtype=torch.float64) CH = 1 << 22 ga, wf = gs.flatten(), ws.flatten() for i in range(0, ga.numel(), CH): x, y = ga[i:i + CH], wf[i:i + CH] dot += (x * y).sum().double() na += (x * x).sum().double() nb += (y * y).sum().double() c = float(dot / (na.sqrt() * nb.sqrt() + 1e-30)) ok = c > 0.99 print(f" [{'PASS' if ok else 'FAIL'}] {key} {tuple(gs.shape)} cos={c:.5f}") fails += 0 if ok else 1 model, missing, unexpected = build_hf_model(sd, os.path.join(HERE, "qwen35_config.json")) unexp_ok = len(unexpected) == 0 print(f" [{'PASS' if unexp_ok else 'FAIL'}] load_state_dict unexpected={len(unexpected)} (应=0)" + (f": {unexpected[:4]}" if unexpected else "")) fails += 0 if unexp_ok else 1 print(f" [info] missing={len(missing)}（其余 30 层 + vision 属正常缺省）") if not args.keep: import shutil shutil.rmtree(tmpdir) print("GGUF 回环", "PASS ✅" if fails == 0 else f"FAIL ({fails})") raise SystemExit(0 if fails == 0 else 1) if __name__ == "__main__": main()