File size: 4,328 Bytes
df84e45 | 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 | """Compare recorded CUA trajectories with real model rollouts on the GUI surface.
python scripts/compare_cua.py --cua artifacts/cua/cua.jsonl --trace artifacts/cua-compare/trace.jsonl
For each side it reports how the episode was driven: turns, calls per turn,
the gesture mix, how numbers got into the editor (typed into fields vs dragged),
scrolls, undos, thought length, and reward. The point is to see whether the
recorded behaviour looks like what a capable model actually does, per layout.
"""
from __future__ import annotations
import argparse
import json
import statistics
from collections import Counter
from pathlib import Path
def _profile(turns: list[dict]) -> dict:
"""`turns`: [{"text": str, "calls": [{"name", "args"}]}] -> behaviour profile."""
calls = [c for t in turns for c in t["calls"]]
names = Counter(c["name"] for c in calls)
typed = [c for c in calls if c["name"] == "type"]
numeric = sum(1 for c in typed if str(c["args"].get("text", "")).replace(".", "").replace("-", "").isdigit())
keys = Counter(str(c["args"].get("keys")) for c in calls if c["name"] == "key")
words = [len(t["text"].split()) for t in turns if t.get("text")]
return {
"turns": len(turns),
"calls": len(calls),
"calls_per_turn": round(len(calls) / max(1, len(turns)), 2),
"mix": {k: round(v / max(1, len(calls)), 3) for k, v in names.most_common()},
"typed_numbers": numeric,
"typed_text": len(typed) - numeric,
"drags": names.get("left_click_drag", 0),
"scrolls": names.get("scroll", 0),
"undo": keys.get("ctrl+z", 0) + keys.get("cmd+z", 0),
"screenshots_requested": names.get("screenshot", 0),
"thought_words_median": statistics.median(words) if words else 0,
"turns_with_thought": round(len(words) / max(1, len(turns)), 2),
}
def from_cua(traj: dict) -> dict:
turns = []
for m in traj["messages"]:
if m["role"] == "assistant":
turns.append({"text": m.get("content") or "", "calls": [
{"name": c["function"]["name"], "args": json.loads(c["function"]["arguments"])}
for c in m.get("tool_calls", [])]})
meta = traj["meta"]
return {"who": "recorder", "skin": meta["skin"], "layout": meta.get("layout"), "task_id": meta["task_id"],
"reward": meta["reward"], **_profile(turns)}
def from_trace(rec: dict) -> dict:
turns = [{"text": t.get("text", ""), "calls": t.get("calls", [])} for t in rec.get("turns", [])]
return {"who": rec["model"], "skin": rec.get("skin"), "layout": "default", "task_id": rec.get("task_id"),
"reward": rec.get("reward"), **_profile(turns)}
def main(argv=None) -> int:
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
ap.add_argument("--cua", action="append", default=[], help="cua.jsonl, or a directory of trajectory .json files")
ap.add_argument("--trace", action="append", default=[], help="rollout trace.jsonl")
ap.add_argument("--out", default=None)
args = ap.parse_args(argv)
rows = []
for src in args.cua:
p = Path(src)
items = ([json.loads(line) for line in p.read_text().splitlines() if line.strip()] if p.is_file()
else [json.loads(f.read_text()) for f in sorted(p.glob("*.json"))])
rows += [from_cua(t) for t in items]
for src in args.trace:
rows += [from_trace(json.loads(line)) for line in Path(src).read_text().splitlines()
if line.strip() and json.loads(line).get("surface") == "gui"]
cols = ["who", "skin", "reward", "turns", "calls", "calls_per_turn", "typed_numbers", "typed_text", "drags",
"scrolls", "undo", "screenshots_requested", "thought_words_median"]
print("| " + " | ".join(cols) + " | top gestures |")
print("|" + "---|" * (len(cols) + 1))
for r in sorted(rows, key=lambda r: (r["skin"] or "", r["who"])):
top = ", ".join(f"{k} {v:.0%}" for k, v in list(r["mix"].items())[:4])
vals = [f"{r[c]:.3f}" if isinstance(r[c], float) and c == "reward" else str(r[c]) for c in cols]
print("| " + " | ".join(vals) + f" | {top} |")
if args.out:
Path(args.out).write_text(json.dumps(rows, indent=1))
return 0
if __name__ == "__main__":
raise SystemExit(main())
|