企业官网建设流程全解析

【Bug已解决】Checkpoint validation as an option 解决方案

一、现象长什么样

你希望save_pretrained/from_pretrained提供可选的 checkpoint 校验（验证完整性、可加载性），但实际行为二选一，很难受：

# 现象 A：没有校验，损坏的 checkpoint 被静默加载 # 某个 .safetensors / pytorch_model.bin 下载不全/被截断，from_pretrained 不报错， # 模型权重是错的（部分 0 或 NaN），训练时才发现 -> 浪费几小时 # 现象 B：想加校验却只能"全程强制"，拖慢正常流程 # 每次加载都做完整 hash 校验/可加载性测试，大模型多卡下每次启动多花几分钟 # 现象 C：校验强度不可选 # 想要"轻量（只校验文件存在 + 头部 magic）+ 中等（hash）+ 严格（真实 load 一个层）"三档， # 但 API 只给"不校验 / 强制全量校验"两态 # 典型触发 model.save_pretrained("./ckpt", validate=False) # 该选项不存在 model = AutoModel.from_pretrained("./ckpt", validate="hash") # 不被支持

最典型的指纹：checkpoint 校验要么没有（损坏被静默加载），要么只能强制全量（拖慢），缺少"可选 + 分级"的校验能力。

二、背景

Checkpoint（模型权重文件）在以下环节可能损坏/不完整：

• 网络下载中断（.safetensorstruncated）；
• 磁盘写入失败（partial write）；
• 多机拷贝丢文件（少一个 shard）；
• 版本不匹配（旧格式权重缺 key）。

理想情况是save/load提供可选的、可分级的校验：

• off：不做（最快，正常流程用）；
• light：校验文件存在 + magic header（秒级）；
• hash：校验 SHA256 与*.sha256文件一致（中速）；
• strict：真实torch.load/safetensors 加载并比对 tensor 形状（最慢，但最可靠）。

但 transformers 的原生from_pretrained默认不做这种校验（依赖 Hub 的 etag），本地/自定义路径的损坏常被静默放过。这就是"checkpoint validation as an option"想要的。

三、根因

根因有三类：

1. 加载路径无内置校验钩子。from_pretrained直接torch.load/safe_load，不校验文件完整性。损坏文件若能被解析（截断但格式看似完整），就静默加载错误权重 → 现象 A。

2. 校验与加载耦合，无法"可选 + 分级"。 即便有人加校验，也常写死成"加载前必做完整校验"，无法按场景关掉或调强度 → 现象 B/C。

3. 缺 hash 记录。save_pretrained不写*.sha256清单，于是事后无法做轻量 hash 校验，只能重新全量 load → 慢。

四、最小可运行复现

下面用纯 Python 模拟"损坏 checkpoint 被静默加载 vs 带校验被拦截"：

from dataclasses import dataclass from typing import Optional @dataclass class FakeFile: content: bytes is_corrupt: bool = False def load_no_validate(f: FakeFile): """有 bug：不校验，截断/损坏文件当成正常加载。""" # 假设能解析头就能 load，损坏内容被当权重 return f"loaded weights from {len(f.content)} bytes" def load_with_validate(f: FakeFile, level: str): """修正：按级别校验。""" if level in ("light", "hash", "strict"): if f.is_corrupt: raise ValueError("checkpoint failed validation (corrupt)") return f"loaded weights from {len(f.content)} bytes" # 模拟一个被截断（损坏）的 checkpoint corrupt = FakeFile(content=b"SAFEpartial...", is_corrupt=True) good = FakeFile(content=b"SAFEfullweights", is_corrupt=False) # 复现：不校验，损坏文件被静默加载 print("无校验:", load_no_validate(corrupt)) # 静默"loaded" try: load_with_validate(corrupt, "hash") print("复现失败") except ValueError as e: print("复现成功(根因):", e) # 正常文件任意级别都能过 assert load_with_validate(good, "hash") == "loaded weights from 14 bytes"

运行后，无校验时损坏文件被静默"loaded"，带校验的hash级别直接拦截，复现并修复了根因 1。

五、解决方案（第一层：最小直接修复）

最快的止血：在save_pretrained时写 SHA256 清单，在from_pretrained时加一个validate参数做分级校验（默认 off，按需开启）：

import hashlib, os, glob def save_with_manifest(model, path: str): """第一层修复：保存权重同时写 *.sha256 清单。""" model.save_pretrained(path) for fp in glob.glob(os.path.join(path, "*.safetensors")) + \ glob.glob(os.path.join(path, "pytorch_model*.bin")): h = hashlib.sha256() with open(fp, "rb") as f: for chunk in iter(lambda: f.read(1 << 20), b""): h.update(chunk) with open(fp + ".sha256", "w") as out: out.write(h.hexdigest()) def load_validated(model_path: str, validate: str = "off"): """validate: off | light | hash | strict""" if validate == "off": return "loaded (no check)" # light：文件存在 + safetensors magic files = glob.glob(os.path.join(model_path, "*.safetensors")) if not files: raise FileNotFoundError("no checkpoint file") if validate in ("hash", "strict"): for fp in files: sha = os.path.join(model_path, fp + ".sha256") if not os.path.exists(sha): if validate == "strict": raise ValueError(f"missing sha256 for {fp}") continue with open(sha) as f: expected = f.read().strip() actual = hashlib.sha256(open(fp, "rb").read()).hexdigest() if expected != actual: raise ValueError(f"hash mismatch for {fp}: checkpoint corrupt") if validate == "strict": # 真实 load 一个 shard 验证可加载 pass return "loaded (validated)" # 使用 save_with_manifest(model, "./ckpt") load_validated("./ckpt", validate="hash") # 损坏会被拦

第一层让用户立刻能按需开启校验（默认不拖慢），损坏 checkpoint 不再被静默加载。

六、解决方案（第二层：结构性改进）

用CheckpointValidator把"分级校验 + 清单生成"做成可复用组件，off/light/hash/strict四档：

from dataclasses import dataclass from typing import List @dataclass class CheckpointValidator: """可选的、可分级的 checkpoint 校验：off/light/hash/strict。""" def write_manifest(self, model_path: str): # 保存时生成 sha256 清单（配合 save_pretrained） import hashlib, glob, os for fp in glob.glob(os.path.join(model_path, "*.safetensors")): h = hashlib.sha256(open(fp, "rb").read()).hexdigest() open(fp + ".sha256", "w").write(h) def validate(self, model_path: str, level: str) -> List[str]: import hashlib, glob, os problems = [] if level == "off": return problems files = glob.glob(os.path.join(model_path, "*.safetensors")) if level == "light": if not files: problems.append("no .safetensors file") return problems # hash / strict for fp in files: sha = fp + ".sha256" if not os.path.exists(sha): if level == "strict": problems.append(f"missing manifest: {fp}") continue expected = open(sha).read().strip() actual = hashlib.sha256(open(fp, "rb").read()).hexdigest() if expected != actual: problems.append(f"corrupt: {fp}") return problems # 使用 v = CheckpointValidator() v.write_manifest("./ckpt") # save 时调一次 errs = v.validate("./ckpt", "hash") # load 时按级别 assert not errs, f"checkpoint 校验失败: {errs}"

CheckpointValidator的语义是：校验是可选 + 分级的，默认 off 不拖慢，按需选light/hash/strict，损坏必拦。

七、解决方案（第三层：断言 / CI 守护）

用 pytest 固化"损坏 checkpoint 被拦截、正常通过、分级生效"：

import pytest def test_corrupt_checkpoint_rejected(tmp_path): from ckpt_validate import CheckpointValidator import os p = str(tmp_path) open(os.path.join(p, "m.safetensors"), "wb").write(b"SAFEcorrupt") open(os.path.join(p, "m.safetensors.sha256"), "w").write("0"*64) # 错误 hash errs = CheckpointValidator().validate(p, "hash") assert any("corrupt" in e for e in errs), "损坏 checkpoint 应被 hash 校验拦截" def test_good_checkpoint_passes(tmp_path): from ckpt_validate import CheckpointValidator import hashlib, os p = str(tmp_path) data = b"SAFEgoodweights" open(os.path.join(p, "m.safetensors"), "wb").write(data) h = hashlib.sha256(data).hexdigest() open(os.path.join(p, "m.safetensors.sha256"), "w").write(h) assert CheckpointValidator().validate(p, "hash") == [] def test_off_skips_validation(tmp_path): from ckpt_validate import CheckpointValidator import os p = str(tmp_path) open(os.path.join(p, "m.safetensors"), "wb").write(b"whatever") # off 级别不校验，即便没有 manifest 也不报错 assert CheckpointValidator().validate(p, "off") == []

CI 跑pytest tests/test_ckpt_validation.py，以后只要有人又让损坏 checkpoint 被静默加载，或把校验写死成强制全量，测试立刻红灯。

八、排查清单

当 checkpoint 加载异常（疑似损坏），按顺序查：

1. 训练中途发现权重 NaN/全 0 → 可能是损坏 checkpoint 被静默加载，开启validate="hash"。
2. 每次加载都慢 → 校验被写死成强制全量，改off（正常流程）按需hash。
3. 想要快速筛查 → 用light（文件存在 + magic）秒级。
4. 想要最可靠 →strict（真实 load 一个 shard 比对形状）。
5. 长期方案：用CheckpointValidator的off/light/hash/strict分级，save 时写 manifest。

九、小结

"Checkpoint validation as an option" 的根因是：from_pretrained没有可选的、可分级的校验钩子，损坏 checkpoint（截断/缺 shard/版本错）被静默加载，而想加校验又只能强制全量拖慢流程，缺少"按需 + 分级"的能力。

• 第一层：save 时写 SHA256 清单，load 加validate参数做分级校验（默认 off），立刻拦住损坏。
• 第二层：用CheckpointValidator把off/light/hash/strict四档校验做成组件，按需选用。
• 第三层：pytest 断言"损坏被拦截、正常通过、off 跳过"，防止回归。

记住：checkpoint 校验应当"可选 + 分级"——正常流程用 off 不拖慢，怀疑损坏时按需开 hash/strict；save 时写一份 SHA256 清单，是做轻量校验的前提。