Upload 4 files

special_tokens_map.json

@@ -0,0 +1 @@
+{}

tokenization_rwkv_world.py

@@ -0,0 +1,505 @@
+# coding=utf-8
+# Copyright 2018 The Open AI Team Authors and The HuggingFace Inc. team.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Tokenization classes for OpenAI GPT."""
+
+import json
+import os
+from typing import TYPE_CHECKING, List, Optional, Tuple, Union
+from transformers.tokenization_utils import AddedToken, PreTrainedTokenizer
+from transformers.utils import logging, to_py_obj
+from transformers.tokenization_utils_base import BatchEncoding
+
+import bisect
+import itertools
+import re
+import unicodedata
+from collections import OrderedDict
+from typing import Any, Dict, List, Optional, Tuple, Union, overload
+
+from transformers.tokenization_utils_base import (
+    ENCODE_KWARGS_DOCSTRING,
+    ENCODE_PLUS_ADDITIONAL_KWARGS_DOCSTRING,
+    INIT_TOKENIZER_DOCSTRING,
+    AddedToken,
+    BatchEncoding,
+    EncodedInput,
+    EncodedInputPair,
+    PreTokenizedInput,
+    PreTokenizedInputPair,
+    PreTrainedTokenizerBase,
+    TextInput,
+    TextInputPair,
+    TruncationStrategy,
+)
+from transformers.utils import PaddingStrategy, TensorType, add_end_docstrings, logging
+
+
+if TYPE_CHECKING:
+    from transformers.pipelines.conversational import Conversation
+
+logger = logging.get_logger(__name__)
+
+VOCAB_FILES_NAMES = {
+    "vocab_file": "rwkv_vocab_v20230424.json",
+}
+
+
+class DATrie:
+    class Node:
+        def __init__(self, is_leaf=False, leaf_data=None, tail=""):
+            self._is_leaf = is_leaf
+            self._leaf_data = leaf_data
+            self._tail = tail
+            self._next_map = {}
+
+        def is_leaf(self):
+            return self._is_leaf
+
+        def set_leaf(self):
+            self._is_leaf = True
+
+        def has_next(self, w):
+            if w in self._next_map:
+                return True
+            return False
+
+        def add_node(self, w, node):
+            self._next_map[w] = node
+
+        def get_node(self, w):
+            if w in self._next_map:
+                return self._next_map[w]
+            return None
+
+        def get_tail(self):
+            return self._tail
+
+        def get_data(self):
+            return self._leaf_data
+
+        def set_data(self, data):
+            self._leaf_data = data
+
+    def __init__(self, special_ids):
+        self.root = self.Node()
+        self.data = {}
+        self.r_data = {}
+        self.special_ids = special_ids
+
+    def insert(self, word, data):
+        self.data[word] = data
+        self.r_data[data] = word
+        idx = 0
+        node = self.root
+        while idx < len(word):
+            w = word[idx]
+            is_leaf = (idx == (len(word) - 1))
+            leaf_data = (data if is_leaf else None)
+            # 不存在则插入
+            if not node.has_next(w):
+                node.add_node(w, self.Node(is_leaf=is_leaf, leaf_data=leaf_data))
+                # last word
+            node = node.get_node(w)
+            idx += 1
+        if not node.is_leaf():
+            node.set_leaf()
+            node.set_data(data)
+
+    def findStrict(self, word):
+        idx = 0
+        node = self.root
+        while node is not None and idx < len(word):
+            w = word[idx]
+            if not node.has_next(w):
+                return None
+                # last word
+            node = node.get_node(w)
+            idx += 1
+        if node.is_leaf():
+            return node.get_data()
+        return None
+
+    def prefix(self, word):
+        idx = 0
+        node = self.root
+        result = []
+        while node is not None and idx < len(word):
+            w = word[idx]
+            if not node.has_next(w):
+                return result
+                # last word
+            node = node.get_node(w)
+            if node.is_leaf():
+                result.append([word[:idx + 1], node.get_data()])
+            idx += 1
+        return result
+
+    def max_prefix(self, content, start_idx):
+        idx = start_idx
+        node = self.root
+        l = len(content)
+        result = [["", ], ]
+        while node is not None and idx < l:
+            w = content[idx]
+            if not node.has_next(w):
+                return result[-1]
+                # last word
+            node = node.get_node(w)
+            if node.is_leaf():
+                result.append([content[start_idx:idx + 1], node.get_data()])
+            idx += 1
+        return result[-1]
+
+    def max_score(self, content, start_idx):
+        idx = start_idx
+        node = self.root
+        l = len(content)
+        result = [["", (3, 0)], ]
+        while node is not None and idx < l:
+            w = content[idx]
+            if not node.has_next(w):
+                break
+                # last word
+            node = node.get_node(w)
+            if node.is_leaf():
+                result.append([content[start_idx:idx + 1], node.get_data()])
+            idx += 1
+        if len(result) > 1:
+            result = sorted(result, key=lambda x: x[1][1])
+        return result[-1]
+
+    def match(self, content, add_unk=True, unk_id=-1, **kwargs):
+        # length
+        l = len(content)
+        i = 0
+        result_list = []
+        while i < l:
+            match_word = self.max_prefix(content=content, start_idx=i)
+            # print(match_word)
+            w = match_word[0]
+            if len(w) > 0:
+                result_list.append(match_word[1])
+                i += len(w)
+            else:
+                if add_unk:
+                    result_list.append(unk_id)
+                i += 1
+        return result_list
+
+    def id2str(self, ids, escape_special_ids=True, end_ids=[], **kwargs):
+        res_str = ""
+        for rid in ids:
+            if rid in self.r_data:
+                if rid in end_ids:
+                    break
+                if escape_special_ids and rid in self.special_ids:
+                    continue
+                rstr = self.r_data[rid]
+                res_str += rstr
+            elif rid == 0:
+                break
+            else:
+                print("ERROR unknown id %d" % rid)
+                res_str += "UNK"
+        return res_str
+
+    def id2str_v2(self, ids, escape_special_ids=True, end_ids=[], **kwargs):
+        res_str = ""
+        for rid in ids:
+            if rid in self.r_data:
+                if rid in end_ids:
+                    break
+                rstr = self.r_data[rid]
+                if escape_special_ids and rid in self.special_ids:
+                    continue
+                res_str += rstr
+            elif rid == 0:
+                break
+            else:
+                print("ERROR unknown id %d" % rid)
+                res_str += "UNK"
+        return res_str
+
+
+class RWKVWorldTokenizer(PreTrainedTokenizer):
+    vocab_files_names = VOCAB_FILES_NAMES
+    model_input_names = ["input_ids", "attention_mask"]
+
+    def __init__(
+            self,
+            vocab_file,
+            errors="replace",
+            **kwargs
+    ):
+        self.add_bos_token = False
+
+        with open(vocab_file, encoding="utf-8") as vocab_handle:
+            self.encoder = json.load(vocab_handle)
+        super().__init__(
+            errors=errors,
+            **kwargs,
+        )
+        self.decoder = {v: k for k, v in self.encoder.items()}
+        self.trie = DATrie(self.all_special_ids)
+        for k, v in self.encoder.items():
+            self.trie.insert(k, v)
+        self.errors = errors  # how to handle errors in decoding
+        self.cache = {}
+
+    @property
+    def vocab_size(self):
+        return len(self.encoder)
+
+    def get_vocab(self):
+        return dict(self.encoder, **self.added_tokens_encoder)
+
+    def build_inputs_with_special_tokens(self, token_ids_0, token_ids_1=None):
+        if self.add_bos_token:
+            bos_token_ids = [self.bos_token_id]
+        else:
+            bos_token_ids = []
+
+        output = bos_token_ids + token_ids_0
+
+        if token_ids_1 is None:
+            return output
+
+        return output + bos_token_ids + token_ids_1
+
+    def get_special_tokens_mask(
+            self, token_ids_0: List[int], token_ids_1: Optional[List[int]] = None,
+            already_has_special_tokens: bool = False
+    ) -> List[int]:
+        """
+        Retrieves sequence ids from a token list that has no special tokens added. This method is called when adding
+        special tokens using the tokenizer `prepare_for_model` or `encode_plus` methods.
+
+        Args:
+            token_ids_0 (`List[int]`):
+                List of IDs.
+            token_ids_1 (`List[int]`, *optional*):
+                Optional second list of IDs for sequence pairs.
+            already_has_special_tokens (`bool`, *optional*, defaults to `False`):
+                Whether or not the token list is already formatted with special tokens for the model.
+
+        Returns:
+            `List[int]`: A list of integers in the range [0, 1]: 1 for a special token, 0 for a sequence token.
+        """
+        if already_has_special_tokens:
+            return super().get_special_tokens_mask(
+                token_ids_0=token_ids_0, token_ids_1=token_ids_1, already_has_special_tokens=True
+            )
+
+        if not self.add_bos_token:
+            return super().get_special_tokens_mask(
+                token_ids_0=token_ids_0, token_ids_1=token_ids_1, already_has_special_tokens=False
+            )
+
+        if token_ids_1 is None:
+            return [1] + ([0] * len(token_ids_0))
+        return [1] + ([0] * len(token_ids_0)) + [1] + ([0] * len(token_ids_1))
+
+    def _tokenize(self, text, **kwargs):
+        """Tokenize a string."""
+        return self.trie.match(text, unk_id=self.unk_token_id, **kwargs)
+
+    def _decode(self,
+               token_ids: Union[int, List[int], "np.ndarray", "torch.Tensor", "tf.Tensor"],
+               skip_special_tokens: bool = False,
+               **kwargs
+               ) -> str:
+
+        # Convert inputs to python lists
+        token_ids = to_py_obj(token_ids)
+        if isinstance(token_ids, int):
+            if token_ids in self.all_special_ids and skip_special_tokens:
+                return ""
+            return self.decoder.get(token_ids, self.unk_token)
+        elif isinstance(token_ids, list):
+            return self.trie.id2str(
+                token_ids,
+                escape_special_ids=skip_special_tokens,
+                **kwargs
+            )
+        else:
+            return token_ids
+
+    def _convert_token_to_id(self, token):
+        """Converts a token (str) in an id using the vocab."""
+        return self.encoder.get(token, self.encoder.get(self.unk_token))
+
+    def _convert_id_to_token(self, index):
+        """Converts an index (integer) in a token (str) using the vocab."""
+        return self.decoder.get(index)
+
+    def save_vocabulary(self, save_directory: str, filename_prefix: Optional[str] = None) -> Tuple[str]:
+        if not os.path.exists(save_directory):
+            os.mkdir(save_directory)
+        if not os.path.isdir(save_directory):
+            logger.error(f"Vocabulary path ({save_directory}) should be a directory")
+            return
+        vocab_file = os.path.join(
+            save_directory, (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]
+        )
+
+        with open(vocab_file, "w", encoding="utf-8") as f:
+            f.write(json.dumps(self.encoder, indent=2, sort_keys=True, ensure_ascii=False) + "\n")
+
+        return (vocab_file,)
+
+    def prepare_for_tokenization(self, text, **kwargs):
+        return (text, kwargs)
+
+    def _encode_plus(
+        self,
+        text: Union[TextInput, EncodedInput],
+        add_special_tokens: bool = True,
+        padding_strategy: PaddingStrategy = PaddingStrategy.DO_NOT_PAD,
+        truncation_strategy: TruncationStrategy = TruncationStrategy.DO_NOT_TRUNCATE,
+        max_length: Optional[int] = None,
+        stride: int = 0,
+        pad_to_multiple_of: Optional[int] = None,
+        return_tensors: Optional[Union[str, TensorType]] = None,
+        return_token_type_ids: Optional[bool] = None,
+        return_attention_mask: Optional[bool] = None,
+        return_overflowing_tokens: bool = False,
+        return_special_tokens_mask: bool = False,
+        return_offsets_mapping: bool = False,
+        return_length: bool = False,
+        verbose: bool = True,
+        **kwargs
+    ) -> BatchEncoding:
+        def get_input_ids(text):
+            if isinstance(text, str):
+                text_id = self.trie.match(text, unk_id=self.unk_token_id)
+                return text_id
+            elif isinstance(text, list) and len(text) > 0 and isinstance(text[0], str):
+                return [self.trie.match(t, unk_id=self.unk_token_id) for t in text]
+            elif isinstance(text, (list, tuple)) and len(text) > 0 and isinstance(text[0], int):
+                return text
+            else:
+                raise ValueError(
+                    "Input is not valid. Should be a string, a list/tuple of strings or a list/tuple of integers."
+                )
+
+        if return_offsets_mapping:
+            raise NotImplementedError(
+                "return_offset_mapping is not available when using Python tokenizers. "
+                "To use this feature, change your tokenizer to one deriving from "
+                "transformers.PreTrainedTokenizerFast. "
+                "More information on available tokenizers at "
+                "https://github.com/huggingface/transformers/pull/2674"
+            )
+
+        first_ids = get_input_ids(text)
+
+        return self.prepare_for_model(
+            first_ids,
+            pair_ids=None,
+            add_special_tokens=add_special_tokens,
+            padding=padding_strategy.value,
+            truncation=truncation_strategy.value,
+            max_length=max_length,
+            stride=stride,
+            pad_to_multiple_of=pad_to_multiple_of,
+            return_tensors=return_tensors,
+            prepend_batch_axis=True,
+            return_attention_mask=return_attention_mask,
+            return_token_type_ids=return_token_type_ids,
+            return_overflowing_tokens=return_overflowing_tokens,
+            return_special_tokens_mask=return_special_tokens_mask,
+            return_length=return_length,
+            verbose=verbose,
+        )
+
+    def _batch_encode_plus(
+        self,
+        batch_text_or_text_pairs: Union[
+            List[TextInput],
+            List[EncodedInput],
+        ],
+        add_special_tokens: bool = True,
+        padding_strategy: PaddingStrategy = PaddingStrategy.DO_NOT_PAD,
+        truncation_strategy: TruncationStrategy = TruncationStrategy.DO_NOT_TRUNCATE,
+        max_length: Optional[int] = None,
+        stride: int = 0,
+        pad_to_multiple_of: Optional[int] = None,
+        return_tensors: Optional[Union[str, TensorType]] = None,
+        return_token_type_ids: Optional[bool] = None,
+        return_attention_mask: Optional[bool] = None,
+        return_overflowing_tokens: bool = False,
+        return_special_tokens_mask: bool = False,
+        return_offsets_mapping: bool = False,
+        return_length: bool = False,
+        verbose: bool = True,
+        **kwargs
+    ) -> BatchEncoding:
+        def get_input_ids(text):
+            if isinstance(text, str):
+                text_id = self.trie.match(text, unk_id=self.unk_token_id)
+                return text_id
+            elif isinstance(text, list) and len(text) > 0 and isinstance(text[0], str):
+                return [self.trie.match(t, unk_id=self.unk_token_id) for t in text]
+            elif isinstance(text, (list, tuple)) and len(text) > 0 and isinstance(text[0], int):
+                return text
+            else:
+                raise ValueError(
+                    "Input is not valid. Should be a string, a list/tuple of strings or a list/tuple of integers."
+                )
+
+        if return_offsets_mapping:
+            raise NotImplementedError(
+                "return_offset_mapping is not available when using Python tokenizers. "
+                "To use this feature, change your tokenizer to one deriving from "
+                "transformers.PreTrainedTokenizerFast."
+            )
+
+        input_ids = []
+        for ids_or_pair_ids in batch_text_or_text_pairs:
+            if not isinstance(ids_or_pair_ids, (list, tuple)):
+                ids, pair_ids = ids_or_pair_ids, None
+            else:
+                ids, pair_ids = ids_or_pair_ids
+
+            first_ids = get_input_ids(ids)
+            second_ids = get_input_ids(pair_ids) if pair_ids is not None else None
+            input_ids.append((first_ids, second_ids))
+
+        batch_outputs = self._batch_prepare_for_model(
+            input_ids,
+            add_special_tokens=add_special_tokens,
+            padding_strategy=padding_strategy,
+            truncation_strategy=truncation_strategy,
+            max_length=max_length,
+            stride=stride,
+            pad_to_multiple_of=pad_to_multiple_of,
+            return_attention_mask=return_attention_mask,
+            return_token_type_ids=return_token_type_ids,
+            return_overflowing_tokens=return_overflowing_tokens,
+            return_special_tokens_mask=return_special_tokens_mask,
+            return_length=return_length,
+            return_tensors=return_tensors,
+            verbose=verbose,
+        )
+
+        return BatchEncoding(batch_outputs)
+
+    def _build_conversation_input_ids(self, conversation: "Conversation") -> List[int]:
+        input_ids = []
+        for is_user, text in conversation.iter_texts():
+            input_ids.extend(self.encode(text, add_special_tokens=False) + [self.eos_token_id])
+        if len(input_ids) > self.model_max_length:
+            input_ids = input_ids[-self.model_max_length:]
+        return input_ids

tokenizer_config.json

@@ -0,0 +1,12 @@
+{
+  "name_or_path": "rwkv-world",
+  "add_prefix_space": false,
+  "tokenizer_class": "RWKVWorldTokenizer",
+  "use_fast": false,
+  "auto_map": {
+    "AutoTokenizer": [
+      "tokenization_rwkv_world.RWKVWorldTokenizer",
+      null
+      ]
+  }
+}