Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import argparse | |
| import csv | |
| import json | |
| from pathlib import Path | |
| from typing import Any, Literal | |
| from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator | |
| class RetrievalEvaluationRecord(BaseModel): | |
| model_config = ConfigDict(extra="forbid", frozen=True) | |
| id: str = Field(min_length=1) | |
| question: str = Field(min_length=1) | |
| ground_truth: str | |
| source_file: str | None = None | |
| gold_pages: list[int] = Field(default_factory=list) | |
| gold_chunk_ids: list[str] = Field(default_factory=list) | |
| answerable: bool | |
| question_type: str | None = None | |
| difficulty: Literal["easy", "medium", "hard"] | None = None | |
| def _strip_required_strings(cls, value: str) -> str: | |
| stripped = value.strip() | |
| if not stripped: | |
| raise ValueError("Value cannot be blank.") | |
| return stripped | |
| def _strip_ground_truth(cls, value: str) -> str: | |
| return value.strip() | |
| def _strip_optional_strings(cls, value: str | None) -> str | None: | |
| if value is None: | |
| return None | |
| stripped = value.strip() | |
| return stripped or None | |
| def _normalize_pages(cls, pages: list[int]) -> list[int]: | |
| if any(page < 1 for page in pages): | |
| raise ValueError("gold_pages must contain positive page numbers.") | |
| return sorted(set(pages)) | |
| def _normalize_chunk_ids(cls, chunk_ids: list[str]) -> list[str]: | |
| return list(dict.fromkeys(chunk_id.strip() for chunk_id in chunk_ids if chunk_id.strip())) | |
| def _validate_grounding(self) -> RetrievalEvaluationRecord: | |
| if self.answerable and not self.ground_truth: | |
| raise ValueError("Answerable records require a non-empty ground_truth.") | |
| if self.gold_pages and not self.source_file: | |
| raise ValueError("source_file is required when gold_pages are provided.") | |
| if not self.answerable and (self.gold_pages or self.gold_chunk_ids): | |
| raise ValueError("Unanswerable records cannot contain gold source annotations.") | |
| return self | |
| def has_gold_sources(self) -> bool: | |
| return bool(self.gold_chunk_ids or (self.source_file and self.gold_pages)) | |
| def annotation_required(self) -> bool: | |
| return ( | |
| (self.answerable and not self.has_gold_sources) | |
| or self.question_type is None | |
| or self.difficulty is None | |
| ) | |
| def load_evaluation_records(path: Path) -> list[RetrievalEvaluationRecord]: | |
| if not path.exists(): | |
| raise FileNotFoundError(f"Evaluation dataset not found: {path}") | |
| if path.suffix.lower() == ".jsonl": | |
| payloads: list[Any] = [] | |
| for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1): | |
| if not line.strip(): | |
| continue | |
| try: | |
| payloads.append(json.loads(line)) | |
| except json.JSONDecodeError as error: | |
| raise ValueError(f"Invalid JSONL at line {line_number}: {error.msg}") from error | |
| elif path.suffix.lower() == ".json": | |
| payload = json.loads(path.read_text(encoding="utf-8")) | |
| if not isinstance(payload, list): | |
| raise ValueError("JSON evaluation dataset must contain a list.") | |
| payloads = payload | |
| else: | |
| raise ValueError("Evaluation dataset must use .json or .jsonl.") | |
| if not payloads: | |
| raise ValueError("Evaluation dataset cannot be empty.") | |
| records: list[RetrievalEvaluationRecord] = [] | |
| seen_ids: set[str] = set() | |
| for index, payload in enumerate(payloads): | |
| if not isinstance(payload, dict): | |
| raise ValueError(f"Evaluation record {index} must be an object.") | |
| record = RetrievalEvaluationRecord.model_validate(payload) | |
| if record.id in seen_ids: | |
| raise ValueError(f"Duplicate evaluation record id: {record.id}") | |
| seen_ids.add(record.id) | |
| records.append(record) | |
| return records | |
| def _record_id(raw_id: str, index: int) -> str: | |
| cleaned = raw_id.strip() | |
| if cleaned.isdigit(): | |
| return f"qa-{int(cleaned):03d}" | |
| return cleaned or f"qa-{index:03d}" | |
| def convert_csv_to_jsonl( | |
| input_path: Path, | |
| output_path: Path, | |
| report_path: Path, | |
| ) -> dict[str, object]: | |
| with input_path.open(encoding="utf-8-sig", newline="") as input_file: | |
| rows = list(csv.DictReader(input_file)) | |
| if not rows: | |
| raise ValueError("Benchmark CSV cannot be empty.") | |
| records: list[RetrievalEvaluationRecord] = [] | |
| for index, row in enumerate(rows, start=1): | |
| question = (row.get("question") or "").strip() | |
| ground_truth = (row.get("ground truth") or row.get("ground_truth") or "").strip() | |
| if not question or not ground_truth: | |
| raise ValueError(f"CSV row {index} requires question and ground truth values.") | |
| records.append( | |
| RetrievalEvaluationRecord( | |
| id=_record_id(row.get("no.", ""), index), | |
| question=question, | |
| ground_truth=ground_truth, | |
| source_file=None, | |
| gold_pages=[], | |
| gold_chunk_ids=[], | |
| answerable=True, | |
| question_type=None, | |
| difficulty=None, | |
| ) | |
| ) | |
| output_path.parent.mkdir(parents=True, exist_ok=True) | |
| output_path.write_text( | |
| "".join(json.dumps(record.model_dump(), ensure_ascii=False) + "\n" for record in records), | |
| encoding="utf-8", | |
| ) | |
| annotation_ids = [record.id for record in records if record.annotation_required] | |
| report: dict[str, object] = { | |
| "input_file": str(input_path), | |
| "output_file": str(output_path), | |
| "total_records": len(records), | |
| "annotation_required_count": len(annotation_ids), | |
| "annotation_required_ids": annotation_ids, | |
| "missing_fields": { | |
| "source_file": len(records), | |
| "gold_pages_or_chunk_ids": len(records), | |
| "question_type": len(records), | |
| "difficulty": len(records), | |
| }, | |
| "note": ( | |
| "Source metadata, question type, and difficulty were not present in the CSV " | |
| "and were intentionally left unannotated." | |
| ), | |
| } | |
| report_path.parent.mkdir(parents=True, exist_ok=True) | |
| report_path.write_text( | |
| json.dumps(report, ensure_ascii=False, indent=2) + "\n", | |
| encoding="utf-8", | |
| ) | |
| return report | |
| def main() -> None: | |
| parser = argparse.ArgumentParser( | |
| description="Convert the legacy benchmark CSV to validated evaluation JSONL." | |
| ) | |
| parser.add_argument("input_csv", type=Path) | |
| parser.add_argument("--output", type=Path, required=True) | |
| parser.add_argument("--report", type=Path, required=True) | |
| args = parser.parse_args() | |
| report = convert_csv_to_jsonl(args.input_csv, args.output, args.report) | |
| print(json.dumps(report, ensure_ascii=False, indent=2)) | |
| if __name__ == "__main__": | |
| main() | |