build.py

"""Stage 2: turn graded runs into routing labels (chat-format SFT data) with a held-out test split.

Label policy (cost-aware oracle):
  * cheapest routable agent that solved the task, counting a win only if the agent is reliable on that source
    (solve rate >= 70%). Wins by unreliable agents are mostly lucky single samples a router cannot predict: v1 sent
    gemma-3-1b 68 held-out tasks and only 53% were solved, and kimi-vl solves just 63% of text tasks that >= 4 other
    agents solve (lightning 95%, glm 98%). gemma-3-1b is excluded as a target outright;
  * if no routable agent solved it -> ensemble of the capable agents among glm / qwen / ultra (aggregated by the
    last one) when at least two can take the task, otherwise the strongest capable agent alone.
The router sees `router_view(...)`: a size / image header plus the start and end of the task.
"""
import json, random
from collections import Counter
from datasets import Dataset, DatasetDict
from common import AGENTS, POOL, ROUTER_SYSTEM, STRONGEST, capable, est_tokens, router_view

random.seed(0)
ROUTABLE = [a for a in AGENTS if a != "gemma-3-1b"]
ENSEMBLE_POOL = ["glm-5.3-flash", "qwen3.8-27b", STRONGEST]
RELIABLE = 0.70
SOLVE_RATE = {}  # (source, agent) -> solve rate, filled in main()


def cost(res, name):
    return POOL[name]["cost"] * (res[name]["prompt_tokens"] + res[name]["completion_tokens"])


def can(task, agent):
    """Capability check; long prompts are stored truncated, so trust the recorded skip instead of re-estimating."""
    return not task["results"][agent].get("skipped") and capable(agent, "" if task.get("prompt_truncated")
                                                                 else task["prompt"], task["has_image"])


def label(task):
    res = task["results"]
    n_correct = sum(res[a]["correct"] for a in AGENTS)
    correct = [a for a in ROUTABLE if res[a]["correct"] and SOLVE_RATE[task["source"], a] >= RELIABLE]
    difficulty = "easy" if n_correct >= 4 else "medium" if n_correct >= 2 else "hard"
    if correct:
        return dict(strategy="single", agents=[min(correct, key=lambda a: cost(res, a))], difficulty=difficulty)
    ens = [a for a in ENSEMBLE_POOL if can(task, a)]
    if len(ens) >= 2 and not task["has_image"]:
        return dict(strategy="ensemble", agents=ens, difficulty="hard")
    fallback = [a for a in ROUTABLE if can(task, a)] or [STRONGEST]
    return dict(strategy="single", agents=[fallback[-1]], difficulty="hard")


def to_example(task):
    lab = label(task)
    target = json.dumps({"domain": task["domain"], "difficulty": lab["difficulty"],
                         "strategy": lab["strategy"], "agents": lab["agents"]})
    view = router_view(task["prompt"], task["has_image"], task["est_tokens"])
    return dict(id=task["id"], source=task["source"], domain=task["domain"], kind=task["kind"],
                gold=task["gold"], prompt=task["prompt"], has_image=task["has_image"],
                est_tokens=task["est_tokens"], prompt_truncated=bool(task.get("prompt_truncated")),
                router_input=view, route=target,
                agent_results=json.dumps({a: {k: task["results"][a].get(k) for k in
                                              ("correct", "completion_tokens", "latency", "skipped")}
                                          for a in AGENTS}),
                messages=[{"role": "system", "content": ROUTER_SYSTEM},
                          {"role": "user", "content": view},
                          {"role": "assistant", "content": target}])


def main():
    rows = [json.loads(l) for l in open("data/runs_v3.jsonl")]
    n_all = len(rows)
    rows = [r for r in rows if not any(v.get("error") for v in r["results"].values())]
    print(f"{n_all} runs, {n_all - len(rows)} dropped for transient API errors")
    for r in rows:
        r.setdefault("has_image", False)
        r.setdefault("est_tokens", est_tokens(r["prompt"]))
    for src in {r["source"] for r in rows}:
        sub = [r for r in rows if r["source"] == src]
        for a in AGENTS:
            SOLVE_RATE[src, a] = sum(r["results"][a]["correct"] for r in sub) / len(sub)
    random.shuffle(rows)
    ex = [to_example(r) for r in rows]
    n_test = min(max(200, len(ex) // 10), len(ex) // 5)
    dd = DatasetDict(train=Dataset.from_list(ex[n_test:]), test=Dataset.from_list(ex[:n_test]))
    dd.save_to_disk("data/shaktimaan_routing")
    print(dd)
    routes = Counter(json.loads(e["route"])["strategy"] + ":" + ",".join(json.loads(e["route"])["agents"])
                     for e in ex)
    print("label distribution:", *routes.most_common(), sep="\n  ")
    for src in sorted({r["source"] for r in rows}):
        sub = [r for r in rows if r["source"] == src]
        print(f"{src:10s} n={len(sub):4d} " + " ".join(
            f"{a}={sum(r['results'][a]['correct'] for r in sub) / len(sub):.2f}" for a in AGENTS))


if __name__ == "__main__":
    main()