rwkv-4-world-7b-arabic / tokenization_rwkv_world.py

Upload 4 files

d0269df 5 months ago

No virus

18 kB

	# 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