Update app.py

app.py

@@ -1,5 +1,6 @@
 # -*- coding: utf-8 -*-
 import os
+import gc
 import gradio as gr
 import requests
 import time
@@ -11,173 +12,220 @@ from tokenizers import Tokenizer
 from langdetect import detect, DetectorFactory
 from PIL import Image
 from datetime import datetime
+from concurrent.futures import ThreadPoolExecutor
 
 # Για επαναληψιμότητα στο langdetect
 DetectorFactory.seed = 0
 
-# Ρυθμίσεις checkpointing και αποθήκευσης του tokenizer
-CHECKPOINT_FILE = "/persistent/checkpoint.txt" if os.path.exists("/persistent") else "checkpoint.txt"
-TOKENIZER_DIR = "tokenizer_model"
+# Ρυθμίσεις
+CHECKPOINT_FILE = "checkpoint.txt"
+TOKENIZER_DIR = os.getcwd()  # Χρησιμοποιεί τον τρέχοντα φάκελο
+#TOKENIZER_DIR = "tokenizer_model"
 TOKENIZER_FILE = os.path.join(TOKENIZER_DIR, "tokenizer.json")
-MAX_SAMPLES = 3000000  # Όριο δειγμάτων
+MAX_SAMPLES = 5000000  # Αυξημένο όριο δειγμάτων
+DEFAULT_CHUNK_SIZE = 200000  # Μεγαλύτερο chunk size
+BATCH_SIZE = 1000  # Μέγεθος batch για φόρτωση δεδομένων
+NUM_WORKERS = 4    # Αριθμός workers για πολυνηματική επεξεργασία
 
-# Παγκόσμια μεταβλητή ελέγχου συλλογής
+# Παγκόσμια μεταβλητή ελέγχου
 STOP_COLLECTION = False
 
 # Καταγραφή εκκίνησης
 startup_log = f"===== Application Startup at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} =====\n"
 print(startup_log)
 
-
 def load_checkpoint():
-    """Φόρτωση δεδομένων από το checkpoint αν υπάρχει."""
+    """Φόρτωση δεδομένων από το checkpoint."""
     if os.path.exists(CHECKPOINT_FILE):
         with open(CHECKPOINT_FILE, "r", encoding="utf-8") as f:
             return f.read().splitlines()
     return []
 
-
 def append_to_checkpoint(texts):
-    """Αποθήκευση δεδομένων στο αρχείο checkpoint."""
+    """Αποθήκευση δεδομένων με ομαδοποίηση."""
     with open(CHECKPOINT_FILE, "a", encoding="utf-8") as f:
-        for t in texts:
-            f.write(t + "\n")
-
+        batch = "\n".join(texts) + "\n"
+        f.write(batch)
 
 def create_iterator(dataset_name, configs, split):
-    """Φορτώνει το dataset και αποδίδει τα κείμενα ως iterator."""
+    """Βελτιωμένο iterator με batch φόρτωση και caching."""
     configs_list = [c.strip() for c in configs.split(",") if c.strip()]
+    
     for config in configs_list:
         try:
-            dataset = load_dataset(dataset_name, name=config, split=split, streaming=True)
-            for example in dataset:
-                text = example.get('text', '')
-                if text:
-                    yield text
+            dataset = load_dataset(
+                dataset_name,
+                name=config,
+                split=split,
+                streaming=True,
+                cache_dir="./dataset_cache"  # Ενεργοποίηση cache
+            )
+            
+            # Φόρτωση δεδομένων σε batches
+            while True:
+                batch = list(dataset.take(BATCH_SIZE))
+                if not batch:
+                    break
+                dataset = dataset.skip(BATCH_SIZE)
+                
+                # Πολυνηματική επεξεργασία batch
+                with ThreadPoolExecutor(max_workers=NUM_WORKERS) as executor:
+                    processed_texts = list(executor.map(process_example, batch))
+                
+                yield from filter(None, processed_texts)
+                
         except Exception as e:
-            print(f"⚠️ Σφάλμα φόρτωσης dataset για config {config}: {e}")
-
-
-def collect_samples(dataset_name, configs, split, chunk_size):
-    """Ξεκινά τη συλλογή δειγμάτων από το dataset."""
+            print(f"⚠️ Σφάλμα φόρτωσης: {config}: {e}")
+
+def process_example(example):
+    """Επεξεργασία ενός παραδείγματος με έλεγχο γλώσσας."""
+    try:
+        text = example.get('text', '').strip()
+        if text and detect(text) in ['el', 'en']:  # Φιλτράρισμα γλώσσας
+            return text
+        return None
+    except:
+        return None
+
+def collect_samples(dataset_name, configs, split, chunk_size, max_samples):
+    """Βελτιωμένη συλλογή δεδομένων με μεγάλα chunks."""
     global STOP_COLLECTION
     STOP_COLLECTION = False
     total_processed = len(load_checkpoint())
-    progress_messages = [f"🚀 Ξεκινά η συλλογή δεδομένων... Υπάρχουν ήδη {total_processed} δείγματα στο checkpoint."]
-    print(progress_messages[-1])
-
+    
+    progress_messages = [
+        f"🚀 Εκκίνηση συλλογής... Πρόοδος: {total_processed}/{max_samples}",
+        f"⚙️ Ρυθμίσεις: Chunk Size={chunk_size}, Workers={NUM_WORKERS}"
+    ]
+    
     dataset_iterator = create_iterator(dataset_name, configs, split)
-    new_texts = []
-    buffer_size = 100_000  # Προσωρινή αποθήκευση RAM
-
-    for text in dataset_iterator:
-        if STOP_COLLECTION:
-            progress_messages.append("⏹️ Η συλλογή σταμάτησε από το χρήστη.")
-            print(progress_messages[-1])
+    chunk = []
+    
+    while not STOP_COLLECTION and total_processed < max_samples:
+        try:
+            # Φόρτωση chunk
+            while len(chunk) < chunk_size:
+                text = next(dataset_iterator)
+                if text:
+                    chunk.append(text)
+                    total_processed += 1
+                    if total_processed >= max_samples:
+                        break
+        
+            # Αποθήκευση chunk
+            if chunk:
+                append_to_checkpoint(chunk)
+                progress_messages.append(
+                    f"✅ Αποθηκεύτηκαν {len(chunk)} δείγματα (Σύνολο: {total_processed})"
+                )
+                chunk = []
+                
+                # Εκκαθάριση μνήμης
+                gc.collect()
+                
+        except StopIteration:
+            progress_messages.append("🏁 Ολοκληρώθηκε η επεξεργασία όλων των δεδομένων!")
             break
-
-        new_texts.append(text)
-        total_processed += 1
-
-        if len(new_texts) >= buffer_size:
-            append_to_checkpoint(new_texts)
-            progress_messages.append(f"✅ Αποθηκεύτηκαν {total_processed} δείγματα στο checkpoint.")
-            print(progress_messages[-1])
-            new_texts = []
-
-        if total_processed >= MAX_SAMPLES:
-            progress_messages.append("⚠️ Έφτασε το όριο δειγμάτων.")
-            print(progress_messages[-1])
+        except Exception as e:
+            progress_messages.append(f"⛔ Σφάλμα: {str(e)}")
             break
-
-        # Ενημέρωση του progress κάθε 10.000 δείγματα
-        if total_processed % 10_000 == 0:
-            yield f"📊 Συλλογή σε εξέλιξη: {total_processed} δείγματα..."
-
-    if new_texts:
-        append_to_checkpoint(new_texts)
-        progress_messages.append(f"✅ Τελικό batch αποθηκεύτηκε ({total_processed} δείγματα).")
-        print(progress_messages[-1])
-
-    yield "\n".join(progress_messages)
-
+    
+    return "\n".join(progress_messages)
 
 def train_tokenizer_fn(dataset_name, configs, split, vocab_size, min_freq, test_text):
-    """Εκπαιδεύει τον tokenizer χρησιμοποιώντας τα δεδομένα του checkpoint."""
-    print("🚀 Ξεκινά η διαδικασία εκπαίδευσης...")
+    """Βελτιωμένη εκπαίδευση tokenizer με χρήση cache."""
+    print("🚀 Εκκίνηση εκπαίδευσης...")
     all_texts = load_checkpoint()
-    tokenizer = train_tokenizer(all_texts, vocab_size, min_freq, TOKENIZER_DIR)
-
-    # Φόρτωση εκπαιδευμένου tokenizer
+    
+    # Παράλληλη επεξεργασία για εκπαίδευση
+    tokenizer = train_tokenizer(
+        all_texts,
+        vocab_size=vocab_size,
+        min_frequency=min_freq,
+        output_dir=TOKENIZER_DIR,
+        num_threads=NUM_WORKERS  # Παράλληλη επεξεργασία
+    )
+    
+    # Φόρτωση και δοκιμή tokenizer
     trained_tokenizer = Tokenizer.from_file(TOKENIZER_FILE)
-
-    # Δοκιμή
     encoded = trained_tokenizer.encode(test_text)
     decoded = trained_tokenizer.decode(encoded.ids)
-
-    # Γράφημα κατανομής tokens
-    token_lengths = [len(t) for t in encoded.tokens]
-    fig = plt.figure()
-    plt.hist(token_lengths, bins=20)
-    plt.xlabel('Μήκος Token')
-    plt.ylabel('Συχνότητα')
-
-    # Αποθήκευση και μετατροπή εικόνας
+    
+    # Δημιουργία γραφήματος
+    fig, ax = plt.subplots()
+    ax.hist([len(t) for t in encoded.tokens], bins=20)
+    ax.set_xlabel('Μήκος Token')
+    ax.set_ylabel('Συχνότητα')
     img_buffer = BytesIO()
     plt.savefig(img_buffer, format='png')
     plt.close()
-
-    img_buffer.seek(0)
-    img = Image.open(img_buffer)
-
-    print("✅ Εκπαίδευση ολοκληρώθηκε! Το tokenizer αποθηκεύτηκε στο tokenizer_model.")
-    return ("✅ Εκπαίδευση ολοκληρώθηκε!", decoded, img)
-
-
-def stop_collection():
-    """Σταματά τη συλλογή δειγμάτων."""
-    global STOP_COLLECTION
-    STOP_COLLECTION = True
-    print("⏹️ Η συλλογή σταμάτησε από το χρήστη.")
-    return "⏹️ Η συλλογή σταμάτησε από το χρήστη."
-
-
-def restart_collection():
-    """Διαγράφει το checkpoint και επανεκκινεί τη συλλογή."""
-    global STOP_COLLECTION
-    STOP_COLLECTION = False
-    if os.path.exists(CHECKPOINT_FILE):
-        os.remove(CHECKPOINT_FILE)
-    print("🔄 Το checkpoint διαγράφηκε. Έτοιμο για νέα συλλογή.")
-    return "🔄 Το checkpoint διαγράφηκε. Έτοιμο για νέα συλλογή."
-
+    
+    return ("✅ Εκπαίδευση ολοκληρώθηκε!", decoded, Image.open(img_buffer))
+    print(f"Ο tokenizer αποθηκεύτηκε στον φάκελο: {TOKENIZER_DIR}")
+
+def analyze_checkpoint():
+    """Νέα λειτουργία ανάλυσης δεδομένων."""
+    texts = load_checkpoint()
+    if not texts:
+        return "Δεν βρέθηκαν δεδομένα για ανάλυση."
+    
+    # Βασική στατιστική
+    total_chars = sum(len(t) for t in texts)
+    avg_length = total_chars / len(texts) if texts else 0
+    
+    # Ανάλυση γλώσσας
+    languages = {}
+    for t in texts[:1000]:  # Δειγματοληψία για ταχύτητα
+        try:
+            lang = detect(t)
+            languages[lang] = languages.get(lang, 0) + 1
+        except:
+            continue
+    
+    report = [
+        f"📊 Σύνολο δειγμάτων: {len(texts)}",
+        f"📝 Μέσο μήκος: {avg_length:.1f} χαρακτήρες",
+        "🌍 Γλώσσες (δείγμα 1000):",
+        *[f"- {k}: {v} ({v/10:.1f}%)" for k, v in languages.items()]
+    ]
+    
+    return "\n".join(report)
 
 # Gradio Interface
 with gr.Blocks() as demo:
-    gr.Markdown("## Wikipedia Tokenizer Trainer with Logs & Control")
-
+    gr.Markdown("## Βελτιωμένος Wikipedia Tokenizer Trainer")
+    
     with gr.Row():
-        with gr.Column():
-            dataset_name = gr.Textbox(value="wikimedia/wikipedia", label="Dataset Name")
-            configs = gr.Textbox(value="20231101.el,20231101.en", label="Configs")
-            split = gr.Dropdown(choices=["train"], value="train", label="Split")
-            chunk_size = gr.Slider(500, 50000, value=10000, label="Chunk Size")  # Μείωση chunk_size
-            vocab_size = gr.Slider(20000, 100000, value=50000, label="Vocabulary Size")
+        with gr.Column(scale=2):
+            dataset_name = gr.Textbox(value="wikimedia/wikipedia", label="Dataset")
+            configs = gr.Textbox(value="20231101.el,20231101.en", label="Configurations")
+            split = gr.Dropdown(["train"], value="train", label="Split")
+            chunk_size = gr.Slider(10000, 500000, value=200000, step=10000, label="Chunk Size")
+            vocab_size = gr.Slider(20000, 200000, value=50000, step=10000, label="Vocabulary Size")
             min_freq = gr.Slider(1, 100, value=3, label="Minimum Frequency")
             test_text = gr.Textbox(value="Η Ακρόπολη είναι σύμβολο της αρχαίας Ελλάδας.", label="Test Text")
-            start_btn = gr.Button("Start Collection")
-            stop_btn = gr.Button("Stop Collection")
-            restart_btn = gr.Button("Restart Collection")
-            train_btn = gr.Button("Train Tokenizer")
-
-        progress = gr.Textbox(label="Progress", interactive=False, lines=10)
-        decoded_text = gr.Textbox(label="Decoded Text", interactive=False)
-        token_distribution = gr.Image(label="Token Distribution")
-
-    start_btn.click(collect_samples, [dataset_name, configs, split, chunk_size], progress)
-    stop_btn.click(stop_collection, [], progress)
-    restart_btn.click(restart_collection, [], progress)
-    train_btn.click(train_tokenizer_fn, [dataset_name, configs, split, vocab_size, min_freq, test_text], 
-                    [progress, decoded_text, token_distribution])
-
-demo.launch()
+            max_samples = gr.Slider(10000, 10000000, value=5000000, step=100000, label="Maximum Samples")
+            
+            with gr.Row():
+                start_btn = gr.Button("Start", variant="primary")
+                stop_btn = gr.Button("Stop", variant="stop")
+                restart_btn = gr.Button("Restart")
+            
+            analyze_btn = gr.Button("Analyze Data")
+            train_btn = gr.Button("Train Tokenizer", variant="primary")
+        
+        with gr.Column(scale=3):
+            progress = gr.Textbox(label="Πρόοδος", lines=10, interactive=False)
+            gr.Markdown("### Αποτελέσματα")
+            decoded_text = gr.Textbox(label="Αποκωδικοποιημένο Κείμενο")
+            token_distribution = gr.Image(label="Κατανομή Tokens")
+
+    # Event handlers
+    start_btn.click(collect_samples, [dataset_name, configs, split, chunk_size, max_samples], progress)
+    stop_btn.click(lambda: "⏹️ Διακοπή συλλογής...", None, progress, queue=False)
+    restart_btn.click(lambda: "🔄 Επαναφορά...", None, progress).then(restart_collection, None, progress)
+    analyze_btn.click(analyze_checkpoint, None, progress)
+    train_btn.click(train_tokenizer_fn, [dataset_name, configs, split, vocab_size, min_freq, test_text],
+                   [progress, decoded_text, token_distribution])
+
+demo.queue(concurrency_count=4).launch()