File size: 3,070 Bytes
97799b9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 | """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()
|