Blame for code/actionman/data.py

"""Datasets, collation (random mask ratio = masked-diffusion training), splits, negatives."""
from __future__ import annotations

import json
import random

import torch

from .features import build_pass_a, build_pass_b, gold_plan
from .schema import Example, example_from_dict, example_to_dict


def load_jsonl(path) -> list[Example]:
    with open(path, encoding="utf-8") as f:
        return [example_from_dict(json.loads(line)) for line in f if line.strip()]


def save_jsonl(path, examples: list[Example]):
    with open(path, "w", encoding="utf-8") as f:
        for ex in examples:
            f.write(json.dumps(example_to_dict(ex), ensure_ascii=False) + "\n")


def featurize(tok, ex: Example, cached: bool = False, markers: bool = False, v3: bool = False) -> dict | None:
    if v3:
        from .v3feat import build_pass_a_v3, build_pass_b_v3, gold_plan_v3
        a = build_pass_a_v3(tok, ex)
        if a is None:
            return None
        b = None
        if ex.calls and len(ex.calls) <= len(a["call_pos"]):
            b = build_pass_b_v3(tok, ex, a, gold_plan_v3(tok, ex, a), with_targets=True, keep_context=cached,
                                markers=markers)
        return {"a": a, "b": b}
    a = build_pass_a(tok, ex)
    if a is None:
        return None
    b = None
    if ex.calls and len(ex.calls) <= len(a["call_pos"]):
        b = build_pass_b(tok, ex, a, gold_plan(a, ex), with_targets=True, keep_context=cached, markers=markers)
    return {"a": a, "b": b}


class FeatureDataset(torch.utils.data.Dataset):
    def __init__(self, tok, examples: list[Example], cached: bool = False, markers: bool = False, v3: bool = False):
        self.tok, self.examples, self.cached, self.markers, self.v3 = tok, examples, cached, markers, v3

    def __len__(self):
        return len(self.examples)

    def __getitem__(self, i):
        return featurize(self.tok, self.examples[i], self.cached, self.markers, self.v3)

    def length(self, i) -> int:
        """Cheap length proxy for bucketing (chars/3)."""
        ex = self.examples[i]
        n = len(ex.query) + sum(len(t.name) + len(t.description) + 20 * len(t.params) for t in ex.tools)
        return n // 3 + 40


def _pad(seqs, value):
    L = max(len(s) for s in seqs)
    ids = torch.full((len(seqs), L), value, dtype=torch.long)
    attn = torch.zeros((len(seqs), L), dtype=torch.long)
    for i, s in enumerate(seqs):
        ids[i, : len(s)] = torch.tensor(s)
        attn[i, : len(s)] = 1
    return ids, attn


def _mask_slots(b, rng, v3_mask: str = "clean"):
    """Masked-diffusion corruption: random ratio r ~ U(0,1], mask that share of slot tokens, reveal the rest.
    v3_mask="legacy" reproduces the original v3 run (s10): every anchor always masked, gate trained on all anchors."""
    ids = list(b["input_ids"])
    flat = [(p, t) for pos, tg in zip(b["slot_pos"], b["slot_targets"]) for p, t in zip(pos, tg)]
    if not flat:                                   # v3: option-only slot block
        return ids, []
    n_mask = max(1, round(rng.uniform(0.0, 1.0) * len(flat)))
    masked = set(rng.sample(range(len(flat)), n_mask))
    always = set()
    if b.get("anchors") and v3_mask == "legacy":
        always = set(b["anchors"])
        b.pop("_clean", None)
    elif b.get("anchors"):
        # v3: half the values are fully masked ("clean"); only their anchors train the copy gate / span pointer,
        # matching inference, where the gate is read once with every slot masked. Other values follow the random
        # mask, so revealed anchors (as in later refinement passes) are seen too.
        clean = [pos for pos in b["slot_pos"] if rng.random() < 0.5]
        always = {p for pos in clean for p in pos}
        b["_clean"] = [pos[0] for pos in clean]
    targets = []
    for idx, (p, t) in enumerate(flat):
        if idx in masked or p in always:
            targets.append((p, t))
        else:
            ids[p] = t
    return ids, targets


def collate(items, tok, rng: random.Random | None = None, cached: bool = False, v3_mask: str = "clean"):
    rng = rng or random
    items = [it for it in items if it is not None]
    if not items:
        return None
    if cached:
        return _collate_cached(items, tok, rng, v3_mask)
    A = [it["a"] for it in items]
    a_ids, a_attn = _pad([a["input_ids"] for a in A], tok.pad_token_id)
    J = max(max(len(a["t_pos"]) for a in A), 1)
    t_pos = torch.full((len(A), J), -1, dtype=torch.long)
    for i, a in enumerate(A):
        t_pos[i, : len(a["t_pos"])] = torch.tensor(a["t_pos"], dtype=torch.long)
    dec = torch.tensor([[J if (l == len(a["t_pos"])) else l for l in a["dec_labels"]] for a in A])
    len_idx, len_labels = [], []
    for i, a in enumerate(A):
        for (ci, j, k, b) in a["len_items"]:
            len_idx.append((i, a["call_pos"][ci], a["arg_pos"][(j, k)]))
            len_labels.append(b)
    batch = {"a_ids": a_ids, "a_attn": a_attn, "t_pos": t_pos, "t_mask": t_pos >= 0,
             "call_pos": torch.tensor([a["call_pos"] for a in A]), "dec_labels": dec,
             "len_idx": torch.tensor(len_idx, dtype=torch.long).reshape(-1, 3),
             "len_labels": torch.tensor(len_labels, dtype=torch.long)}

    # pass B with a random mask ratio per example (masked-diffusion objective)
    B = [it["b"] for it in items if it["b"] is not None and (it["b"]["slot_pos"] or it["b"].get("opt_slots"))]
    if B:
        seqs, sb, sp, sl = [], [], [], []
        for bi, b in enumerate(B):
            ids, targets = _mask_slots(b, rng, v3_mask)
            for p, t in targets:
                sb.append(bi); sp.append(p); sl.append(t)
            seqs.append(ids)
        b_ids, b_attn = _pad(seqs, tok.pad_token_id)
        batch.update({"b_ids": b_ids, "b_attn": b_attn, "slot_b": torch.tensor(sb), "slot_p": torch.tensor(sp),
                      "slot_labels": torch.tensor(sl)})
    else:
        batch["b_ids"] = None
    if "q_pos" in A[0]:
        from .v3feat import v3_extras
        batch.update(v3_extras(A, list(enumerate(B))))
    return batch


def _collate_cached(items, tok, rng, v3_mask="clean"):
    """One sequence per example: pass-A context + slot block, read with the block-causal mask."""
    A, seqs, ctx, sb, sp, sl = [], [], [], [], [], []
    for i, it in enumerate(items):
        a, b = it["a"], it["b"]
        A.append(a)
        ctx.append(len(a["input_ids"]))
        if b is not None and (b["slot_pos"] or b.get("opt_slots")):
            ids, targets = _mask_slots(b, rng, v3_mask)
            for p, t in targets:
                sb.append(i); sp.append(p); sl.append(t)
            seqs.append(ids)
        else:
            seqs.append(a["input_ids"])
    ids, attn = _pad(seqs, tok.pad_token_id)
    J = max(max(len(a["t_pos"]) for a in A), 1)
    t_pos = torch.full((len(A), J), -1, dtype=torch.long)
    for i, a in enumerate(A):
        t_pos[i, : len(a["t_pos"])] = torch.tensor(a["t_pos"], dtype=torch.long)
    dec = torch.tensor([[J if (l == len(a["t_pos"])) else l for l in a["dec_labels"]] for a in A])
    len_idx, len_labels = [], []
    for i, a in enumerate(A):
        for (ci, j, k, b) in a["len_items"]:
            len_idx.append((i, a["call_pos"][ci], a["arg_pos"][(j, k)]))
            len_labels.append(b)
    out = {"a_ids": ids, "a_attn": attn, "t_pos": t_pos, "t_mask": t_pos >= 0,
            "call_pos": torch.tensor([a["call_pos"] for a in A]), "dec_labels": dec,
            "len_idx": torch.tensor(len_idx, dtype=torch.long).reshape(-1, 3),
            "len_labels": torch.tensor(len_labels, dtype=torch.long),
            "ctx_len": torch.tensor(ctx), "b_ids": None, "slot_b": torch.tensor(sb, dtype=torch.long),
            "slot_p": torch.tensor(sp, dtype=torch.long), "slot_labels": torch.tensor(sl, dtype=torch.long)}
    if "q_pos" in A[0]:
        from .v3feat import v3_extras
        out.update(v3_extras(A, [(i, it["b"]) for i, it in enumerate(items)
                                 if it["b"] is not None and (it["b"]["slot_pos"] or it["b"].get("opt_slots"))]))
    return out


def token_batches(lengths: list[int], tokens_per_batch: int, rng: random.Random, max_bs: int = 256):
    """Length-bucketed batches (sorted by length, chunked by token budget, shuffled)."""
    order = sorted(range(len(lengths)), key=lambda i: lengths[i])
    batches, cur, cur_max = [], [], 0
    for i in order:
        m = max(cur_max, lengths[i])
        if cur and (m * (len(cur) + 1) > tokens_per_batch or len(cur) >= max_bs):
            batches.append(cur); cur, m = [], lengths[i]
        cur.append(i); cur_max = m
    if cur:
        batches.append(cur)
    rng.shuffle(batches)
    return batches


def hard_negatives(examples: list[Example], n: int, rng: random.Random) -> list[Example]:
    """Same query, same domain: drop every tool the gold answer uses, keep its related neighbours -> no call."""
    out = []
    pool = [e for e in examples if e.calls and len(e.tools) > len({c.name for c in e.calls})]
    rng.shuffle(pool)
    for e in pool[:n]:
        needed = {c.name for c in e.calls}
        tools = [t for t in e.tools if t.name not in needed]
        out.append(Example(e.query, tools, [], id=f"hardneg_{len(out)}", source="xlam_hardneg", gold=[]))
    return out


def irrelevance_negatives(examples: list[Example], n: int, rng: random.Random) -> list[Example]:
    """Pair real queries with tool lists from other examples that don't contain any needed tool."""
    out = []
    pool = [e for e in examples if e.tools]
    while len(out) < n:
        q, other = rng.choice(pool), rng.choice(pool)
        needed = {c.name for c in q.calls}
        tools = [t for t in other.tools if t.name not in needed]
        if not tools or other is q:
            continue
        out.append(Example(q.query, tools, [], id=f"neg_{len(out)}", source="xlam_neg", gold=[]))
    return out