From e762753ec18fb15e49eeb37e0ea2fffb08bd31c4 Mon Sep 17 00:00:00 2001 From: xs1997zju <106210907+xs1997zju@users.noreply.github.com> Date: Thu, 8 Oct 2026 12:36:46 +0000 Subject: [PATCH 1/2] [Test] Check GLM-5.3 end-to-end logits alongside loss --- doc/glm53_forward_parity.md | 53 +++++ tests/model/glm53_forward_parity.py | 258 +++++++++++++++++++++ tests/model/prepare_glm53_parity_inputs.py | 82 +++++++ tests/model/test_glm53_text_moe.py | 13 +- tests/model/test_logits_metrics.py | 37 +++ xtuner/_testing/logits.py | 27 +++ 6 files changed, 469 insertions(+), 1 deletion(-) create mode 100644 doc/glm53_forward_parity.md create mode 100644 tests/model/glm53_forward_parity.py create mode 100644 tests/model/prepare_glm53_parity_inputs.py create mode 100644 tests/model/test_logits_metrics.py create mode 100644 xtuner/_testing/logits.py diff --git a/doc/glm53_forward_parity.md b/doc/glm53_forward_parity.md new file mode 100644 index 0000000000..bf6deb273b --- /dev/null +++ b/doc/glm53_forward_parity.md @@ -0,0 +1,53 @@ +# GLM 5.3 Flash forward loss and logits parity + +A scalar cross entropy can hide changes in full-vocabulary logits. The existing real-checkpoint text test now retains its public loss-path check and also checks the last eight next-token positions against native Transformers. `tests/model/glm53_forward_parity.py` adds image/text replay with FSDP, EP and SP using the same preprocessed inputs on both implementations. + +This is eval/no-grad validation: no packing, backward, optimizer step, or MTP. Use a reduced GLM checkpoint that fits a single GPU for the HF reference. The checkpoint is supplied by the caller; it is not downloaded by the test. + +## Backend and precision contract + +XT uses FlashAttention for the vision tower and torch reference DSA/indexer for the language tower. HF explicitly uses eager: Transformers 5.17 rejects `flash_attention_2` for `Glm5NextVisionModel`. This is an implementation/backend comparison, not a claim that both frameworks use identical kernels. BF16 model compute and FP32 CE are used; TF32 is disabled by the replay runner. CPU metric reductions use FP64. + +For each sample, select the final eight supervised positions (or all if fewer than eight), retaining the entire vocabulary at each position. Labels are shifted exactly once. Check shape, finiteness, fixture SHA256, supervised-token count, global positions and target labels before comparing. Sequence padding is excluded. With SP, gather only sampled logits and sum NLL/count within the SP group; EP/world are not loss reduction groups. DP groups deliberately replay identical inputs. + +The logits check requires **each sample** to have relative L2 below 0.05 and cosine above 0.998, independently of the existing loss-vector cosine >0.97 and mean relative loss error <0.03. These are BF16 regression bounds, not bitwise-parity guarantees. The thresholds are fixed in the helper, not adjusted by the replay runner to fit results. Reports include per-sample maximum absolute error as a diagnostic. A CPU regression verifies that a common logits offset, which leaves CE unchanged, fails the logits check. + +## Prepare portable image fixtures + +Run from the repository root in the XTuner environment. Supply absolute paths for your model, XTuner-format image JSONL, media root and a new output directory: + +```bash +python tests/model/prepare_glm53_parity_inputs.py \ + --model /models/GLM-5.3-Flash-25B \ + --jsonl /data/medpix/train.jsonl --media-root /data/medpix \ + --limit 20 --output /results/glm53/inputs +``` + +The preparer uses the real GLM tokenizer/processor once (max pixels 1048576, max length 16384). It writes raw, unshifted `input_ids`, `labels`, `mm_token_type_ids`, `pixel_values`, `image_grid_thw`, and shifted-input `logit_positions` as safetensors. The manifest stores relative filenames, hashes, sample IDs and supervised counts. Dataset/images/weights are external fixtures and are not added to Git. This replay entry point currently supports image inputs, not video/audio. + +## Run sequentially + +Use the same checkpoint for all runs. All output directories must be new; the runner refuses overwrites. Both entry points run from the XTuner checkout with its test dependencies installed; separate processes release model memory between implementations. + +```bash +# Native HF reference on one GPU. +torchrun --standalone --nproc-per-node=1 tests/model/glm53_forward_parity.py \ + --backend hf --model /models/GLM-5.3-Flash-25B \ + --inputs /results/glm53/inputs --output /results/glm53/hf + +# XT FSDP EP1/SP1 baseline. +torchrun --standalone --nproc-per-node=1 tests/model/glm53_forward_parity.py \ + --backend xt --model /models/GLM-5.3-Flash-25B \ + --inputs /results/glm53/inputs --reference /results/glm53/hf \ + --ep-size 1 --sp-size 1 --output /results/glm53/xt-ep1-sp1 + +# Run both SP1 and SP2 on eight GPUs, with distinct output directories. +torchrun --standalone --nproc-per-node=8 tests/model/glm53_forward_parity.py \ + --backend xt --model /models/GLM-5.3-Flash-25B \ + --inputs /results/glm53/inputs --reference /results/glm53/hf \ + --ep-size 8 --sp-size 2 --output /results/glm53/xt-ep8-sp2 +``` + +Set `--sp-size 1` and another output directory for EP8/SP1. The program exits nonzero on input-contract, logits, or loss-check failure. Each run records metadata, sampled tensors and `loss.jsonl`; comparisons also write `comparison.json`. This real-checkpoint replay is opt-in and is not advertised as an automatically provisioned CI test. + +Run the inexpensive metric regressions with `pytest tests/model/test_logits_metrics.py`. The existing real-checkpoint text regression remains in `TestGlm53TextMoEAccuracy.test_fsdp_accuracy`, with `GLM_5_3_FLASH_PATH` and `XTUNER_TEST_WORLD_SIZE` selecting the local checkpoint and GPU count. diff --git a/tests/model/glm53_forward_parity.py b/tests/model/glm53_forward_parity.py new file mode 100644 index 0000000000..69658cf05c --- /dev/null +++ b/tests/model/glm53_forward_parity.py @@ -0,0 +1,258 @@ +"""Replay serialized GLM image/text inputs through HF or XTuner; compare CE and logits. + +Run using torchrun, including --nproc-per-node=1 for the HF reference. +See doc/glm53_forward_parity.md for the input contract and commands. +""" + +import argparse +import hashlib +import json +import os +from datetime import timedelta +from pathlib import Path + +import torch +import torch.distributed as dist +import torch.nn.functional as F +from safetensors.torch import load_file, save_file +from torch.distributed.device_mesh import init_device_mesh +from torch.distributed.tensor import DTensor + +from xtuner._testing.logits import check_logits +from xtuner.v1.config import FSDPConfig +from xtuner.v1.data_proto import SequenceContext +from xtuner.v1.model.compose.glm53 import Glm53BaseConfig + + +def main() -> None: + """所有 DP 组回放相同样本；NLL 和计数只在 SP 组求和一次。""" + parser = argparse.ArgumentParser() + parser.add_argument("--model", required=True) + parser.add_argument("--inputs", type=Path, required=True) + parser.add_argument("--backend", choices=("hf", "xt"), required=True) + parser.add_argument("--reference", type=Path) + parser.add_argument("--ep-size", type=int, default=1) + parser.add_argument("--sp-size", type=int, choices=(1, 2), default=1) + parser.add_argument("--limit", type=int, default=20) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + if args.limit < 1 or args.ep_size < 1: + raise ValueError("limit and ep-size must be positive") + if args.reference is not None: + reference_metadata = json.loads((args.reference / "metadata-rank0.json").read_text()) + if Path(reference_metadata["model"]).resolve() != Path(args.model).resolve(): + raise ValueError("Reference checkpoint path differs") + torch.backends.cuda.matmul.allow_tf32 = False + torch.cuda.set_device(int(os.environ["LOCAL_RANK"])) + dist.init_process_group("nccl", timeout=timedelta(minutes=10)) + rank, world = dist.get_rank(), dist.get_world_size() + if world % args.ep_size or world % args.sp_size: + raise ValueError("EP/SP must divide world size") + if args.backend == "hf" and (world, args.ep_size, args.sp_size) != (1, 1, 1): + raise ValueError("HF reference requires world=EP=SP=1") + if args.output.exists(): + raise FileExistsError(f"请使用新的输出目录：{args.output}") + dist.barrier() + if rank == 0: + args.output.mkdir(parents=True) + dist.barrier() + torch.manual_seed(1234) + mesh = init_device_mesh("cuda", (world // args.sp_size, args.sp_size), mesh_dim_names=("dp", "sp")) + sp_mesh = mesh["sp"] + if args.backend == "hf": + from transformers import AutoConfig, Glm5NextForConditionalGeneration + + config = AutoConfig.from_pretrained(args.model) + config.text_config.num_nextn_predict_layers = 0 + model = Glm5NextForConditionalGeneration.from_pretrained( + args.model, + config=config, + dtype=torch.bfloat16, + device_map="cuda", + attn_implementation="eager", + ) + assert model.config.vision_config._attn_implementation == "eager" + else: + with torch.device("meta"): + cfg = Glm53BaseConfig.from_hf(args.model) + cfg.text_config.mtp_config = None + cfg.text_config.num_nextn_predict_layers = 0 + cfg.text_config.dispatcher = "all2all" + cfg.text_config.ep_size = args.ep_size + cfg.text_config.compile_cfg = False + cfg.compile_cfg = False + cfg.text_config.attention.sparse_mla_backend = "torch" + cfg.text_config.attention.indexer_backend = "torch" + cfg.vision_config.attn_impl = "flash_attention" + model = cfg.build()._to_device_dtype(dtype=torch.bfloat16, skip_buffers_dtype=True) + model.fully_shard(FSDPConfig(ep_size=args.ep_size, cpu_offload=False)) + model.from_hf(args.model, strict=False) + model.eval() + metadata = { + "backend": args.backend, + "model": args.model, + "rank": rank, + "world": world, + "ep": args.ep_size, + "sp": args.sp_size, + "sp_ranks": sp_mesh.mesh.tolist(), + "vision_attention": "eager" if args.backend == "hf" else "flash_attention", + "text_attention": "eager" if args.backend == "hf" else "torch", + "torch": torch.__version__, + "cuda": torch.version.cuda, + "eval": True, + "no_grad": True, + "packing": False, + "mtp": 0, + } + if args.backend == "xt": + ep_meshes = [m.ep_mesh for m in model.modules() if getattr(m, "ep_mesh", None) is not None] + if not ep_meshes or any(m.size() != args.ep_size for m in ep_meshes): + raise ValueError("Actual expert mesh does not match requested EP") + metadata["ep_ranks"] = ep_meshes[0].mesh.tolist() + metadata["expert_shards"] = [ + {"name": n, "global_shape": list(p.shape), "local_shape": list(p.to_local().shape)} + for n, p in model.named_parameters() + if isinstance(p, DTensor) and "experts." in n and "shared_experts" not in n + ][:3] + (args.output / f"metadata-rank{rank}.json").write_text(json.dumps(metadata, indent=2)) + manifest = json.loads((args.inputs / "manifest.json").read_text()) + samples = manifest["samples"][: args.limit] + if len({s["id"] for s in samples}) != len(samples) or len({s["index"] for s in samples}) != len(samples): + raise ValueError("Duplicate input sample IDs or indices") + if len(samples) != args.limit: + raise ValueError("输入样本数不足") + log = (args.output / "loss.jsonl").open("w") if rank == 0 else None + comparisons = [] + reference_rows = None + if args.reference is not None: + reference_rows = [json.loads(line) for line in (args.reference / "loss.jsonl").read_text().splitlines()] + if len(reference_rows) != len(samples): + raise ValueError("Reference sample count differs") + try: + with torch.no_grad(): + for ordinal, sample in enumerate(samples): + path = Path(sample["file"]) + if not path.is_absolute(): + path = args.inputs / path + if hashlib.sha256(path.read_bytes()).hexdigest() != sample["sha256"]: + raise ValueError("输入 hash 改变") + data = load_file(str(path), device=f"cuda:{torch.cuda.current_device()}") + ids = data["input_ids"][:, :-1].contiguous() + labels = data["labels"][:, 1:].contiguous() + ctx = SequenceContext.from_input_ids((ids,), device="cuda") + ctx.mm_token_type_ids = data["mm_token_type_ids"][:, :-1].contiguous() + # split 只补尾部到 SP 的倍数；真实样本仍只有一条，padding 不参与 CE。 + if args.sp_size > 1: + ctx = ctx.split(sp_mesh) + ctx.pixel_values = data["pixel_values"].to(torch.bfloat16) + ctx.image_grid_thw = data["image_grid_thw"] + ctx.num_img_tokens = [[int(ctx.image_grid_thw.prod(-1).sum())]] + if args.backend == "hf": + logits = model( + input_ids=ids, + pixel_values=ctx.pixel_values, + image_grid_thw=data["image_grid_thw"], + use_cache=False, + ).logits + else: + logits = model(seq_ctx=ctx, loss_ctx=None).logits + local_length = ctx.input_ids.shape[1] + start = sp_mesh.get_local_rank() * local_length + padded_labels = F.pad(labels, (0, local_length * args.sp_size - labels.shape[1]), value=-100) + targets = padded_labels[:, start : start + local_length] + if logits.shape[:2] != targets.shape: + raise ValueError(f"logits/labels mismatch: {logits.shape}, {targets.shape}") + valid = targets != -100 + nll = ( + F.cross_entropy(logits[valid].float(), targets[valid], reduction="sum") + if valid.any() + else logits.new_zeros((), dtype=torch.float32) + ) + stats = torch.stack((nll.double(), valid.sum().double())) + # EP 组不是 loss 归约组；DP 是重复验证样本，不再次求和。 + if args.sp_size > 1: + dist.all_reduce(stats, group=sp_mesh.get_group()) + if int(stats[1].item()) != sample["valid_tokens"] or not torch.isfinite(stats).all(): + raise ValueError("全局监督分母或 NLL 非法") + positions = data["logit_positions"] + expected_positions = torch.nonzero(labels[0] != -100).flatten()[-8:] + if not torch.equal(positions, expected_positions): + raise ValueError("Expected the final eight supervised global positions") + local_mask = (positions >= start) & (positions < start + local_length) + chosen = torch.zeros((positions.numel(), logits.shape[-1]), device=logits.device, dtype=torch.float32) + chosen[local_mask] = logits[0, positions[local_mask] - start].float() + if args.sp_size > 1: + dist.all_reduce(chosen, group=sp_mesh.get_group()) + if not torch.isfinite(chosen).all(): + raise ValueError("抽样 logits 非有限") + # 记录不同 DP 复制组的偏差；它们应得到相同样本统计。 + low, high = stats.clone(), stats.clone() + dist.all_reduce(low, op=dist.ReduceOp.MIN) + dist.all_reduce(high, op=dist.ReduceOp.MAX) + row = { + "sample_id": sample["id"], + "index": sample["index"], + "input_sha256": sample["sha256"], + "valid_tokens": int(stats[1].item()), + "nll_sum": stats[0].item(), + "mean_ce": (stats[0] / stats[1]).item(), + "dp_nll_span": (high[0] - low[0]).item(), + "global_input_length": ids.shape[1], + "padded_length": local_length * args.sp_size, + "ep": args.ep_size, + "sp": args.sp_size, + } + if reference_rows is not None: + ref = reference_rows[ordinal] + for key in ("sample_id", "input_sha256", "valid_tokens"): + if row[key] != ref[key]: + raise ValueError(f"Reference contract mismatch: {key}") + reference = load_file(str(args.reference / f"{sample['index']:03d}-logits.safetensors")) + if not torch.equal(reference["positions"], positions.cpu()) or not torch.equal( + reference["labels"], labels[0, positions].cpu() + ): + raise ValueError("Reference positions/labels differ") + metrics = check_logits(chosen, reference["logits"]) + comparisons.append( + { + "sample_id": sample["id"], + "reference_ce": ref["mean_ce"], + "candidate_ce": row["mean_ce"], + **metrics, + } + ) + if rank == 0: + print(json.dumps(row), flush=True) + log.write(json.dumps(row) + "\n") + log.flush() + save_file( + { + "positions": positions.cpu(), + "labels": labels[0, positions].cpu(), + "logits": chosen.cpu().contiguous(), + }, + str(args.output / f"{sample['index']:03d}-logits.safetensors"), + ) + del logits, chosen, data, ctx + if comparisons: + expected = torch.tensor([r["reference_ce"] for r in comparisons], dtype=torch.float64) + actual = torch.tensor([r["candidate_ce"] for r in comparisons], dtype=torch.float64) + mean_relative = ((actual - expected).abs() / expected.abs()).mean().item() + cosine = F.cosine_similarity(actual, expected, dim=0).item() + if rank == 0: + (args.output / "comparison.json").write_text( + json.dumps( + {"mean_ce_relative": mean_relative, "loss_cosine": cosine, "samples": comparisons}, indent=2 + ) + ) + assert mean_relative < 0.03 and cosine > 0.97, (mean_relative, cosine) + finally: + if log is not None: + log.close() + dist.destroy_process_group() + + +if __name__ == "__main__": + with torch.backends.cudnn.flags(allow_tf32=False): + main() diff --git a/tests/model/prepare_glm53_parity_inputs.py b/tests/model/prepare_glm53_parity_inputs.py new file mode 100644 index 0000000000..b9f8b1983b --- /dev/null +++ b/tests/model/prepare_glm53_parity_inputs.py @@ -0,0 +1,82 @@ +"""Offline shared VLM inputs, using the actual XTuner GLM53 tokenizer.""" + +import argparse +import hashlib +import json +from pathlib import Path + +import torch +from safetensors.torch import save_file + +from transformers import AutoTokenizer +from xtuner.v1.datasets.mllm_tokenize_fn.glm53_vl_tokenize_fn import Glm53VLTokenizeFunction + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--model", required=True) + parser.add_argument("--jsonl", type=Path, required=True) + parser.add_argument("--media-root", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--limit", type=int, default=20) + args = parser.parse_args() + MODEL, OUT, ROOT = args.model, args.output, args.media_root + if args.limit < 1: + raise ValueError("limit must be positive") + OUT.mkdir(parents=True, exist_ok=False) + tok = AutoTokenizer.from_pretrained(MODEL, local_files_only=True) + fn = Glm53VLTokenizeFunction(tok, MODEL, str(args.jsonl), max_pixels=1048576, max_length=16384) + samples = [] + with args.jsonl.open() as source: + for index, line in enumerate(source): + if index == args.limit: + break + record = json.loads(line) + item = fn(record, media_root=str(ROOT)) + tensors = { + "input_ids": torch.tensor(item["input_ids"], dtype=torch.long).reshape(1, -1), + "labels": torch.tensor(item["labels"], dtype=torch.long).reshape(1, -1), + "pixel_values": item["pixel_values"].cpu().contiguous(), + "image_grid_thw": item["image_grid_thw"].cpu().contiguous(), + "mm_token_type_ids": torch.as_tensor(item["mm_token_type_ids"]).reshape(1, -1).contiguous(), + } + pos = torch.nonzero(tensors["labels"][0, 1:] != -100).flatten()[-8:] + tensors["logit_positions"] = pos + assert pos.numel() > 0 + assert ( + int((tensors["input_ids"] == fn.processor.image_token_id).sum()) + == int(tensors["image_grid_thw"].prod(dim=-1).sum()) // fn.merge_unit + ) + path = OUT / f"{index:03d}.safetensors" + save_file(tensors, str(path), metadata={"shift": "raw_unshifted; use input_ids[:,:-1] and labels[:,1:]"}) + sample = { + "index": index, + "id": record["id"], + "file": path.name, + "sha256": hashlib.sha256(path.read_bytes()).hexdigest(), + "length": tensors["input_ids"].numel(), + "valid_tokens": int((tensors["labels"][:, 1:] != -100).sum()), + "logit_positions": pos.tolist(), + "image_grid_thw": tensors["image_grid_thw"].tolist(), + } + samples.append(sample) + sample["tensors"] = { + key: {"shape": list(value.shape), "dtype": str(value.dtype)} for key, value in tensors.items() + } + print(json.dumps(sample), flush=True) + if len(samples) != args.limit: + raise ValueError("Insufficient input samples") + (OUT / "manifest.json").write_text( + json.dumps( + { + "model": MODEL, + "contract": "raw input_ids and raw labels; shift exactly once; logits positions index shifted input; final 8 supervised positions", + "samples": samples, + }, + indent=2, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/tests/model/test_glm53_text_moe.py b/tests/model/test_glm53_text_moe.py index fd8c233b50..ff8e3bad52 100644 --- a/tests/model/test_glm53_text_moe.py +++ b/tests/model/test_glm53_text_moe.py @@ -38,6 +38,7 @@ from transformers import AutoTokenizer, Glm5NextForConditionalGeneration from xtuner._testing import DeterministicDDPTestCase +from xtuner._testing.logits import check_logits from xtuner.v1.config import FSDPConfig from xtuner.v1.data_proto import SequenceContext from xtuner.v1.loss.ce_loss import CELossConfig @@ -357,12 +358,16 @@ def test_fsdp_accuracy(self, dispatcher, ep_size): "就像老树拥抱归巢的鸟儿，内存管理应该给予每个对象足够的安全感", ] tokenizer = AutoTokenizer.from_pretrained(GLM_5_3_FLASH_PATH) + hf_model.eval() expected_losses = [] + expected_logits = [] for text in text_list: input_ids = tokenizer(text, return_tensors="pt").input_ids.to("cuda") with torch.no_grad(): output = hf_model(input_ids=input_ids, labels=input_ids.clone()) expected_losses.append(output.loss) + # Last eight next-token positions, including the full vocabulary. + expected_logits.append(output.logits[0, :-1][-8:].detach().cpu()) del hf_model torch.cuda.empty_cache() @@ -381,8 +386,9 @@ def test_fsdp_accuracy(self, dispatcher, ep_size): model.fully_shard(fsdp_config=fsdp_config) model.from_hf(GLM_5_3_FLASH_PATH, strict=False) + model.eval() losses = [] - for text in text_list: + for sample_index, text in enumerate(text_list): input_ids = tokenizer(text, return_tensors="pt").input_ids.to("cuda") shift_input_ids = input_ids[:, :-1] shifted_labels = input_ids[:, 1:] @@ -396,6 +402,11 @@ def test_fsdp_accuracy(self, dispatcher, ep_size): with torch.no_grad(): output = model(seq_ctx=seq_ctx, loss_ctx={"lm": loss_ctx}) losses.append(output["loss"]) + # Preserve the public loss path above; additionally exercise inference logits. + with torch.no_grad(): + logits = model(seq_ctx=seq_ctx, loss_ctx=None).logits + metrics = check_logits(logits[0, -8:], expected_logits[sample_index]) + print(f"GLM53 text sample={sample_index} logits={metrics}", flush=True) self._check_loss_curve( losses=torch.tensor(losses), losses_ref=torch.tensor(expected_losses), sim_tol=3e-2, rtol=3e-2 diff --git a/tests/model/test_logits_metrics.py b/tests/model/test_logits_metrics.py new file mode 100644 index 0000000000..d0e00652d4 --- /dev/null +++ b/tests/model/test_logits_metrics.py @@ -0,0 +1,37 @@ +"""The logits check must catch errors that cross entropy cannot detect.""" + +import pytest +import torch +import torch.nn.functional as F + +from xtuner._testing.logits import check_logits + + +def test_identical_logits(): + x = torch.arange(40).reshape(4, 10).float() + assert check_logits(x, x)["relative_l2"] == 0 + + +def test_common_offset_preserves_ce_but_fails_logits(): + x = torch.arange(40).reshape(4, 10).float() / 10 + y = torch.tensor([0, 1, 2, 3]) + torch.testing.assert_close(F.cross_entropy(x, y), F.cross_entropy(x + 10, y)) + with pytest.raises(AssertionError): + check_logits(x + 10, x) + + +def test_position_permutation_is_rejected(): + x = torch.eye(8) + with pytest.raises(AssertionError): + check_logits(x.roll(1, 0), x) + + +@pytest.mark.parametrize("value", [float("nan"), float("inf")]) +def test_nonfinite_logits_are_rejected(value): + with pytest.raises(AssertionError): + check_logits(torch.full((2, 4), value), torch.ones(2, 4)) + + +def test_wrong_shape_is_rejected(): + with pytest.raises(AssertionError): + check_logits(torch.ones(2, 4), torch.ones(3, 4)) diff --git a/xtuner/_testing/logits.py b/xtuner/_testing/logits.py new file mode 100644 index 0000000000..9ca217aa7f --- /dev/null +++ b/xtuner/_testing/logits.py @@ -0,0 +1,27 @@ +"""Diagnostics for sampled, full-vocabulary end-to-end logits.""" + +import torch +import torch.nn.functional as F + + +def check_logits(actual, expected, *, max_relative_l2=0.05, min_cosine=0.998): + """Check each sample separately; use FP64 only for offline reductions. + + Inputs are [sampled_positions, vocab], not hidden states or target-token logits. + Callers must select the same globally indexed, non-padding positions on both sides. + The 5% L2 bound is independent of the existing mean loss-curve tolerance. + """ + assert actual.shape == expected.shape and actual.ndim == 2 and actual.numel() > 0 + assert torch.isfinite(actual).all() and torch.isfinite(expected).all(), "non-finite logits" + actual = actual.detach().to(device="cpu", dtype=torch.float64).flatten() + expected = expected.detach().to(device="cpu", dtype=torch.float64).flatten() + norm = expected.norm() + assert norm > 0, "zero reference norm" + delta = actual - expected + metrics = { + "relative_l2": (delta.norm() / norm).item(), + "cosine": F.cosine_similarity(actual, expected, dim=0).item(), + "max_abs": delta.abs().max().item(), + } + assert metrics["relative_l2"] < max_relative_l2 and metrics["cosine"] > min_cosine, metrics + return metrics From 43979f1754c682cbdc6a7e6135ab6df93ad2539c Mon Sep 17 00:00:00 2001 From: xs1997zju <106210907+xs1997zju@users.noreply.github.com> Date: Thu, 8 Oct 2026 13:02:00 +0000 Subject: [PATCH 2/2] [Test] Limit logits parity PR to the existing HF accuracy test --- doc/glm53_forward_parity.md | 53 ----- tests/model/glm53_forward_parity.py | 258 --------------------- tests/model/prepare_glm53_parity_inputs.py | 82 ------- 3 files changed, 393 deletions(-) delete mode 100644 doc/glm53_forward_parity.md delete mode 100644 tests/model/glm53_forward_parity.py delete mode 100644 tests/model/prepare_glm53_parity_inputs.py diff --git a/doc/glm53_forward_parity.md b/doc/glm53_forward_parity.md deleted file mode 100644 index bf6deb273b..0000000000 --- a/doc/glm53_forward_parity.md +++ /dev/null @@ -1,53 +0,0 @@ -# GLM 5.3 Flash forward loss and logits parity - -A scalar cross entropy can hide changes in full-vocabulary logits. The existing real-checkpoint text test now retains its public loss-path check and also checks the last eight next-token positions against native Transformers. `tests/model/glm53_forward_parity.py` adds image/text replay with FSDP, EP and SP using the same preprocessed inputs on both implementations. - -This is eval/no-grad validation: no packing, backward, optimizer step, or MTP. Use a reduced GLM checkpoint that fits a single GPU for the HF reference. The checkpoint is supplied by the caller; it is not downloaded by the test. - -## Backend and precision contract - -XT uses FlashAttention for the vision tower and torch reference DSA/indexer for the language tower. HF explicitly uses eager: Transformers 5.17 rejects `flash_attention_2` for `Glm5NextVisionModel`. This is an implementation/backend comparison, not a claim that both frameworks use identical kernels. BF16 model compute and FP32 CE are used; TF32 is disabled by the replay runner. CPU metric reductions use FP64. - -For each sample, select the final eight supervised positions (or all if fewer than eight), retaining the entire vocabulary at each position. Labels are shifted exactly once. Check shape, finiteness, fixture SHA256, supervised-token count, global positions and target labels before comparing. Sequence padding is excluded. With SP, gather only sampled logits and sum NLL/count within the SP group; EP/world are not loss reduction groups. DP groups deliberately replay identical inputs. - -The logits check requires **each sample** to have relative L2 below 0.05 and cosine above 0.998, independently of the existing loss-vector cosine >0.97 and mean relative loss error <0.03. These are BF16 regression bounds, not bitwise-parity guarantees. The thresholds are fixed in the helper, not adjusted by the replay runner to fit results. Reports include per-sample maximum absolute error as a diagnostic. A CPU regression verifies that a common logits offset, which leaves CE unchanged, fails the logits check. - -## Prepare portable image fixtures - -Run from the repository root in the XTuner environment. Supply absolute paths for your model, XTuner-format image JSONL, media root and a new output directory: - -```bash -python tests/model/prepare_glm53_parity_inputs.py \ - --model /models/GLM-5.3-Flash-25B \ - --jsonl /data/medpix/train.jsonl --media-root /data/medpix \ - --limit 20 --output /results/glm53/inputs -``` - -The preparer uses the real GLM tokenizer/processor once (max pixels 1048576, max length 16384). It writes raw, unshifted `input_ids`, `labels`, `mm_token_type_ids`, `pixel_values`, `image_grid_thw`, and shifted-input `logit_positions` as safetensors. The manifest stores relative filenames, hashes, sample IDs and supervised counts. Dataset/images/weights are external fixtures and are not added to Git. This replay entry point currently supports image inputs, not video/audio. - -## Run sequentially - -Use the same checkpoint for all runs. All output directories must be new; the runner refuses overwrites. Both entry points run from the XTuner checkout with its test dependencies installed; separate processes release model memory between implementations. - -```bash -# Native HF reference on one GPU. -torchrun --standalone --nproc-per-node=1 tests/model/glm53_forward_parity.py \ - --backend hf --model /models/GLM-5.3-Flash-25B \ - --inputs /results/glm53/inputs --output /results/glm53/hf - -# XT FSDP EP1/SP1 baseline. -torchrun --standalone --nproc-per-node=1 tests/model/glm53_forward_parity.py \ - --backend xt --model /models/GLM-5.3-Flash-25B \ - --inputs /results/glm53/inputs --reference /results/glm53/hf \ - --ep-size 1 --sp-size 1 --output /results/glm53/xt-ep1-sp1 - -# Run both SP1 and SP2 on eight GPUs, with distinct output directories. -torchrun --standalone --nproc-per-node=8 tests/model/glm53_forward_parity.py \ - --backend xt --model /models/GLM-5.3-Flash-25B \ - --inputs /results/glm53/inputs --reference /results/glm53/hf \ - --ep-size 8 --sp-size 2 --output /results/glm53/xt-ep8-sp2 -``` - -Set `--sp-size 1` and another output directory for EP8/SP1. The program exits nonzero on input-contract, logits, or loss-check failure. Each run records metadata, sampled tensors and `loss.jsonl`; comparisons also write `comparison.json`. This real-checkpoint replay is opt-in and is not advertised as an automatically provisioned CI test. - -Run the inexpensive metric regressions with `pytest tests/model/test_logits_metrics.py`. The existing real-checkpoint text regression remains in `TestGlm53TextMoEAccuracy.test_fsdp_accuracy`, with `GLM_5_3_FLASH_PATH` and `XTUNER_TEST_WORLD_SIZE` selecting the local checkpoint and GPU count. diff --git a/tests/model/glm53_forward_parity.py b/tests/model/glm53_forward_parity.py deleted file mode 100644 index 69658cf05c..0000000000 --- a/tests/model/glm53_forward_parity.py +++ /dev/null @@ -1,258 +0,0 @@ -"""Replay serialized GLM image/text inputs through HF or XTuner; compare CE and logits. - -Run using torchrun, including --nproc-per-node=1 for the HF reference. -See doc/glm53_forward_parity.md for the input contract and commands. -""" - -import argparse -import hashlib -import json -import os -from datetime import timedelta -from pathlib import Path - -import torch -import torch.distributed as dist -import torch.nn.functional as F -from safetensors.torch import load_file, save_file -from torch.distributed.device_mesh import init_device_mesh -from torch.distributed.tensor import DTensor - -from xtuner._testing.logits import check_logits -from xtuner.v1.config import FSDPConfig -from xtuner.v1.data_proto import SequenceContext -from xtuner.v1.model.compose.glm53 import Glm53BaseConfig - - -def main() -> None: - """所有 DP 组回放相同样本；NLL 和计数只在 SP 组求和一次。""" - parser = argparse.ArgumentParser() - parser.add_argument("--model", required=True) - parser.add_argument("--inputs", type=Path, required=True) - parser.add_argument("--backend", choices=("hf", "xt"), required=True) - parser.add_argument("--reference", type=Path) - parser.add_argument("--ep-size", type=int, default=1) - parser.add_argument("--sp-size", type=int, choices=(1, 2), default=1) - parser.add_argument("--limit", type=int, default=20) - parser.add_argument("--output", type=Path, required=True) - args = parser.parse_args() - if args.limit < 1 or args.ep_size < 1: - raise ValueError("limit and ep-size must be positive") - if args.reference is not None: - reference_metadata = json.loads((args.reference / "metadata-rank0.json").read_text()) - if Path(reference_metadata["model"]).resolve() != Path(args.model).resolve(): - raise ValueError("Reference checkpoint path differs") - torch.backends.cuda.matmul.allow_tf32 = False - torch.cuda.set_device(int(os.environ["LOCAL_RANK"])) - dist.init_process_group("nccl", timeout=timedelta(minutes=10)) - rank, world = dist.get_rank(), dist.get_world_size() - if world % args.ep_size or world % args.sp_size: - raise ValueError("EP/SP must divide world size") - if args.backend == "hf" and (world, args.ep_size, args.sp_size) != (1, 1, 1): - raise ValueError("HF reference requires world=EP=SP=1") - if args.output.exists(): - raise FileExistsError(f"请使用新的输出目录：{args.output}") - dist.barrier() - if rank == 0: - args.output.mkdir(parents=True) - dist.barrier() - torch.manual_seed(1234) - mesh = init_device_mesh("cuda", (world // args.sp_size, args.sp_size), mesh_dim_names=("dp", "sp")) - sp_mesh = mesh["sp"] - if args.backend == "hf": - from transformers import AutoConfig, Glm5NextForConditionalGeneration - - config = AutoConfig.from_pretrained(args.model) - config.text_config.num_nextn_predict_layers = 0 - model = Glm5NextForConditionalGeneration.from_pretrained( - args.model, - config=config, - dtype=torch.bfloat16, - device_map="cuda", - attn_implementation="eager", - ) - assert model.config.vision_config._attn_implementation == "eager" - else: - with torch.device("meta"): - cfg = Glm53BaseConfig.from_hf(args.model) - cfg.text_config.mtp_config = None - cfg.text_config.num_nextn_predict_layers = 0 - cfg.text_config.dispatcher = "all2all" - cfg.text_config.ep_size = args.ep_size - cfg.text_config.compile_cfg = False - cfg.compile_cfg = False - cfg.text_config.attention.sparse_mla_backend = "torch" - cfg.text_config.attention.indexer_backend = "torch" - cfg.vision_config.attn_impl = "flash_attention" - model = cfg.build()._to_device_dtype(dtype=torch.bfloat16, skip_buffers_dtype=True) - model.fully_shard(FSDPConfig(ep_size=args.ep_size, cpu_offload=False)) - model.from_hf(args.model, strict=False) - model.eval() - metadata = { - "backend": args.backend, - "model": args.model, - "rank": rank, - "world": world, - "ep": args.ep_size, - "sp": args.sp_size, - "sp_ranks": sp_mesh.mesh.tolist(), - "vision_attention": "eager" if args.backend == "hf" else "flash_attention", - "text_attention": "eager" if args.backend == "hf" else "torch", - "torch": torch.__version__, - "cuda": torch.version.cuda, - "eval": True, - "no_grad": True, - "packing": False, - "mtp": 0, - } - if args.backend == "xt": - ep_meshes = [m.ep_mesh for m in model.modules() if getattr(m, "ep_mesh", None) is not None] - if not ep_meshes or any(m.size() != args.ep_size for m in ep_meshes): - raise ValueError("Actual expert mesh does not match requested EP") - metadata["ep_ranks"] = ep_meshes[0].mesh.tolist() - metadata["expert_shards"] = [ - {"name": n, "global_shape": list(p.shape), "local_shape": list(p.to_local().shape)} - for n, p in model.named_parameters() - if isinstance(p, DTensor) and "experts." in n and "shared_experts" not in n - ][:3] - (args.output / f"metadata-rank{rank}.json").write_text(json.dumps(metadata, indent=2)) - manifest = json.loads((args.inputs / "manifest.json").read_text()) - samples = manifest["samples"][: args.limit] - if len({s["id"] for s in samples}) != len(samples) or len({s["index"] for s in samples}) != len(samples): - raise ValueError("Duplicate input sample IDs or indices") - if len(samples) != args.limit: - raise ValueError("输入样本数不足") - log = (args.output / "loss.jsonl").open("w") if rank == 0 else None - comparisons = [] - reference_rows = None - if args.reference is not None: - reference_rows = [json.loads(line) for line in (args.reference / "loss.jsonl").read_text().splitlines()] - if len(reference_rows) != len(samples): - raise ValueError("Reference sample count differs") - try: - with torch.no_grad(): - for ordinal, sample in enumerate(samples): - path = Path(sample["file"]) - if not path.is_absolute(): - path = args.inputs / path - if hashlib.sha256(path.read_bytes()).hexdigest() != sample["sha256"]: - raise ValueError("输入 hash 改变") - data = load_file(str(path), device=f"cuda:{torch.cuda.current_device()}") - ids = data["input_ids"][:, :-1].contiguous() - labels = data["labels"][:, 1:].contiguous() - ctx = SequenceContext.from_input_ids((ids,), device="cuda") - ctx.mm_token_type_ids = data["mm_token_type_ids"][:, :-1].contiguous() - # split 只补尾部到 SP 的倍数；真实样本仍只有一条，padding 不参与 CE。 - if args.sp_size > 1: - ctx = ctx.split(sp_mesh) - ctx.pixel_values = data["pixel_values"].to(torch.bfloat16) - ctx.image_grid_thw = data["image_grid_thw"] - ctx.num_img_tokens = [[int(ctx.image_grid_thw.prod(-1).sum())]] - if args.backend == "hf": - logits = model( - input_ids=ids, - pixel_values=ctx.pixel_values, - image_grid_thw=data["image_grid_thw"], - use_cache=False, - ).logits - else: - logits = model(seq_ctx=ctx, loss_ctx=None).logits - local_length = ctx.input_ids.shape[1] - start = sp_mesh.get_local_rank() * local_length - padded_labels = F.pad(labels, (0, local_length * args.sp_size - labels.shape[1]), value=-100) - targets = padded_labels[:, start : start + local_length] - if logits.shape[:2] != targets.shape: - raise ValueError(f"logits/labels mismatch: {logits.shape}, {targets.shape}") - valid = targets != -100 - nll = ( - F.cross_entropy(logits[valid].float(), targets[valid], reduction="sum") - if valid.any() - else logits.new_zeros((), dtype=torch.float32) - ) - stats = torch.stack((nll.double(), valid.sum().double())) - # EP 组不是 loss 归约组；DP 是重复验证样本，不再次求和。 - if args.sp_size > 1: - dist.all_reduce(stats, group=sp_mesh.get_group()) - if int(stats[1].item()) != sample["valid_tokens"] or not torch.isfinite(stats).all(): - raise ValueError("全局监督分母或 NLL 非法") - positions = data["logit_positions"] - expected_positions = torch.nonzero(labels[0] != -100).flatten()[-8:] - if not torch.equal(positions, expected_positions): - raise ValueError("Expected the final eight supervised global positions") - local_mask = (positions >= start) & (positions < start + local_length) - chosen = torch.zeros((positions.numel(), logits.shape[-1]), device=logits.device, dtype=torch.float32) - chosen[local_mask] = logits[0, positions[local_mask] - start].float() - if args.sp_size > 1: - dist.all_reduce(chosen, group=sp_mesh.get_group()) - if not torch.isfinite(chosen).all(): - raise ValueError("抽样 logits 非有限") - # 记录不同 DP 复制组的偏差；它们应得到相同样本统计。 - low, high = stats.clone(), stats.clone() - dist.all_reduce(low, op=dist.ReduceOp.MIN) - dist.all_reduce(high, op=dist.ReduceOp.MAX) - row = { - "sample_id": sample["id"], - "index": sample["index"], - "input_sha256": sample["sha256"], - "valid_tokens": int(stats[1].item()), - "nll_sum": stats[0].item(), - "mean_ce": (stats[0] / stats[1]).item(), - "dp_nll_span": (high[0] - low[0]).item(), - "global_input_length": ids.shape[1], - "padded_length": local_length * args.sp_size, - "ep": args.ep_size, - "sp": args.sp_size, - } - if reference_rows is not None: - ref = reference_rows[ordinal] - for key in ("sample_id", "input_sha256", "valid_tokens"): - if row[key] != ref[key]: - raise ValueError(f"Reference contract mismatch: {key}") - reference = load_file(str(args.reference / f"{sample['index']:03d}-logits.safetensors")) - if not torch.equal(reference["positions"], positions.cpu()) or not torch.equal( - reference["labels"], labels[0, positions].cpu() - ): - raise ValueError("Reference positions/labels differ") - metrics = check_logits(chosen, reference["logits"]) - comparisons.append( - { - "sample_id": sample["id"], - "reference_ce": ref["mean_ce"], - "candidate_ce": row["mean_ce"], - **metrics, - } - ) - if rank == 0: - print(json.dumps(row), flush=True) - log.write(json.dumps(row) + "\n") - log.flush() - save_file( - { - "positions": positions.cpu(), - "labels": labels[0, positions].cpu(), - "logits": chosen.cpu().contiguous(), - }, - str(args.output / f"{sample['index']:03d}-logits.safetensors"), - ) - del logits, chosen, data, ctx - if comparisons: - expected = torch.tensor([r["reference_ce"] for r in comparisons], dtype=torch.float64) - actual = torch.tensor([r["candidate_ce"] for r in comparisons], dtype=torch.float64) - mean_relative = ((actual - expected).abs() / expected.abs()).mean().item() - cosine = F.cosine_similarity(actual, expected, dim=0).item() - if rank == 0: - (args.output / "comparison.json").write_text( - json.dumps( - {"mean_ce_relative": mean_relative, "loss_cosine": cosine, "samples": comparisons}, indent=2 - ) - ) - assert mean_relative < 0.03 and cosine > 0.97, (mean_relative, cosine) - finally: - if log is not None: - log.close() - dist.destroy_process_group() - - -if __name__ == "__main__": - with torch.backends.cudnn.flags(allow_tf32=False): - main() diff --git a/tests/model/prepare_glm53_parity_inputs.py b/tests/model/prepare_glm53_parity_inputs.py deleted file mode 100644 index b9f8b1983b..0000000000 --- a/tests/model/prepare_glm53_parity_inputs.py +++ /dev/null @@ -1,82 +0,0 @@ -"""Offline shared VLM inputs, using the actual XTuner GLM53 tokenizer.""" - -import argparse -import hashlib -import json -from pathlib import Path - -import torch -from safetensors.torch import save_file - -from transformers import AutoTokenizer -from xtuner.v1.datasets.mllm_tokenize_fn.glm53_vl_tokenize_fn import Glm53VLTokenizeFunction - - -def main(): - parser = argparse.ArgumentParser() - parser.add_argument("--model", required=True) - parser.add_argument("--jsonl", type=Path, required=True) - parser.add_argument("--media-root", type=Path, required=True) - parser.add_argument("--output", type=Path, required=True) - parser.add_argument("--limit", type=int, default=20) - args = parser.parse_args() - MODEL, OUT, ROOT = args.model, args.output, args.media_root - if args.limit < 1: - raise ValueError("limit must be positive") - OUT.mkdir(parents=True, exist_ok=False) - tok = AutoTokenizer.from_pretrained(MODEL, local_files_only=True) - fn = Glm53VLTokenizeFunction(tok, MODEL, str(args.jsonl), max_pixels=1048576, max_length=16384) - samples = [] - with args.jsonl.open() as source: - for index, line in enumerate(source): - if index == args.limit: - break - record = json.loads(line) - item = fn(record, media_root=str(ROOT)) - tensors = { - "input_ids": torch.tensor(item["input_ids"], dtype=torch.long).reshape(1, -1), - "labels": torch.tensor(item["labels"], dtype=torch.long).reshape(1, -1), - "pixel_values": item["pixel_values"].cpu().contiguous(), - "image_grid_thw": item["image_grid_thw"].cpu().contiguous(), - "mm_token_type_ids": torch.as_tensor(item["mm_token_type_ids"]).reshape(1, -1).contiguous(), - } - pos = torch.nonzero(tensors["labels"][0, 1:] != -100).flatten()[-8:] - tensors["logit_positions"] = pos - assert pos.numel() > 0 - assert ( - int((tensors["input_ids"] == fn.processor.image_token_id).sum()) - == int(tensors["image_grid_thw"].prod(dim=-1).sum()) // fn.merge_unit - ) - path = OUT / f"{index:03d}.safetensors" - save_file(tensors, str(path), metadata={"shift": "raw_unshifted; use input_ids[:,:-1] and labels[:,1:]"}) - sample = { - "index": index, - "id": record["id"], - "file": path.name, - "sha256": hashlib.sha256(path.read_bytes()).hexdigest(), - "length": tensors["input_ids"].numel(), - "valid_tokens": int((tensors["labels"][:, 1:] != -100).sum()), - "logit_positions": pos.tolist(), - "image_grid_thw": tensors["image_grid_thw"].tolist(), - } - samples.append(sample) - sample["tensors"] = { - key: {"shape": list(value.shape), "dtype": str(value.dtype)} for key, value in tensors.items() - } - print(json.dumps(sample), flush=True) - if len(samples) != args.limit: - raise ValueError("Insufficient input samples") - (OUT / "manifest.json").write_text( - json.dumps( - { - "model": MODEL, - "contract": "raw input_ids and raw labels; shift exactly once; logits positions index shifted input; final 8 supervised positions", - "samples": samples, - }, - indent=2, - ) - ) - - -if __name__ == "__main__": - main()