reference/pipeline.py

"""Run a chunked D1A Core ML trunk end to end: host embeddings -> chunk 0 .. chunk N (h, per-layer inputs and the shared
keys/values passed along) -> hidden states at the picks -> pointer head on the host.
Usage as a library: Pipeline(export_dir, chunk_dir, L, units).probs(enc) -> [per-question probabilities]."""
import glob
import os
import re
import sys
import time

import coremltools as ct
import numpy as np
import torch

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))   # d1a_coreml.py sits next to this file
from d1a_coreml import Export, HostEmbed, masks_for  # noqa: E402

P_PICKS = 128


class Pipeline:
    def __init__(self, export_dir, chunk_dir, L, units="CPU_AND_NE"):
        self.ex = Export(export_dir)
        self.host = HostEmbed(self.ex)
        self.hq, self.hk = self.ex.head()
        self.T = self.ex.d1a["temperature"]
        self.L = L
        cu = getattr(ct.ComputeUnit, units)
        self.chunks = []
        if chunk_dir.endswith(".mlpackage"):   # one multifunction package: functions c00_07, c07_15, ...
            names = ["c00_07", "c07_15", "c15_22", "c22_29", "c29_35"]
            for name in names:
                a, b = map(int, name[1:].split("_"))
                m = ct.models.MLModel(chunk_dir, compute_units=cu, function_name=name)
                fn = next(f for f in m.get_spec().description.functions if f.name == name)
                self.chunks.append((a, b, m, [i.name for i in fn.input], [o.name for o in fn.output]))
            return
        files = sorted(glob.glob(f"{chunk_dir}/L{L}_*.mlpackage"))
        for f in files:
            a, b = map(int, re.search(r"_(\d+)-(\d+)\.mlpackage$", f).groups())
            m = ct.models.MLModel(f, compute_units=cu)
            spec = m.get_spec()
            self.chunks.append((a, b, m, [i.name for i in spec.description.input], [o.name for o in spec.description.output]))

    def hidden(self, enc):
        L, n = self.L, len(enc["ids"])
        # right-pad with the pad token's own embeddings (as d1a's torch backend pads ids), never zero rows: an all-zero row
        # makes every RMSNorm compute 0 / sqrt(eps), and eps (1e-6) is subnormal in fp16, which the Neural Engine flushes
        ids = list(enc["ids"]) + [self.ex.d1a.get("pad_id", 0)] * (L - n)
        emb, ple = self.host(ids)
        e = emb.numpy().astype(np.float16)
        p = ple.numpy().astype(np.float16)
        mf, ms, pos = masks_for(enc, self.ex.tc.sliding_window, L=L)
        picks = list(enc["decide_idx"]) + [o for oo in enc["opt_idx"] for o in oo]
        pk = np.zeros(P_PICKS, np.int32); pk[: len(picks)] = picks
        common = {"pos": pos.numpy().astype(np.int32), "mask_full": mf.clamp(min=-1e4).numpy().astype(np.float16),
                  "mask_slide": ms.clamp(min=-1e4).numpy().astype(np.float16)}
        state = {"emb": e, "ple_tok": p, "picks": pk}
        for a, b, m, ins, outs in self.chunks:
            feed = {k: np.ascontiguousarray(common[k] if k in common else state[k]) for k in ins}
            feed = {k: (v.astype(np.float16) if v.dtype == np.float32 else v) for k, v in feed.items()}   # chunk outputs come back fp32
            out = {k: np.array(v, copy=True) for k, v in m.predict(feed).items()}   # Core ML may reuse its output buffers on the next call
            if "hidden" in out:
                return torch.tensor(out["hidden"][: len(picks)], dtype=torch.float32)
            state["h_in"] = out["h"]
            if "ple_out" in out:
                state["ple"] = out["ple_out"]
            for k, v in out.items():
                if k.endswith("_out") and k != "ple_out":
                    state[k[: -len("_out")]] = v
        raise RuntimeError("no chunk returned hidden")

    def probs(self, enc):
        h = self.hidden(enc)
        Q = len(enc["decide_idx"])
        out, j = [], Q
        with torch.no_grad():
            for qi, oo in enumerate(enc["opt_idx"]):
                z = (self.hk(h[j: j + len(oo)]) @ self.hq(h[qi])) / (self.hq.out_features ** 0.5)
                out.append(torch.softmax(z / self.T, -1).numpy()); j += len(oo)
        return out