| """Evaluate real /analyze outputs. Input JSONL: {text, label: human|ai}.""" |
| import argparse |
| import json |
| import urllib.request |
| from pathlib import Path |
|
|
|
|
| def metrics(rows, threshold): |
| human = [s for label, s in rows if label == 'human'] |
| ai = [s for label, s in rows if label == 'ai'] |
| fp = sum(s >= threshold for s in human) |
| tp = sum(s >= threshold for s in ai) |
| auc = sum((a > h) + .5 * (a == h) for a in ai for h in human) / (len(ai)*len(human)) if ai and human else None |
| return {'n': len(rows), 'threshold': threshold, 'false_positive_rate': fp/len(human) if human else None, |
| 'detection_rate': tp/len(ai) if ai else None, 'auroc': auc, |
| 'confusion_matrix': {'tp': tp, 'fp': fp, 'tn': len(human)-fp, 'fn': len(ai)-tp}} |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument('input', type=Path) |
| parser.add_argument('--url', default='http://127.0.0.1:8000') |
| parser.add_argument('--threshold', type=float, default=.5) |
| parser.add_argument('--output', type=Path, required=True) |
| args = parser.parse_args() |
| if not 0 <= args.threshold <= 1: |
| parser.error('threshold must be between 0 and 1') |
| grouped, latency, unavailable = {}, {}, {} |
| n = inconclusive = failed = 0 |
| for line in args.input.read_text().splitlines(): |
| if not line.strip(): |
| continue |
| item = json.loads(line) |
| if item['label'] not in ('human', 'ai'): |
| raise ValueError('Labels must be human or ai; mixed authorship needs separate evaluation.') |
| n += 1 |
| request = urllib.request.Request(args.url+'/analyze', data=json.dumps({'text':item['text'],'language':'en'}).encode(), headers={'Content-Type':'application/json'}) |
| try: |
| with urllib.request.urlopen(request, timeout=240) as response: |
| report = json.load(response) |
| except Exception: |
| failed += 1 |
| continue |
| if report['status'] == 'inconclusive': |
| inconclusive += 1 |
| for result in report.get('results', []): |
| key = result['model_id'] |
| grouped.setdefault(key, []).append((item['label'], result['ai_score'])) |
| latency.setdefault(key, []).append(result['duration_ms']) |
| for error in report.get('errors', []): |
| key = error['model_id'] |
| unavailable[key] = unavailable.get(key, 0) + 1 |
| out = {'documents': n, 'failed_requests': failed, 'failure_rate': failed/n if n else None, |
| 'inconclusive_rate': inconclusive/n if n else None, 'model_errors': unavailable, |
| 'warning': 'Small samples are smoke tests, not evidence of competitive accuracy. Reported metrics exclude failed/inconclusive inputs; inspect coverage.', |
| 'models': {k: {**metrics(v,args.threshold), 'mean_duration_ms':sum(latency[k])/len(latency[k]), 'document_coverage':len(v)/n} for k,v in grouped.items()}} |
| args.output.write_text(json.dumps(out, indent=2)+'\n') |
| print(json.dumps(out, indent=2)) |
|
|
| if __name__ == '__main__': |
| main() |
|
|
