File size: 5,314 Bytes
09ccad2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | #!/usr/bin/env python3
"""候选学生（GGUF 量化 / uncensored 微调 / 任意 HF 目录）vs bf16 原版 的特征对齐测试。
这是"换学生"的照妖镜——适配器训练时的输入是 bf16 原版的 hidden_states[-1]，
候选版隐藏态的任何漂移都会直接传导到生成的 conditioning。
用法:
python check_alignment.py <bf16 原版目录> <候选: *.gguf 文件 或 HF 目录>
[--prompts 3] [--device cuda]
判定:
cos > 0.99 合格（量化级噪声，直接用）
0.95-0.99 有可见漂移，评估后可用（建议先跑一次端到端生成对比）
< 0.95 漂移显著，需用候选学生重提特征 + 重训/微调适配器
说明:
GGUF 走 transformers 内置加载器（>=4.45，AutoModel.from_pretrained(".gguf")），
隐藏态语义与 bf16 原版完全一致（都是 transformers 路径），测的是量化+微调漂移本身。
"""
from __future__ import annotations
import argparse
import os
import sys
import torch
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
from student import StudentTextEncoder, load_student, find_language_model # noqa: E402
from adapter_clip import H3_HIDDEN # noqa: E402
def load_candidate(path: str, device: torch.device, baseline_mem: float = 0.0):
from transformers import AutoConfig, AutoModel
if path.endswith(".gguf"):
from gguf_qwen35 import load_gguf_model_quantized
model, info = load_gguf_model_quantized(path)
print(f" [candidate] 自定义 GGUF 加载器(量化驻留): {info}", flush=True)
model = model.to(torch.bfloat16).to(device) # GGMLTensor 的 to 忽略 dtype, 只移设备
if device.type == "cuda":
used = torch.cuda.memory_allocated() / 2**30 - baseline_mem
print(f" [candidate] 净显存: {used:.2f} GiB", flush=True)
return model
# HF 目录
cfg = AutoConfig.from_pretrained(path)
model = AutoModel.from_config(cfg)
try:
sd = {k[len("model."):] if k.startswith("model.") else k: v
for k, v in torch.load(os.path.join(path, "pytorch_model.bin"), map_location="cpu").items()}
missing, _ = model.load_state_dict(sd, strict=False)
assert not missing, f"missing: {missing[:5]}"
except FileNotFoundError:
model = AutoModel.from_pretrained(path, torch_dtype=torch.bfloat16)
return model.to(device)
@torch.no_grad()
def hidden_fn(lm, tok, prompt: str, device: torch.device) -> torch.Tensor:
ids = tok(prompt, add_special_tokens=False)["input_ids"]
out = lm(input_ids=torch.tensor([ids], device=device), output_hidden_states=True)
return out.hidden_states[-1][0].float() # [S, D] 与训练同路径
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("reference", help="bf16 原版 Qwen3.5-4B HF 目录")
ap.add_argument("candidate", help="候选: .gguf 或 HF 目录")
ap.add_argument("--prompts", type=int, default=3)
ap.add_argument("--device", default="")
args = ap.parse_args()
from transformers import AutoTokenizer
dev = torch.device(args.device) if args.device else torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"device: {dev}")
print("[1/2] loading reference (bf16)...", flush=True)
ref_model = load_student(args.reference).to(torch.bfloat16).to(dev)
ref_lm = find_language_model(ref_model)
ref_tok = AutoTokenizer.from_pretrained(args.reference)
print("[2/2] loading candidate...", flush=True)
baseline = torch.cuda.memory_allocated() / 2**30 if dev.type == "cuda" else 0.0
cand_model = load_candidate(args.candidate, dev, baseline)
cand_lm = find_language_model(cand_model)
# .gguf 候选无自带 tokenizer: 复用参考模型的（同一 Qwen3.5-4B 词表）
cand_tok = ref_tok if args.candidate.endswith(".gguf") else AutoTokenizer.from_pretrained(args.candidate)
demo = [
"A cinematic shot of a tiny robot repairing a broken music box in a giant's attic",
"A surfer riding a massive neon wave at sunset, spray frozen in the air, dramatic backlight",
"Paper collage birds flocking over a misty city skyline, soft morning light",
]
print(f"\ncosine 对比 {args.prompts} 条 prompt（hidden_states[-1], [S, 2560]）:")
worst = 1.0
for i, p in enumerate(demo[: args.prompts]):
h_ref = hidden_fn(ref_lm, ref_tok, p, dev)
h_can = hidden_fn(cand_lm, cand_tok, p, dev)
n = min(h_ref.shape[0], h_can.shape[0])
if h_ref.shape[0] != h_can.shape[0]:
print(f" [注意] prompt {i}: 长度不同 ref={h_ref.shape[0]} cand={h_can.shape[0]}（截断到 {n} 比较）")
a, b = h_ref[:n].flatten(), h_can[:n].flatten()
cos = (a * b).sum() / (a.norm() * b.norm())
worst = min(worst, cos.item())
print(f" prompt {i}: cos={cos.item():.4f}")
print(f"\n最差 cos = {worst:.4f} -> ", end="")
if worst > 0.99:
print("合格（量化级噪声，直接用）✅")
elif worst >= 0.95:
print("有可见漂移，评估后可用（建议先端到端生成对比）⚠️")
else:
print("漂移显著，需用候选学生重提特征 + 重训/微调适配器 ❌")
if __name__ == "__main__":
main()
|