adaface/arc2face_models.py

import torch
import torch.nn as nn
from transformers import CLIPTextModel
from transformers.models.clip.modeling_clip import CLIPAttention
from typing import Optional, Tuple, Union
from transformers.modeling_outputs import BaseModelOutputWithPooling
from transformers.modeling_attn_mask_utils import AttentionMaskConverter
from diffusers import (
    StableDiffusionPipeline,
    UNet2DConditionModel,
    DDIMScheduler,
)
# from transformers.models.clip.modeling_clip import _make_causal_mask, _expand_mask
_make_causal_mask = AttentionMaskConverter._make_causal_mask
_expand_mask = AttentionMaskConverter._expand_mask

from .util import perturb_tensor

def create_arc2face_pipeline(base_model_path="models/sd15-dste8-vae.safetensors", 
                             dtype=torch.float16, unet_only=False):
    unet = UNet2DConditionModel.from_pretrained(
        'models/arc2face', subfolder="arc2face", torch_dtype=dtype
    )
    if unet_only:
        return unet

    text_encoder = CLIPTextModelWrapper.from_pretrained(
        'models/arc2face', subfolder="encoder", torch_dtype=dtype
    )
    noise_scheduler = DDIMScheduler(
        num_train_timesteps=1000,
        beta_start=0.00085,
        beta_end=0.012,
        beta_schedule="scaled_linear",
        clip_sample=False,
        set_alpha_to_one=False,
        steps_offset=1,
    )    
    pipeline = StableDiffusionPipeline.from_single_file(
            base_model_path,
            text_encoder=text_encoder,
            unet=unet,
            torch_dtype=dtype,
            safety_checker=None
        )
    pipeline.scheduler = noise_scheduler
    return pipeline

# Extend CLIPAttention by using multiple k_proj and v_proj in each head.
# To avoid too much increase of computation, we don't extend q_proj.
class CLIPAttentionMKV(nn.Module):
    """Multi-headed attention from 'Attention Is All You Need' paper"""

    def __init__(self, config, multiplier=2):
        super().__init__()
        self.config = config
        self.embed_dim = config.hidden_size
        self.num_heads = config.num_attention_heads
        self.head_dim = self.embed_dim // self.num_heads
        if self.head_dim * self.num_heads != self.embed_dim:
            raise ValueError(
                f"embed_dim must be divisible by num_heads (got `embed_dim`: {self.embed_dim} and `num_heads`:"
                f" {self.num_heads})."
            )
        self.scale = self.head_dim**-0.5
        self.dropout = config.attention_dropout
        self.multiplier  = multiplier

        self.k_proj   = nn.Linear(self.embed_dim, self.embed_dim * self.multiplier)
        self.v_proj   = nn.Linear(self.embed_dim, self.embed_dim * self.multiplier)
        self.q_proj   = nn.Linear(self.embed_dim, self.embed_dim)
        self.out_proj = nn.Linear(self.embed_dim, self.embed_dim)

    # The (approximately) repeated token features are repeated along the last dim in tensor
    # (multiplier * num_heads * head_dim), and then reshaped to (bsz, -1, num_heads, head_dim).
    # Therefore, the "multiplier" dim is tucked into the seq_len dim, which looks like
    # [token1_emb, token1_emb, token2_emb, token2_emb, ..., tokenN_emb, tokenN_emb].
    def _shape(self, tensor: torch.Tensor, seq_len: int, bsz: int):
        return tensor.view(bsz, seq_len, self.num_heads, self.head_dim).transpose(1, 2).contiguous()

    # clip_attn_layer is usually self.
    def extend_weights(self, clip_attn_layer, layer_idx, multiplier, perturb_std=0.2, 
                       perturb_std_is_relative=True, perturb_keep_norm=False, verbose=False):
        ORIG_V_SHAPE    = list(clip_attn_layer.v_proj.weight.shape)
        ORIG_V_SHAPE_D0 = ORIG_V_SHAPE[0]
        ORIG_K_SHAPE    = list(clip_attn_layer.k_proj.weight.shape)
        ORIG_K_SHAPE_D0 = ORIG_K_SHAPE[0]

        self.multiplier *= multiplier

        # q_proj and out_proj are the same as the original CLIPAttention.
        self.q_proj.weight.data   = clip_attn_layer.q_proj.weight.data.clone()
        self.q_proj.bias.data     = clip_attn_layer.q_proj.bias.data.clone()
        self.out_proj.weight.data = clip_attn_layer.out_proj.weight.data.clone()
        self.out_proj.bias.data   = clip_attn_layer.out_proj.bias.data.clone()

        # bias doesn't need noise perturbation, as after the weights are noised, 
        # different copies of the weight/bias will receive different gradients, 
        # making the bias terms diverge and identifiable after training.
        self.k_proj.bias.data     = clip_attn_layer.k_proj.bias.data.repeat(multiplier)
        self.v_proj.bias.data     = clip_attn_layer.v_proj.bias.data.repeat(multiplier)

        self.k_proj.weight.data   = clip_attn_layer.k_proj.weight.data.repeat(multiplier, 1)
        self.v_proj.weight.data   = clip_attn_layer.v_proj.weight.data.repeat(multiplier, 1)

        # Correct the out_features attribute of k_proj and v_proj.
        self.k_proj.out_features = self.k_proj.weight.shape[0]
        self.v_proj.out_features = self.v_proj.weight.shape[0]

        if perturb_std > 0:
            # Adding noise to the extra copies of the weights (keep the first copy unchanged).
            self.v_proj.weight.data[ORIG_V_SHAPE_D0:] = \
                perturb_tensor(self.v_proj.weight.data[ORIG_V_SHAPE_D0:], 
                                    perturb_std, perturb_std_is_relative, perturb_keep_norm, verbose=verbose)
            if verbose:
                NEW_V_SHAPE     = list(self.v_proj.weight.shape)
                NOISED_V_SHAPE  = list(self.v_proj.weight.data[ORIG_V_SHAPE_D0:].shape)
                print(f"Layer {layer_idx}: {NOISED_V_SHAPE} in {NEW_V_SHAPE} of v_proj is added with {perturb_std} noise")

            # Adding noise to the extra copies of the weights.
            self.k_proj.weight.data[ORIG_K_SHAPE_D0:] = \
                perturb_tensor(self.k_proj.weight.data[ORIG_K_SHAPE_D0:], 
                                    perturb_std, perturb_std_is_relative, perturb_keep_norm, verbose=verbose)
            if verbose:
                NEW_K_SHAPE     = list(self.k_proj.weight.shape)
                NOISED_K_SHAPE  = list(self.k_proj.weight.data[ORIG_K_SHAPE_D0:].shape)
                print(f"Layer {layer_idx}: {NOISED_K_SHAPE} in {NEW_K_SHAPE} of k_proj is added with {perturb_std} noise")

    def squeeze_weights(self, clip_attn_layer, divisor):
        if self.multiplier % divisor != 0:
            breakpoint()
        self.multiplier //= divisor

        self.k_proj.bias.data     = clip_attn_layer.k_proj.bias.data.reshape(divisor, -1).mean(dim=0)
        self.v_proj.bias.data     = clip_attn_layer.v_proj.bias.data.reshape(divisor, -1).mean(dim=0)

        self.k_proj.weight.data   = clip_attn_layer.k_proj.weight.data.reshape(divisor, -1, self.k_proj.weight.shape[1]).mean(dim=0)
        self.v_proj.weight.data   = clip_attn_layer.v_proj.weight.data.reshape(divisor, -1, self.v_proj.weight.shape[1]).mean(dim=0)

        # Correct the out_features attribute of k_proj and v_proj.
        self.k_proj.out_features = self.k_proj.weight.shape[0]
        self.v_proj.out_features = self.v_proj.weight.shape[0]


    def forward(
        self,
        hidden_states: torch.Tensor,
        attention_mask: Optional[torch.Tensor] = None,
        causal_attention_mask: Optional[torch.Tensor] = None,
        output_attentions: Optional[bool] = False,
    ) -> Tuple[torch.Tensor, Optional[torch.Tensor], Optional[Tuple[torch.Tensor]]]:
        """Input shape: Batch x Time x Channel"""

        bsz, tgt_len, embed_dim = hidden_states.size()

        query_states = self.q_proj(hidden_states) * self.scale
        # For key_states and value_states, the multiplier is absorbed into the seq_len (dim 1, shape specified as -1).
        # [token0_head_emb, token0_head_emb, token1_head_emb, token1_head_emb, ..., tokenN-1_head_emb, tokenN-1_head_emb].
        key_states   = self._shape(self.k_proj(hidden_states), -1, bsz)
        value_states = self._shape(self.v_proj(hidden_states), -1, bsz)

        proj_shape = (bsz * self.num_heads, -1, self.head_dim)
        query_states = self._shape(query_states, tgt_len, bsz).view(*proj_shape)
        key_states   = key_states.view(*proj_shape)
        value_states = value_states.view(*proj_shape)

        src_len = key_states.size(1)
        # src_len0 is the original src_len without the multiplier.
        src_len0 = src_len // self.multiplier
        attn_weights = torch.bmm(query_states, key_states.transpose(1, 2).contiguous())

        if attn_weights.size() != (bsz * self.num_heads, tgt_len, src_len):
            raise ValueError(
                f"Attention weights should be of size {(bsz * self.num_heads, tgt_len, src_len)}, but is"
                f" {attn_weights.size()}"
            )

        # apply the causal_attention_mask first
        if causal_attention_mask is not None:
            if causal_attention_mask.size() != (bsz, 1, tgt_len, src_len0):
                raise ValueError(
                    f"Attention mask should be of size {(bsz, 1, tgt_len, src_len0)}, but is"
                    f" {causal_attention_mask.size()}"
                )
            # The last dim of attn_weights corresponds to [token0, token0, token1, token1, ..., tokenN-1, tokenN-1].
            # If reshaping it as (self.multiplier, src_len0), it will become 
            # [[token0, token0, token1, token1, ..., tokenN//2], [tokenN//2+1, tokenN//2+1, ..., tokenN-1, tokenN-1]],
            # and the mask will be applied to wrong elements.
            # If reshaping it as (src_len0, self.multiplier), it will become 
            # [[token0, token1, ..., tokenN-1], [token0, token1, ..., tokenN-1]], and then
            # the mask at element i will mask all the multiplier elements at i, which is desired.
            attn_weights = attn_weights.view(bsz, self.num_heads, tgt_len, src_len0, self.multiplier) + causal_attention_mask.unsqueeze(4)
            attn_weights = attn_weights.view(bsz * self.num_heads, tgt_len, src_len)

        if attention_mask is not None:
            if attention_mask.size() != (bsz, 1, tgt_len, src_len0):
                raise ValueError(
                    f"Attention mask should be of size {(bsz, 1, tgt_len, src_len0)}, but is {attention_mask.size()}"
                )
            attn_weights = attn_weights.view(bsz, self.num_heads, tgt_len, src_len0, self.multiplier) + attention_mask.unsqueeze(4)
            attn_weights = attn_weights.view(bsz * self.num_heads, tgt_len, src_len)

        attn_weights = nn.functional.softmax(attn_weights, dim=-1)

        if output_attentions:
            # this operation is a bit awkward, but it's required to
            # make sure that attn_weights keeps its gradient.
            # In order to do so, attn_weights have to reshaped
            # twice and have to be reused in the following
            attn_weights_reshaped = attn_weights.view(bsz, self.num_heads, tgt_len, src_len)
            attn_weights = attn_weights_reshaped.view(bsz * self.num_heads, tgt_len, src_len)
        else:
            attn_weights_reshaped = None

        attn_probs = nn.functional.dropout(attn_weights, p=self.dropout, training=self.training)

        attn_output = torch.bmm(attn_probs, value_states)

        if attn_output.size() != (bsz * self.num_heads, tgt_len, self.head_dim):
            raise ValueError(
                f"`attn_output` should be of size {(bsz, self.num_heads, tgt_len, self.head_dim)}, but is"
                f" {attn_output.size()}"
            )

        attn_output = attn_output.view(bsz, self.num_heads, tgt_len, self.head_dim)
        attn_output = attn_output.transpose(1, 2).contiguous()
        attn_output = attn_output.reshape(bsz, tgt_len, embed_dim)

        attn_output = self.out_proj(attn_output)

        return attn_output, attn_weights_reshaped
    
class CLIPTextModelWrapper(CLIPTextModel):
    # Adapted from https://github.com/huggingface/transformers/blob/v4.34.1/src/transformers/models/clip/modeling_clip.py#L812
    # Modified to accept precomputed token embeddings "input_token_embs" as input or calculate them from input_ids and return them.
    def forward(
        self,
        input_ids: Optional[torch.Tensor] = None,
        attention_mask: Optional[torch.Tensor] = None,
        position_ids: Optional[torch.Tensor] = None,
        output_attentions: Optional[bool] = None,
        output_hidden_states: Optional[bool] = None,
        return_dict: Optional[bool] = None,
        input_token_embs: Optional[torch.Tensor] = None,
        hidden_state_layer_weights: Optional[torch.Tensor] = None,
        return_token_embs: Optional[bool] = False,
    ) -> Union[Tuple, torch.Tensor, BaseModelOutputWithPooling]:

        if return_token_embs:
            return self.text_model.embeddings.token_embedding(input_ids)
    
        return_dict = return_dict if return_dict is not None else self.config.use_return_dict
    
        output_attentions = output_attentions if output_attentions is not None else self.text_model.config.output_attentions
        output_hidden_states = (
            output_hidden_states if output_hidden_states is not None else self.text_model.config.output_hidden_states
        )
        if hidden_state_layer_weights is not None:
            output_hidden_states = True
        return_dict = return_dict if return_dict is not None else self.text_model.config.use_return_dict
    
        if input_ids is None:
            raise ValueError("You have to specify input_ids")
    
        input_shape = input_ids.size()
        input_ids = input_ids.view(-1, input_shape[-1])
    
        hidden_states = self.text_model.embeddings(input_ids=input_ids, position_ids=position_ids, inputs_embeds=input_token_embs)
    
        # CLIP's text model uses causal mask, prepare it here.
        # https://github.com/openai/CLIP/blob/cfcffb90e69f37bf2ff1e988237a0fbe41f33c04/clip/model.py#L324
        causal_attention_mask = _make_causal_mask(input_shape, hidden_states.dtype, device=hidden_states.device)
        # expand attention_mask
        if attention_mask is not None:
            # [bsz, seq_len] -> [bsz, 1, tgt_seq_len, src_seq_len]
            attention_mask = _expand_mask(attention_mask, hidden_states.dtype)
    
        encoder_outputs = self.text_model.encoder(
            inputs_embeds=hidden_states,
            attention_mask=attention_mask,
            causal_attention_mask=causal_attention_mask,
            output_attentions=output_attentions,
            # output_hidden_states is False by default, and only True if hidden_state_layer_weights is provided.
            output_hidden_states=output_hidden_states,      
            return_dict=return_dict,
        )
    
        # If output_hidden_states is True, then encoder_outputs[0] is last_hidden_state [1, 22, 768].
        # encoder_outputs[1] is hidden_states, which is a tuple of 13 hidden states, each being [1, 22, 768].
        # encoder_outputs[0] == encoder_outputs[1][12].
        if hidden_state_layer_weights is None:
            last_hidden_state = encoder_outputs[0]
        else:
            num_hidden_state_layers = len(hidden_state_layer_weights)
            last_hidden_states = encoder_outputs[1][-num_hidden_state_layers:]
            hidden_state_layer_weights = hidden_state_layer_weights.to(last_hidden_states[0].dtype)
            # Normalize the weights of to sum to 1 across layers.
            # hidden_state_layer_weights: [3, 1] or [3, 768].
            hidden_state_layer_weights = hidden_state_layer_weights / hidden_state_layer_weights.sum(dim=0, keepdim=True)
            # [3, 1/768] -> [3, 1, 1, 1/768]
            hidden_state_layer_weights = hidden_state_layer_weights.unsqueeze(1).unsqueeze(1)
            # A weighted sum of last_hidden_states.
            # [3, 1, 22, 768] * [3, 1, 1, 1/768] -> [3, 1, 22, 768] -> [1, 22, 768]
            last_hidden_state = (torch.stack(last_hidden_states, dim=0) * hidden_state_layer_weights).sum(dim=0)
            
        last_hidden_state = self.text_model.final_layer_norm(last_hidden_state)

        # self.text_model.eos_token_id == 2 is True.
        if self.text_model.eos_token_id == 2:
            # The `eos_token_id` was incorrect before PR #24773: Let's keep what have been done here.
            # A CLIP model with such `eos_token_id` in the config can't work correctly with extra new tokens added
            # ------------------------------------------------------------
            # text_embeds.shape = [batch_size, sequence_length, transformer.width]
            # take features from the eot embedding (eot_token is the highest number in each sequence)
            # casting to torch.int for onnx compatibility: argmax doesn't support int64 inputs with opset 14
            pooled_output = last_hidden_state[
                torch.arange(last_hidden_state.shape[0], device=last_hidden_state.device),
                input_ids.to(dtype=torch.int, device=last_hidden_state.device).argmax(dim=-1),
            ]
        else:
            # The config gets updated `eos_token_id` from PR #24773 (so the use of exta new tokens is possible)
            pooled_output = last_hidden_state[
                torch.arange(last_hidden_state.shape[0], device=last_hidden_state.device),
                # We need to get the first position of `eos_token_id` value (`pad_token_ids` might equal to `eos_token_id`)
                (input_ids.to(dtype=torch.int, device=last_hidden_state.device) == self.text_model.eos_token_id)
                .int()
                .argmax(dim=-1),
            ]
    
        if not return_dict:
            return (last_hidden_state, pooled_output) + encoder_outputs[1:]
    
        return BaseModelOutputWithPooling(
            last_hidden_state=last_hidden_state,
            pooler_output=pooled_output,
            hidden_states=encoder_outputs.hidden_states,
            attentions=encoder_outputs.attentions,
        )
    
    # Applied to all attention layers in the encoder, if the corresponding multiplier is not 1.
    # The layer indexed by end_layer_idx is not included.
    # If both layer indices are -1, then apply to all layers (0-11).
    def extend_clip_attention_MKV_multiplier(self, prompt2token_proj_attention_multipliers, perturb_std=0.1):
        num_extended_layers = 0

        for layer_idx, layer in enumerate(self.text_model.encoder.layers):
            multiplier = prompt2token_proj_attention_multipliers[layer_idx]
            if multiplier == 1:
                continue
            # This shouldn't happen, unless self_attn has already been extended as CLIPAttentionMKV.
            if not isinstance(layer.self_attn, (CLIPAttention, CLIPAttentionMKV)):
                breakpoint()
            old_attn_layer = layer.self_attn
            if not isinstance(old_attn_layer, CLIPAttentionMKV):
                layer.self_attn = CLIPAttentionMKV(old_attn_layer.config, 1)
            # Extends the v_proj and k_proj weights in the self_attn layer.
            layer.self_attn.extend_weights(old_attn_layer, layer_idx, multiplier, perturb_std, verbose=True)
            num_extended_layers += 1
    
        return num_extended_layers

    # Applied to layers [begin_layer_idx, end_layer_idx) in the encoder.
    # The layer indexed by end_layer_idx is not included.
    # If both layer indices are -1, then apply to all layers (0-11).
    def squeeze_clip_attention_MKV_divisor(self, prompt2token_proj_attention_divisors):
        num_squeezed_layers = 0

        for layer_idx, layer in enumerate(self.text_model.encoder.layers):
            divisor = prompt2token_proj_attention_divisors[layer_idx]
            if divisor == 1:
                continue
            # This shouldn't happen, unless self_attn has already been extended as CLIPAttentionMKV.
            if not isinstance(layer.self_attn, (CLIPAttention, CLIPAttentionMKV)):
                breakpoint()
            old_attn_layer = layer.self_attn
            if not isinstance(old_attn_layer, CLIPAttentionMKV):
                layer.self_attn = CLIPAttentionMKV(old_attn_layer.config, 1)
            # Squeeze the k_proj and v_proj  weights in the self_attn layer.
            layer.self_attn.squeeze_weights(old_attn_layer, divisor)
            num_squeezed_layers += 1
    
        return num_squeezed_layers