rag/scripts/validate_eval_dataset.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
验证 eval_dataset_v2.json 的完整性和格式正确性
"""
import json
import sys
from pathlib import Path
from collections import Counter

PROJECT_ROOT = Path(__file__).parent.parent
DATASET_PATH = PROJECT_ROOT / "tests" / "eval_dataset_v2.json"


def validate():
    errors = []
    warnings = []

    with open(DATASET_PATH, 'r', encoding='utf-8') as f:
        data = json.load(f)

    queries = data.get("queries", [])
    print(f"[OK] Dataset loaded: {len(queries)} queries")

    # 1. Check required fields
    required = {"id", "query", "query_type", "relevant_docs", "reference_answer", "expected_keywords", "difficulty"}
    for q in queries:
        missing = required - set(q.keys())
        if missing:
            errors.append(f"  {q.get('id', '?')}: missing fields: {missing}")

    # 2. Check unique IDs
    ids = [q["id"] for q in queries]
    dupes = [id for id, cnt in Counter(ids).items() if cnt > 1]
    if dupes:
        errors.append(f"  Duplicate IDs: {dupes}")

    # 3. Check query types
    valid_types = {"simple_fact", "enumeration", "definition", "comparison",
                   "reasoning", "table_data", "cross_doc", "negative", "paraphrase"}
    type_counts = Counter(q["query_type"] for q in queries)
    invalid_types = set(type_counts.keys()) - valid_types
    if invalid_types:
        errors.append(f"  Invalid query_types: {invalid_types}")

    print(f"\n[INFO] Query type distribution:")
    for t, c in sorted(type_counts.items(), key=lambda x: -x[1]):
        marker = " [INVALID]" if t in (invalid_types or set()) else ""
        print(f"  {t}: {c}{marker}")

    # 4. Check difficulty distribution
    diff_counts = Counter(q["difficulty"] for q in queries)
    print(f"\n[INFO] Difficulty distribution:")
    for d in ["easy", "medium", "hard"]:
        print(f"  {d}: {diff_counts.get(d, 0)}")

    # 5. Check document coverage
    doc_counts = Counter()
    for q in queries:
        for doc in q["relevant_docs"]:
            doc_counts[doc] += 1
    if not any("negative" in q["query_type"] for q in queries):
        warnings.append("  No negative test cases")

    print(f"\n[INFO] Document coverage:")
    for doc, cnt in sorted(doc_counts.items(), key=lambda x: -x[1]):
        print(f"  {doc}: {cnt} queries")
    neg_count = sum(1 for q in queries if q["query_type"] == "negative")
    print(f"  (negative/out-of-scope): {neg_count} queries")

    # 6. Check expected_keywords
    empty_kw = [q["id"] for q in queries if not q.get("expected_keywords")]
    if empty_kw:
        warnings.append(f"  Empty expected_keywords: {empty_kw}")

    # 7. Check paraphrase references
    paraphrases = [q for q in queries if q["query_type"] == "paraphrase"]
    for p in paraphrases:
        ref = p.get("paraphrase_of")
        if ref and ref not in ids:
            errors.append(f"  {p['id']}: paraphrase_of '{ref}' not found")

    # 8. Check reference_answer length
    short_refs = [q["id"] for q in queries if len(q.get("reference_answer", "")) < 10]
    if short_refs:
        warnings.append(f"  Very short reference_answer: {short_refs}")

    # Summary
    print(f"\n{'='*60}")
    if errors:
        print(f"[ERROR] {len(errors)} error(s):")
        for e in errors:
            print(e)
    if warnings:
        print(f"[WARN] {len(warnings)} warning(s):")
        for w in warnings:
            print(w)
    if not errors and not warnings:
        print("[PASS] Dataset validation passed!")
    print(f"{'='*60}")

    return len(errors) == 0


if __name__ == "__main__":
    if sys.platform == 'win32':
        sys.stdout.reconfigure(encoding='utf-8', errors='replace')
    ok = validate()
    sys.exit(0 if ok else 1)