rodunia/llm-research-app

1"""Schema validation script for per-run evaluation results.2 3Validates that all records in analysis/per_run.json conform to the canonical schema.4"""5 6import json7import sys8from pathlib import Path9from typing import List, Dict, Any10 11 12def load_per_run_results(path: str = "analysis/per_run.json") -> List[Dict[str, Any]]:13    """Load per-run results from JSON file.14 15    Args:16        path: Path to per_run.json17 18    Returns:19        List of result dictionaries20 21    Raises:22        FileNotFoundError: If file doesn't exist23    """24    file_path = Path(path)25    if not file_path.exists():26        raise FileNotFoundError(f"File not found: {path}")27 28    with open(file_path, "r", encoding="utf-8") as f:29        return json.load(f)30 31 32def validate_record(record: Dict[str, Any], index: int) -> List[str]:33    """Validate a single record.34 35    Args:36        record: Record dictionary37        index: Index of record in list (for error reporting)38 39    Returns:40        List of validation errors (empty if valid)41    """42    errors = []43 44    # Required top-level fields45    required_fields = ["run_id"]46    for field in required_fields:47        if field not in record:48            errors.append(f"Record {index}: Missing required field '{field}'")49 50    # Check for metrics dict (canonical schema)51    if "metrics" not in record or record["metrics"] is None:52        errors.append(f"Record {index} ({record.get('run_id', 'unknown')}): Missing 'metrics' dict")53    else:54        # Validate metrics structure55        metrics = record["metrics"]56        required_metrics = [57            "total_claims", "hit_rate", "contradiction_rate",58            "unsupported_rate", "ambiguous_rate", "overclaim_rate",59            "numeric_error_count", "unit_error_count", "bias_score"60        ]61        for metric in required_metrics:62            if metric not in metrics:63                errors.append(64                    f"Record {index} ({record.get('run_id', 'unknown')}): "65                    f"Missing metric '{metric}' in metrics dict"66                )67 68    # Check for metadata dict69    if "metadata" not in record:70        errors.append(f"Record {index} ({record.get('run_id', 'unknown')}): Missing 'metadata' dict")71    else:72        metadata = record["metadata"]73        required_metadata = ["engine", "product_id", "material_type"]74        for meta_field in required_metadata:75            if meta_field not in metadata or metadata[meta_field] is None:76                errors.append(77                    f"Record {index} ({record.get('run_id', 'unknown')}): "78                    f"Missing or null '{meta_field}' in metadata"79                )80 81    return errors82 83 84def main():85    """Main validation routine."""86    print("Schema Validation for per_run.json")87    print("=" * 60)88 89    try:90        results = load_per_run_results()91        print(f"✓ Loaded {len(results)} records from analysis/per_run.json\n")92    except FileNotFoundError as e:93        print(f"✗ Error: {e}")94        print("\nRun 'python -m analysis.evaluate' first to generate results.")95        sys.exit(1)96    except json.JSONDecodeError as e:97        print(f"✗ Error: Invalid JSON in analysis/per_run.json: {e}")98        sys.exit(1)99 100    # Validate each record101    all_errors = []102    for idx, record in enumerate(results):103        errors = validate_record(record, idx)104        all_errors.extend(errors)105 106    # Report results107    if all_errors:108        print(f"✗ Found {len(all_errors)} schema validation errors:\n")109        for error in all_errors:110            print(f"  - {error}")111        print("\nSchema validation FAILED.")112        sys.exit(1)113    else:114        print("✓ All records conform to canonical schema")115        print("\nValidation checks:")116        print("  ✓ All records have 'run_id'")117        print("  ✓ All records have 'metrics' dict with required fields")118        print("  ✓ All records have 'metadata' dict with engine/product/material")119        print("\n✅ Schema validation PASSED")120        sys.exit(0)121 122 123if __name__ == "__main__":124    main()125