322 lines
11 KiB
Python
322 lines
11 KiB
Python
# Copyright (c) Facebook, Inc. and its affiliates.
|
|
#
|
|
# This source code is licensed under the MIT license found in the
|
|
# LICENSE file in the root directory of this source tree.
|
|
from typing import Optional, TypeVar, Union, overload
|
|
|
|
import torch
|
|
import torch.nn.functional as F
|
|
from torch import Tensor, device, dtype, nn
|
|
|
|
import bitsandbytes as bnb
|
|
from bitsandbytes.optim import GlobalOptimManager
|
|
|
|
T = TypeVar("T", bound="torch.nn.Module")
|
|
|
|
|
|
class StableEmbedding(torch.nn.Embedding):
|
|
def __init__(
|
|
self,
|
|
num_embeddings: int,
|
|
embedding_dim: int,
|
|
padding_idx: Optional[int] = None,
|
|
max_norm: Optional[float] = None,
|
|
norm_type: float = 2.0,
|
|
scale_grad_by_freq: bool = False,
|
|
sparse: bool = False,
|
|
_weight: Optional[Tensor] = None,
|
|
device=None,
|
|
dtype=None,
|
|
) -> None:
|
|
super().__init__(
|
|
num_embeddings,
|
|
embedding_dim,
|
|
padding_idx,
|
|
max_norm,
|
|
norm_type,
|
|
scale_grad_by_freq,
|
|
sparse,
|
|
_weight,
|
|
device,
|
|
dtype,
|
|
)
|
|
self.norm = torch.nn.LayerNorm(embedding_dim, device=device)
|
|
GlobalOptimManager.get_instance().register_module_override(
|
|
self, "weight", {"optim_bits": 32}
|
|
)
|
|
|
|
def reset_parameters(self) -> None:
|
|
torch.nn.init.xavier_uniform_(self.weight)
|
|
self._fill_padding_idx_with_zero()
|
|
|
|
""" !!! This is a redefinition of _fill_padding_idx_with_zero in torch.nn.Embedding
|
|
to make the Layer compatible with Pytorch < 1.9.
|
|
This means that if this changes in future PyTorch releases this need to change too
|
|
which is cumbersome. However, with this we can ensure compatibility with previous
|
|
PyTorch releases.
|
|
"""
|
|
|
|
def _fill_padding_idx_with_zero(self) -> None:
|
|
if self.padding_idx is not None:
|
|
with torch.no_grad():
|
|
self.weight[self.padding_idx].fill_(0)
|
|
|
|
def forward(self, input: Tensor) -> Tensor:
|
|
emb = F.embedding(
|
|
input,
|
|
self.weight,
|
|
self.padding_idx,
|
|
self.max_norm,
|
|
self.norm_type,
|
|
self.scale_grad_by_freq,
|
|
self.sparse,
|
|
)
|
|
|
|
# always apply layer norm in full precision
|
|
emb = emb.to(torch.get_default_dtype())
|
|
|
|
return self.norm(emb).to(self.weight.dtype)
|
|
|
|
|
|
class Embedding(torch.nn.Embedding):
|
|
def __init__(
|
|
self,
|
|
num_embeddings: int,
|
|
embedding_dim: int,
|
|
padding_idx: Optional[int] = None,
|
|
max_norm: Optional[float] = None,
|
|
norm_type: float = 2.0,
|
|
scale_grad_by_freq: bool = False,
|
|
sparse: bool = False,
|
|
_weight: Optional[Tensor] = None,
|
|
) -> None:
|
|
super().__init__(
|
|
num_embeddings,
|
|
embedding_dim,
|
|
padding_idx,
|
|
max_norm,
|
|
norm_type,
|
|
scale_grad_by_freq,
|
|
sparse,
|
|
_weight,
|
|
)
|
|
GlobalOptimManager.get_instance().register_module_override(
|
|
self, "weight", {"optim_bits": 32}
|
|
)
|
|
|
|
def reset_parameters(self) -> None:
|
|
torch.nn.init.xavier_uniform_(self.weight)
|
|
self._fill_padding_idx_with_zero()
|
|
|
|
""" !!! This is a redefinition of _fill_padding_idx_with_zero in torch.nn.Embedding
|
|
to make the Layer compatible with Pytorch < 1.9.
|
|
This means that if this changes in future PyTorch releases this need to change too
|
|
which is cumbersome. However, with this we can ensure compatibility with previous
|
|
PyTorch releases.
|
|
"""
|
|
|
|
def _fill_padding_idx_with_zero(self) -> None:
|
|
if self.padding_idx is not None:
|
|
with torch.no_grad():
|
|
self.weight[self.padding_idx].fill_(0)
|
|
|
|
def forward(self, input: Tensor) -> Tensor:
|
|
emb = F.embedding(
|
|
input,
|
|
self.weight,
|
|
self.padding_idx,
|
|
self.max_norm,
|
|
self.norm_type,
|
|
self.scale_grad_by_freq,
|
|
self.sparse,
|
|
)
|
|
|
|
return emb
|
|
|
|
class FP4Params(torch.nn.Parameter):
|
|
def __new__(cls, data=None, requires_grad=True, quant_state=None, blocksize=64, compress_statistics=True):
|
|
cls.quant_state = None
|
|
cls.blocksize = blocksize
|
|
cls.compress_statistics = compress_statistics
|
|
if data is None:
|
|
data = torch.empty(0)
|
|
return torch.Tensor._make_subclass(cls, data, requires_grad)
|
|
|
|
def cuda(self, device):
|
|
w = self.data.contiguous().half().cuda(device)
|
|
w_fp4, quant_state = bnb.functional.quantize_fp4(w, blocksize=self.blocksize, compress_statistics=self.compress_statistics)
|
|
self.data = w_fp4
|
|
self.quant_state = quant_state
|
|
|
|
return self
|
|
|
|
@overload
|
|
def to(self: T, device: Optional[Union[int, device]] = ..., dtype: Optional[Union[dtype, str]] = ..., non_blocking: bool = ...,) -> T:
|
|
...
|
|
|
|
@overload
|
|
def to(self: T, dtype: Union[dtype, str], non_blocking: bool = ...) -> T:
|
|
...
|
|
|
|
@overload
|
|
def to(self: T, tensor: Tensor, non_blocking: bool = ...) -> T:
|
|
...
|
|
|
|
def to(self, *args, **kwargs):
|
|
device, dtype, non_blocking, convert_to_format = torch._C._nn._parse_to(*args, **kwargs)
|
|
|
|
if (device is not None and device.type == "cuda" and self.data.device.type == "cpu"):
|
|
return self.cuda(device)
|
|
else:
|
|
new_param = FP4Params(super().to(device=device, dtype=dtype, non_blocking=non_blocking),
|
|
requires_grad=self.requires_grad, quant_state=self.quant_state)
|
|
|
|
return new_param
|
|
|
|
|
|
class LinearFP4(nn.Linear):
|
|
def __init__(self, input_features, output_features, bias=True, compute_dtype=None, compress_statistics=True):
|
|
super().__init__(input_features, output_features, bias)
|
|
self.state = bnb.MatmulLtState()
|
|
self.weight = FP4Params(self.weight.data, requires_grad=False, compress_statistics=compress_statistics)
|
|
self.compute_dtype = compute_dtype
|
|
|
|
def init_8bit_state(self):
|
|
pass
|
|
|
|
def forward(self, x: torch.Tensor):
|
|
self.state.is_training = self.training
|
|
|
|
# weights are cast automatically as Int8Params, but the bias has to be cast manually
|
|
if self.bias is not None and self.bias.dtype != x.dtype:
|
|
self.bias.data = self.bias.data.to(x.dtype)
|
|
|
|
if getattr(self.weight, 'quant_state', None) is None:
|
|
print('FP4 quantization state not initialized. Please call .cuda() or .to(device) on the LinearFP4 layer first.')
|
|
inp_dtype = x.dtype
|
|
if self.compute_dtype is not None:
|
|
x = x.to(self.compute_dtype)
|
|
|
|
bias = None if self.bias is None else self.bias.half()
|
|
out = bnb.matmul_fp4(x, self.weight.t(), bias=bias, quant_state=self.weight.quant_state)
|
|
|
|
out = out.to(inp_dtype)
|
|
|
|
return out
|
|
|
|
|
|
class Int8Params(torch.nn.Parameter):
|
|
def __new__(
|
|
cls,
|
|
data=None,
|
|
requires_grad=True,
|
|
has_fp16_weights=False,
|
|
CB=None,
|
|
SCB=None,
|
|
):
|
|
cls.has_fp16_weights = has_fp16_weights
|
|
cls.CB = None
|
|
cls.SCB = None
|
|
if data is None:
|
|
data = torch.empty(0)
|
|
return torch.Tensor._make_subclass(cls, data, requires_grad)
|
|
|
|
def cuda(self, device):
|
|
if self.has_fp16_weights:
|
|
return super().cuda(device)
|
|
else:
|
|
# we store the 8-bit rows-major weight
|
|
# we convert this weight to the turning/ampere weight during the first inference pass
|
|
B = self.data.contiguous().half().cuda(device)
|
|
CB, CBt, SCB, SCBt, coo_tensorB = bnb.functional.double_quant(B)
|
|
del CBt
|
|
del SCBt
|
|
self.data = CB
|
|
setattr(self, "CB", CB)
|
|
setattr(self, "SCB", SCB)
|
|
|
|
return self
|
|
|
|
@overload
|
|
def to(
|
|
self: T,
|
|
device: Optional[Union[int, device]] = ...,
|
|
dtype: Optional[Union[dtype, str]] = ...,
|
|
non_blocking: bool = ...,
|
|
) -> T:
|
|
...
|
|
|
|
@overload
|
|
def to(self: T, dtype: Union[dtype, str], non_blocking: bool = ...) -> T:
|
|
...
|
|
|
|
@overload
|
|
def to(self: T, tensor: Tensor, non_blocking: bool = ...) -> T:
|
|
...
|
|
|
|
def to(self, *args, **kwargs):
|
|
device, dtype, non_blocking, convert_to_format = torch._C._nn._parse_to(
|
|
*args, **kwargs
|
|
)
|
|
|
|
if (
|
|
device is not None
|
|
and device.type == "cuda"
|
|
and self.data.device.type == "cpu"
|
|
):
|
|
return self.cuda(device)
|
|
else:
|
|
new_param = Int8Params(
|
|
super().to(
|
|
device=device, dtype=dtype, non_blocking=non_blocking
|
|
),
|
|
requires_grad=self.requires_grad,
|
|
has_fp16_weights=self.has_fp16_weights,
|
|
)
|
|
new_param.CB = self.CB
|
|
new_param.SCB = self.SCB
|
|
|
|
return new_param
|
|
|
|
|
|
|
|
class Linear8bitLt(nn.Linear):
|
|
def __init__(self, input_features, output_features, bias=True, has_fp16_weights=True,
|
|
memory_efficient_backward=False, threshold=0.0, index=None):
|
|
super().__init__(input_features, output_features, bias)
|
|
assert not memory_efficient_backward, "memory_efficient_backward is no longer required and the argument is deprecated in 0.37.0 and will be removed in 0.39.0"
|
|
self.state = bnb.MatmulLtState()
|
|
self.index = index
|
|
|
|
self.state.threshold = threshold
|
|
self.state.has_fp16_weights = has_fp16_weights
|
|
self.state.memory_efficient_backward = memory_efficient_backward
|
|
if threshold > 0.0 and not has_fp16_weights:
|
|
self.state.use_pool = True
|
|
|
|
self.weight = Int8Params(self.weight.data, has_fp16_weights=has_fp16_weights, requires_grad=has_fp16_weights)
|
|
|
|
def init_8bit_state(self):
|
|
self.state.CB = self.weight.CB
|
|
self.state.SCB = self.weight.SCB
|
|
self.weight.CB = None
|
|
self.weight.SCB = None
|
|
|
|
def forward(self, x: torch.Tensor):
|
|
self.state.is_training = self.training
|
|
if self.weight.CB is not None:
|
|
self.init_8bit_state()
|
|
|
|
# weights are cast automatically as Int8Params, but the bias has to be cast manually
|
|
if self.bias is not None and self.bias.dtype != x.dtype:
|
|
self.bias.data = self.bias.data.to(x.dtype)
|
|
|
|
out = bnb.matmul(x, self.weight, bias=self.bias, state=self.state)
|
|
if not self.state.has_fp16_weights:
|
|
if self.state.CB is not None and self.state.CxB is not None:
|
|
# we converted 8-bit row major to turing/ampere format in the first inference pass
|
|
# we no longer need the row-major weight
|
|
del self.state.CB
|
|
self.weight.data = self.state.CxB
|
|
return out
|