hobbylm/diffusion.py

"""Pure masked-diffusion (LLaDA / MDLM) conversion utilities for the MoE LLM.

The model itself only flips one thing for diffusion: attention becomes bidirectional
(model.py, gated on cfg.diffusion). Everything else lives here:

  forward_mask : the forward (noising) process used at train time.
  generate     : iterative-denoising sampler with semi-autoregressive blocks.

The TRAIN loss is model.diffusion_cross_entropy (fused/chunked for big batches). The
unfused `diffusion_loss` below is for tests / sanity checks only.
"""
from __future__ import annotations

import torch
import torch.nn.functional as F
from torch import Tensor


def forward_mask(input_ids: Tensor, mask_id: int, eps: float = 1e-3,
                 prompt_lens: Tensor | None = None, generator: torch.Generator | None = None):
    """LLaDA forward process.

    One mask ratio t ~ U(eps, 1) per sequence; mask each token iid with prob t.
    Returns (noisy, labels, p_mask):
      noisy  : input with masked positions replaced by mask_id
      labels : original token at masked positions, -1 (ignore_index) elsewhere
      p_mask : per-token mask probability t (broadcast), used for the 1/p reweighting
    `prompt_lens` (B,) optionally protects a prompt prefix from being masked/scored (SFT).
    """
    b, l = input_ids.shape
    dev = input_ids.device
    t = torch.rand(b, device=dev, generator=generator) * (1 - eps) + eps   # (b,)
    p_mask = t[:, None].expand(b, l).contiguous()                          # (b, l)
    mask = torch.rand(b, l, device=dev, generator=generator) < p_mask
    if prompt_lens is not None:
        pos = torch.arange(l, device=dev)[None, :]
        mask &= pos >= prompt_lens[:, None]
    # guarantee >=1 masked token per sequence so no micro-batch contributes a zero loss
    none_masked = ~mask.any(dim=1)
    if none_masked.any():
        rows = none_masked.nonzero(as_tuple=True)[0]
        lo = 0 if prompt_lens is None else int(prompt_lens.min().item())
        j = torch.randint(lo, l, (rows.numel(),), device=dev, generator=generator)
        mask[rows, j] = True
    noisy = torch.where(mask, torch.full_like(input_ids, mask_id), input_ids)
    labels = torch.where(mask, input_ids, torch.full_like(input_ids, -1))
    return noisy, labels, p_mask


def diffusion_loss(logits: Tensor, labels: Tensor, p_mask: Tensor) -> Tensor:
    """Unfused LLaDA loss (tests): sum_{masked} CE / p_mask, normalized by B*L."""
    b, l, _ = logits.shape
    m = labels != -1
    if int(m.sum()) == 0:
        return logits.sum() * 0.0
    ce = F.cross_entropy(logits[m].float(), labels[m], reduction="none")
    return (ce / p_mask[m]).sum() / (b * l)


def get_num_transfer_tokens(n: int, steps: int) -> list[int]:
    """Spread n unmask events as evenly as possible across `steps` (sums to n)."""
    base = n // steps
    out = [base] * steps
    for i in range(n - base * steps):
        out[i] += 1
    return out


def add_gumbel_noise(logits: Tensor, temperature: float,
                     generator: torch.Generator | None = None) -> Tensor:
    """Gumbel-max categorical sampling (LLaDA). argmax of this == a sample at `temperature`.
    temperature<=0 -> identity (argmax == greedy)."""
    if temperature <= 0:
        return logits
    logits = logits.to(torch.float64)
    noise = torch.rand(logits.shape, dtype=torch.float64, device=logits.device, generator=generator)
    gumbel = (-torch.log(noise + 1e-12)) ** temperature
    return logits.exp() / gumbel


def _rep_penalty(blk: Tensor, present_ids: Tensor, penalty: float) -> Tensor:
    """CTRL-style penalty across the canvas: damp logits of tokens already present (prompt +
    committed) so the denoiser stops filling many slots with the same token. In-place on blk."""
    if penalty == 1.0 or present_ids.numel() == 0:
        return blk
    col = blk[:, present_ids]
    blk[:, present_ids] = torch.where(col > 0, col / penalty, col * penalty)
    return blk


@torch.no_grad()
def generate(model, prompt_ids: Tensor, gen_len: int = 256, block: int = 32, steps: int = 64,
             mask_id: int = 50257, temperature: float = 0.0, rep_penalty: float = 1.0,
             remask_steps: int = 0, remask_frac: float = 0.3, valid_vocab: int = 50257,
             eos_id: int | None = None, generator: torch.Generator | None = None) -> Tensor:
    """Semi-autoregressive iterative denoising. prompt_ids: (1, P). Returns generated ids (1, <=gen_len).

    Each block of `block` masked slots is filled over ~`steps*block/gen_len` steps, committing the
    highest-confidence still-masked positions each step (low-confidence-remasking selection). Then
    `remask_steps` refinement passes re-mask the lowest-confidence committed tokens and re-predict
    them with full bidirectional context — this is what lets the model fix repetition/mistakes.
    Sentinels (>= valid_vocab, incl. mask_id) are banned from being emitted. Blocks are causal
    w.r.t. each other (a block attends to the committed prefix + itself), bidirectional within.
    """
    was_training = model.training
    model.eval()
    dev = prompt_ids.device
    x = torch.cat([prompt_ids, torch.full((1, gen_len), mask_id, device=dev, dtype=prompt_ids.dtype)], dim=1)
    P = prompt_ids.shape[1]

    def block_logits(b1: int, b0: int) -> Tensor:
        """Forward prefix+block; return (blk_len, V) logits with sentinels banned + rep-penalty."""
        logits, _ = model(x[:, :b1])
        blk = logits[0, b0:b1].float()
        blk[:, valid_vocab:] = -float("inf")                  # never emit mask/sentinel ids
        present = torch.unique(x[0, :b1])
        present = present[(present < valid_vocab) & (present != mask_id)]
        return _rep_penalty(blk, present, rep_penalty)

    def predict(blk: Tensor):
        prob = blk.softmax(-1)
        pred = add_gumbel_noise(blk, temperature, generator).argmax(-1) if temperature > 0 else blk.argmax(-1)
        return pred, prob

    for b0 in range(P, P + gen_len, block):
        b1 = min(b0 + block, P + gen_len)
        blk_len = b1 - b0
        sb = max(1, round(steps * blk_len / gen_len))
        sched = get_num_transfer_tokens(blk_len, sb)
        # --- fill: commit the most-confident still-masked positions over sb steps ---
        for s in range(sb):
            pred, prob = predict(block_logits(b1, b0))
            conf = prob.gather(-1, pred.unsqueeze(-1)).squeeze(-1)
            still = x[0, b0:b1] == mask_id
            conf = torch.where(still, conf, torch.full_like(conf, -1.0))
            k = min(sched[s], int(still.sum()))
            if k <= 0:
                continue
            idx = conf.topk(k).indices
            x[0, b0 + idx] = pred[idx].to(x.dtype)
        # --- refine: re-mask the least-confident committed tokens and re-predict them ---
        for _ in range(remask_steps):
            blk = block_logits(b1, b0)
            prob = blk.softmax(-1)
            cur = x[0, b0:b1]
            cur_conf = prob.gather(-1, cur.unsqueeze(-1)).squeeze(-1)   # confidence in current tokens
            r = max(1, int(blk_len * remask_frac))
            x[0, b0 + cur_conf.topk(r, largest=False).indices] = mask_id
            pred, _ = predict(block_logits(b1, b0))
            still = (x[0, b0:b1] == mask_id).nonzero(as_tuple=True)[0]
            x[0, b0 + still] = pred[still].to(x.dtype)
        if eos_id is not None and bool((x[0, b0:b1] == eos_id).any()):
            rel = int((x[0, b0:b1] == eos_id).nonzero(as_tuple=True)[0][0].item())
            if was_training:
                model.train()
            return x[:, P:b0 + rel + 1]
    if was_training:
        model.train()
    return x[:, P:]