codefuse-ai/CodeFuse-DevOps-Model-7B-Chat

1# Copyright (c) Alibaba Cloud.2#3# This source code is licensed under the license found in the4# LICENSE file in the root directory of this source tree.5 6"""Tokenization classes for QWen."""7 8import base649import logging10import os11import unicodedata12from typing import Collection, Dict, List, Set, Tuple, Union13 14import tiktoken15from transformers import PreTrainedTokenizer, AddedToken16 17logger = logging.getLogger(__name__)18 19 20VOCAB_FILES_NAMES = {"vocab_file": "qwen.tiktoken"}21 22PAT_STR = r"""(?i:'s|'t|'re|'ve|'m|'ll|'d)|[^\r\n\p{L}\p{N}]?\p{L}+|\p{N}| ?[^\s\p{L}\p{N}]+[\r\n]*|\s*[\r\n]+|\s+(?!\S)|\s+"""23ENDOFTEXT = "<|endoftext|>"24IMSTART = "<|im_start|>"25IMEND = "<|im_end|>"26# as the default behavior is changed to allow special tokens in27# regular texts, the surface forms of special tokens need to be28# as different as possible to minimize the impact29EXTRAS = tuple((f"<|extra_{i}|>" for i in range(205)))30SPECIAL_TOKENS = (31    ENDOFTEXT,32    IMSTART,33    IMEND,34) + EXTRAS35 36 37def _load_tiktoken_bpe(tiktoken_bpe_file: str) -> Dict[bytes, int]:38    with open(tiktoken_bpe_file, "rb") as f:39        contents = f.read()40    return {41        base64.b64decode(token): int(rank)42        for token, rank in (line.split() for line in contents.splitlines() if line)43    }44 45class QWenTokenizer(PreTrainedTokenizer):46    """QWen tokenizer."""47 48    vocab_files_names = VOCAB_FILES_NAMES49 50    def __init__(51        self,52        vocab_file,53        errors="replace",54        **kwargs,55    ):56        super().__init__(**kwargs)57 58        self.errors = errors  # how to handle errors in decoding59 60        self.mergeable_ranks = _load_tiktoken_bpe(vocab_file)  # type: dict[bytes, int]61        self.special_tokens = {62            token: index63            for index, token in enumerate(64                SPECIAL_TOKENS, start=len(self.mergeable_ranks)65            )66        }67 68        enc = tiktoken.Encoding(69            "Qwen",70            pat_str=PAT_STR,71            mergeable_ranks=self.mergeable_ranks,72            special_tokens=self.special_tokens,73        )74        assert (75            len(self.mergeable_ranks) + len(self.special_tokens) == enc.n_vocab76        ), f"{len(self.mergeable_ranks) + len(self.special_tokens)} != {enc.n_vocab} in encoding"77 78        self.decoder = {79            v: k for k, v in self.mergeable_ranks.items()80        }  # type: dict[int, bytes|str]81        self.decoder.update({v: k for k, v in self.special_tokens.items()})82 83        self.tokenizer = enc  # type: tiktoken.Encoding84 85        self.eod_id = self.tokenizer.eot_token86        self.im_start_id = self.special_tokens[IMSTART]87        self.im_end_id = self.special_tokens[IMEND]88 89    def __len__(self) -> int:90        return self.tokenizer.n_vocab91 92    def get_vocab(self) -> Dict[bytes, int]:93        return self.mergeable_ranks94 95    def convert_tokens_to_ids(96        self, tokens: Union[bytes, str, List[Union[bytes, str]]]97    ) -> List[int]:98        ids = []99        if isinstance(tokens, (str, bytes)):100            if tokens in self.special_tokens:101                return self.special_tokens[tokens]102            else:103                return self.mergeable_ranks.get(tokens)104        for token in tokens:105            if token in self.special_tokens:106                ids.append(self.special_tokens[token])107            else:108                ids.append(self.mergeable_ranks.get(token))109        return ids110 111    def _add_tokens(self, new_tokens: Union[List[str], List[AddedToken]], special_tokens: bool = False) -> int:112        if not special_tokens and new_tokens:113            raise ValueError('Adding regular tokens is not supported')114        for token in new_tokens:115            surface_form = token.content if isinstance(token, AddedToken) else token116            if surface_form not in SPECIAL_TOKENS:117                raise ValueError('Adding unknown special tokens is not supported')118        return 0119 120    def save_vocabulary(self, save_directory: str, **kwargs) -> Tuple[str]:121        """122        Save only the vocabulary of the tokenizer (vocabulary).123 124        Returns:125            `Tuple(str)`: Paths to the files saved.126        """127        file_path = os.path.join(save_directory, "qwen.tiktoken")128        with open(file_path, "w", encoding="utf8") as w:129            for k, v in self.mergeable_ranks.items():130                line = base64.b64encode(k).decode("utf8") + " " + str(v) + "\n"131                w.write(line)132        return (file_path,)133 134    def tokenize(135        self,136        text: str,137        allowed_special: Union[Set, str] = "all",138        disallowed_special: Union[Collection, str] = (),139        **kwargs,140    ) -> List[Union[bytes, str]]:141        """142        Converts a string in a sequence of tokens.143 144        Args:145            text (`str`):146                The sequence to be encoded.147            allowed_special (`Literal["all"]` or `set`):148                The surface forms of the tokens to be encoded as special tokens in regular texts.149                Default to "all".150            disallowed_special (`Literal["all"]` or `Collection`):151                The surface forms of the tokens that should not be in regular texts and trigger errors.152                Default to an empty tuple.153 154            kwargs (additional keyword arguments, *optional*):155                Will be passed to the underlying model specific encode method.156 157        Returns:158            `List[bytes|str]`: The list of tokens.159        """160        tokens = []161        text = unicodedata.normalize("NFC", text)162 163        # this implementation takes a detour: text -> token id -> token surface forms164        for t in self.tokenizer.encode(165            text, allowed_special=allowed_special, disallowed_special=disallowed_special166        ):167            tokens.append(self.decoder[t])168        return tokens169 170    def convert_tokens_to_string(self, tokens: List[Union[bytes, str]]) -> str:171        """172        Converts a sequence of tokens in a single string.173        """174        text = ""175        temp = b""176        for t in tokens:177            if isinstance(t, str):178                if temp:179                    text += temp.decode("utf-8", errors=self.errors)180                    temp = b""181                text += t182            elif isinstance(t, bytes):183                temp += t184            else:185                raise TypeError("token should only be of type types or str")186        if temp:187            text += temp.decode("utf-8", errors=self.errors)188        return text189 190    @property191    def vocab_size(self):192        return self.tokenizer.n_vocab193 194    def _convert_id_to_token(self, index: int) -> Union[bytes, str]:195        """Converts an id to a token, special tokens included"""196        if index in self.decoder:197            return self.decoder[index]198        raise ValueError("unknown ids")199 200    def _convert_token_to_id(self, token: Union[bytes, str]) -> int:201        """Converts a token to an id using the vocab, special tokens included"""202        if token in self.special_tokens:203            return self.special_tokens[token]204        if token in self.mergeable_ranks:205            return self.mergeable_ranks[token]206        raise ValueError("unknown token")207 208    def _tokenize(self, text: str, **kwargs):209        """210        Converts a string in a sequence of tokens (string), using the tokenizer. Split in words for word-based211        vocabulary or sub-words for sub-word-based vocabularies (BPE/SentencePieces/WordPieces).212 213        Do NOT take care of added tokens.214        """215        raise NotImplementedError216 217    def _decode(218        self,219        token_ids: Union[int, List[int]],220        skip_special_tokens: bool = False,221        errors: str = None,222        **kwargs,223    ) -> str:224        if isinstance(token_ids, int):225            token_ids = [token_ids]226        if skip_special_tokens:227            token_ids = [i for i in token_ids if i < self.eod_id]228        return self.tokenizer.decode(token_ids, errors=errors or self.errors)229