| |
| """合成迷你 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() |
|
|
