File size: 3,677 Bytes
5094348 | 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 | """Curated AIFlow Ink v1과 사람 annotation을 P Formula v1 JSONL로 엄격 변환한다."""
from __future__ import annotations
import argparse
from datetime import datetime, timezone
import json
from pathlib import Path
import sys
from typing import Any
import torch
PROJECT_ROOT = Path(__file__).parents[1]
SOURCE_ROOT = PROJECT_ROOT / "src"
if str(SOURCE_ROOT) not in sys.path:
sys.path.insert(0, str(SOURCE_ROOT))
from math_grid_drawer.research.p_curation import read_p_records
from math_grid_drawer.research.p_formula_gate06 import audit_p_formula_records06
from math_grid_drawer.research.p_formula_intake06 import (
materialize_p_formula_records06,
read_p_formula_annotations06,
write_p_formula_jsonl06,
)
def _checkpoint_labels06(path: Path) -> tuple[str, ...]:
"""필요 변수: 신뢰된 로컬 0.6 checkpoint. 작동 원리: 중복 없는 고정 378 exact vocabulary만 반환한다."""
payload: dict[str, Any] = torch.load(
path,
map_location="cpu",
weights_only=False,
)
raw_labels = payload.get("exact_labels")
if not isinstance(raw_labels, (list, tuple)):
raise ValueError("checkpoint에 exact_labels 계약이 없습니다.")
labels = tuple(str(label) for label in raw_labels)
if len(labels) != 378 or len(set(labels)) != 378 or any(not label for label in labels):
raise ValueError("P Formula materialize에는 중복 없는 378 exact label checkpoint가 필요합니다.")
return labels
def main() -> None:
"""필요 변수: curated root·annotation JSONL·출력·감사 report. 작동 원리: 전체 product gate 통과 후에만 원자적으로 materialize한다."""
parser = argparse.ArgumentParser(description="Materialize AIFlow P Formula v1")
parser.add_argument("--input", type=Path, required=True)
parser.add_argument("--annotations", type=Path, required=True)
parser.add_argument("--checkpoint", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
parser.add_argument("--report", type=Path, required=True)
parser.add_argument("--minimum-independent-sources", type=int, default=2)
parser.add_argument("--overwrite", action="store_true")
args = parser.parse_args()
intake_records = read_p_records(args.input)
annotations = read_p_formula_annotations06(args.annotations)
labels = _checkpoint_labels06(args.checkpoint)
records = materialize_p_formula_records06(
intake_records,
annotations,
allowed_labels=labels,
)
report = audit_p_formula_records06(
records,
minimum_independent_sources=args.minimum_independent_sources,
)
report.update({
"generated_at": datetime.now(timezone.utc).isoformat(),
"input": str(args.input),
"annotations": str(args.annotations),
"checkpoint": str(args.checkpoint),
"vocabulary_labels": len(labels),
"output": str(args.output),
"materialized": False,
})
args.report.parent.mkdir(parents=True, exist_ok=True)
args.report.write_text(
json.dumps(report, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
if not report["eligible_for_product_evaluation"]:
print(json.dumps(report, ensure_ascii=False, indent=2))
raise SystemExit(2)
write_p_formula_jsonl06(args.output, records, overwrite=args.overwrite)
report["materialized"] = True
args.report.write_text(
json.dumps(report, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
print(json.dumps(report, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()
|