from einops import rearrange, repeat from torch import nn from typing import Union import torch import torch.nn.functional as F import typing as tp import numpy as np import warnings def default(val: tp.Any, d: tp.Any) -> tp.Any: return val if val is not None else d def flatten(x, x_len): x_f = x.view(-1, *x.shape[2:]) return x_f def ema_inplace(moving_avg, new, decay): if isinstance(decay, torch.Tensor): moving_avg.data.mul_(decay).add_(new * (1 - decay)) else: moving_avg.data.mul_(decay).add_(new, alpha=(1 - decay)) def laplace_smoothing(x, n_categories: int, epsilon: float = 1e-5): return (x + epsilon) / (x.sum() + n_categories * epsilon) def uniform_init(*shape: int): t = torch.empty(shape) nn.init.kaiming_uniform_(t) return t def sample_vectors(samples, num: int): num_samples, device = samples.shape[0], samples.device if num_samples >= num: indices = torch.randperm(num_samples, device=device)[:num] else: indices = torch.randint(0, num_samples, (num,), device=device) return samples[indices] class EuclideanCodebook(nn.Module): def __init__( self, dim: int, codebook_size: int, decay: float = 0.99, epsilon: float = 1e-5, threshold_ema_dead_code: float = 1.0, n_cache_iters: int = 1, ): super().__init__() self.decay = decay init_fn: tp.Union[tp.Callable[..., torch.Tensor], tp.Any] = uniform_init embed = init_fn(codebook_size, dim) self.codebook_size = codebook_size self.epsilon = epsilon self.threshold_ema_dead_code = threshold_ema_dead_code self.update_iter = 0 self.n_cache_iters = n_cache_iters self.cache_vectors = [] self.cache_indices = [] if isinstance(self.decay, (tuple, list)): self.embed_avg_cache = [] self.register_buffer("diff_avg_long", torch.zeros(codebook_size) + 1e-5) self.register_buffer("diff_avg_short", torch.zeros(codebook_size) + 1e-5) self.register_buffer("inited", torch.Tensor([True])) self.register_buffer("cluster_size", torch.zeros(codebook_size)) self.register_buffer("embed", embed) self.register_buffer("embed_avg", embed.clone()) @torch.jit.ignore def init_embed_(self, data): if self.inited: return def replace_(self, samples, mask, dists=None): reset_cluster_size = min( self.threshold_ema_dead_code + 1, self.threshold_ema_dead_code * 1.1 ) modified_codebook = torch.where( mask[..., None], sample_vectors(samples, self.codebook_size), self.embed ) modified_codebook_avg = torch.where( mask[..., None], modified_codebook * reset_cluster_size, self.embed_avg ) modified_cluster_size = torch.where( mask, torch.full_like(self.cluster_size, reset_cluster_size), self.cluster_size, ) self.embed.data.copy_(modified_codebook) self.embed_avg.data.copy_(modified_codebook_avg) self.cluster_size.data.copy_(modified_cluster_size) def expire_codes_(self, batch_samples, dists=None): self.update_iter += 1 if self.threshold_ema_dead_code == 0: return elif self.threshold_ema_dead_code < 1: threshold_ema_dead_code = ( sum(self.cluster_size) * self.threshold_ema_dead_code ) else: threshold_ema_dead_code = self.threshold_ema_dead_code expired_codes = self.cluster_size < threshold_ema_dead_code if not torch.any(expired_codes): return batch_samples = rearrange(batch_samples, "... d -> (...) d") self.replace_(batch_samples, mask=expired_codes, dists=dists) def preprocess(self, x): x = rearrange(x, "... d -> (...) d") return x def quantize(self, x): embed = self.embed.t() dist = -( x.pow(2).sum(1, keepdim=True) - 2 * x @ embed + embed.pow(2).sum(0, keepdim=True) ) embed_ind = dist.max(dim=-1).indices return embed_ind, dist def postprocess_emb(self, embed_ind, shape): return embed_ind.view(*shape[:-1]) def dequantize(self, embed_ind): quantize = F.embedding(embed_ind, self.embed) return quantize def encode(self, x): shape = x.shape # pre-process x = self.preprocess(x) # quantize embed_ind, dist = self.quantize(x) # post-process embed_ind = self.postprocess_emb(embed_ind, shape) return embed_ind, dist def decode(self, embed_ind): quantize = self.dequantize(embed_ind) return quantize def forward(self, x, x_len, enable_vq=True, update_codebook=True, masking=False): x_org, shape, dtype = x, x.shape, x.dtype x = self.preprocess(x) embed_ind, dist = self.quantize(x) embed_ind = self.postprocess_emb(embed_ind, shape) dist = dist.view(shape[0], shape[1], dist.shape[-1]) quantize = self.dequantize(embed_ind) if self.training and update_codebook: if enable_vq: quantize = x_org + (quantize - x_org).detach() else: quantize = x_org # Get flatten embedding indices and distances if masking: x_f = torch.cat( [e[: int(e_len)] for e, e_len in zip(x_org, x_len)], dim=0 ) embed_ind_f = torch.cat( [e[: int(e_len)] for e, e_len in zip(embed_ind, x_len)], dim=0 ) dist_f = torch.cat( [e[: int(e_len)] for e, e_len in zip(dist, x_len)], dim=0 ) q_f = torch.cat( [e[: int(e_len)] for e, e_len in zip(quantize.detach(), x_len)], dim=0, ) commit_loss = F.mse_loss(q_f, x_f) else: x_f = x_org.view(-1, x_org.shape[-1]).contiguous() embed_ind_f = embed_ind.view(-1).contiguous() dist_f = dist.view(-1).contiguous() commit_loss = F.mse_loss(quantize.detach(), x_org) self.init_embed_(x_f) # We do the expiry of code at that point as buffers are in sync # and all the workers will take the same decision. self.expire_codes_(x_f, dist_f) # Calculate codebook statistics embed_onehot = F.one_hot(embed_ind_f, self.codebook_size).type(dtype) embed_onehot_sum = embed_onehot.sum(0) embed_sum = x_f.t() @ embed_onehot # EMA updating ema_inplace(self.cluster_size, embed_onehot_sum, self.decay) ema_inplace(self.embed_avg, embed_sum.t(), self.decay) cluster_size = ( laplace_smoothing(self.cluster_size, self.codebook_size, self.epsilon) * self.cluster_size.sum() ) embed_normalized = self.embed_avg / cluster_size.unsqueeze(1) self.embed.data.copy_(embed_normalized) else: commit_loss = torch.tensor( 0.0, device=quantize.device, requires_grad=self.training ) return quantize, commit_loss, embed_ind class MultiHeadEuclideanCodebook(nn.Module): def __init__( self, dim: Union[int, list], codebook_size: list, n_groups: int = 1, dropout_rate_per_group: float = 0, ordered: bool = False, ordered_axis: str = "sequence", method: str = "product", **kwargs, ): super().__init__() self.codebook_sizes = codebook_size self.codebook_dims = dim self.n_groups = n_groups self.n_heads_per_group = len(codebook_size) // n_groups self.dropout_rate_per_group = dropout_rate_per_group self.ordered = ordered self.ordered_axis = ordered_axis self.method = method assert len(codebook_size) % n_groups == 0 self.codebooks = nn.ModuleList() dim = self.codebook_dims for i, size in enumerate(self.codebook_sizes): if isinstance(self.codebook_dims, list): dim = ( self.codebook_dims[i] if method == "product" else sum(self.codebook_dims) ) self.codebooks.append(EuclideanCodebook(dim, size, **kwargs)) def decode(self, embed_ind): if self.n_groups == 1 or len(embed_ind.shape) == 2: embed_ind = embed_ind.unsqueeze(-1) actual_n_groups = embed_ind.shape[-1] if actual_n_groups < self.n_groups: print( f"The actual number of heads ({actual_n_groups}) is smaller than the pre-designed ({self.n_groups})!" ) embed_ind = F.pad( embed_ind, (0, self.n_groups - actual_n_groups), "replicate" ) # assert embed_ind.shape[-1] == self.n_groups index_heads, codebook_heads, scale_heads = zip( *[ ( embed_ind[..., i // self.n_heads_per_group], self.codebooks[i : i + self.n_heads_per_group], self.codebook_sizes[i : i + self.n_heads_per_group], ) for i in range(0, len(self.codebook_sizes), self.n_heads_per_group) ] ) quantize_heads, quantize_groups = [], [] for i in range(self.n_groups): embed_ind, codebooks, scales = ( index_heads[i], codebook_heads[i], scale_heads[i], ) inv_scales = list(torch.tensor([1] + scales[:-1]).cumprod(dim=0))[::-1] inv_quantizes = [] for codebook, scale in zip(codebooks[::-1], inv_scales): index, embed_ind = embed_ind // scale, embed_ind % scale quantize = codebook.dequantize(index) inv_quantizes.append(quantize) quantizes = inv_quantizes[::-1] group_embeddings = torch.cat(quantizes, dim=-1) quantize_groups.append(group_embeddings) quantize_heads += quantizes if self.method == "product": if actual_n_groups < self.n_groups: for i in range(actual_n_groups, self.n_groups): quantize_groups[i].zero_() quantize = torch.cat(quantize_groups, dim=-1) elif self.method == "residual": quantize = sum(quantize_heads) return quantize def forward(self, x, x_len, enable_vq=True, update_codebook=True): # Pre-process x = self._preprocess(x) # Quantize quants, losses, indices = self._quantize( x, x_len, enable_vq=enable_vq, update_codebook=update_codebook ) # Integrate quant, loss, index = self._integrate( quants, losses, indices, update_codebook=update_codebook ) return quant, loss, index def _preprocess(self, x): if self.method == "product" and isinstance(self.codebook_dims, (list, tuple)): x = torch.split(x, self.codebook_dims, dim=-1) return x def _quantize(self, x, x_len, enable_vq, update_codebook): if self.method == "product": quants, losses, indices = zip( *[ codebook( chunk, x_len, enable_vq=enable_vq, update_codebook=update_codebook, ) for chunk, codebook in zip(x, self.codebooks) ] ) elif self.method == "residual": quants, losses, indices = [], [], [] residual = x for codebook in self.codebooks: quant, loss, index = codebook( residual, x_len, enable_vq=enable_vq, update_codebook=update_codebook, ) residual = residual - quant quants.append(quant) losses.append(loss) indices.append(index) return quants, losses, indices def _integrate(self, quants, losses, indices, update_codebook=True): (B, T, D), M = quants[0].shape, len(quants) device = quants[0].device # Average loss loss = sum(losses) / len(losses) # Get indices if self.n_groups == 1: scale = ( torch.tensor([1] + self.codebook_sizes[:-1]).cumprod(dim=0).to(device) ) index = (torch.stack(indices, dim=-1) * scale).sum(dim=-1) else: index_heads, scale_heads = zip( *[ ( indices[i : i + self.n_heads_per_group], torch.tensor( [1] + self.codebook_sizes[i : i + self.n_heads_per_group - 1] ) .cumprod(dim=0) .to(device), ) for i in range(0, len(quants), self.n_heads_per_group) ] ) index = torch.stack( [ (torch.stack(x, dim=-1) * s).sum(dim=-1) for x, s in zip(index_heads, scale_heads) ], dim=-1, ) # Add dropout quant_groups = self._dropout(quants, enabled=update_codebook) # Combine quantized features if self.method == "product": quant = torch.cat(quant_groups, dim=-1) elif self.method == "residual": quant = torch.cat(quant_groups, dim=-1).view(B, T, M, D).sum(dim=2) return quant, loss, index def _dropout(self, quants, enabled=True): if enabled and self.training and self.ordered: if self.dropout_rate_per_group == 0: threshold = [ (i // self.n_heads_per_group * 1.0 / self.n_groups) for i in range(0, len(quants), self.n_heads_per_group) ] elif self.dropout_rate_per_group == "exp": x = [np.exp(4 * i / self.n_groups) for i in range(self.n_groups)] x = np.asarray(x) / sum(x) threshold = np.cumsum(np.asarray([0] + x)) else: x = np.asarray(self.dropout_rate_per_group) / sum( self.dropout_rate_per_group ) threshold = np.cumsum(np.asarray([0] + x)) if self.ordered_axis == "sequence": rate = torch.rand((quants[0].shape[0], 1, 1), device=quants[0].device) elif self.ordered_axis == "frame": rate = torch.rand( (quants[0].shape[0], quants[0].shape[1], 1), device=quants[0].device ) quant_groups = [] for i in range(0, len(quants), self.n_heads_per_group): quant_group = torch.cat(quants[i : i + self.n_heads_per_group], dim=-1) is_kept = threshold[i // self.n_heads_per_group] <= rate quant_group = torch.where( is_kept, quant_group, torch.zeros_like(quant_group) ) quant_groups.append(quant_group) elif self.ordered: quant_groups = [] for i in range(0, len(quants), self.n_heads_per_group): quant_group = torch.cat(quants[i : i + self.n_heads_per_group], dim=-1) quant_groups.append(quant_group) else: quant_groups = quants return quant_groups class VectorQuantization(nn.Module): def __init__( self, dim: int, codebook_size: Union[int, list], codebook_dim: Union[int, list] = None, decay: float = 0.99, epsilon: float = 1e-5, threshold_ema_dead_code: float = 1.0, commitment_weight: float = 1.0, requires_projection: bool = False, norm: str = "none", **kwargs, ): super().__init__() _codebook_dim: Union[int, list] = default(codebook_dim, dim) requires_projection = _codebook_dim != dim or requires_projection proj_dim = ( sum(_codebook_dim) if isinstance(_codebook_dim, list) else _codebook_dim ) if requires_projection: self.project_in = nn.Linear(dim, proj_dim) self.project_out = nn.Linear(proj_dim, dim) if norm == "weight_norm": self.project_in = torch.nn.utils.weight_norm(self.project_in) self.project_out = torch.nn.utils.weight_norm(self.project_out) else: self.norm = None self.project_in = nn.Identity() self.project_out = nn.Identity() self.epsilon = epsilon self.commitment_weight = commitment_weight self.codebook_size = codebook_size codebook_class = ( EuclideanCodebook if isinstance(codebook_size, int) else MultiHeadEuclideanCodebook ) self._codebook = codebook_class( dim=_codebook_dim, codebook_size=codebook_size, decay=decay, epsilon=epsilon, threshold_ema_dead_code=threshold_ema_dead_code, **kwargs, ) self.codebook_size = codebook_size @property def codebook(self): return self._codebook.embed def encode(self, x, x_len=None): x = rearrange(x, "b d n -> b n d") x = self.project_in(x) embed_in = self._codebook.encode(x) return embed_in def decode(self, embed_ind, embed_len=None): quantize = self._codebook.decode(embed_ind) quantize = self.project_out(quantize) quantize = rearrange(quantize, "b n d -> b d n") return quantize def decode_latent(self, latent, latent_len=None): if latent_len is None: latent_len = ( torch.Tensor([latent.shape[1]] * latent.shape[0]) .to(latent.device) .int() ) quantize, _, _ = self._codebook(latent, latent_len) quantize = self.project_out(quantize) return quantize @torch.cuda.amp.autocast(dtype=torch.float32) def forward( self, x, x_len, enable_vq=True, update_codebook=True, return_pre_quant=False, return_dict=False, ): device = x.device x = self.project_in(x) quantize, commit_loss, embed_ind = self._codebook( x, x_len, enable_vq=enable_vq, update_codebook=update_codebook ) if self.training and update_codebook: loss = torch.tensor(0.0, device=device, requires_grad=True) if self.commitment_weight > 0: loss = loss + commit_loss * self.commitment_weight else: loss = torch.tensor(0.0, device=device, requires_grad=False) embed = quantize quantize = self.project_out(quantize) if return_dict: return { "quantize": quantize, "loss": loss, "embed": embed, "embed_ind": embed_ind, } elif return_pre_quant: pre_quantize = self.project_out(x) return (pre_quantize, quantize), loss, embed_ind else: return quantize, loss, embed_ind