models/diffusion.py

import itertools
import math
import torch
import numpy as np
import pytorch_lightning as L
import torchmetrics
from dataclasses import dataclass
import dit, ema
import noise_schedule  # Assuming this is part of the MDLM repository

LOG2 = math.log(2)

@dataclass
class Loss:
    loss: torch.FloatTensor
    nlls: torch.FloatTensor
    token_mask: torch.FloatTensor

class NLL(torchmetrics.MeanMetric):
    pass

class BPD(NLL):
    def compute(self) -> torch.Tensor:
        """Computes the bits per dimension.
        Returns:
          bpd
        """
        return self.mean_value / self.weight / LOG2

class Perplexity(NLL):
    def compute(self) -> torch.Tensor:
        """Computes the Perplexity.
        Returns:
         Perplexity
        """
        return torch.exp(self.mean_value / self.weight)

# Based on MDLM repo
class Diffusion(L.LightningModule):
    def __init__(self, config, latent_dim, tokenizer):
        super().__init__()
        self.config = config
        self.latent_dim = latent_dim
        self.tokenizer = tokenizer

        self.backbone = dit.DIT(self.config, vocab_size=self.latent_dim)
        self.T = self.config.T
        self.subs_masking = self.config.SUBS_MASKING
        self.antithetic_sampling = self.config.Training.ANTITHETIC_SAMPLING
        self.mask_index = self.tokenizer.mask_token_id

        self.softplus = torch.nn.Softplus()
        metrics = torchmetrics.MetricCollection({
            'nll': NLL(),
            'bpd': BPD(),
            'ppl': Perplexity(),
        })
        metrics.set_dtype(torch.float64)
        self.train_metrics = metrics.clone(prefix='train/')
        self.valid_metrics = metrics.clone(prefix='val/')
        self.test_metrics = metrics.clone(prefix='test/')

        self.noise = noise_schedule.get_noise(self.config, dtype=self.dtype)
        self.lr = self.config.Optim.LR
        self.sampling_eps = self.config.Training.SAMPLING_EPS
        self.time_conditioning = self.config.TIME_CONDITIONING
        self.neg_infinity = -1000000.0


    ############ FORWARD DIFFUSION #########
    def subs_parameterization(self, logits, noised_latents):
        # log prob at the mask index = - infinity
        logits[:, :, self.mask_index] += self.neg_infinity
        
        # Normalize the logits such that x.exp() is
        # a probability distribution over vocab_size.
        logits = logits - torch.logsumexp(logits, dim=-1,
                                        keepdim=True)

        # Apply updates directly in the logits matrix.
        # For the logits of the unmasked tokens, set all values
        # to -infinity except for the indices corresponding to
        # the unmasked tokens.
        unmasked_indices = (noised_latents != self.mask_index)
        logits[unmasked_indices] = self.neg_infinity
        logits[unmasked_indices, noised_latents[unmasked_indices]] = 0
        return logits

    def forward(self, latents, sigma):
        latents = latents.long()
        with torch.cuda.amp.autocast(dtype=torch.float32):
            logits = self.backbone(latents, sigma)
        print(logits)
        optimized_logits = self.subs_parameterization(logits, latents)
        return optimized_logits
    
    def q_xt(self, latents, move_chance):
        """
        Computes the noisy sample xt.
        Args:
            x: int torch.Tensor with shape (batch_size, diffusion_model_input_length), input. 
            move_chance: float torch.Tensor with shape (batch_size, 1).
        """
        latents = latents.mean(dim=1) # [bsz x seq_len x 1280] --> [bsz x 1280] as per args
        move_indices = torch.rand(* latents.shape, device=latents.device) < move_chance
        noised_latents = torch.where(move_indices, self.mask_index, latents)
        return noised_latents

    def sample_timestep(self, n, device):
        _eps_t = torch.rand(n, device=device)
        if self.antithetic_sampling:
            offset = torch.arange(n, device=device) / n
            _eps_t = (_eps_t / n + offset) % 1
        t = (1 - self.sampling_eps) * _eps_t + self.sampling_eps
        # if self.importance_sampling:
        #     return self.noise.importance_sampling_transformation(t)
        return t


    def d3pm_loss(self, model_output, xt, x0, t):
        """Computes the D3PM loss between noisy latents and the original input at a given time step."""
        dt = 1 / self.T

        if torch.is_tensor(t):
            t = t[:, None]
            assert t.ndim == 2
            t = t.clamp(0., 1. - 1e-4)
        alpha_t = 1 - t + torch.zeros_like(xt)
        alpha_s = 1 - (t - dt) + torch.zeros_like(xt)

        x0 = x0.to(torch.int64)
        log_x_theta_at_x0 = torch.gather(model_output, -1, x0[:, :, None]).squeeze(-1)
        log_x_theta_at_m = model_output[:, :, self.mask_index]
        x_theta_at_m = log_x_theta_at_m.exp()
        
        term_1_coef = dt / t
        term_1_log_nr = torch.log(alpha_t * x_theta_at_m / t + 1)
        term_1_log_dr = log_x_theta_at_x0
        
        term_2_coef = 1 - dt / t
        term_2_log_nr = term_1_log_nr
        term_2_log_dr = torch.log(alpha_s * x_theta_at_m / (t - dt) + 1)

        L_vb_masked = (
            term_1_coef * (term_1_log_nr - term_1_log_dr)
            + term_2_coef * (term_2_log_nr - term_2_log_dr))

        L_vb = L_vb_masked * (xt == self.mask_index)

        return self.T * L_vb

    def forward_diffusion(self, latents):
        """Forward diffusion process, adds noise to the latents."""

        t = self.sample_timestep(latents.shape[0], latents.device)
        if self.T > 0:
            t = (t * self.T).to(torch.int)
            t = t / self.T
            # t \in {1/T, 2/T, ..., 1}
            t += (1 / self.T)

        sigma, dsigma = self.noise(t)
        unet_conditioning = sigma[:, None]
        move_chance = 1 - torch.exp(-sigma[:, None])
        
        noised_latents = self.q_xt(latents, move_chance)
        model_output = self.forward(noised_latents, unet_conditioning)
    
        if self.T > 0:
            diffusion_loss = self.d3pm_loss(model_output=model_output, xt=noised_latents, x0=latents, t=t)
            return diffusion_loss
        # SUBS parameterization, continuous time.
        else:
            log_p_theta = torch.gather(input=model_output, dim=-1, index=latents[:, :, None]).squeeze(-1)
            return - log_p_theta * (dsigma / torch.expm1(sigma))[:, None]
    

    ######### LOSS CALCULATIONS #########
    def maybe_sub_sample(self, x0, attention_mask):
        # seqlen = x0.shape[1]
        # print(seqlen)
        # if seqlen > self.config.model.length:
        #     assert seqlen == 2 * self.config.model.length
        #     # cropping is needed for text8-crop dataset
        #     # try the same starting point for now
        #     start = np.random.choice(self.config.model.length)
        #     end = start + self.config.model.length
        #     input_tokens = x0[:, start: end]
        #     output_tokens = x0[:, start + 1: end + 1]
        #     new_attention_mask = attention_mask[:, start: end]

        #     # Helps with validation PPL, since the val
        #     # examples will all start and end with BOS/EOS
        #     input_tokens[:, 0] = self.tokenizer.bos_token_id
        #     output_tokens[:, -1] = self.tokenizer.eos_token_id
    
        # elif self.parameterization == 'ar':
        #     input_tokens = x0[:, :-1]
        #     output_tokens = x0[:, 1:]
        #     new_attention_mask = attention_mask[:, 1:]
        # else:
        input_tokens = x0
        output_tokens = None
        new_attention_mask = attention_mask
    
        return input_tokens, output_tokens, new_attention_mask

    def compute_loss(self, latents, attention_mask):
        """"Average of MLM losses to stabilize training"""
        (input_tokens, output_tokens, attention_mask) = self.maybe_sub_sample(latents, attention_mask)
        loss = self.forward_diffusion(input_tokens)

        nlls = loss * attention_mask
        count = attention_mask.sum()
        batch_nll = nlls.sum()
        token_nll = batch_nll / count

        return Loss(loss=token_nll, nlls=nlls, token_mask=attention_mask)


    ######### TRAINING #########
    def training_step(self, batch, batch_idx):
        latents, attention_mask = batch
        loss = self.compute_loss(latents, attention_mask)
        return loss

    def configure_optimizers(self):
        optimizer = torch.optim.Adam(self.parameters(), lr=self.lr)
        return optimizer

    def validation_step(self, batch):
        latents, attention_mask = batch
        loss = self.compute_loss(latents, attention_mask)
        return loss
    

    ######### GENERATION #########
    def sample_prior(self, *batch_dims):
        return self.mask_index * torch.ones(* batch_dims, dtype=torch.int64)

    def sample_categorical(categorical_probs):
        gumbel_norm = (1e-10 - (torch.rand_like(categorical_probs) + 1e-10).log())
        return (categorical_probs / gumbel_norm).argmax(dim=-1)

    def ddpm_caching_update(self, x, t, dt, p_x0=None):
        assert self.config.noise.type == 'loglinear'
        sigma_t, _ = self.noise(t)
        if t.ndim > 1:
            t = t.squeeze(-1)
        assert t.ndim == 1
        move_chance_t = t[:, None, None]
        move_chance_s = (t - dt)[:, None, None]
        assert move_chance_t.ndim == 3, move_chance_t.shape
        if p_x0 is None:
            p_x0 = self.forward(x, sigma_t).exp()
    
        assert move_chance_t.ndim == p_x0.ndim
        q_xs = p_x0 * (move_chance_t - move_chance_s)
        q_xs[:, :, self.mask_index] = move_chance_s[:, :, 0]
        _x = self.sample_categorical(q_xs)
        
        copy_flag = (x != self.mask_index).to(x.dtype)
        return p_x0, copy_flag * x + (1 - copy_flag) * _x


    @torch.no_grad()
    def sample_subs_guidance(self, n_samples, stride_length, num_strides, dt=0.001):
        ones = torch.ones(n_samples, dtype=self.dtype,device=self.device)
        num_steps = int(1 / dt)
        sampling_steps = 0
        intermediate_tokens = []
        target = None

        for _ in range(num_strides + 1):
            p_x0_cache = None
            x = self._sample_prior(n_samples,self.config.model.length).to(self.device)
            
            if target is not None:
                x[:, : -stride_length] = target
            
            for i in range(num_steps + 1):
                p_x0_cache, x_next = self.ddpm_caching_update(x=x, t=(1 - i * dt) * ones, dt=dt, p_x0=p_x0_cache)
                if (not torch.allclose(x_next, x) or self.time_conditioning):
                    p_x0_cache = None
                    sampling_steps += 1
                x = x_next
            x = self.forward(x, 0 * ones).argmax(dim=-1)
            intermediate_tokens.append(x[:, :stride_length].cpu().numpy())
            target = x[:, stride_length:]
    
        intermediate_tokens.append(target.cpu().numpy())
        intermediate_text_samples = []
        sequence_lengths = ((np.concatenate(intermediate_tokens, axis=1)[:, 1:]
                                 == self.tokenizer.eos_token_id).cumsum(-1) == 0).sum(-1)
    
        for i in range(2, len(intermediate_tokens) + 1):
            intermediate_text_samples.append(self.tokenizer.decode(np.concatenate(intermediate_tokens[:i], axis=1)))
        
        return (sampling_steps, intermediate_text_samples,
            sequence_lengths)

    def restore_model_and_semi_ar_sample(self, stride_length, num_strides, dt=0.001):
        """Generate samples from the model."""
        # Lightning auto-casting is not working in this method for some reason
        self.backbone.eval()
        self.noise.eval()
    
        (sampling_steps, samples, sequence_lengths) = self.sample_subs_guidance(n_samples=self.config.Loader.BATCH_SIZE,stride_length=stride_length,num_strides=num_strides,dt=dt)

        self.backbone.train()
        self.noise.train()
        return sampling_steps, samples, sequence_lengths