ramixpe/1.8testing

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)))30# changed to use actual index to avoid misconfiguration with vocabulary expansion31SPECIAL_START_ID = 15164332SPECIAL_TOKENS = tuple(33    enumerate(34        (35            (36                ENDOFTEXT,37                IMSTART,38                IMEND,39            )40            + EXTRAS41        ),42        start=SPECIAL_START_ID,43    )44)45SPECIAL_TOKENS_SET = set(t for i, t in SPECIAL_TOKENS)46 47 48def _load_tiktoken_bpe(tiktoken_bpe_file: str) -> Dict[bytes, int]:49    with open(tiktoken_bpe_file, "rb") as f:50        contents = f.read()51    return {52        base64.b64decode(token): int(rank)53        for token, rank in (line.split() for line in contents.splitlines() if line)54    }55 56 57class QWenTokenizer(PreTrainedTokenizer):58    """QWen tokenizer."""59 60    vocab_files_names = VOCAB_FILES_NAMES61 62    def __init__(63        self,64        vocab_file,65        errors="replace",66        extra_vocab_file=None,67        **kwargs,68    ):69        super().__init__(**kwargs)70 71        # how to handle errors in decoding UTF-8 byte sequences72        # use ignore if you are in streaming inference73        self.errors = errors  74 75        self.mergeable_ranks = _load_tiktoken_bpe(vocab_file)  # type: Dict[bytes, int]76        self.special_tokens = {77            token: index78            for index, token in SPECIAL_TOKENS79        }80 81        # try load extra vocab from file82        if extra_vocab_file is not None:83            used_ids = set(self.mergeable_ranks.values()) | set(self.special_tokens.values())84            extra_mergeable_ranks = _load_tiktoken_bpe(extra_vocab_file)85            for token, index in extra_mergeable_ranks.items():86                if token in self.mergeable_ranks:87                    logger.info(f"extra token {token} exists, skipping")88                    continue89                if index in used_ids:90                    logger.info(f'the index {index} for extra token {token} exists, skipping')91                    continue92                self.mergeable_ranks[token] = index93            # the index may be sparse after this, but don't worry tiktoken.Encoding will handle this94 95        enc = tiktoken.Encoding(96            "Qwen",97            pat_str=PAT_STR,98            mergeable_ranks=self.mergeable_ranks,99            special_tokens=self.special_tokens,100        )101        assert (102            len(self.mergeable_ranks) + len(self.special_tokens) == enc.n_vocab103        ), f"{len(self.mergeable_ranks) + len(self.special_tokens)} != {enc.n_vocab} in encoding"104 105        self.decoder = {106            v: k for k, v in self.mergeable_ranks.items()107        }  # type: dict[int, bytes|str]108        self.decoder.update({v: k for k, v in self.special_tokens.items()})109 110        self.tokenizer = enc  # type: tiktoken.Encoding111 112        self.eod_id = self.tokenizer.eot_token113        self.im_start_id = self.special_tokens[IMSTART]114        self.im_end_id = self.special_tokens[IMEND]115 116    def __getstate__(self):117        # for pickle lovers118        state = self.__dict__.copy()119        del state["tokenizer"]120        return state121 122    def __setstate__(self, state):123        # tokenizer is not python native; don't pass it; rebuild it124        self.__dict__.update(state)125        enc = tiktoken.Encoding(126            "Qwen",127            pat_str=PAT_STR,128            mergeable_ranks=self.mergeable_ranks,129            special_tokens=self.special_tokens,130        )131        self.tokenizer = enc132 133    def __len__(self) -> int:134        return self.tokenizer.n_vocab135 136    def get_vocab(self) -> Dict[bytes, int]:137        return self.mergeable_ranks138 139    def convert_tokens_to_ids(140        self, tokens: Union[bytes, str, List[Union[bytes, str]]]141    ) -> List[int]:142        ids = []143        if isinstance(tokens, (str, bytes)):144            if tokens in self.special_tokens:145                return self.special_tokens[tokens]146            else:147                return self.mergeable_ranks.get(tokens)148        for token in tokens:149            if token in self.special_tokens:150                ids.append(self.special_tokens[token])151            else:152                ids.append(self.mergeable_ranks.get(token))153        return ids154 155    def _add_tokens(156        self,157        new_tokens: Union[List[str], List[AddedToken]],158        special_tokens: bool = False,159    ) -> int:160        if not special_tokens and new_tokens:161            raise ValueError("Adding regular tokens is not supported")162        for token in new_tokens:163            surface_form = token.content if isinstance(token, AddedToken) else token164            if surface_form not in SPECIAL_TOKENS_SET:165                raise ValueError("Adding unknown special tokens is not supported")166        return 0167 168    def save_vocabulary(self, save_directory: str, **kwargs) -> Tuple[str]:169        """170        Save only the vocabulary of the tokenizer (vocabulary).171 172        Returns:173            `Tuple(str)`: Paths to the files saved.174        """175        file_path = os.path.join(save_directory, "qwen.tiktoken")176        with open(file_path, "w", encoding="utf8") as w:177            for k, v in self.mergeable_ranks.items():178                line = base64.b64encode(k).decode("utf8") + " " + str(v) + "\n"179                w.write(line)180        return (file_path,)181 182    def tokenize(183        self,184        text: str,185        allowed_special: Union[Set, str] = "all",186        disallowed_special: Union[Collection, str] = (),187        **kwargs,188    ) -> List[Union[bytes, str]]:189        """190        Converts a string in a sequence of tokens.191 192        Args:193            text (`str`):194                The sequence to be encoded.195            allowed_special (`Literal["all"]` or `set`):196                The surface forms of the tokens to be encoded as special tokens in regular texts.197                Default to "all".198            disallowed_special (`Literal["all"]` or `Collection`):199                The surface forms of the tokens that should not be in regular texts and trigger errors.200                Default to an empty tuple.201 202            kwargs (additional keyword arguments, *optional*):203                Will be passed to the underlying model specific encode method.204 205        Returns:206            `List[bytes|str]`: The list of tokens.207        """208        tokens = []209        text = unicodedata.normalize("NFC", text)210 211        # this implementation takes a detour: text -> token id -> token surface forms212        for t in self.tokenizer.encode(213            text, allowed_special=allowed_special, disallowed_special=disallowed_special214        ):215            tokens.append(self.decoder[t])216        return tokens217 218    def convert_tokens_to_string(self, tokens: List[Union[bytes, str]]) -> str:219        """220        Converts a sequence of tokens in a single string.221        """222        text = ""223        temp = b""224        for t in tokens:225            if isinstance(t, str):226                if temp:227                    text += temp.decode("utf-8", errors=self.errors)228                    temp = b""229                text += t230            elif isinstance(t, bytes):231                temp += t232            else:233                raise TypeError("token should only be of type types or str")234        if temp:235            text += temp.decode("utf-8", errors=self.errors)236        return text237 238    @property239    def vocab_size(self):240        return self.tokenizer.n_vocab241 242    def _convert_id_to_token(self, index: int) -> Union[bytes, str]:243        """Converts an id to a token, special tokens included"""244        if index in self.decoder:245            return self.decoder[index]246        raise ValueError("unknown ids")247 248    def _convert_token_to_id(self, token: Union[bytes, str]) -> int:249        """Converts a token to an id using the vocab, special tokens included"""250        if token in self.special_tokens:251            return self.special_tokens[token]252        if token in self.mergeable_ranks:253            return self.mergeable_ranks[token]254        raise ValueError("unknown token")255 256    def _tokenize(self, text: str, **kwargs):257        """258        Converts a string in a sequence of tokens (string), using the tokenizer. Split in words for word-based259        vocabulary or sub-words for sub-word-based vocabularies (BPE/SentencePieces/WordPieces).260 261        Do NOT take care of added tokens.262        """263        raise NotImplementedError264 265    def _decode(266        self,267        token_ids: Union[int, List[int]],268        skip_special_tokens: bool = False,269        errors: str = None,270        **kwargs,271    ) -> str:272        if isinstance(token_ids, int):273            token_ids = [token_ids]274        if skip_special_tokens:275            token_ids = [i for i in token_ids if i < self.eod_id]276        return self.tokenizer.decode(token_ids, errors=errors or self.errors)277