123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965 |
- from abc import abstractmethod
- from typing import Dict, List, Optional, Tuple
- import torch
- import torch.nn.functional as F
- from loguru import logger
- from torch.nn.parameter import Parameter, UninitializedParameter
- # yapf: disable
- from aphrodite.distributed import (divide,
- get_current_tp_rank_partition_offset,
- get_current_tp_rank_partition_size,
- get_tensor_model_parallel_rank,
- get_tensor_model_parallel_world_size,
- split_tensor_along_last_dim,
- tensor_model_parallel_all_gather,
- tensor_model_parallel_all_reduce)
- # yapf: enable
- from aphrodite.modeling.utils import set_weight_attrs
- from aphrodite.quantization.base_config import (QuantizationConfig,
- QuantizeMethodBase)
- def adjust_marlin_shard(param, shard_size, shard_offset):
- marlin_tile_size = getattr(param, "marlin_tile_size", None)
- if marlin_tile_size is None:
- return shard_size, shard_offset
- return shard_size * marlin_tile_size, shard_offset * marlin_tile_size
- def adjust_bitsandbytes_shard(param: Parameter,
- qkv_offsets: Dict[str, Tuple[int, int]],
- loaded_shard_id: str) -> Tuple[int, int]:
- """Adjust the quantization offsets and sizes for BitsAndBytes sharding."""
- total, _ = qkv_offsets["total"]
- orig_offset, orig_size = qkv_offsets[loaded_shard_id]
- quantized_total = param.data.shape[0]
- quantized_offset = orig_offset * quantized_total // total
- quantized_size = orig_size * quantized_total // total
- return quantized_size, quantized_offset
- def adjust_scalar_to_fused_array(param, loaded_weight, shard_id):
- """For fused modules (QKV and MLP) we have an array of length
- N that holds 1 scale for each "logical" matrix. So the param
- is an array of length N. The loaded_weight corresponds to
- one of the shards on disk. Here, we slice the param based on
- the shard_id for loading.
- """
- qkv_idxs = {"q": 0, "k": 1, "v": 2}
- if isinstance(shard_id, str):
- shard_id = qkv_idxs[shard_id]
- elif not isinstance(shard_id, int):
- raise ValueError(f"Unknown Shard Id {shard_id}")
- # AutoFP8 scales do not have a shape
- # compressed-tensors scales do have a shape
- if len(loaded_weight.shape) != 0:
- assert loaded_weight.shape[0] == 1
- loaded_weight = loaded_weight[0]
- return param[shard_id], loaded_weight
- class LinearMethodBase(QuantizeMethodBase):
- """Base class for different (maybe quantized) linear methods."""
- @abstractmethod
- def create_weights(self, layer: torch.nn.Module,
- input_size_per_partition: int,
- output_partition_sizes: List[int], input_size: int,
- output_size: int, params_dtype: torch.dtype,
- **extra_weight_attrs):
- """Create weights for a linear layer.
- The weights will be set as attributes of the layer.
- Args:
- layer: The layer that is using the LinearMethodBase factory.
- input_size_per_partition: Size of the weight input dim on rank X.
- output_partition_sizes: Sizes of the output dim of each logical
- weight on rank X. E.g., output_partition_sizes for QKVLinear
- is a list contains the width of Wq, Wk, Wv on rank X.
- input_size: Size of the input dim of the weight across all ranks.
- output_size: Size of the output dim of the weight across all ranks.
- params_dtype: Datatype of the parameters.
- """
- raise NotImplementedError
- @abstractmethod
- def apply(self,
- layer: torch.nn.Module,
- x: torch.Tensor,
- bias: Optional[torch.Tensor] = None) -> torch.Tensor:
- """Apply the weights in layer to the input tensor.
- Expects create_weights to have been called before on the layer."""
- raise NotImplementedError
- class UnquantizedLinearMethod(LinearMethodBase):
- """Linear method without quantization."""
- def create_weights(self, layer: torch.nn.Module,
- input_size_per_partition: int,
- output_partition_sizes: List[int], input_size: int,
- output_size: int, params_dtype: torch.dtype,
- **extra_weight_attrs):
- weight = Parameter(torch.empty(sum(output_partition_sizes),
- input_size_per_partition,
- dtype=params_dtype),
- requires_grad=False)
- set_weight_attrs(weight, {"input_dim": 1, "output_dim": 0})
- layer.register_parameter("weight", weight)
- set_weight_attrs(weight, extra_weight_attrs)
- def apply(self,
- layer: torch.nn.Module,
- x: torch.Tensor,
- bias: Optional[torch.Tensor] = None) -> torch.Tensor:
- return F.linear(x, layer.weight, bias)
- class LinearBase(torch.nn.Module):
- """Base linear layer.
- Args:
- input_size: input dimension of the linear layer.
- output_size: output dimension of the linear layer.
- bias: If true, add bias.
- skip_bias_add: If true, skip adding bias but instead return it.
- params_dtype: Data type for the parameters.
- quant_config: Quantization configure.
- """
- def __init__(
- self,
- input_size: int,
- output_size: int,
- skip_bias_add: bool = False,
- params_dtype: Optional[torch.dtype] = None,
- quant_config: Optional[QuantizationConfig] = None,
- prefix: str = "",
- ):
- super().__init__()
- # Keep input parameters
- self.input_size = input_size
- self.output_size = output_size
- self.skip_bias_add = skip_bias_add
- if params_dtype is None:
- params_dtype = torch.get_default_dtype()
- self.params_dtype = params_dtype
- if quant_config is None:
- self.quant_method: Optional[
- QuantizeMethodBase] = UnquantizedLinearMethod()
- else:
- self.quant_method = quant_config.get_quant_method(self,
- prefix=prefix)
- def forward(self, x: torch.Tensor) -> torch.Tensor:
- raise NotImplementedError
- class ReplicatedLinear(LinearBase):
- """Replicated linear layer.
- Args:
- input_size: input dimension of the linear layer.
- output_size: output dimension of the linear layer.
- bias: If true, add bias.
- skip_bias_add: If true, skip adding bias but instead return it.
- params_dtype: Data type for the parameters.
- quant_config: Quantization configure.
- prefix: The name of the layer in the state dict, including all parents
- (e.g. model.layers.0.qkv_proj)
- """
- def __init__(self,
- input_size: int,
- output_size: int,
- bias: bool = True,
- skip_bias_add: bool = False,
- params_dtype: Optional[torch.dtype] = None,
- quant_config: Optional[QuantizationConfig] = None,
- prefix: str = ""):
- super().__init__(input_size,
- output_size,
- skip_bias_add,
- params_dtype,
- quant_config,
- prefix=prefix)
- # All the linear layer supports quant method.
- assert self.quant_method is not None
- self.quant_method.create_weights(self,
- self.input_size, [self.output_size],
- self.input_size,
- self.output_size,
- self.params_dtype,
- prefix=prefix)
- if bias:
- self.bias = Parameter(
- torch.empty(self.output_size, dtype=self.params_dtype))
- set_weight_attrs(self.bias, {"output_dim": 0})
- else:
- self.register_parameter("bias", None)
- def weight_loader(self, param: Parameter, loaded_weight: torch.Tensor):
- # If the weight on disk does not have a shape, give it one
- # (such scales for AutoFp8).
- if len(loaded_weight.shape) == 0:
- loaded_weight = loaded_weight.reshape(1)
- assert param.size() == loaded_weight.size()
- param.data.copy_(loaded_weight)
- def forward(self, x: torch.Tensor) -> torch.Tensor:
- bias = self.bias if not self.skip_bias_add else None
- assert self.quant_method is not None
- output = self.quant_method.apply(self, x, bias)
- output_bias = self.bias if self.skip_bias_add else None
- return output, output_bias
- def extra_repr(self) -> str:
- s = f"in_features={self.input_size}"
- s += f", output_features={self.output_size}"
- s += f", bias={self.bias is not None}"
- return s
- class ColumnParallelLinear(LinearBase):
- """Linear layer with column parallelism.
- The linear layer is defined as Y = XA + b. A is parallelized along
- its second dimension as A = [A_1, ..., A_p].
- Args:
- input_size: first dimension of matrix A.
- output_size: second dimension of matrix A.
- bias: If true, add bias.
- gather_output: If true, call all-gather on output and make Y available
- to all GPUs, otherwise, every GPU will have its output
- which is Y_i = XA_i
- skip_bias_add: This was added to enable performance optimizations where
- bias can be fused with other element-wise operations. we
- skip adding bias but instead return it.
- params_dtype: Data type for the parameters.
- quant_config: Quantization configure.
- output_sizes: list of output sizes packed into one output, like for QKV
- the list would be size 3.
- prefix: The name of the layer in the state dict, including all parents
- (e.g. model.layers.0.qkv_proj)
- """
- def __init__(self,
- input_size: int,
- output_size: int,
- bias: bool = True,
- gather_output: bool = False,
- skip_bias_add: bool = False,
- params_dtype: Optional[torch.dtype] = None,
- quant_config: Optional[QuantizationConfig] = None,
- output_sizes: Optional[List[int]] = None,
- prefix: str = ""):
- super().__init__(input_size, output_size, skip_bias_add, params_dtype,
- quant_config, prefix)
- self.gather_output = gather_output
- # Divide the weight matrix along the last dimension.
- tp_rank = get_tensor_model_parallel_rank()
- tp_size = get_tensor_model_parallel_world_size()
- assert self.quant_method is not None
- if quant_config is None:
- self.output_size_per_partition = get_current_tp_rank_partition_size(
- output_size, tp_rank, tp_size)
- else:
- self.output_size_per_partition = divide(self.output_size, tp_size)
- self.output_partition_sizes = [self.output_size_per_partition]
- # If QKV or MergedColumn, use output size of each partition.
- if hasattr(self, "output_sizes"):
- if quant_config is None:
- self.output_partition_sizes = [
- get_current_tp_rank_partition_size(output_size, tp_rank,
- tp_size)
- for output_size in self.output_sizes
- ]
- else:
- self.output_partition_sizes = [
- divide(output_size, tp_size)
- for output_size in self.output_sizes
- ]
- if output_sizes is None:
- output_sizes = [output_size]
- self.quant_method.create_weights(
- layer=self,
- input_size_per_partition=self.input_size,
- output_partition_sizes=self.output_partition_sizes,
- input_size=self.input_size,
- output_size=self.output_size,
- params_dtype=self.params_dtype,
- weight_loader=self.weight_loader,
- prefix=prefix)
- if bias:
- self.bias = Parameter(
- torch.empty(self.output_size_per_partition,
- dtype=params_dtype))
- set_weight_attrs(self.bias, {
- "output_dim": 0,
- "weight_loader": self.weight_loader,
- })
- else:
- self.register_parameter("bias", None)
- def weight_loader(self, param: Parameter, loaded_weight: torch.Tensor):
- tp_rank = get_tensor_model_parallel_rank()
- output_dim = getattr(param, "output_dim", None)
- # Special case for GGUF
- is_gguf_weight = getattr(param, "is_gguf_weight", False)
- is_gguf_weight_type = getattr(param, "is_gguf_weight_type", False)
- if is_gguf_weight_type:
- param.weight_type = loaded_weight.item()
- # Materialize GGUF UninitializedParameter
- if is_gguf_weight and isinstance(param, UninitializedParameter):
- param.materialize(loaded_weight.shape, dtype=loaded_weight.dtype)
- param_data = param.data
- if output_dim is not None:
- shard_size = param_data.shape[output_dim]
- start_idx = tp_rank * shard_size
- loaded_weight = loaded_weight.narrow(output_dim, start_idx,
- shard_size)
- # Special case for loading scales off disk, which often do not
- # have a shape (such as in the case of AutoFP8).
- if len(loaded_weight.shape) == 0:
- loaded_weight = loaded_weight.reshape(1)
- assert param_data.shape == loaded_weight.shape
- param_data.copy_(loaded_weight)
- def forward(self, input_):
- bias = self.bias if not self.skip_bias_add else None
- # Matrix multiply.
- assert self.quant_method is not None
- output_parallel = self.quant_method.apply(self, input_, bias)
- if self.gather_output:
- # All-gather across the partitions.
- output = tensor_model_parallel_all_gather(output_parallel)
- else:
- output = output_parallel
- output_bias = self.bias if self.skip_bias_add else None
- return output, output_bias
- def extra_repr(self) -> str:
- s = f"in_features={self.input_size}"
- s += f", output_features={self.output_size_per_partition}"
- s += f", bias={self.bias is not None}"
- s += f", tp_size={get_tensor_model_parallel_world_size()}"
- s += f", gather_output={self.gather_output}"
- return s
- class MergedColumnParallelLinear(ColumnParallelLinear):
- """Packed linear layers with column parallelism.
- Similar to ColumnParallelLinear, but the weight matrix is concatenated
- along the output dimension. When the weight matrix is loaded, the
- different partitions are sharded separately.
- Args:
- input_size: input dimension of the linear layer.
- output_sizes: list of output dimensions of the linear layer.
- bias: If true, add bias.
- gather_output: If true, call all-gather on output and make the output
- available to all GPUs, otherwise, every GPU will have
- its own output.
- skip_bias_add: This was added to enable performance optimizations where
- bias can be fused with other element-wise operations. we
- skip adding bias but instead return it.
- params_dtype: Data type for the parameters.
- quant_config: Quantization configure.
- prefix: The name of the layer in the state dict, including all parents
- (e.g. model.layers.0.qkv_proj)
- """
- def __init__(self,
- input_size: int,
- output_sizes: List[int],
- bias: bool = True,
- gather_output: bool = False,
- skip_bias_add: bool = False,
- params_dtype: Optional[torch.dtype] = None,
- quant_config: Optional[QuantizationConfig] = None,
- prefix: str = ""):
- self.output_sizes = output_sizes
- self.quant_config = quant_config
- if quant_config is not None:
- tp_size = get_tensor_model_parallel_world_size()
- assert all(output_size % tp_size == 0
- for output_size in output_sizes)
- super().__init__(input_size=input_size,
- output_size=sum(output_sizes),
- bias=bias,
- gather_output=gather_output,
- skip_bias_add=skip_bias_add,
- params_dtype=params_dtype,
- quant_config=quant_config,
- prefix=prefix)
- def weight_loader(self,
- param: Parameter,
- loaded_weight: torch.Tensor,
- loaded_shard_id: Optional[int] = None):
- # Special case for GGUF
- # initialize GGUF param after we know the quantize type
- is_gguf_weight = getattr(param, "is_gguf_weight", False)
- is_gguf_weight_type = getattr(param, "is_gguf_weight_type", False)
- if is_gguf_weight_type:
- param.data[loaded_shard_id].copy_(loaded_weight)
- param.shard_weight_type[loaded_shard_id] = loaded_weight.item()
- return
- if is_gguf_weight and isinstance(param, UninitializedParameter):
- from gguf.constants import GGML_QUANT_SIZES
- ori_shape = param.tensor_shape
- weight_types = self.qweight_type.shard_weight_type.values()
- row_size = []
- for weight_type in weight_types:
- block_size, type_size = GGML_QUANT_SIZES[weight_type]
- row_size.append(ori_shape[1] // block_size * type_size)
- q_shape = (ori_shape[0], max(row_size))
- param.materialize(q_shape, dtype=loaded_weight.dtype)
- param_data = param.data
- output_dim = getattr(param, "output_dim", None)
- # Special case for AQLM codebooks.
- is_metadata = getattr(param, "is_metadata", False)
- # Special case for per-tensor scale to load scalar into fused array.
- needs_scalar_to_array = getattr(param, "needs_scalar_to_array", False)
- if loaded_shard_id is None:
- # Loaded weight is already fused on disk (qkv/mlp).
- if output_dim is None:
- if needs_scalar_to_array:
- param_data, loaded_weight = adjust_scalar_to_fused_array(
- param_data, loaded_weight, 0)
- assert param_data.shape == loaded_weight.shape
- param_data.copy_(loaded_weight)
- return
- current_shard_offset = 0
- shard_offsets: List[Tuple[int, int, int]] = []
- for i, output_size in enumerate(self.output_sizes):
- shard_offsets.append((i, current_shard_offset, output_size))
- current_shard_offset += output_size
- packed_dim = getattr(param, "packed_dim", None)
- for shard_id, shard_offset, shard_size in shard_offsets:
- # Special case for Quantization.
- # If quantized, we need to adjust the offset and size to account
- # for the packing.
- if packed_dim == output_dim:
- shard_size = shard_size // param.pack_factor
- shard_offset = shard_offset // param.pack_factor
- # Special case for Marlin.
- shard_size, shard_offset = adjust_marlin_shard(
- param, shard_size, shard_offset)
- loaded_weight_shard = loaded_weight.narrow(
- output_dim, shard_offset, shard_size)
- self.weight_loader(param, loaded_weight_shard, shard_id)
- return
- assert loaded_shard_id < len(self.output_sizes)
- tp_rank = get_tensor_model_parallel_rank()
- tp_size = get_tensor_model_parallel_world_size()
- if output_dim is not None:
- if self.quant_config is None:
- shard_offset = sum(
- get_current_tp_rank_partition_size(output_size, tp_rank,
- tp_size)
- for output_size in self.output_sizes[:loaded_shard_id])
- shard_size = get_current_tp_rank_partition_size(
- self.output_sizes[loaded_shard_id], tp_rank, tp_size)
- else:
- shard_offset = sum(
- self.output_sizes[:loaded_shard_id]) // tp_size
- shard_size = self.output_sizes[loaded_shard_id] // tp_size
- # Special case for quantization.
- # If quantized, we need to adjust the offset and size to account
- # for the packing.
- packed_dim = getattr(param, "packed_dim", None)
- if packed_dim == output_dim:
- shard_size = shard_size // param.pack_factor
- shard_offset = shard_offset // param.pack_factor
- # Special case for Marlin.
- shard_size, shard_offset = adjust_marlin_shard(
- param, shard_size, shard_offset)
- use_bitsandbytes = getattr(param, "use_bitsandbytes", False)
- if use_bitsandbytes:
- shard_size = loaded_weight.shape[output_dim]
- shard_offset = loaded_weight.shape[output_dim] * \
- loaded_shard_id
- if is_gguf_weight:
- tp_size = get_tensor_model_parallel_world_size()
- output_dim = getattr(param, "output_dim", None)
- shard_shape = list(loaded_weight.shape)
- shard_shape[output_dim] = shard_shape[output_dim] // tp_size
- param.shard_id.append(loaded_shard_id)
- param.shard_size[loaded_shard_id] = shard_shape
- input_dim = getattr(param, "input_dim", None)
- input_size = loaded_weight.shape[input_dim]
- param_data = param_data.narrow(input_dim, 0, input_size)
- param_data = param_data.narrow(output_dim, shard_offset,
- shard_size)
- if self.quant_config is None:
- start_idx = get_current_tp_rank_partition_offset(
- loaded_weight.shape[output_dim], tp_rank, tp_size)
- else:
- start_idx = tp_rank * shard_size
- loaded_weight = loaded_weight.narrow(output_dim, start_idx,
- shard_size)
- # Special case for AQLM codebooks.
- elif is_metadata:
- # metadata indicates fixed size concatenated along dim 0
- shard_size = loaded_weight.shape[0]
- shard_offset = loaded_shard_id * shard_size
- param_data = param_data.narrow(0, shard_offset, shard_size)
- # Special case for per-tensor scales in fused case.
- elif needs_scalar_to_array:
- param_data, loaded_weight = adjust_scalar_to_fused_array(
- param_data, loaded_weight, loaded_shard_id)
- else:
- ignore_warning = getattr(param, "ignore_warning", False)
- if not ignore_warning:
- logger.warning(
- "Loading a weight without `output_dim` attribute in "
- "MergedColumnParallelLinear, assume the weight is "
- "the same for all partitions.")
- assert param_data.shape == loaded_weight.shape
- param_data.copy_(loaded_weight)
- class QKVParallelLinear(ColumnParallelLinear):
- """Linear layers for the attention's QKV transformation.
- Linear layers for the linear transformation of the query, key, and value
- vectors in the attention layer. The weight matrix is concatenated along
- the output dimension. The layer is parallelized along the head dimension.
- When the number of key/value heads is smaller than the number of query
- heads (e.g., multi-query/grouped-query attention), the key/value head may
- be replicated while the query heads are partitioned.
- Args:
- hidden_size: input hidden state size of the transformer.
- head_size: size of each attention head.
- total_num_heads: total number of attention query heads.
- total_num_kv_heads: total number of attention key/value heads. If
- None, assume total_num_kv_heads = total_num_heads.
- bias: If true, add bias.
- skip_bias_add: This was added to enable performance optimizations where
- bias can be fused with other element-wise operations. we
- skip adding bias but instead return it.
- params_dtype: Data type for the parameters.
- quant_config: Quantization configure.
- prefix: The name of the layer in the state dict, including all parents
- (e.g. model.layers.0.qkv_proj)
- """
- def __init__(self,
- hidden_size: int,
- head_size: int,
- total_num_heads: int,
- total_num_kv_heads: Optional[int] = None,
- bias: bool = True,
- skip_bias_add: bool = False,
- params_dtype: Optional[torch.dtype] = None,
- quant_config: Optional[QuantizationConfig] = None,
- prefix: str = ""):
- self.hidden_size = hidden_size
- self.head_size = head_size
- self.total_num_heads = total_num_heads
- self.quant_config = quant_config
- if total_num_kv_heads is None:
- total_num_kv_heads = total_num_heads
- self.total_num_kv_heads = total_num_kv_heads
- # Divide the weight matrix along the last dimension.
- tp_size = get_tensor_model_parallel_world_size()
- tp_rank = get_tensor_model_parallel_rank()
- if quant_config is None:
- self.num_heads_per_kv_head = (self.total_num_heads //
- self.total_num_kv_heads)
- self.num_kv_heads = get_current_tp_rank_partition_size(
- self.total_num_kv_heads, tp_rank, tp_size)
- self.num_heads = self.num_kv_heads * self.num_heads_per_kv_head
- self.num_kv_head_replicas = 1
- else:
- self.num_heads = divide(self.total_num_heads, tp_size)
- if tp_size >= self.total_num_kv_heads:
- self.num_kv_heads = 1
- self.num_kv_head_replicas = divide(tp_size,
- self.total_num_kv_heads)
- elif tp_size < self.total_num_kv_heads and quant_config is not None:
- self.num_kv_heads = divide(self.total_num_kv_heads, tp_size)
- self.num_kv_head_replicas = 1
- input_size = self.hidden_size
- output_size = (self.num_heads +
- 2 * self.num_kv_heads) * tp_size * self.head_size
- self.output_sizes = [
- self.num_heads * self.head_size * tp_size, # q_proj
- self.num_kv_heads * self.head_size * tp_size, # k_proj
- self.num_kv_heads * self.head_size * tp_size, # v_proj
- ]
- super().__init__(input_size=input_size,
- output_size=output_size,
- bias=bias,
- gather_output=False,
- skip_bias_add=skip_bias_add,
- params_dtype=params_dtype,
- quant_config=quant_config,
- prefix=prefix)
- def weight_loader(self,
- param: Parameter,
- loaded_weight: torch.Tensor,
- loaded_shard_id: Optional[str] = None):
- # Special case for GGUF
- # initialize GGUF param after we know the quantize type
- is_gguf_weight = getattr(param, "is_gguf_weight", False)
- is_gguf_weight_type = getattr(param, "is_gguf_weight_type", False)
- if is_gguf_weight_type and loaded_shard_id is not None:
- idx_map = {"q": 0, "k": 1, "v": 2}
- param.data[idx_map[loaded_shard_id]].copy_(loaded_weight)
- param.shard_weight_type[loaded_shard_id] = loaded_weight.item()
- return
- if is_gguf_weight and isinstance(param, UninitializedParameter):
- from gguf.constants import GGML_QUANT_SIZES
- ori_shape = param.tensor_shape
- weight_types = self.qweight_type.shard_weight_type.values()
- row_size = []
- for weight_type in weight_types:
- block_size, type_size = GGML_QUANT_SIZES[weight_type]
- row_size.append(ori_shape[1] // block_size * type_size)
- q_shape = (ori_shape[0], max(row_size))
- param.materialize(q_shape, dtype=loaded_weight.dtype)
- param_data = param.data
- output_dim = getattr(param, "output_dim", None)
- # Special case for AQLM codebooks.
- is_metadata = getattr(param, "is_metadata", False)
- # Special case for per-tensor scales in fused case.
- needs_scalar_to_array = getattr(param, "needs_scalar_to_array", False)
- if loaded_shard_id is None:
- # Loaded weight is already fused on disk (qkv/mlp).
- if output_dim is None:
- if needs_scalar_to_array:
- param_data, loaded_weight = adjust_scalar_to_fused_array(
- param_data, loaded_weight, 0)
- assert param_data.shape == loaded_weight.shape
- param_data.copy_(loaded_weight)
- return
- shard_offsets = [
- # (shard_id, shard_offset, shard_size)
- ("q", 0, self.total_num_heads * self.head_size),
- ("k", self.total_num_heads * self.head_size,
- self.total_num_kv_heads * self.head_size),
- ("v", (self.total_num_heads + self.total_num_kv_heads) *
- self.head_size, self.total_num_kv_heads * self.head_size),
- ]
- packed_dim = getattr(param, "packed_dim", None)
- for shard_id, shard_offset, shard_size in shard_offsets:
- # Special case for Quantized Weights.
- # If quantized, we need to adjust the offset and size to account
- # for the packing.
- if packed_dim == output_dim:
- shard_size = shard_size // param.pack_factor
- shard_offset = shard_offset // param.pack_factor
- # Special case for Marlin.
- shard_size, shard_offset = adjust_marlin_shard(
- param, shard_size, shard_offset)
- loaded_weight_shard = loaded_weight.narrow(
- output_dim, shard_offset, shard_size)
- self.weight_loader(param, loaded_weight_shard, shard_id)
- return
- tp_rank = get_tensor_model_parallel_rank()
- assert loaded_shard_id in ["q", "k", "v"]
- # If output dim is defined, use the default loading process.
- if output_dim is not None:
- if loaded_shard_id == "q":
- shard_offset = 0
- shard_size = self.num_heads * self.head_size
- if self.quant_config is None:
- multiple_of = self.head_size * self.num_heads_per_kv_head
- elif loaded_shard_id == "k":
- shard_offset = self.num_heads * self.head_size
- shard_size = self.num_kv_heads * self.head_size
- if self.quant_config is None:
- multiple_of = self.head_size
- elif loaded_shard_id == "v":
- shard_offset = (self.num_heads +
- self.num_kv_heads) * self.head_size
- shard_size = self.num_kv_heads * self.head_size
- if self.quant_config is None:
- multiple_of = self.head_size
- # Special case for Quantized Weights.
- # If quantized, we need to adjust the offset and size to account
- # for the packing.
- packed_dim = getattr(param, "packed_dim", None)
- if packed_dim == output_dim:
- shard_size = shard_size // param.pack_factor
- shard_offset = shard_offset // param.pack_factor
- if self.quant_config is None:
- multiple_of = multiple_of // param.pack_factor
- # Special case for Marlin.
- shard_size, shard_offset = adjust_marlin_shard(
- param, shard_size, shard_offset)
- use_bitsandbytes = getattr(param, "use_bitsandbytes", False)
- if use_bitsandbytes:
- orig_qkv_offsets = {
- "q": (0, self.num_heads * self.head_size),
- "k": (self.num_heads * self.head_size,
- self.num_kv_heads * self.head_size),
- "v":
- ((self.num_heads + self.num_kv_heads) * self.head_size,
- self.num_kv_heads * self.head_size),
- "total":
- ((self.num_heads + 2 * self.num_kv_heads) * self.head_size,
- 0)
- }
- shard_size, shard_offset = adjust_bitsandbytes_shard(
- param, orig_qkv_offsets, loaded_shard_id)
- if is_gguf_weight:
- tp_size = get_tensor_model_parallel_world_size()
- output_dim = getattr(param, "output_dim", None)
- shard_shape = list(loaded_weight.shape)
- shard_shape[output_dim] = shard_shape[output_dim] // tp_size
- param.shard_id.append(loaded_shard_id)
- param.shard_size[loaded_shard_id] = shard_shape
- input_dim = getattr(param, "input_dim", None)
- input_size = loaded_weight.shape[input_dim]
- param_data = param_data.narrow(input_dim, 0, input_size)
- param_data = param_data.narrow(output_dim, shard_offset,
- shard_size)
- if self.quant_config is None:
- tp_size = get_tensor_model_parallel_world_size()
- total_size = loaded_weight.shape[output_dim]
- start_idx = get_current_tp_rank_partition_offset(
- total_size, tp_rank, tp_size, multiple_of=multiple_of)
- else:
- if loaded_shard_id == "q":
- shard_id = tp_rank
- else:
- shard_id = tp_rank // self.num_kv_head_replicas
- start_idx = shard_id * shard_size
- loaded_weight = loaded_weight.narrow(output_dim, start_idx,
- shard_size)
- # Special case for for AQLM codebooks.
- elif is_metadata:
- # metadata indicates fixed size concatenated along dim 0
- shard_size = loaded_weight.shape[0]
- shard_index = ["q", "k", "v"].index(loaded_shard_id)
- param_data = param_data.narrow(0, shard_index * shard_size,
- shard_size)
- # Special case for per-tensor scales in fused case.
- elif needs_scalar_to_array:
- param_data, loaded_weight = adjust_scalar_to_fused_array(
- param_data, loaded_weight, loaded_shard_id)
- else:
- ignore_warning = getattr(param, "ignore_warning", False)
- if not ignore_warning:
- logger.warning(
- "Loading a weight without `output_dim` attribute in "
- "QKVParallelLinear, assume the weight is the same "
- "for all partitions.")
- assert param_data.shape == loaded_weight.shape
- param_data.copy_(loaded_weight)
- class RowParallelLinear(LinearBase):
- """Linear layer with row parallelism.
- The linear layer is defined as Y = XA + b. A is parallelized along
- its first dimension and X along its second dimension as:
- - -
- | A_1 |
- | . |
- A = | . | X = [X_1, ..., X_p]
- | . |
- | A_p |
- - -
- Arguments:
- input_size: first dimension of matrix A.
- output_size: second dimension of matrix A.
- bias: If true, add bias. Note that bias is not parallelized.
- input_is_parallel: If true, we assume that the input is already
- split across the GPUs and we do not split
- again.
- skip_bias_add: This was added to enable performance optimization where
- bias can be fused with other element-wise operations.
- We skip adding bias but instead return it.
- params_dtype: Data type for the parameters.
- quant_config: Quantization configure.
- partition_multiple_of: Partitions will be divided,
- so each partition is a multiple of this number.
- """
- def __init__(self,
- input_size: int,
- output_size: int,
- bias: bool = True,
- input_is_parallel: bool = True,
- skip_bias_add: bool = False,
- params_dtype: Optional[torch.dtype] = None,
- reduce_results: bool = True,
- quant_config: Optional[QuantizationConfig] = None,
- partition_multiple_of: int = 1,
- prefix: str = ""):
- super().__init__(input_size, output_size, skip_bias_add, params_dtype,
- quant_config, prefix)
- self.input_is_parallel = input_is_parallel
- self.reduce_results = reduce_results
- self.quant_config = quant_config
- # Divide the weight matrix along the last dimension.
- self.tp_rank = get_tensor_model_parallel_rank()
- self.tp_size = get_tensor_model_parallel_world_size()
- self.tp_rank = get_tensor_model_parallel_rank()
- if quant_config is None:
- self.partition_multiple_of = partition_multiple_of
- self.input_size_per_partition = get_current_tp_rank_partition_size(
- input_size, self.tp_rank, self.tp_size, partition_multiple_of)
- else:
- self.input_size_per_partition = divide(input_size, self.tp_size)
- assert self.quant_method is not None
- self.quant_method.create_weights(
- layer=self,
- input_size_per_partition=self.input_size_per_partition,
- output_partition_sizes=[self.output_size],
- input_size=self.input_size,
- output_size=self.output_size,
- params_dtype=self.params_dtype,
- weight_loader=self.weight_loader,
- prefix=prefix)
- if not reduce_results and (bias and not skip_bias_add):
- raise ValueError("When not reduce the results, adding bias to the "
- "results can lead to incorrect results")
- if bias:
- self.bias = Parameter(
- torch.empty(self.output_size, dtype=params_dtype))
- set_weight_attrs(self.bias, {
- "output_dim": 0,
- "weight_loader": self.weight_loader,
- })
- else:
- self.register_parameter("bias", None)
- def weight_loader(self, param: Parameter, loaded_weight: torch.Tensor):
- tp_size = get_tensor_model_parallel_world_size()
- input_dim = getattr(param, "input_dim", None)
- # Special case for GGUF
- is_gguf_weight = getattr(param, "is_gguf_weight", False)
- is_gguf_weight_type = getattr(param, "is_gguf_weight_type", False)
- if is_gguf_weight_type:
- param.weight_type = loaded_weight.item()
- # Materialize GGUF UninitializedParameter
- if is_gguf_weight and isinstance(param, UninitializedParameter):
- weight_shape = list(loaded_weight.shape)
- if input_dim:
- weight_shape[input_dim] = weight_shape[input_dim] // tp_size
- param.materialize(tuple(weight_shape), dtype=loaded_weight.dtype)
- param_data = param.data
- if input_dim is not None:
- shard_size = param_data.shape[input_dim]
- if self.quant_config is None:
- start_idx = get_current_tp_rank_partition_offset(
- self.input_size,
- self.tp_rank,
- self.tp_size,
- multiple_of=self.partition_multiple_of)
- else:
- start_idx = self.tp_rank * shard_size
- loaded_weight = loaded_weight.narrow(input_dim, start_idx,
- shard_size)
- # Special case for loading scales off disk, which often do not
- # have a shape (such as in the case of AutoFP8).
- if len(loaded_weight.shape) == 0:
- loaded_weight = loaded_weight.reshape(1)
- assert param_data.shape == loaded_weight.shape
- param_data.copy_(loaded_weight)
- def forward(self, input_):
- if self.input_is_parallel:
- input_parallel = input_
- else:
- tp_rank = get_tensor_model_parallel_rank()
- splitted_input = split_tensor_along_last_dim(
- input_, num_partitions=self.tp_size)
- input_parallel = splitted_input[tp_rank].contiguous()
- # Matrix multiply.
- assert self.quant_method is not None
- # Only fuse bias add into GEMM for rank 0 (this ensures that
- # bias will not get added more than once in TP>1 case)
- bias_ = None if (self.tp_rank > 0 or self.skip_bias_add) else self.bias
- output_parallel = self.quant_method.apply(self,
- input_parallel,
- bias=bias_)
- if self.reduce_results and self.tp_size > 1:
- output = tensor_model_parallel_all_reduce(output_parallel)
- else:
- output = output_parallel
- output_bias = self.bias if self.skip_bias_add else None
- return output, output_bias
- def extra_repr(self) -> str:
- s = f"input_features={self.input_size_per_partition}"
- s += f", output_features={self.output_size}"
- s += f", bias={self.bias is not None}"
- s += f", tp_size={self.tp_size}"
- s += f", reduce_results={self.reduce_results}"
- return s
|