Analyse (partial) results from a https://github.com/NVIDIA/SkillEvaluator run

analyze_results.py

# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
"""Summarise a SkillEvaluator tier3 run from its per-trial scores.

SkillEvaluator only reports a score for a condition (with or without the skill)
when every trial in it succeeded. One agent timeout or provider error leaves the
report at N/A and no ``lift.json`` is written, even though the other trials were
graded. This script reads each trial's ``reward.json`` instead. It skips trials
that errored or have no reward, and says how many were skipped.

Usage::

    python analyze_results.py RUN_DIR[+RUN_DIR...] [RUN_DIR ...]

``RUN_DIR`` is a run directory such as ``~/skilleval-results/cuml/latest``. Run
directories joined with ``+`` are merged, for example a main run and a fill-in
run of the same skill. Each argument is summarised separately.
"""

import json
import sys
from collections import Counter, defaultdict
from pathlib import Path
from statistics import mean

ARMS = ("with-skill", "without-skill")
# "overall" includes skill_execution and skill_efficiency, which measure use of
# the skill itself, so the content metrics are the fairer comparison.
METRICS = ("overall", "accuracy", "goal_accuracy", "behavior_check", "skill_execution")
CONTENT_METRICS = ("accuracy", "goal_accuracy", "behavior_check")


def agent_dir(run):
    """The per-agent results directory of a run (e.g. opencode/ or claude-code/)."""
    dirs = [d for d in Path(run).iterdir() if (d / "with-skill").is_dir()]
    if len(dirs) != 1:
        sys.exit(f"expected one agent directory in {run}, found {dirs}")
    return dirs[0]


def trial_info(trial):
    result = json.loads((trial / "result.json").read_text())
    exc = result.get("exception_info")
    exc_type = exc.get("exception_type") if isinstance(exc, dict) else exc
    reward_path = trial / "reward.json"
    reward = json.loads(reward_path.read_text()) if reward_path.exists() else None
    traj_path = trial / "trajectory.json"
    calls = []
    if traj_path.exists():
        traj = json.loads(traj_path.read_text())
        calls = [
            c.get("function_name")
            for s in traj.get("steps", [])
            for c in (s.get("tool_calls") or [])
        ]
    return {
        "trial": trial.name,
        "case": trial.name.split("__")[0],
        "error": exc_type or (None if reward else "no reward.json"),
        "reward": reward,
        "skill_called": "skill" in {c.lower() for c in calls if c},
    }


def fmt(x, signed=False):
    if x is None:
        return "-"
    return f"{x:+.2f}" if signed else f"{x:.2f}"


def content_score(reward):
    values = [reward[m] for m in CONTENT_METRICS if reward.get(m) is not None]
    return mean(values) if values else None


def summarise(spec):
    runs = [Path(p).expanduser() for p in spec.split("+")]
    trials = defaultdict(list)  # arm -> [info]
    for arm in ARMS:
        for run in runs:
            for trial in sorted((agent_dir(run) / arm / "trials").iterdir()):
                trials[arm].append(trial_info(trial))
    scored = {arm: [i for i in trials[arm] if not i["error"]] for arm in ARMS}

    print(f"## {' + '.join(str(r) for r in runs)}\n")
    model = next((i["reward"].get("model") for i in scored["with-skill"]), None)
    if model:
        print(f"agent model: {model}\n")

    print(f"{'metric':16s} {'with skill':>12s} {'without':>12s} {'lift':>7s}")
    counts = [f"{len(scored[a])}/{len(trials[a])}" for a in ARMS]
    print(f"{'scored trials':16s} {counts[0]:>12s} {counts[1]:>12s}")
    for m in METRICS:
        means = {
            arm: mean(vals) if (vals := [i["reward"][m] for i in scored[arm]
                                         if i["reward"].get(m) is not None]) else None
            for arm in ARMS
        }
        lift = (
            means["with-skill"] - means["without-skill"]
            if None not in means.values()
            else None
        )
        print(f"{m:16s} {fmt(means['with-skill']):>12s} "
              f"{fmt(means['without-skill']):>12s} {fmt(lift, signed=True):>7s}")
    loaded = sum(i["skill_called"] for i in scored["with-skill"])
    print(f"\nskill loaded in {loaded}/{len(scored['with-skill'])} scored with-skill trials")

    print("\nPer case (content = mean of " + ", ".join(CONTENT_METRICS) + "):")
    print(f"{'case':6s} {'with n':>6s} {'overall':>7s} {'content':>7s} | "
          f"{'w/o n':>5s} {'overall':>7s} {'content':>7s} | {'lift':>6s} {'content':>7s}")
    by_case = defaultdict(lambda: defaultdict(list))
    for arm in ARMS:
        for i in trials[arm]:
            by_case[i["case"]][arm].append(i)
    lifts, content_lifts = [], []
    for case in sorted(by_case):
        cells, overall, content = [], {}, {}
        for arm in ARMS:
            infos = by_case[case][arm]
            ok = [i["reward"] for i in infos if not i["error"]]
            overall[arm] = mean(r["overall"] for r in ok) if ok else None
            cs = [c for r in ok if (c := content_score(r)) is not None]
            content[arm] = mean(cs) if cs else None
            cells.append(f"{len(ok)}/{len(infos):<{5 if arm == ARMS[0] else 3}} "
                         f"{fmt(overall[arm]):>7s} {fmt(content[arm]):>7s}")
        lift = content_lift = None
        if None not in overall.values():
            lift = overall["with-skill"] - overall["without-skill"]
            lifts.append(lift)
        if None not in content.values():
            content_lift = content["with-skill"] - content["without-skill"]
            content_lifts.append(content_lift)
        print(f"{case:6s} {cells[0]} | {cells[1]} | "
              f"{fmt(lift, signed=True):>6s} {fmt(content_lift, signed=True):>7s}")
    if lifts:
        positive = sum(lift > 0 for lift in lifts)
        print(f"\nmean per-case lift: overall {mean(lifts):+.3f}, content "
              f"{mean(content_lifts):+.3f}; overall lift positive in {positive}/{len(lifts)} cases")

    errors = [(arm, i) for arm in ARMS for i in trials[arm] if i["error"]]
    if errors:
        print(f"\nUnscored trials ({len(errors)}, excluded above):")
        for arm, i in errors:
            print(f"  {arm:13s} {i['trial']:16s} {i['error']}")
        print("  by type:", dict(Counter(i["error"] for _, i in errors)))
    print()


if __name__ == "__main__":
    if len(sys.argv) < 2:
        sys.exit(__doc__)
    for spec in sys.argv[1:]:
        summarise(spec)