scripts/diffusion.py

import itertools
import math
import torch
import torch.nn as nn
import numpy as np
import pytorch_lightning as L
import torchmetrics
from dataclasses import dataclass
from esm_utils import load_esm2_model
from transformers import AutoModel, AutoTokenizer
import dit, ema
import sys
import config
import wandb
import noise_schedule  # Assuming this is part of the MDLM repository

wandb_key = "2b76a2fa2c1cdfddc5f443602c17b011fefb0a8f"
wandb.login(key=wandb_key)
wandb.init(project=config.Wandb.PROJECT, group=config.Wandb.GROUP)

LOG2 = math.log(2)

# Goal is to build an MDLM head on the BERT-style ESM model
# Wrap the ESM model to obtain embeddings and ignore sigma to work with MDLM codebase
class WrapESM(nn.Module):
    def __init__(self, esm_model_path):
        super(WrapESM, self).__init__()
        self.esm_tokenizer, self.esm_model, _ = load_esm2_model(esm_model_path)

        ### Only fine-tune the last 3 layers of ESM
        # Count number of encoder layers
        model_layers = len(self.esm_model.esm.encoder.layer)

        # Disable parameter updates for all layers
        for param in self.esm_model.parameters():
            param.requires_grad = False

        # Now that all parameters are disabled, only enable updates for the last 3 layers
        for i, layer in enumerate(self.esm_model.esm.encoder.layer):
            if i >= model_layers-config.ESM_LAYERS:
                for module in layer.attention.self.key.modules():
                    for param in module.parameters():
                        param.requires_grad = True
                for module in layer.attention.self.query.modules():
                    for param in module.parameters():
                        param.requires_grad = True
                for module in layer.attention.self.value.modules():
                    for param in module.parameters():
                        param.requires_grad = True
        
    def forward(self, latents, sigma):
        return latents

@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.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.T = self.config.T
        self.lr = self.config.Optim.LR
        self.backbone = WrapESM(self.config.MODEL_NAME)
        self.noise = noise_schedule.get_noise(self.config, dtype=self.dtype)
        self.time_conditioning = self.config.TIME_CONDITIONING
        self.subs_masking = self.config.SUBS_MASKING
        self.mask_index = self.tokenizer.mask_token_id
        self.antithetic_sampling = self.config.Training.ANTITHETIC_SAMPLING
        self.sampling_eps = self.config.Training.SAMPLING_EPS
        self.neg_infinity = -1000000.0


    ############ FORWARD DIFFUSION #########
    def subs_parameterization(self, logits, noised_latents):
        print(logits.size()) # [bsz x bsz x seq_len]
        logits = logits.float()
        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)

        unmasked_indices = (noised_latents != self.mask_index)
        logits[unmasked_indices] = self.neg_infinity
        logits[~unmasked_indices] = 0
        
        return logits

        # # -inf probability of selecting a masked token
        # unmasked_indices = (noised_latents != self.mask_index)
        # logits[unmasked_indices] = self.neg_infinity

        # # Carry over unmasked tokens
        # bsz, seq_len, input_dim = logits.shape
        # for batch_idx in range(bsz):
        #     for residue in range(seq_len):
        #         logits[batch_idx, residue, noised_latents[batch_idx, residue]] = 0
        
        # return logits

    def forward(self, latents, sigma):
        latents = latents.long()
        logits = self.backbone(latents, sigma)
        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 = torch.mean(latents, dim=2) # [bsz x seq_len x 1280] --> [bsz x seq_len] as per markdown
        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 forward_diffusion(self, x0):
        """Forward diffusion process, adds noise to the latents."""

        t = self.sample_timestep(x0.shape[0], x0.device)
        sigma, dsigma = self.noise(t)
        unet_conditioning = sigma[:, None]
        move_chance = 1 - torch.exp(-sigma[:, None, None])

        xt = self.q_xt(x0, move_chance)
        model_output = self.forward(xt, unet_conditioning)
        print(f'model out: {model_output}')
        print(f'model out dim: {model_output.size()}') # [bsz x bsz x seq_len]
    
        # SUBS parameterization, continuous time.
        idx = torch.mean(x0, dim=2).long()[:, :, None]
        print(f'idx: {idx}')
        print(f'idx dim: {idx.size()}') # [bsz x seq_len x 1]

        log_p_theta = torch.gather(input=model_output, dim=-1, index=idx).squeeze(-1)
        scale = (dsigma / torch.expm1(sigma))[:, None]
        return - log_p_theta * scale


    ######### LOSS CALCULATIONS #########
    def compute_loss(self, latents, attention_mask):
        """"Average of MLM losses to stabilize training"""
        loss = self.forward_diffusion(latents)

        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):
        latents, attention_mask = batch
        loss = self.compute_loss(latents, attention_mask)
        wandb.log({"train_loss": loss.loss.item()})
        return loss.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)
        wandb.log({"val_loss": loss.loss.item()})
        return loss.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