| """Check file integrity, model loading, and the three decision interfaces. |
| |
| Usage: python validate_checkpoint.py --model-dir /path/to/snapshot --output VALIDATION.json |
| The prompts below are synthetic interface checks, not benchmark items. |
| """ |
| import argparse |
| import hashlib |
| import importlib.metadata |
| import json |
| import math |
| from datetime import datetime, timezone |
| from pathlib import Path |
|
|
|
|
| def sha256(path): |
| digest = hashlib.sha256() |
| with path.open("rb") as stream: |
| for chunk in iter(lambda: stream.read(1024 * 1024), b""): |
| digest.update(chunk) |
| return digest.hexdigest() |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--model-dir", type=Path, required=True) |
| parser.add_argument("--output", type=Path, required=True) |
| args = parser.parse_args() |
| manifest = json.loads((args.model_dir / "MANIFEST.json").read_text()) |
| for entry in manifest["files"]: |
| path = args.model_dir / entry["path"] |
| assert path.stat().st_size == entry["bytes"], entry["path"] |
| assert sha256(path) == entry["sha256"], entry["path"] |
|
|
| import torch |
| from laya import Agent |
|
|
| torch.set_num_threads(2) |
| model = Agent(str(args.model_dir), device="cpu") |
| probes = [ |
| ( |
| "用户问：体检报告里的这个指标是什么意思？", |
| {"intent": {"type": "choice", "instructions": "判断用户的问题主要属于哪类意图。", |
| "criteria": {"A": "检查报告解读", "B": "预约挂号", "C": "费用咨询"}}}, |
| ), |
| ( |
| {"问题一": "如何预约体检？", "问题二": "体检怎样预约？"}, |
| {"match": {"type": "noul", "instructions": "两个问题表达的需求是否相同？", |
| "criteria": {"false": "不同", "true": "相同"}}}, |
| ), |
| ( |
| {"查询": "体检预约方式", "标题": "体检预约流程与联系方式"}, |
| {"relevance": {"type": "score", "instructions": "评价标题与查询的相关程度。", |
| "criteria": ["不相关", "部分相关", "高度相关"]}}, |
| ), |
| ] |
| checks = [] |
| for state, questions in probes: |
| result = model.system_one(state=state, questions=questions) |
| qid, question = next(iter(questions.items())) |
| answer = result["answers"][qid] |
| kind = question["type"] |
| assert answer["type"] == kind |
| if kind == "noul": |
| assert math.isfinite(answer["noul"]) and 0 <= answer["noul"] <= 1 |
| else: |
| probabilities = answer["probabilities"] |
| expected = set(question["criteria"]) if kind == "choice" else {"0", "1", "2"} |
| assert set(probabilities) == expected |
| assert all(math.isfinite(v) and 0 <= v <= 1 for v in probabilities.values()) |
| assert math.isclose(sum(probabilities.values()), 1, abs_tol=0.0005) |
| if kind == "choice": |
| assert answer["choice"] in expected |
| else: |
| assert math.isfinite(answer["score"]) and 0 <= answer["score"] <= 2 |
| checks.append({"type": kind, "status": "passed"}) |
|
|
| |
| for entry in manifest["files"]: |
| assert sha256(args.model_dir / entry["path"]) == entry["sha256"], entry["path"] |
| report = { |
| "status": "passed", |
| "checked_at_utc": datetime.now(timezone.utc).isoformat(), |
| "source_repo_id": manifest["source_repo_id"], |
| "source_revision": manifest["source_revision"], |
| "weights_sha256": next(x["sha256"] for x in manifest["files"] if x["path"] == "model.safetensors"), |
| "file_integrity": "all manifest files match before and after loading", |
| "model_load": "strict state-dict restore through laya.Agent", |
| "device": "cpu", |
| "versions": {name: importlib.metadata.version(name) for name in |
| ("laya", "torch", "transformers", "safetensors", "huggingface-hub", "tokenizers")}, |
| "smoke_checks": checks, |
| "benchmark_evaluation": "not_run", |
| "calibration_evaluation": "not_run", |
| "long_context_evaluation": "not_run", |
| } |
| args.output.parent.mkdir(parents=True, exist_ok=True) |
| args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n") |
| print(json.dumps(report, ensure_ascii=False, indent=2)) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
