Diplomovka/perun/ministral_3b_aya/scripts/evaluate.py
2026-11-12 16:51:42 +01:00

302 lines
14 KiB
Python

#!/usr/bin/env python3
"""Evaluate the trained adapter on the prepared eval or test split; report loss, F1 and exact match."""
from __future__ import annotations
import argparse
from collections import Counter, defaultdict
import hashlib
import json
import math
import os
from pathlib import Path
import subprocess
import sys
import unicodedata
from train import ROOT, read_config, write_config, verify_data, run_logged
def normalized_tokens(text: str) -> list[str]:
text = unicodedata.normalize("NFKC", text).casefold()
return "".join(
" " if unicodedata.category(char).startswith("P") else char
for char in text
).split()
def score_pair(prediction: str, reference: str) -> dict:
predicted = normalized_tokens(prediction)
expected = normalized_tokens(reference)
if not expected:
return {
"token_f1": None,
"normalized_exact_match": None,
"scored": False,
"skip_reason": "reference_has_zero_normalized_tokens",
"prediction_tokens": len(predicted),
"reference_tokens": 0,
}
shared = sum((Counter(predicted) & Counter(expected)).values())
f1 = 200.0 * shared / (len(predicted) + len(expected))
return {
"token_f1": f1,
"normalized_exact_match": 100.0 if predicted == expected else 0.0,
"scored": True,
"skip_reason": None,
"prediction_tokens": len(predicted),
"reference_tokens": len(expected),
}
def read_jsonl(path: Path) -> list[dict]:
result = []
with path.open("r", encoding="utf-8-sig") as handle:
for line_number, line in enumerate(handle, 1):
if not line.strip():
continue
try:
row = json.loads(line)
except json.JSONDecodeError as exc:
raise ValueError(f"{path}:{line_number}: invalid JSON: {exc.msg}") from exc
if not isinstance(row, dict):
raise ValueError(f"{path}:{line_number}: expected a JSON object")
result.append(row)
return result
def required_text(row: dict, field: str, context: str) -> str:
value = row.get(field)
if not isinstance(value, str):
raise ValueError(f"{context}: {field!r} must contain a string")
return value
def reference_text(row: dict, index: int) -> str:
for field in ("output", "response", "label"):
if field in row:
return required_text(row, field, f"reference row {index}")
raise ValueError(f"reference row {index}: missing output/response/label")
def sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for block in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
def verify_training_dataset(adapter: Path, data_dir: Path) -> Path:
run_dir = adapter.parent
if adapter.name.startswith("checkpoint-") and not (run_dir / "run.json").is_file():
run_dir = run_dir.parent
record_path = run_dir / "run.json"
if not record_path.is_file():
raise ValueError(f"Missing training record: {record_path}. Keep the adapter with its original run files.")
record = json.loads(record_path.read_text(encoding="utf-8"))
if record.get("data_manifest_sha256") != sha256(data_dir / "manifest.json"):
raise ValueError("The prepared dataset differs from this adapter's training data. "
"Use the original dataset and its eval/test split recorded in run.json.")
return run_dir
def summarize(rows: list[dict]) -> dict:
scored = [row for row in rows if row["scored"]]
count = len(scored)
return {
"count_total": len(rows),
"count_scored": count,
"count_skipped_zero_token_references": len(rows) - count,
"count_empty_predictions_raw": sum(not row["prediction"].strip() for row in rows),
"count_zero_token_predictions": sum(row["prediction_tokens"] == 0 for row in rows),
"count_nonempty_predictions_raw": sum(bool(row["prediction"].strip()) for row in rows),
"count_truncated_or_modified_labels": sum(
not row["label_matches_full_reference"] for row in rows
),
"count_labels_different_before_normalization": sum(
not row["label_matches_full_reference_raw"] for row in rows
),
"token_f1": sum(row["token_f1"] for row in scored) / count if count else None,
"normalized_exact_match": (
sum(row["normalized_exact_match"] for row in scored) / count if count else None
),
"prediction_tokens_total": sum(row["prediction_tokens"] for row in rows),
"reference_tokens_total": sum(row["reference_tokens"] for row in rows),
}
def score_files(
predictions_path: Path,
references_path: Path,
output_dir: Path,
metadata_path: Path | None = None,
) -> dict:
predictions = read_jsonl(predictions_path)
references = read_jsonl(references_path)
if not references:
raise ValueError("The reference file contains no examples")
if len(predictions) != len(references):
raise ValueError(
f"Row count mismatch: {len(predictions)} predictions != {len(references)} references. "
"No scores written; evaluate the complete split in the original order."
)
metadata = read_jsonl(metadata_path) if metadata_path is not None else [{} for _ in references]
if len(metadata) != len(references):
raise ValueError(
f"Row count mismatch: {len(metadata)} metadata rows != {len(references)} references"
)
rows = []
groups = defaultdict(list)
for index, (prediction_row, reference_row, meta) in enumerate(
zip(predictions, references, metadata)
):
prediction = required_text(prediction_row, "predict", f"prediction row {index}")
label = required_text(prediction_row, "label", f"prediction row {index}")
reference = reference_text(reference_row, index)
if "row_index" in meta and meta["row_index"] != index:
raise ValueError(f"metadata row {index}: row_index does not match the reference order")
source_metadata = meta.get("source_metadata", {})
if not isinstance(source_metadata, dict):
raise ValueError(f"metadata row {index}: source_metadata must be an object")
task_type = meta.get(
"task_type", source_metadata.get("task_type", reference_row.get("task_type", "unknown"))
)
if not isinstance(task_type, str):
raise ValueError(f"metadata row {index}: task_type must be a string")
row = {
"id": meta.get("stable_id", meta.get("id", reference_row.get("id", index))),
"index": index,
"task_type": task_type,
"prediction": prediction,
"reference": reference,
"exported_label": label,
"label_matches_full_reference": normalized_tokens(label) == normalized_tokens(reference),
"label_matches_full_reference_raw": label == reference,
**score_pair(prediction, reference),
}
rows.append(row)
groups[task_type].append(row)
metrics = {
"schema_version": 1,
"metric_scale": "0..100",
"metric_version": "aya_overlap_v1",
"overall": summarize(rows),
"by_task_type": {name: summarize(group) for name, group in sorted(groups.items())},
"inputs": {
"predictions": {"path": str(predictions_path.resolve()), "sha256": sha256(predictions_path)},
"references": {"path": str(references_path.resolve()), "sha256": sha256(references_path)},
},
}
if metadata_path is not None:
metrics["inputs"]["metadata"] = {"path": str(metadata_path.resolve()), "sha256": sha256(metadata_path)}
if output_dir.exists() and (not output_dir.is_dir() or any(output_dir.iterdir())):
raise ValueError(f"Output directory must be new or empty: {output_dir}")
output_dir.mkdir(parents=True, exist_ok=True)
with (output_dir / "per_sample.jsonl").open("w", encoding="utf-8", newline="\n") as handle:
for row in rows:
handle.write(json.dumps(row, ensure_ascii=False, allow_nan=False) + "\n")
(output_dir / "metrics.json").write_text(
json.dumps(metrics, ensure_ascii=False, indent=2, allow_nan=False) + "\n", encoding="utf-8"
)
return metrics
def evaluation_configs(training: dict, adapter: Path, output_dir: Path, split: str, max_new_tokens: int):
keys = ("model_name_or_path", "model_revision", "stage", "finetuning_type", "template",
"dataset_dir", "cutoff_len", "packing", "train_on_prompt",
"bf16", "infer_dtype", "flash_attn", "use_unsloth", "rope_scaling", "seed", "data_seed",
"preprocessing_num_workers", "dataloader_num_workers", "report_to")
base = {key: training[key] for key in keys if key in training}
base.update({"adapter_name_or_path": str(adapter), "eval_dataset": f"aya_{split}",
"use_unsloth_gc": False, "overwrite_output_dir": False})
loss = dict(base, output_dir=str(output_dir / "loss"), do_eval=True,
predict_with_generate=False, per_device_eval_batch_size=8)
generation = dict(base, output_dir=str(output_dir / "generation"), do_predict=True,
predict_with_generate=True, per_device_eval_batch_size=1,
do_sample=False, num_beams=1, max_new_tokens=max_new_tokens)
return loss, generation
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("config", type=Path, help="The same model YAML used for training.")
parser.add_argument("--adapter", type=Path, help="Defaults to output_dir in the model YAML.")
parser.add_argument("--output-dir", type=Path, required=True)
parser.add_argument("--split", choices=("eval", "test"), default="eval")
parser.add_argument("--max-new-tokens", type=int, default=256)
parser.add_argument("--dry-run", action="store_true")
args = parser.parse_args()
if args.max_new_tokens < 1:
parser.error("--max-new-tokens must be positive.")
config = read_config(args.config.resolve())
data_dir = Path(config["dataset_dir"])
verify_data(data_dir, (args.split,))
adapter = args.adapter.resolve() if args.adapter else Path(config["output_dir"])
if not (adapter / "adapter_config.json").is_file():
raise ValueError(f"Missing trained adapter: {adapter}")
run_dir = verify_training_dataset(adapter, data_dir)
used_config = run_dir / "used_config.yaml"
if not used_config.is_file():
raise ValueError(f"Missing recorded training configuration: {used_config}")
if used_config.is_file():
trained = read_config(used_config)
for key in ("model_name_or_path", "model_revision", "template", "cutoff_len", "rope_scaling"):
if trained.get(key) != config.get(key):
raise ValueError(f"Evaluation setting {key} differs from this adapter's training; use {used_config}.")
output_dir = args.output_dir.resolve()
if output_dir.exists() and any(output_dir.iterdir()):
raise ValueError(f"Evaluation output directory must be new or empty: {output_dir}")
loss, generation = evaluation_configs(config, adapter, output_dir, args.split, args.max_new_tokens)
if args.dry_run:
print(json.dumps({"split": args.split, "adapter": str(adapter), "evaluation_gpus": 1,
"loss_config": loss, "generation_config": generation}, ensure_ascii=False, indent=2))
print("Evaluation configuration/data checks passed. No inference started.")
return 0
output_dir.mkdir(parents=True, exist_ok=True)
record_path = output_dir / "evaluation.json"
record = {"status": "running", "split": args.split, "adapter": str(adapter),
"config": str(args.config.resolve()), "gpus": 1, "backend": "Unsloth", "precision": "bf16"}
record_path.write_text(json.dumps(record, indent=2) + "\n", encoding="utf-8")
env = os.environ.copy()
env.update({"PYTHONUNBUFFERED": "1", "TOKENIZERS_PARALLELISM": "false"})
try:
for name, settings in (("eval_loss", loss), ("eval_generate", generation)):
path = output_dir / f"{name}.yaml"
write_config(path, settings)
command = [sys.executable, "-m", "torch.distributed.run", "--standalone", "--nproc_per_node=1",
str(ROOT / "scripts/train.py"), str(path), "--worker"]
run_logged(command, output_dir / f"{name}.log", env)
metrics = score_files(output_dir / "generation/generated_predictions.jsonl",
data_dir / f"{args.split}.jsonl", output_dir / "scores",
data_dir / f"{args.split}.metadata.jsonl")
reported = json.loads((output_dir / "loss/eval_results.json").read_text(encoding="utf-8"))
value = reported.get("eval_loss")
if value is None or not math.isfinite(value):
raise ValueError("LLaMA-Factory did not report a finite eval_loss.")
summary = {"model": config["model_name_or_path"], "split": args.split,
"eval_loss": value, "perplexity": math.exp(value) if value < 700 else None,
"token_f1": metrics["overall"]["token_f1"],
"normalized_exact_match": metrics["overall"]["normalized_exact_match"],
"examples": metrics["overall"]["count_total"], "metric_scale": "0..100",
"max_new_tokens": args.max_new_tokens, "cutoff_len": config["cutoff_len"],
"generation_context_limit": config["cutoff_len"]}
(output_dir / "summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
record["status"] = "complete"
print(json.dumps(summary, ensure_ascii=False, indent=2))
except BaseException:
record["status"] = "failed_or_interrupted"
raise
finally:
record_path.write_text(json.dumps(record, indent=2) + "\n", encoding="utf-8")
return 0
if __name__ == "__main__":
try:
raise SystemExit(main())
except (OSError, ValueError, RuntimeError, ImportError, subprocess.CalledProcessError) as error:
print(f"ERROR: {error}", file=sys.stderr)
raise SystemExit(1)