Upload baoule_tokenizer.py with huggingface_hub

baoule_tokenizer.py

@@ -0,0 +1,414 @@
+
+import unicodedata
+import regex as re
+from datasets import load_dataset
+import time
+import os
+
+def get_stats(ids, stats=None):
+    """
+    Calcule la fréquence des paires d'ids consécutifs.
+    Conserve la même logique que la version originale car cette fonction est indépendante 
+    des spécificités de la langue.
+    """
+    stats = {} if stats is None else stats
+    for pair in zip(ids, ids[1:]): 
+        stats[pair] = stats.get(pair, 0) + 1
+    return stats
+
+def merge(ids, pair, idx):
+    """
+    Fusionne les paires d'ids identifiées.
+    Conserve la même logique que la version originale car cette fonction gère
+    uniquement la fusion des tokens déjà identifiés.
+    """
+    newids = []
+    i = 0
+    while i < len(ids):
+        if ids[i] == pair[0] and i < len(ids) - 1 and ids[i+1] == pair[1]:
+            newids.append(idx)
+            i += 2
+        else:
+            newids.append(ids[i])
+            i += 1
+    return newids
+
+def replace_control_characters(s: str) -> str:
+    """
+    Remplace les caractères de contrôle, avec une attention particulière aux 
+    caractères spéciaux du baoulé.
+    """
+    chars = []
+    for ch in s:
+        # Gestion spéciale des caractères baoulé
+        if ch in ['ɛ', 'ɔ', 'ŋ', 'ɲ']:  # Caractères spéciaux baoulé
+            chars.append(ch)
+        # Gestion standard des caractères de contrôle
+        elif unicodedata.category(ch)[0] != "C":
+            chars.append(ch)
+        else:
+            chars.append(f"\u{ord(ch):04x}")
+    return "".join(chars)
+
+def render_token(t: bytes) -> str:
+    """
+    Décode les tokens en gérant les caractères spéciaux du baoulé.
+    """
+    try:
+        # Tentative de décodage standard
+        s = t.decode('utf-8', errors='replace')
+        # Gestion des caractères spéciaux baoulé
+        s = replace_control_characters(s)
+        return s
+    except UnicodeDecodeError:
+        # En cas d'échec, retourne le caractère de remplacement
+        return '�'
+
+
+
+class BaouleTokenizer:
+    def __init__(self):
+        # Initialisation des attributs obligatoires
+        self.special_chars = {
+            'ɛ': 256,
+            'ɔ': 257,
+            'ŋ': 258,
+            'ɲ': 259
+        }
+        
+        self.pattern = r"(?i:'n|gb|kp|ny|[ɛɔ]n)|(?i:'s|'t|'re|'ve|'m|'ll|'d)|[^
+\p{L}\p{N}]?\p{L}+|\p{N}{1,3}| ?[^\s\p{L}\p{N}]+[
+]*|\s*[
+]+|\s+(?!\S)|\s+"
+        self.compiled_pattern = re.compile(self.pattern)
+        
+        self.special_tokens = {
+            #'<|baoule|>': 1101,
+            #'<|french|>': 1102,
+            #'<|end|>': 1103,
+            #'<|unknown|>': 1104,
+            #'<|pad|>': 1105,
+            # or
+            '<|begin_of_text|>': 1101,
+            '<|end_of_text|>': 1102,
+            '<|start_header_id|>': 1103,
+            '<|end_header_id|>': 1104,
+            '<|eot_id|>': 1105
+        }
+        
+        self.merges = {}
+        self.vocab = self._build_vocab()
+
+    def train(self, dataset, vocab_size):
+        assert vocab_size >= 260  # 256 ASCII + 4 caractères spéciaux baoulé minimum
+
+        # Extraction des textes baoulé du dataset HuggingFace
+        text_chunks = []
+        for item in dataset['train']:
+            chunks = re.findall(self.compiled_pattern, item['baoule'])
+            text_chunks.extend(chunks)
+
+        # Conversion en ids avec gestion spéciale des caractères baoulé
+        ids = []
+        for chunk in text_chunks:
+            chunk_ids = []
+            i = 0
+            while i < len(chunk):
+                # Vérification des digraphes
+                if i < len(chunk) - 1:
+                    digraph = chunk[i:i+2]
+                    if digraph in ['gb', 'kp', 'ny']:
+                        chunk_ids.append(ord(digraph[0]))
+                        chunk_ids.append(ord(digraph[1]))
+                        i += 2
+                        continue
+                
+                # Gestion des caractères spéciaux baoulé
+                char = chunk[i]
+                if char in self.special_chars:
+                    chunk_ids.append(self.special_chars[char])
+                else:
+                    # Encodage UTF-8 standard pour les autres caractères
+                    chunk_ids.extend(list(char.encode("utf-8")))
+                i += 1
+            ids.append(chunk_ids)
+
+        # Calcul des fusions
+        num_merges = vocab_size - (260 + len(self.special_tokens))
+        merges = {}
+        vocab = {idx: bytes([idx]) for idx in range(256)}
+        vocab.update({idx: char.encode('utf-8') for char, idx in self.special_chars.items()})
+
+        for i in range(num_merges):
+            stats = {}
+            for chunk_ids in ids:
+                get_stats(chunk_ids, stats)
+            if not stats:
+                break
+            pair = max(stats, key=stats.get)
+            idx = 260 + len(self.special_tokens) + i
+            ids = [merge(chunk_ids, pair, idx) for chunk_ids in ids]
+            merges[pair] = idx
+            vocab[idx] = vocab[pair[0]] + vocab[pair[1]]
+
+        self.merges = merges
+        self.vocab = vocab
+
+    def _build_vocab(self):
+        # Vocabulaire de base incluant ASCII et caractères spéciaux baoulé
+        vocab = {idx: bytes([idx]) for idx in range(256)}
+        vocab.update({idx: char.encode('utf-8') for char, idx in self.special_chars.items()})
+        
+        # Ajout des fusions
+        for (p0, p1), idx in self.merges.items():
+            vocab[idx] = vocab[p0] + vocab[p1]
+        
+        # Ajout des tokens spéciaux
+        for special, idx in self.special_tokens.items():
+            vocab[idx] = special.encode("utf-8")
+
+        return vocab
+
+    def save(self, file_prefix):
+        # Sauvegarde du modèle
+        model_file = file_prefix + ".model"
+        with open(model_file, 'w') as f:
+            f.write("baoule tokenizer v1.0
+")
+            f.write(f"{self.pattern}
+")
+            
+            # Sauvegarde des caractères spéciaux baoulé
+            f.write(f"{len(self.special_chars)}
+")
+            for char, idx in self.special_chars.items():
+                f.write(f"{char} {idx}
+")
+            
+            # Sauvegarde des tokens spéciaux
+            f.write(f"{len(self.special_tokens)}
+")
+            for token, idx in self.special_tokens.items():
+                f.write(f"{token} {idx}
+")
+            
+            # Sauvegarde des fusions
+            for idx1, idx2 in self.merges:
+                f.write(f"{idx1} {idx2}
+")
+
+        # Sauvegarde du vocabulaire
+        vocab_file = file_prefix + ".vocab"
+        inverted_merges = {idx: pair for pair, idx in self.merges.items()}
+        with open(vocab_file, "w", encoding="utf-8") as f:
+            for idx, token in self.vocab.items():
+                s = render_token(token)
+                if idx in inverted_merges:
+                    idx0, idx1 = inverted_merges[idx]
+                    s0 = render_token(self.vocab[idx0])
+                    s1 = render_token(self.vocab[idx1])
+                    f.write(f"[{s0}][{s1}] -> [{s}] {idx}
+")
+                else:
+                    f.write(f"[{s}] {idx}
+")
+    def load(self, model_file):
+      merges = {}
+      special_tokens = {}
+      special_chars = {}
+      
+      with open(model_file, 'r', encoding="utf-8") as f:
+          version = f.readline().strip()
+          self.pattern = f.readline().strip()
+          self.compiled_pattern = re.compile(self.pattern)
+          
+          # Lecture des caractères spéciaux baoulé
+          num_special_chars = int(f.readline().strip())
+          for _ in range(num_special_chars):
+              char, char_idx = f.readline().strip().split()
+              special_chars[char] = int(char_idx)
+          
+          # Lecture des tokens spéciaux
+          num_special = int(f.readline().strip())
+          for _ in range(num_special):
+              special, special_idx = f.readline().strip().split()
+              special_tokens[special] = int(special_idx)
+          
+          # Création du vocabulaire de base
+          base_vocab = {}
+          # Ajouter les caractères ASCII
+          for i in range(256):
+              base_vocab[i] = bytes([i])
+          # Ajouter les caractères spéciaux
+          for char, idx in special_chars.items():
+              base_vocab[idx] = char.encode('utf-8')
+          # Ajouter les tokens spéciaux
+          for token, idx in special_tokens.items():
+              base_vocab[idx] = token.encode('utf-8')
+              
+          # Lecture des fusions
+          for line in f:
+              try:
+                  idx1, idx2 = map(int, line.strip().split())
+                  if idx1 not in base_vocab or idx2 not in base_vocab:
+                      print(f"Warning: skipping fusion for indices {idx1}, {idx2} - not found in vocabulary")
+                      continue
+                  next_idx = len(base_vocab)
+                  merges[(idx1, idx2)] = next_idx
+                  base_vocab[next_idx] = base_vocab[idx1] + base_vocab[idx2]
+              except Exception as e:
+                  print(f"Error processing line: {line.strip()}")
+                  print(f"Current vocabulary keys: {sorted(base_vocab.keys())}")
+                  raise e
+
+      self.merges = merges
+      self.special_tokens = special_tokens
+      self.special_chars = special_chars
+      self.vocab = base_vocab
+      
+      return self
+
+    def encode(self, text):
+        """
+        Encode le texte baoulé en liste d'identifiants entiers.
+        Gère les caractères spéciaux baoulé et les digraphes.
+        """
+        # Pattern pour identifier les tokens spéciaux
+        special_pattern = "(" + "|".join(re.escape(k) for k in self.special_tokens) + ")"
+        special_chunks = re.split(special_pattern, text)
+        
+        ids = []
+
+        for part in special_chunks:
+            # Gestion des tokens spéciaux
+            if part in self.special_tokens:
+                ids.append(self.special_tokens[part])
+            elif part:  # Ignorer les parties vides
+                # Découpage du texte en chunks selon le pattern
+                text_chunks = re.findall(self.compiled_pattern, part)
+
+                for chunk in text_chunks:
+                    chunk_ids = []
+                    i = 0
+                    
+                    # Traitement caractère par caractère avec gestion des digraphes
+                    while i < len(chunk):
+                        # Vérification des digraphes baoulé (gb, kp, ny)
+                        if i < len(chunk) - 1:
+                            digraph = chunk[i:i+2]
+                            if digraph.lower() in ['gb', 'kp', 'ny']:
+                                chunk_ids.extend([ord(digraph[0]), ord(digraph[1])])
+                                i += 2
+                                continue
+                        
+                        # Vérification des voyelles nasales
+                        if i < len(chunk) - 1 and chunk[i+1] == 'n':
+                            current_char = chunk[i]
+                            if current_char in 'aɛiɔu':
+                                # Traiter la voyelle nasale comme une unité
+                                nasal_vowel = chunk[i:i+2]
+                                chunk_ids.extend(list(nasal_vowel.encode('utf-8')))
+                                i += 2
+                                continue
+                        
+                        # Gestion des caractères spéciaux baoulé
+                        current_char = chunk[i]
+                        if current_char in self.special_chars:
+                            chunk_ids.append(self.special_chars[current_char])
+                        else:
+                            # Encodage UTF-8 standard pour les autres caractères
+                            chunk_ids.extend(list(current_char.encode('utf-8')))
+                        i += 1
+
+                    # Application des fusions (byte-pair encoding)
+                    while len(chunk_ids) >= 2:
+                        stats = get_stats(chunk_ids)
+                        pair = min(stats, key=lambda p: self.merges.get(p, float('inf')))
+                        
+                        if pair not in self.merges:
+                            break
+                            
+                        idx = self.merges[pair]
+                        chunk_ids = merge(chunk_ids, pair, idx)
+                    
+                    ids.extend(chunk_ids)
+        
+        return ids
+
+    def decode(self, ids):
+        """
+        Décode une liste d'identifiants en texte baoulé.
+        Gère la reconstruction des caractères spéciaux et des digraphes.
+        """
+        part_bytes = []
+        inverse_special_tokens = {v: k for k, v in self.special_tokens.items()}
+        inverse_special_chars = {v: k for k, v in self.special_chars.items()}
+
+        i = 0
+        while i < len(ids):
+            current_id = ids[i]
+            
+            # Gestion des tokens spéciaux
+            if current_id in inverse_special_tokens:
+                part_bytes.append(inverse_special_tokens[current_id].encode('utf-8'))
+                i += 1
+                continue
+                
+            # Gestion des caractères spéciaux baoulé
+            if current_id in inverse_special_chars:
+                part_bytes.append(inverse_special_chars[current_id].encode('utf-8'))
+                i += 1
+                continue
+                
+            # Gestion du vocabulaire standard
+            if current_id in self.vocab:
+                # Vérification des digraphes potentiels
+                if i < len(ids) - 1:
+                    next_id = ids[i + 1]
+                    current_bytes = self.vocab[current_id]
+                    if next_id in self.vocab:
+                        next_bytes = self.vocab[next_id]
+                        combined = current_bytes + next_bytes
+                        # Vérification si c'est un digraphe baoulé
+                        try:
+                            combined_str = combined.decode('utf-8')
+                            if combined_str.lower() in ['gb', 'kp', 'ny']:
+                                part_bytes.append(combined)
+                                i += 2
+                                continue
+                        except UnicodeDecodeError:
+                            pass
+                
+                part_bytes.append(self.vocab[current_id])
+                i += 1
+            else:
+                raise ValueError(f"ID de token invalide: {current_id}")
+
+        # Reconstruction du texte final
+        text_bytes = b''.join(part_bytes)
+        text = text_bytes.decode('utf-8', errors='replace')
+        
+        return text
+
+
+
+
+# Chargement du dataset depuis HuggingFace
+dataset = load_dataset("Adjoumani/translations_french_baoule_V1")
+
+# Configuration
+vocab_size = 512
+output_dir = "./models"
+os.makedirs(output_dir, exist_ok=True)
+
+# Initialisation et entraînement
+tokenizer = BaouleTokenizer()
+start_time = time.time()
+tokenizer.train(dataset, vocab_size)
+end_time = time.time()
+
+# Sauvegarde
+tokenizer.save(f"{output_dir}/baoule_tokenizer")
+
+print(f"Durée d'entraînement: {end_time-start_time:.2f} secondes")
+print(f"Modèle sauvegardé dans: {output_dir}/baoule_tokenizer")