custom_st.py

from typing import List, Dict, Tuple, Union, Any, Optional

import os
import json
import torch

from torch import nn
from transformers import AutoConfig, AutoModel, AutoTokenizer
from transformers.utils import is_flash_attn_2_available


INSTRUCTION_CONFIG = {
    "nl2code": {
        "query": "Find the most relevant code snippet given the following query:\n",
        "passage": "Candidate code snippet:\n"
    },
    "qa": {
        "query": "Find the most relevant answer given the following question:\n",
        "passage": "Candidate answer:\n"
    },
    "code2code": {
        "query": "Find an equivalent code snippet given the following code snippet:\n",
        "passage": "Candidate code snippet:\n"
    },
    "code2nl": {
        "query": "Find the most relevant comment given the following code snippet:\n",
        "passage": "Candidate comment:\n"
    },
    "code2completion": {
        "query": "Find the most relevant completion given the following start of code snippet:\n",
        "passage": "Candidate completion:\n"
    }
}


def batch(iterable, n=1):
    items = len(iterable)
    for ndx in range(0, items, n):
        yield iterable[ndx : min(ndx + n, items)]


def last_token_pooling(model_output, attention_mask):
    token_embeddings = model_output[0]
    left_padding = (attention_mask[:, -1].sum() == attention_mask.shape[0])
    if left_padding:
        return token_embeddings[:, -1]
    else:
        sequence_lengths = attention_mask.sum(dim=1) - 1
        batch_size = token_embeddings.shape[0]
        return token_embeddings[torch.arange(batch_size, device=token_embeddings.device), sequence_lengths].float()


class Transformer(nn.Module):
    def __init__(
        self,
        model_name_or_path: str,
        max_seq_length: int = None,
        model_args: Dict[str, Any] = None,
        tokenizer_args: Dict[str, Any] = None,
        config_args: Dict[str, Any] = None,
        cache_dir: str = None,
        do_lower_case: bool = False,
        tokenizer_name_or_path: str = None,
        **kwargs,
    ) -> None:
        super().__init__()
        self.config_keys = ["max_seq_length", "do_lower_case"]
        self.do_lower_case = do_lower_case
        if model_args is None:
            model_args = {}
        if tokenizer_args is None:
            tokenizer_args = {}
        if config_args is None:
            config_args = {}
        
        self.config = AutoConfig.from_pretrained(model_name_or_path, **config_args, cache_dir=cache_dir)

        self.task_names = self.config.task_names

        self.default_task = model_args.pop('default_task', None)

        model_args["attn_implementation"] = "flash_attention_2" if is_flash_attn_2_available() else "sdpa"

        self.auto_model = AutoModel.from_pretrained(model_name_or_path, config=self.config, cache_dir=cache_dir, **model_args)

        if max_seq_length is not None and "model_max_length" not in tokenizer_args:
            tokenizer_args["model_max_length"] = max_seq_length
        self.tokenizer = AutoTokenizer.from_pretrained(
            tokenizer_name_or_path if tokenizer_name_or_path is not None else model_name_or_path,
            cache_dir=cache_dir,
            **tokenizer_args,
        )

        # No max_seq_length set. Try to infer from model
        if max_seq_length is None:
            if (
                hasattr(self.auto_model, "config")
                and hasattr(self.auto_model.config, "max_position_embeddings")
                and hasattr(self.tokenizer, "model_max_length")
            ):
                max_seq_length = min(self.auto_model.config.max_position_embeddings, self.tokenizer.model_max_length)

        self.max_seq_length = max_seq_length

        if tokenizer_name_or_path is not None:
            self.auto_model.config.tokenizer_class = self.tokenizer.__class__.__name__
    

    @property
    def default_task(self):
        return self._default_task


    @default_task.setter
    def default_task(self, task: Union[None, str]):
        self._validate_task(task)
        self._default_task = task
        

    def _validate_task(self, task: str):
        if task and task not in self.task_names:
            raise ValueError(
                f"Unsupported task '{task}'. "
                f"Supported tasks are: {', '.join(self.config.task_names)}."
            )
    

    def forward(
        self,
        features: Dict[str, torch.Tensor],
        task: Optional[str] = None
    ) -> Dict[str, torch.Tensor]:
        """
        Forward pass through the model.
        """
        features.pop('prompt_length', None)
        output_states = self.auto_model.forward(
            **features,
            output_attentions=False,
            return_dict=True
        )
        output_tokens = output_states[0]
        features.update({"token_embeddings": output_tokens, "attention_mask": features["attention_mask"]})
        return features


    def get_word_embedding_dimension(self) -> int:
        return self.auto_model.config.hidden_size
    

    def tokenize(
        self,
        texts: Union[List[str], List[dict], List[Tuple[str, str]]],
        padding: Union[str, bool] = True
    ) -> Dict[str, torch.Tensor]:
        """Tokenizes a text and maps tokens to token-ids"""
        output = {}
        if isinstance(texts[0], str):
            to_tokenize = [texts]
        elif isinstance(texts[0], dict):
            to_tokenize = []
            output["text_keys"] = []
            for lookup in texts:
                text_key, text = next(iter(lookup.items()))
                to_tokenize.append(text)
                output["text_keys"].append(text_key)
            to_tokenize = [to_tokenize]
        else:
            batch1, batch2 = [], []
            for text_tuple in texts:
                batch1.append(text_tuple[0])
                batch2.append(text_tuple[1])
            to_tokenize = [batch1, batch2]

        # strip
        to_tokenize = [[str(s).strip() for s in col] for col in to_tokenize]

        # Lowercase
        if self.do_lower_case:
            to_tokenize = [[s.lower() for s in col] for col in to_tokenize]

        output.update(
            self.tokenizer(
                *to_tokenize,
                padding=padding,
                truncation=True,
                return_tensors="pt",
                max_length=self.max_seq_length,
            )
        )
        return output
    

    def get_config_dict(self) -> Dict[str, Any]:
         return {key: self.__dict__[key] for key in self.config_keys}


    def save(self, output_path: str, safe_serialization: bool = True) -> None:
        self.auto_model.save_pretrained(output_path, safe_serialization=safe_serialization)
        self.tokenizer.save_pretrained(output_path)

        with open(os.path.join(output_path, "sentence_transformer_config.json"), "w") as fOut:
            json.dump(self.get_config_dict(), fOut, indent=2)
    

    @classmethod
    def load(cls, input_path: str) -> "Transformer":
        config_name = "sentence_transformer_config.json"
        stransformer_config_path = os.path.join(input_path, config_name)
        with open(stransformer_config_path) as fIn:
            config = json.load(fIn)
        # Don't allow configs to set trust_remote_code
        if "model_args" in config and "trust_remote_code" in config["model_args"]:
            config["model_args"].pop("trust_remote_code")
        if "tokenizer_args" in config and "trust_remote_code" in config["tokenizer_args"]:
            config["tokenizer_args"].pop("trust_remote_code")
        if "config_args" in config and "trust_remote_code" in config["config_args"]:
            config["config_args"].pop("trust_remote_code")
        return cls(model_name_or_path=input_path, **config)