Shen Feiyu
add 1s
faadabf
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