Model/model.py

# -*- coding: utf-8 -*-

from __future__ import annotations

import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.utils.checkpoint import checkpoint

from Model.blocks import StandardBlock
from Model.config import RDTConfig
from Model.layers.rmsnorm import RMSNorm
from Model.recurrent import RecurrentCore
from Model.vision import VisionInjector


class RDTForCausalLM(nn.Module):
    def __init__(self, cfg: RDTConfig, patch_pixels: int = 14 * 14 * 3):
        super().__init__()

        self.cfg = cfg
        self.patch_pixels = patch_pixels

        self.embed = nn.Embedding(
            cfg.vocab_size,
            cfg.d_model,
            padding_idx=cfg.pad_id,
        )

        self.vision = VisionInjector(cfg, patch_pixels)

        self.prelude = nn.ModuleList(
            StandardBlock(cfg, layer_idx=i) for i in range(cfg.n_prelude)
        )

        self.recurrent = RecurrentCore(cfg)

        self.coda = nn.ModuleList(
            StandardBlock(cfg, layer_idx=cfg.n_prelude + i) for i in range(cfg.n_coda)
        )

        self.final_norm = RMSNorm(cfg.d_model, eps=cfg.rmsnorm_eps)
        self.lm_head = nn.Linear(cfg.d_model, cfg.vocab_size, bias=False)

        self.bidirectional = cfg.bidirectional

        if self.bidirectional:
            self.reverse_head = nn.Linear(cfg.d_model, cfg.vocab_size, bias=False)
        else:
            self.reverse_head = None

        self.apply(self._init_weights)

        if cfg.tie_word_embeddings:
            self.lm_head.weight = self.embed.weight
            if self.reverse_head is not None:
                self.reverse_head.weight = self.embed.weight

    def forward(
        self,
        input_ids: torch.Tensor,
        attention_mask: torch.Tensor | None = None,
        labels: torch.Tensor | None = None,
        pixel_values: torch.Tensor | None = None,
        word_pos: torch.Tensor | None = None,
        morph_depth: torch.Tensor | None = None,
        steps: int | None = None,
        bptt_window: int | None = None,
        return_logits: bool = True,
        loss_chunk_size: int | None = None,
    ) -> dict[str, torch.Tensor | dict | None]:
        self._check_inputs(input_ids, attention_mask, labels)
        if loss_chunk_size is None:
            loss_chunk_size = self.cfg.loss_chunk_size
        elif loss_chunk_size <= 0:
            raise ValueError("loss_chunk_size must be positive")

        bsz, seq_len = input_ids.shape

        if attention_mask is None:
            attention_mask = (input_ids != self.cfg.pad_id).long()

        if word_pos is None or morph_depth is None:
            word_pos, morph_depth = self._default_morph_info(
                input_ids=input_ids,
                attention_mask=attention_mask,
            )

        h = self.embed(input_ids)

        if pixel_values is not None:
            h = self.vision(h, input_ids, pixel_values)

        for block in self.prelude:
            h = self._maybe_ckpt(
                block,
                h,
                word_pos=word_pos,
                morph_depth=morph_depth,
                attn_mask=attention_mask,
                causal=True,
            )

        e0 = h

        h, rec_info = self.recurrent(
            e0,
            word_pos=word_pos,
            morph_depth=morph_depth,
            attn_mask=attention_mask,
            causal=True,
            steps=steps,
            bptt_window=bptt_window,
        )

        for block in self.coda:
            h = self._maybe_ckpt(
                block,
                h,
                word_pos=word_pos,
                morph_depth=morph_depth,
                attn_mask=attention_mask,
                causal=True,
            )

        h = self.final_norm(h)
        logits = None

        loss = None
        loss_parts: dict[str, float] = {}

        if labels is not None:
            if not return_logits and loss_chunk_size is not None:
                loss, loss_parts = self._losses_chunked(
                    h,
                    labels,
                    rec_info,
                    loss_chunk_size,
                )
            else:
                logits = self.lm_head(h)
                loss, loss_parts = self._losses(h, logits, labels, rec_info)
        elif return_logits:
            logits = self.lm_head(h)

        return {
            "loss": loss,
            "logits": logits if return_logits else None,
            "loss_parts": loss_parts,
            "rec_info": rec_info,
        }

    def _losses(
        self,
        h: torch.Tensor,
        logits: torch.Tensor,
        labels: torch.Tensor,
        rec_info: dict,
    ) -> tuple[torch.Tensor, dict[str, float]]:
        forward = self._causal_loss(logits, labels)
        loss = forward
        parts = {"forward": float(forward.detach())}

        if self.reverse_head is not None:
            rev_logits = self.reverse_head(h)
            reverse = self._reverse_loss(rev_logits, labels)
            loss = loss + self.cfg.reverse_loss_weight * reverse
            parts["reverse"] = float(reverse.detach())

        ponder = rec_info.get("ponder_cost")
        if self.cfg.use_act and isinstance(ponder, torch.Tensor):
            loss = loss + self.cfg.act_ponder_cost * ponder
            parts["ponder"] = float(ponder.detach())

        return loss, parts

    def _losses_chunked(
        self,
        h: torch.Tensor,
        labels: torch.Tensor,
        rec_info: dict,
        chunk_size: int,
    ) -> tuple[torch.Tensor, dict[str, float]]:
        forward = self._chunked_causal_loss(h, labels, self.lm_head, chunk_size)
        loss = forward
        parts = {"forward": float(forward.detach())}

        if self.reverse_head is not None:
            reverse = self._chunked_reverse_loss(
                h,
                labels,
                self.reverse_head,
                chunk_size,
            )
            loss = loss + self.cfg.reverse_loss_weight * reverse
            parts["reverse"] = float(reverse.detach())

        ponder = rec_info.get("ponder_cost")
        if self.cfg.use_act and isinstance(ponder, torch.Tensor):
            loss = loss + self.cfg.act_ponder_cost * ponder
            parts["ponder"] = float(ponder.detach())

        return loss, parts

    def _causal_loss(self, logits: torch.Tensor, labels: torch.Tensor) -> torch.Tensor:
        return self._token_loss(
            logits[:, :-1].reshape(-1, logits.size(-1)),
            labels[:, 1:].reshape(-1),
        )

    def _reverse_loss(
        self,
        rev_logits: torch.Tensor,
        labels: torch.Tensor,
    ) -> torch.Tensor:
        return self._token_loss(
            rev_logits[:, 1:].reshape(-1, rev_logits.size(-1)),
            labels[:, :-1].reshape(-1),
        )

    def _token_loss(self, logits: torch.Tensor, targets: torch.Tensor) -> torch.Tensor:
        valid = targets != self.cfg.ignore_index
        if logits.numel() == 0 or not bool(valid.any()):
            return logits.sum() * 0.0

        return F.cross_entropy(
            logits,
            targets,
            ignore_index=self.cfg.ignore_index,
        )

    def _chunked_causal_loss(
        self,
        h: torch.Tensor,
        labels: torch.Tensor,
        head: nn.Linear,
        chunk_size: int,
    ) -> torch.Tensor:
        return self._chunked_token_loss(
            h[:, :-1].reshape(-1, h.size(-1)),
            labels[:, 1:].reshape(-1),
            head,
            chunk_size,
        )

    def _chunked_reverse_loss(
        self,
        h: torch.Tensor,
        labels: torch.Tensor,
        head: nn.Linear,
        chunk_size: int,
    ) -> torch.Tensor:
        return self._chunked_token_loss(
            h[:, 1:].reshape(-1, h.size(-1)),
            labels[:, :-1].reshape(-1),
            head,
            chunk_size,
        )

    def _chunked_token_loss(
        self,
        hidden: torch.Tensor,
        targets: torch.Tensor,
        head: nn.Linear,
        chunk_size: int,
    ) -> torch.Tensor:
        valid = targets != self.cfg.ignore_index
        if hidden.numel() == 0 or not bool(valid.any()):
            return hidden.sum() * 0.0

        hidden = hidden[valid]
        targets = targets[valid]
        loss_sum = hidden.new_zeros(())

        for start in range(0, hidden.size(0), chunk_size):
            end = min(start + chunk_size, hidden.size(0))
            logits = F.linear(hidden[start:end], head.weight, head.bias)
            loss_sum = loss_sum + F.cross_entropy(
                logits,
                targets[start:end],
                reduction="sum",
            )

        return loss_sum / targets.numel()

    def _maybe_ckpt(
        self,
        block: nn.Module,
        h: torch.Tensor,
        word_pos: torch.Tensor | None,
        morph_depth: torch.Tensor | None,
        attn_mask: torch.Tensor | None,
        causal: bool,
    ) -> torch.Tensor:
        if (
            getattr(self.cfg, "grad_ckpt_prelude_coda", False)
            and self.training
            and h.requires_grad
        ):
            def _fn(x):
                return block(
                    x,
                    word_pos=word_pos,
                    morph_depth=morph_depth,
                    attn_mask=attn_mask,
                    causal=causal,
                )

            return checkpoint(_fn, h, use_reentrant=False)
        return block(
            h,
            word_pos=word_pos,
            morph_depth=morph_depth,
            attn_mask=attn_mask,
            causal=causal,
        )

    def _default_morph_info(
        self,
        input_ids: torch.Tensor,
        attention_mask: torch.Tensor,
    ) -> tuple[torch.Tensor, torch.Tensor]:
        """Derive ``(word_pos, morph_depth)`` from boundary token IDs.

        This is a vectorized counterpart of
        :func:`Tokenizer.pretraining.derive_morph_info_from_boundary_ids`.
        Callers that already supply ``word_pos`` / ``morph_depth`` skip this.

        Semantics (see tokenizer-side reference for details):

        * ``word_boundary_id`` opens a new word at depth 0.
        * ``morpheme_boundary_id`` keeps the current word and bumps depth.
        * Other special tokens (ids in ``[0, 256)``) reset depth to 0 and
          stay on the current ``word_pos``; the next non-special content
          token will inherit ``word_pos`` until a new ``word_boundary``.
        * Padding positions (``attention_mask == 0``) keep their derived
          ``word_pos`` so downstream RoPE never sees ``-1``; the mask
          itself is what zeros out padded contributions.
        """

        device = input_ids.device
        bsz, seq_len = input_ids.shape
        cfg = self.cfg

        wb = int(cfg.word_boundary_id)
        mb = int(cfg.morpheme_boundary_id)
        special_hi = 256

        is_wb = input_ids == wb
        is_mb = input_ids == mb
        is_special = (input_ids >= 0) & (input_ids < special_hi)
        is_content = ~is_special
        is_other_special = is_special & ~is_wb & ~is_mb

        # word_pos: cumulative count of word_boundary occurrences with a
        # per-row shift. If the very first word_boundary appears before
        # any content token, the first wb anchors word 0 (shift = -1).
        # Otherwise content tokens implicitly open word 0 and the first
        # wb opens word 1 (shift = 0).
        wb_cum = is_wb.long().cumsum(dim=1)

        inf = seq_len + 1
        any_wb = is_wb.any(dim=1)
        any_content = is_content.any(dim=1)

        first_wb = torch.where(
            any_wb,
            is_wb.long().argmax(dim=1),
            torch.full((bsz,), inf, device=device, dtype=torch.long),
        )
        first_content = torch.where(
            any_content,
            is_content.long().argmax(dim=1),
            torch.full((bsz,), inf, device=device, dtype=torch.long),
        )
        shift = torch.where(
            first_wb < first_content,
            torch.full((bsz,), -1, device=device, dtype=torch.long),
            torch.zeros(bsz, device=device, dtype=torch.long),
        )

        word_pos = (wb_cum + shift.unsqueeze(-1)).clamp(min=0)

        # morph_depth: cumulative morpheme_boundary count since the most
        # recent reset (word_boundary or other special). The reset
        # position itself reads depth 0.
        cum_inc = is_mb.long().cumsum(dim=1)
        reset_positions = is_wb | is_other_special
        reset_value = torch.where(
            reset_positions, cum_inc, torch.full_like(cum_inc, -1)
        )
        last_reset, _ = reset_value.cummax(dim=1)
        depth = cum_inc - last_reset.clamp(min=0)
        depth = torch.where(reset_positions, torch.zeros_like(depth), depth)

        if cfg.max_morph_depth > 0:
            depth = depth.clamp(max=cfg.max_morph_depth - 1)

        return word_pos.to(torch.long), depth.to(torch.long)


    def _check_inputs(
        self,
        input_ids: torch.Tensor,
        attention_mask: torch.Tensor | None,
        labels: torch.Tensor | None,
    ) -> None:
        if input_ids.ndim != 2:
            raise ValueError("input_ids must have shape [B, L]")

        if input_ids.numel() == 0:
            raise ValueError("input_ids cannot be empty")

        # Dtype + shape checks only; value-range checks are skipped here
        # to avoid host syncs on the hot path. The embedding lookup will
        # raise an out-of-range error if vocab bounds are violated, and
        # we rely on the data pipeline to keep ids well-formed.
        if input_ids.dtype not in (torch.int32, torch.int64):
            raise TypeError("input_ids must be int32 or int64")

        if attention_mask is not None and attention_mask.shape != input_ids.shape:
            raise ValueError("attention_mask must have shape [B, L]")

        if labels is not None and labels.shape != input_ids.shape:
            raise ValueError("labels must have shape [B, L]")

        if input_ids.shape[1] > self.cfg.max_seq_len:
            raise ValueError("sequence length exceeds max_seq_len")

    def _init_weights(self, module: nn.Module) -> None:
        std = self.cfg.init_std

        if isinstance(module, nn.Linear):
            nn.init.normal_(module.weight, mean=0.0, std=std)
            if module.bias is not None:
                nn.init.zeros_(module.bias)

        elif isinstance(module, nn.Embedding):
            nn.init.normal_(module.weight, mean=0.0, std=std)
            if module.padding_idx is not None:
                with torch.no_grad():
                    module.weight[module.padding_idx].zero_()

    @torch.no_grad()
    def count_params(self, trainable_only: bool = False) -> int:
        seen: set[int] = set()
        total = 0

        for param in self.parameters():
            if trainable_only and not param.requires_grad:
                continue

            ptr = param.data_ptr()
            if ptr in seen:
                continue

            seen.add(ptr)
            total += param.numel()

        return total


def _check() -> None:
    from Model.config import tiny_config

    torch.manual_seed(0)

    cfg = tiny_config()
    model = RDTForCausalLM(cfg)

    print("RDTForCausalLM")
    print(f"  params: {model.count_params():,}")
    print(f"  actual_layers: {cfg.actual_layers}")
    print(f"  effective_depth: {cfg.effective_depth}")

    bsz, seq_len = 2, 32
    input_ids = torch.randint(256, 24576, (bsz, seq_len))
    input_ids[:, 0] = cfg.bos_id

    attention_mask = torch.ones(bsz, seq_len, dtype=torch.long)
    labels = input_ids.clone()

    out = model(
        input_ids=input_ids,
        attention_mask=attention_mask,
        labels=labels,
    )

    print("Text")
    print(f"  logits: {tuple(out['logits'].shape)}")
    print(f"  loss: {out['loss'].item():.6f}")
    print(f"  loss_parts: {out['loss_parts']}")
    print(f"  steps_used: {out['rec_info'].get('steps_used')}")

    out["loss"].backward()
    grad_sq = 0.0
    for p in model.parameters():
        if p.grad is not None:
            grad_sq += p.grad.norm().item() ** 2

    print(f"  grad_norm: {grad_sq**0.5:.6f}")

    ids2 = torch.tensor(
        [
            [
                cfg.bos_id,
                300,
                cfg.image_start_id,
                cfg.image_patch_id,
                cfg.image_patch_id,
                cfg.image_end_id,
                301,
                cfg.eos_id,
            ]
        ]
    )

    labels2 = ids2.clone()
    labels2[ids2 == cfg.image_start_id] = cfg.ignore_index
    labels2[ids2 == cfg.image_patch_id] = cfg.ignore_index
    labels2[ids2 == cfg.image_end_id] = cfg.ignore_index

    pixel_values = torch.randn(2, model.patch_pixels)

    out2 = model(
        input_ids=ids2,
        labels=labels2,
        pixel_values=pixel_values,
    )

    print("ImageText")
    print(f"  loss: {out2['loss'].item():.6f}")

    model.eval()
    with torch.no_grad():
        for steps in [2, 8]:
            out_step = model(input_ids=input_ids, steps=steps)
            print(
                f"  steps={steps}, logits_norm={out_step['logits'].norm().item():.6f}"
            )


if __name__ == "__main__":
    _check()