tinymind-native-colab-handoff/bundle/evaluation/hf_pure_auto_training.py

"""Train/eval runner for the Hugging Face pure auto-refinery."""

from __future__ import annotations

from datetime import datetime, timezone
import json
from pathlib import Path

from data.hf_pure_auto_refinery import HFPureAutoRefinery, HFPureSource
from evaluation.knowledge_full_cycle import SourceTraceIndex, audit_pure_records, evaluate_natural_answer_style
from evaluation.local_evidence import run_local_train_eval_bundle


def _read_jsonl(path: str | Path) -> list[dict]:
    p = Path(path)
    if not p.exists() or not p.read_text(encoding="utf-8").strip():
        return []
    return [json.loads(line) for line in p.read_text(encoding="utf-8").splitlines() if line.strip()]


def _markdown(report: dict) -> str:
    train = report.get("train_eval", {})
    return "\n".join(
        [
            "# TinyMind HF Pure Auto-Refine Training",
            "",
            f"- HF pure gate: {report['hf_pure_gate']['passed']}",
            f"- Records: {report['dataset_manifest']['records_written']}",
            f"- Blocked records: {report['dataset_manifest']['blocked_records']}",
            f"- Audit purity: {report['audit']['purity_score']:.2%}",
            f"- Natural answer score: {report['natural_answer_style']['score']:.2%}",
            f"- Train steps: {train.get('steps', 'missing')}",
            f"- Eval loss: {train.get('eval_loss', 'missing')}",
            f"- Perplexity: {train.get('perplexity', 'missing')}",
            f"- HF token present: {report['security']['hf_token_present']}",
            f"- API key saved: {report['security']['api_key_saved']}",
            f"- World-best claim allowed: {report['claim_gate']['world_best_claim_allowed']}",
            "",
            "This is a reproducible HF-sourced local train/eval gate, not an official external leaderboard result.",
            "",
        ]
    )


def run_hf_pure_auto_refine_train(
    out_dir: str | Path,
    sources: list[str] | None = None,
    preset: str = "default",
    rows_per_source: int = 20,
    train_steps: int = 16,
    seed: int = 20260523,
    offline: bool = False,
) -> dict:
    out = Path(out_dir)
    out.mkdir(parents=True, exist_ok=True)
    data_dir = out / "dataset"
    train_dir = out / "train_eval"

    parsed_sources = [HFPureSource.parse(item) for item in sources] if sources else None
    manifest = HFPureAutoRefinery(
        sources=parsed_sources,
        preset=preset,
        rows_per_source=rows_per_source,
        eval_ratio=0.2,
        offline=offline,
    ).write_jsonl(data_dir)
    rows = _read_jsonl(manifest["train_path"]) + _read_jsonl(manifest["eval_path"])
    audit = audit_pure_records(rows)
    natural = evaluate_natural_answer_style(rows)
    source_index = SourceTraceIndex.from_records(rows)
    source_meta = source_index.write(out / "hf_pure_source_trace_index.json")

    train = None
    if rows:
        train = run_local_train_eval_bundle(
            train_dir,
            train_steps=train_steps,
            context_lengths=(32, 128, 1024),
            seed=seed,
            records=rows,
        )

    report = {
        "schema_version": "tinymind-hf-pure-auto-training-v1",
        "created_at": datetime.now(timezone.utc).isoformat(),
        "goal": "HF-sourced high-purity data filtering with real local TinyMind train/eval evidence",
        "dataset_manifest": manifest,
        "audit": audit,
        "natural_answer_style": natural,
        "source_trace": {
            **source_meta,
            "example_queries": [
                {"query": "Thai verified answer evidence", "hits": source_index.query("Thai verified answer evidence")},
                {"query": "math reasoning verified procedure", "hits": source_index.query("math reasoning verified procedure")},
            ],
        },
        "train_eval": (train or {}).get("train_eval", {}),
        "artifacts": (train or {}).get("artifacts", {}),
        "security": {
            "hf_token_present": bool(manifest.get("hf_token_present")),
            "api_key_saved": False,
            "token_value_logged": False,
        },
        "hf_pure_gate": {
            "passed": bool(rows and audit["blocked_records"] == 0 and natural["score"] >= 0.5 and train),
            "policy": "dataset rows must pass CEV audit, natural answer audit, source trace creation, and local train/eval artifact generation",
        },
        "claim_gate": {
            "world_best_claim_allowed": False,
            "perfect_no_error_claim_allowed": False,
            "external_rank_claim_allowed": False,
            "reason": "HF data smoke training is local evidence only; official external rank requires submitted benchmark/provider results.",
        },
    }
    json_path = out / "hf_pure_auto_training_report.json"
    md_path = out / "hf_pure_auto_training_report.md"
    report["json_path"] = str(json_path)
    report["markdown_path"] = str(md_path)
    json_path.write_text(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8")
    md_path.write_text(_markdown(report), encoding="utf-8")
    return report