tokenizer.py 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152
  1. # The implementation is adopted from CLIP, made publicly available
  2. # under MIT License at https://github.com/openai/CLIP
  3. import gzip
  4. import html
  5. from functools import lru_cache
  6. import ftfy
  7. import regex as re
  8. import torch
  9. @lru_cache()
  10. def bytes_to_unicode():
  11. bs = list(range(ord('!'),
  12. ord('~') + 1)) + list(range(
  13. ord('¡'),
  14. ord('¬') + 1)) + list(range(ord('®'),
  15. ord('ÿ') + 1))
  16. cs = bs[:]
  17. n = 0
  18. for b in range(2**8):
  19. if b not in bs:
  20. bs.append(b)
  21. cs.append(2**8 + n)
  22. n += 1
  23. cs = [chr(n) for n in cs]
  24. return dict(zip(bs, cs))
  25. def get_pairs(word):
  26. pairs = set()
  27. prev_char = word[0]
  28. for char in word[1:]:
  29. pairs.add((prev_char, char))
  30. prev_char = char
  31. return pairs
  32. def basic_clean(text):
  33. text = ftfy.fix_text(text)
  34. text = html.unescape(html.unescape(text))
  35. return text.strip()
  36. def whitespace_clean(text):
  37. text = re.sub(r'\s+', ' ', text)
  38. text = text.strip()
  39. return text
  40. class SimpleTokenizer(object):
  41. def __init__(self, bpe_path):
  42. self.byte_encoder = bytes_to_unicode()
  43. self.byte_decoder = {v: k for k, v in self.byte_encoder.items()}
  44. merges = gzip.open(bpe_path).read().decode('utf-8').split('\n')
  45. merges = merges[1:49152 - 256 - 2 + 1]
  46. merges = [tuple(merge.split()) for merge in merges]
  47. vocab = list(bytes_to_unicode().values())
  48. vocab = vocab + [v + '</w>' for v in vocab]
  49. for merge in merges:
  50. vocab.append(''.join(merge))
  51. vocab.extend(['<|startoftext|>', '<|endoftext|>'])
  52. self.encoder = dict(zip(vocab, range(len(vocab))))
  53. self.decoder = {v: k for k, v in self.encoder.items()}
  54. self.bpe_ranks = dict(zip(merges, range(len(merges))))
  55. self.cache = {
  56. '<|startoftext|>': '<|startoftext|>',
  57. '<|endoftext|>': '<|endoftext|>'
  58. }
  59. self.pat = re.compile(
  60. r"""<\|startoftext\|>|<\|endoftext\|>|'s|'t|'re|'ve|'m|'ll|'d|[\p{L}]+|[\p{N}]|[^\s\p{L}\p{N}]+""",
  61. re.IGNORECASE)
  62. def bpe(self, token):
  63. if token in self.cache:
  64. return self.cache[token]
  65. word = tuple(token[:-1]) + (token[-1] + '</w>', )
  66. pairs = get_pairs(word)
  67. if not pairs:
  68. return token + '</w>'
  69. while True:
  70. bigram = min(
  71. pairs, key=lambda pair: self.bpe_ranks.get(pair, float('inf')))
  72. if bigram not in self.bpe_ranks:
  73. break
  74. first, second = bigram
  75. new_word = []
  76. i = 0
  77. while i < len(word):
  78. try:
  79. j = word.index(first, i)
  80. new_word.extend(word[i:j])
  81. i = j
  82. except ValueError:
  83. new_word.extend(word[i:])
  84. break
  85. if word[i] == first and i < len(word) - 1 and word[
  86. i + 1] == second:
  87. new_word.append(first + second)
  88. i += 2
  89. else:
  90. new_word.append(word[i])
  91. i += 1
  92. new_word = tuple(new_word)
  93. word = new_word
  94. if len(word) == 1:
  95. break
  96. else:
  97. pairs = get_pairs(word)
  98. word = ' '.join(word)
  99. self.cache[token] = word
  100. return word
  101. def encode(self, text):
  102. bpe_tokens = []
  103. text = whitespace_clean(basic_clean(text)).lower()
  104. for token in re.findall(self.pat, text):
  105. token = ''.join(self.byte_encoder[b]
  106. for b in token.encode('utf-8'))
  107. bpe_tokens.extend(self.encoder[bpe_token]
  108. for bpe_token in self.bpe(token).split(' '))
  109. return bpe_tokens
  110. def decode(self, tokens):
  111. text = ''.join([self.decoder[token] for token in tokens])
  112. text = bytearray([self.byte_decoder[c] for c in text]).decode(
  113. 'utf-8', errors='replace').replace('</w>', ' ')
  114. return text
  115. def tokenize(self, texts, context_length=77):
  116. if isinstance(texts, str):
  117. texts = [texts]
  118. sot_token = self.encoder['<|startoftext|>']
  119. eot_token = self.encoder['<|endoftext|>']
  120. all_tokens = [[sot_token] + self.encode(text) + [eot_token]
  121. for text in texts]
  122. result = torch.zeros(len(all_tokens), context_length, dtype=torch.int)
  123. for i, tokens in enumerate(all_tokens):
  124. if len(tokens) > context_length:
  125. tokens = tokens[:context_length]
  126. tokens[-1] = eot_token
  127. result[i, :len(tokens)] = torch.tensor(tokens)
  128. return result