#!/usr/bin/env python3 """Contrôle structurel d'un jeu RAG. Python 3.10+, bibliothèque standard uniquement.""" import argparse import copy import hashlib import json import sys from collections import Counter from pathlib import Path def validate(data): errors = [] def check(condition, message): if not condition: errors.append(message) def index(rows, name): ids = [row['id'] for row in rows] check(len(ids) == len(set(ids)), f'{name}: identifiant dupliqué') check(all(isinstance(i, str) and i.strip() for i in ids), f'{name}: identifiant vide') return {row['id']: row for row in rows} check(data['schema_version'] == 1, 'Version de schéma non prise en charge') check(bool(data['corpus_version']) and bool(data['guidelines_version']), 'Versions manquantes') passages = index(data['passages'], 'Passages') questions = index(data['questions'], 'Questions') check(bool(passages) and bool(questions), 'Corpus ou questions vides') for p in passages.values(): check(bool(p['document_id']) and bool(p['version']), f"{p['id']}: référence documentaire vide") check(hashlib.sha256(p['text'].encode('utf-8')).hexdigest() == p['sha256'], f"{p['id']}: empreinte du texte incorrecte") groups = {} for q in questions.values(): check(q['split'] in {'dev', 'test'}, f"{q['id']}: split inconnu") check(bool(q['group_id']), f"{q['id']}: groupe vide") check(q['corpus_version'] == data['corpus_version'], f"{q['id']}: version du corpus incohérente") check(q['answerability'] in {'answerable', 'unanswerable', 'clarify'}, f"{q['id']}: statut inconnu") check(bool(q['expected_behavior'].strip()), f"{q['id']}: comportement attendu absent") groups.setdefault(q['group_id'], set()).add(q['split']) for group, splits in groups.items(): check(len(splits) == 1, f'{group}: groupe réparti entre dev et test') pairs = {} disagreements = 0 unjudged = 0 for a in data['annotations']: key = (a['question_id'], a['passage_id']) check(key not in pairs, f'{key}: paire dupliquée') check(key[0] in questions and key[1] in passages, f'{key}: identifiant inconnu') pairs[key] = a for field in ('annotator_a', 'annotator_b', 'final'): value = a[field] check(value is None or (type(value) is int and value in {0, 1, 2}), f'{key}: note invalide dans {field}') x, y, final = a['annotator_a'], a['annotator_b'], a['final'] if x is not None and y is not None and x != y: disagreements += 1 if final is None: unjudged += 1 else: check(x is not None and y is not None, f'{key}: avis indépendant manquant') if x != y or final != x: check(bool(a.get('adjudicator')) and bool(a.get('reason', '').strip()), f'{key}: arbitrage non documenté') for q in questions.values(): check(any(key[0] == q['id'] for key in pairs), f"{q['id']}: aucune paire examinée") if q['answerability'] == 'answerable': check(bool(q['reference_facts']) and bool(q['evidence_sets']), f"{q['id']}: réponse ou preuve manquante") for evidence in q['evidence_sets']: check(bool(evidence) and len(evidence) == len(set(evidence)), f"{q['id']}: ensemble de preuves vide ou dupliqué") for pid in evidence: a = pairs.get((q['id'], pid)) check(pid in passages and a is not None and a['final'] in (1, 2), f"{q['id']}/{pid}: preuve absente, non jugée ou inutile") else: check(not q['reference_facts'] and not q['evidence_sets'], f"{q['id']}: réponse définitive malgré abstention/clarification") total = len(data['annotations']) paired = sum(a['annotator_a'] is not None and a['annotator_b'] is not None for a in data['annotations']) return {'valid': not errors, 'errors': errors, 'questions': len(questions), 'passages': len(passages), 'groups': len(groups), 'splits': dict(Counter(q['split'] for q in questions.values())), 'answerability': dict(Counter(q['answerability'] for q in questions.values())), 'declared_pairs': total, 'double_annotated_pairs': paired, 'disagreements_before_adjudication': disagreements, 'agreement_before_adjudication': (paired - disagreements) / paired if paired else None, 'unjudged_declared_pairs': unjudged, 'unlisted_pairs': len(questions) * len(passages) - len(pairs), 'warning': 'Validité structurelle seulement. Paires absentes et null restent non jugées. Aucun score de RAG, aucune validation sémantique ni représentativité garantie.'} def self_test(data): base = validate(data) assert base['valid'], base assert base['unjudged_declared_pairs'] == 1 assert base['unlisted_pairs'] == 25 assert base['disagreements_before_adjudication'] == 2 tests = [ ('groupe partagé', lambda d: d['questions'][1].update(split='test')), ('texte modifié', lambda d: d['passages'][0].update(text='Autre contenu')), ('arbitrage absent', lambda d: d['annotations'][0].update(reason='')), ('preuve non jugée', lambda d: d['annotations'][1].update(final=None)), ('identifiant dupliqué', lambda d: d['questions'].append(copy.deepcopy(d['questions'][0]))), ('version incohérente', lambda d: d['questions'][0].update(corpus_version='autre')), ('note invalide', lambda d: d['annotations'][0].update(annotator_a=True)), ('avis absent', lambda d: d['annotations'][1].update(annotator_b=None)), ] for label, mutate in tests: broken = copy.deepcopy(data) mutate(broken) assert not validate(broken)['valid'], label return {'self_test': 'passed', 'negative_cases': len(tests), 'baseline': base} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('file', nargs='?', default=str(Path(__file__).with_name('jeu-annotation.json'))) parser.add_argument('--self-test', action='store_true', help='Tester les erreurs sur le jeu pédagogique intact') args = parser.parse_args() try: data = json.loads(Path(args.file).read_text(encoding='utf-8')) report = self_test(data) if args.self_test else validate(data) except (OSError, ValueError, KeyError, TypeError, AttributeError, AssertionError) as error: print(json.dumps({'valid': False, 'error': str(error)}, ensure_ascii=False, indent=2)) return 1 print(json.dumps(report, ensure_ascii=False, indent=2)) return 0 if args.self_test or report['valid'] else 1 if __name__ == '__main__': sys.exit(main())