File size: 3,079 Bytes
9b2f1cf | 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 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 | #!/usr/bin/env python3
"""Summarize a prime-rl eval traces.jsonl (flat or nested Harbor traces)."""
from __future__ import annotations
import json
import sys
from collections import Counter
from pathlib import Path
def _iter_traces(path: Path):
with path.open() as f:
for line in f:
line = line.strip()
if not line:
continue
try:
row = json.loads(line)
except json.JSONDecodeError:
continue
traces = row.get("traces")
if isinstance(traces, list) and traces:
for t in traces:
if isinstance(t, dict):
yield t
else:
yield row
def _name(t: dict) -> str:
data = (t.get("task") or {}).get("data") or {}
return str(data.get("name") or t.get("task_id") or t.get("id") or "?")
def _score(t: dict) -> float | None:
rewards = t.get("rewards")
if isinstance(rewards, dict):
solved = rewards.get("solved")
if isinstance(solved, dict) and "score" in solved:
try:
return float(solved["score"])
except (TypeError, ValueError):
pass
if isinstance(solved, (int, float)):
return float(solved)
r = t.get("reward")
if r is None:
r = (t.get("metrics") or {}).get("reward")
if r is None:
return None
try:
return float(r)
except (TypeError, ValueError):
return None
def _err(t: dict) -> str | None:
errors = t.get("errors") or t.get("error") or t.get("err")
if isinstance(errors, list) and errors:
return str(errors[0])[:120]
if errors:
return str(errors)[:120]
return None
def main() -> None:
path = Path(sys.argv[1] if len(sys.argv) > 1 else "traces.jsonl")
if path.is_dir():
cands = list(path.rglob("traces.jsonl"))
if not cands:
print("no traces.jsonl under", path)
sys.exit(1)
path = cands[0]
n = n_ok = n_err = n_scored = 0
stops: Counter[str] = Counter()
errs: Counter[str] = Counter()
solved: list[str] = []
for t in _iter_traces(path):
n += 1
err = _err(t)
if err:
n_err += 1
errs[err[:80]] += 1
sc = _score(t)
if sc is not None:
n_scored += 1
if sc > 0.5:
n_ok += 1
solved.append(_name(t))
stops[str(t.get("stop_condition") or "?")] += 1
mark = "OK" if (sc is not None and sc > 0.5) else ("ERR" if err else "no")
print(f" {mark:3s} {_name(t):52s} stop={t.get('stop_condition')} score={sc}")
print(f"file {path}")
print(f"rows {n} scored {n_scored} errors {n_err} solved {n_ok} {solved}")
if n_scored:
print(f"pass_rate {n_ok / n_scored:.4f}")
print("stops", dict(stops))
if errs:
print("errors:")
for k, v in errs.most_common(8):
print(f" {v:4d} {k}")
if __name__ == "__main__":
main()
|