Buckets:
| """Analysis of the Flint x ProgramBench sweep: solo PIV vs team PIV (Cohere Sec 4, Table 4 style). | |
| Reads sweep/manifest.json (written by sweep.py) and each entry's run dir (`run_dir` in the entry, | |
| else runs/<run_id>), and writes results/summary.md and results/summary.json: | |
| - per condition: mean reward (graded `score.score` = tests passed / total), % resolved (== 1), | |
| % almost (>= 0.95), wall time, model calls, tool calls, tokens, workers used (agents other than | |
| the lead with >= 1 model call), delegations (lead posts), reports (worker posts), worker share | |
| of model calls, format errors (mini-swe-agent FormatError messages in the final snapshot's | |
| participant state), who submitted, and cost: task-job CPU hours (manifest) and the serve job's | |
| GPU hours, split over conditions by summed run wall time (an estimate: tokens are priced $0); | |
| - paired over tasks: per-task team - solo reward (seeds averaged per task and condition), mean, | |
| 95% bootstrap CI (10k resamples, seed 0) and an exact two-sided sign test; | |
| - every manifest entry without a graded run is listed as missing, never dropped silently. | |
| Run with the swarmlab env's python: `python -I scripts/analyze.py [--manifest M] [--out DIR]`. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import csv | |
| import json | |
| import math | |
| import random | |
| from collections import Counter, defaultdict | |
| from pathlib import Path | |
| from statistics import mean | |
| from typing import Any | |
| from swarmlab.experiment import Run | |
| # _Blobs and _safe_unpickle are private (no public reader of participant state) | |
| from swarmlab.export import _Blobs, _safe_unpickle, read_events | |
| from swarmlab.world.programbench import NOT_THE_SUBMISSION, submission_of | |
| EXP = Path(__file__).resolve().parent.parent | |
| def format_errors(run_dir: Path) -> int | None: | |
| """FormatError messages in every participant's state in the last snapshot (None: unreadable).""" | |
| snaps = sorted((run_dir / "snapshots").glob("*.json")) | |
| if not snaps: | |
| return None | |
| blobs, n = _Blobs(run_dir / "blobs"), 0 | |
| for key, sha in json.loads(snaps[-1].read_text()).get("plugins", {}).items(): | |
| if key.startswith("participant:"): | |
| data = blobs.get(sha) | |
| state = _safe_unpickle(data) if data is not None else None | |
| if not isinstance(state, dict): | |
| return None | |
| n += sum(1 for m in state.get("messages") or [] | |
| if ((m.get("extra") or {}).get("interrupt_type")) == "FormatError") | |
| return n | |
| def run_stats(run_dir: Path) -> dict[str, Any]: | |
| run = Run(run_dir) | |
| score = run.score or {} | |
| role: dict[str, str] = {} | |
| calls: Counter = Counter() | |
| posts: Counter = Counter() | |
| tokens = [0, 0] | |
| tools = 0 | |
| t0 = t1 = None | |
| submitter = None | |
| for ev in read_events(run_dir / "events.jsonl"): # raw: Run.events drops ts | |
| t, agent = ev.get("type"), ev.get("agent") | |
| if t == "run_started": | |
| t0 = ev["ts"] | |
| elif t == "run_ended": | |
| t1 = ev["ts"] | |
| elif t == "turn_started" and agent and ev.get("role"): | |
| role.setdefault(agent, ev["role"]) | |
| elif t == "inference_attempt" and ev.get("category", "swarm") == "swarm": | |
| calls[agent] += 1 | |
| elif t == "inference_response": | |
| u = ev.get("usage") or {} | |
| tokens[0] += u.get("prompt_tokens") or 0 | |
| tokens[1] += u.get("completion_tokens") or 0 | |
| elif t == "tool_called": | |
| tools += 1 | |
| elif t == "post": | |
| posts[agent] += 1 | |
| elif t == "action_committed" and submitter is None and ev.get("accepted"): | |
| fb = ev.get("feedback") or {} | |
| if (isinstance(fb, dict) and submission_of(fb.get("output") or "", fb.get("returncode")) | |
| is not None and NOT_THE_SUBMISSION not in (fb.get("exception_info") or "")): | |
| submitter = agent | |
| roles = run.meta.get("spec", {}).get("participant_roles") or [] | |
| for i, r in enumerate(roles): # agents that never took a turn still get their role | |
| role.setdefault(f"a{i:03d}", r) | |
| lead = [a for a, r in role.items() if r == "lead"] | |
| workers = [a for a, r in role.items() if r == "worker"] | |
| total = sum(calls.values()) | |
| return { | |
| "run_id": run.id, "status": run.status, "end_reason": run.end_reason, | |
| "graded": bool(score.get("graded")), "reward": score.get("score") if score.get("graded") else None, | |
| "passed": score.get("passed"), "total": score.get("total"), | |
| "submitted": score.get("submitted"), | |
| "wall_s": (t1 - t0) if t0 and t1 else None, | |
| "model_calls": total, "tool_calls": tools, | |
| "prompt_tokens": tokens[0], "completion_tokens": tokens[1], | |
| "workers_used": sum(1 for a in workers if calls[a] > 0) if lead else None, | |
| "delegations": sum(posts[a] for a in lead) if lead else None, | |
| "reports": sum(posts[a] for a in workers) if lead else None, | |
| "worker_share": (sum(calls[a] for a in workers) / total) if lead and total else None, | |
| "format_errors": format_errors(run_dir), | |
| "submitted_by": role.get(submitter, submitter) if submitter else "none", | |
| } | |
| def avg(xs: list[Any]) -> float | None: | |
| xs = [x for x in xs if x is not None] | |
| return mean(xs) if xs else None | |
| def bootstrap_ci(diffs: list[float], n: int = 10_000, seed: int = 0) -> tuple[float, float] | None: | |
| if len(diffs) < 2: | |
| return None | |
| rng = random.Random(seed) | |
| means = sorted(mean(rng.choices(diffs, k=len(diffs))) for _ in range(n)) | |
| return means[int(0.025 * n)], means[int(0.975 * n) - 1] | |
| def sign_test(diffs: list[float]) -> dict[str, Any]: | |
| pos, neg = sum(d > 0 for d in diffs), sum(d < 0 for d in diffs) | |
| n = pos + neg | |
| tail = sum(math.comb(n, k) for k in range(min(pos, neg) + 1)) / 2 ** n if n else 1.0 | |
| return {"team_better": pos, "solo_better": neg, "ties": len(diffs) - n, | |
| "p_two_sided": min(1.0, 2 * tail)} | |
| def fmt(x: Any, p: int = 3) -> str: | |
| if x is None: | |
| return "-" | |
| return f"{x:.{p}f}" if isinstance(x, float) else str(x) | |
| def main(argv: list[str] | None = None) -> None: | |
| ap = argparse.ArgumentParser(description=__doc__.split("\n")[0]) | |
| ap.add_argument("--exp-dir", type=Path, default=EXP) | |
| ap.add_argument("--manifest", type=Path, help="default: EXP/sweep/manifest.json") | |
| ap.add_argument("--out", type=Path, help="default: EXP/results") | |
| a = ap.parse_args(argv) | |
| exp = a.exp_dir.resolve() | |
| man = json.loads((a.manifest or exp / "sweep" / "manifest.json").read_text()) | |
| out = a.out or exp / "results" | |
| with (exp / "tasks" / "programbench-tasks.csv").open() as fh: | |
| difficulty = {row["instance_id"]: row["difficulty"] for row in csv.DictReader(fh)} | |
| rows, missing = [], [] | |
| for e in man["runs"]: | |
| d = Path(e.get("run_dir") or f"runs/{e['run_id']}") | |
| d = d if d.is_absolute() else exp / d | |
| if not (d / "run.json").exists(): | |
| missing.append({**{k: e.get(k) for k in ("key", "status", "job_id")}, | |
| "why": "no run dir"}) | |
| continue | |
| st = {**run_stats(d), "task": e["task"], "condition": e["condition"], "seed": e["seed"], | |
| "cpu_job_h": (e.get("running_secs") or 0) / 3600 or None} | |
| rows.append(st) | |
| if st["reward"] is None: | |
| missing.append({"key": e["key"], "status": e.get("status"), "job_id": e.get("job_id"), | |
| "why": "not graded"}) | |
| serve_h = (man.get("serve", {}).get("running_secs") or 0) / 3600 | |
| wall_all = sum(r["wall_s"] or 0 for r in rows) or 1.0 | |
| conds: dict[str, Any] = {} | |
| for cond in sorted({e["condition"] for e in man["runs"]}): | |
| rs = [r for r in rows if r["condition"] == cond] | |
| graded = [r["reward"] for r in rs if r["reward"] is not None] | |
| wall = sum(r["wall_s"] or 0 for r in rs) | |
| conds[cond] = { | |
| "planned": sum(e["condition"] == cond for e in man["runs"]), "runs": len(rs), | |
| "graded": len(graded), "mean_reward": avg(graded), | |
| "pct_resolved": 100 * sum(g >= 1.0 for g in graded) / len(graded) if graded else None, | |
| "pct_almost": 100 * sum(g >= 0.95 for g in graded) / len(graded) if graded else None, | |
| **{f"mean_{k}": avg([r[k] for r in rs]) for k in ( | |
| "wall_s", "model_calls", "tool_calls", "prompt_tokens", "completion_tokens", | |
| "workers_used", "delegations", "reports", "worker_share", "format_errors")}, | |
| "cpu_job_h": sum(r["cpu_job_h"] or 0 for r in rs), | |
| "serve_gpu_h_share_est": serve_h * wall / wall_all, | |
| "submitted_by": dict(Counter(r["submitted_by"] for r in rs)), | |
| } | |
| per_task: dict[str, dict[str, list[float]]] = defaultdict(lambda: defaultdict(list)) | |
| for r in rows: | |
| if r["reward"] is not None: | |
| per_task[r["task"]][r["condition"]].append(r["reward"]) | |
| pairs = [] | |
| for task in dict.fromkeys(e["task"] for e in man["runs"]): | |
| solo, team = avg(per_task[task]["solo-piv"]), avg(per_task[task]["team-piv"]) | |
| pairs.append({"task": task, "difficulty": difficulty.get(task, "?"), "solo": solo, | |
| "team": team, "diff": team - solo if None not in (solo, team) else None}) | |
| diffs = [p["diff"] for p in pairs if p["diff"] is not None] | |
| paired = {"n_tasks": len(diffs), "mean_diff": avg(diffs), "ci95": bootstrap_ci(diffs), | |
| **sign_test(diffs)} | |
| summary = {"conditions": conds, "paired": paired, "per_task": pairs, "missing": missing, | |
| "serve": {**man.get("serve", {}), "gpu_h": serve_h}, "runs": rows} | |
| out.mkdir(parents=True, exist_ok=True) | |
| (out / "summary.json").write_text(json.dumps(summary, indent=1, default=str) + "\n") | |
| L = ["# Flint x ProgramBench: solo PIV vs team PIV", "", | |
| (f"Serve job {man.get('serve', {}).get('job_id') or '-'}: {serve_h:.2f} GPU h " | |
| f"(${man.get('serve', {}).get('cost_usd') or 0:.2f}); tokens are priced $0."), "", | |
| "## Per condition", ""] | |
| cols = [("planned", "planned", 0), ("graded", "graded", 0), ("reward", "mean_reward", 3), | |
| ("% resolved", "pct_resolved", 1), ("% almost", "pct_almost", 1), | |
| ("wall min", "mean_wall_s", 1), ("model calls", "mean_model_calls", 1), | |
| ("tool calls", "mean_tool_calls", 1), ("workers", "mean_workers_used", 2), | |
| ("delegations", "mean_delegations", 1), ("reports", "mean_reports", 1), | |
| ("worker share", "mean_worker_share", 2), ("format err", "mean_format_errors", 2), | |
| ("CPU job h", "cpu_job_h", 2), ("GPU h (est.)", "serve_gpu_h_share_est", 2), | |
| ("submitted by", "submitted_by", 0)] | |
| L.append("| condition | " + " | ".join(c for c, _, _ in cols) + " |") | |
| L.append("|---" * (len(cols) + 1) + "|") | |
| for cond, c in conds.items(): | |
| cells = [] | |
| for _, k, p in cols: | |
| v = c[k] / 60 if k == "mean_wall_s" and c[k] is not None else c[k] | |
| cells.append(", ".join(f"{a} {n}" for a, n in v.items()) if isinstance(v, dict) | |
| else fmt(v, p)) | |
| L.append(f"| {cond} | " + " | ".join(cells) + " |") | |
| ci = paired["ci95"] | |
| ci_text = f"[{fmt(ci[0])}, {fmt(ci[1])}]" if ci else "- (n < 2)" | |
| L += ["", ("Means are over runs; team-only columns are `-` for solo. GPU h per condition " | |
| "splits the serve job's hours by summed run wall time."), "", | |
| "## Paired comparison (team - solo, per task)", "", | |
| (f"Tasks with both graded: {paired['n_tasks']}; mean diff " | |
| f"{fmt(paired['mean_diff'])}; 95% bootstrap CI {ci_text}; sign test: team better " | |
| f"{paired['team_better']}, solo better {paired['solo_better']}, ties " | |
| f"{paired['ties']}, p = {fmt(paired['p_two_sided'])}."), "", | |
| "| task | difficulty | solo | team | diff |", "|---|---|---|---|---|"] | |
| L += [f"| {p['task']} | {p['difficulty']} | {fmt(p['solo'])} | {fmt(p['team'])} | " | |
| f"{fmt(p['diff'])} |" for p in pairs] | |
| L += ["", f"## Missing or ungraded ({len(missing)})", ""] | |
| L += [f"- {m['key']}: {m['why']} (status {m['status']}, job {m['job_id'] or '-'})" | |
| for m in missing] or ["- none"] | |
| (out / "summary.md").write_text("\n".join(L) + "\n") | |
| print("\n".join(L)) | |
| print(f"\nwrote {out / 'summary.md'} and {out / 'summary.json'}") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 12.4 kB
- Xet hash:
- 68f58d646660ada1a6d7fe977d2ac27dd8012ba1cd586d664aef7e3a9ac97f42
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.