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 AutoModelForMaskedLM, 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 using pre-existing ESM LM head
# Wrap the ESM model to obtain logits and ignore sigma to work with MDLM codebase
class WrapESM(nn.Module):
    def __init__(self, esm_model_path=config.MODEL_NAME):
        super(WrapESM, self).__init__()
        self.model = AutoModelForMaskedLM.from_pretrained(esm_model_path)
        self.tokenizer = AutoTokenizer.from_pretrained(esm_model_path)
    
    # def __getattr__(self, name):
    #     return getattr(self.model, name)

    def __call__(self, *args, **kwargs):
        return self.model(*args, **kwargs)
    
    def freeze_model(self):
       # Disable parameter updates for all layers
        for param in self.model.parameters():
            param.requires_grad = False

    def unfreeze_n_layers(self):
        # Count number of encoder layers
        model_layers = len(self.model.esm.encoder.layer)

        # Enable parameter updates for the last 3 encoder layers
        for i, layer in enumerate(self.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, sigma, **inputs):
        return self.model(**inputs)

    def save_model(self, save_dir):
        self.model.save_pretrained(save_dir)
        self.tokenizer.save_pretrained(save_dir)

    def load_model(self, load_dir):
        self.model = AutoModel.from_pretrained(load_dir)
        self.tokenizer = AutoTokenizer.from_pretrained(load_dir)

@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, tokenizer):
        super().__init__()
        self.config = config
        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 compute_loss(self, latents, attention_mask, val):
        """"Average of MLM losses to stabilize training"""
        self.noise.eval() if val else self.noise.train()
        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)

    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])

        xt = self.q_xt(x0, move_chance)
        model_output = self.forward(xt, unet_conditioning)

        # SUBS parameterization, continuous time.
        log_p_theta = torch.gather(input=model_output, dim=-1, index=x0[:, :, None]).squeeze(-1)
        scale = (dsigma / torch.expm1(sigma))[:, None]
        loss = - log_p_theta * scale
        return loss

    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
        return t
    
    def q_xt(self, x, 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).
        """
        move_indices = torch.rand(* x.shape, device=x.device) < move_chance
        xt = torch.where(move_indices, self.mask_index, x) # Use variable masking rate to mask tokens (introduce noise)
        return xt

    def forward(self, latents, sigma):
        esm_outputs = self.backbone(latents, sigma)
        optimized_logits = self.subs_parameterization(esm_outputs.logits, latents)
        return optimized_logits

    def subs_parameterization(self, logits, xt):
        logits[:, :, self.mask_index] += self.neg_infinity
        logits = logits - torch.logsumexp(logits, dim=-1, keepdim=True)

        unmasked_indices = (xt != self.mask_index)
        logits[unmasked_indices] = self.neg_infinity
        logits[unmasked_indices, xt[unmasked_indices]] = 0
        return logits


    ######### 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):
        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