Update tokenize_and_upload_mistral.py

tokenize_and_upload_mistral.py

@@ -16,16 +16,22 @@ MODEL_NAME = "mistralai/Mistral-7B-Instruct-v0.2"
 HF_TOKEN = os.getenv("HF_TOKEN")
 SOURCE_DATASET_ID = "UcsTurkey/turkish-general-culture-chunks"
 TRAIN_TARGET_DATASET_ID = "UcsTurkey/turkish-general-culture-tokenized"
+RAG_TARGET_DATASET_ID = "UcsTurkey/turkish-general-culture-rag"
 BUFFER_SIZE = 5
-START_CHUNK_NUMBER = 776
-PROCESS_CHUNK_COUNT = 2
+START_CHUNK_NUMBER = 0
+PROCESS_CHUNK_COUNT = 776
+
+GENERATE_TRAIN_DATA = False
+GENERATE_RAG_DATA = True
 
 CHUNK_FOLDER = "/data/chunks"
-PARQUET_FOLDER = "/data/tokenized_chunks"
+TRAIN_FOLDER = "/data/tokenized_chunks"
+RAG_FOLDER = "/data/rag_chunks"
 CACHE_DIR = "/data/.hf_cache"
 
 os.makedirs(CHUNK_FOLDER, exist_ok=True)
-os.makedirs(PARQUET_FOLDER, exist_ok=True)
+os.makedirs(TRAIN_FOLDER, exist_ok=True)
+os.makedirs(RAG_FOLDER, exist_ok=True)
 os.makedirs(CACHE_DIR, exist_ok=True)
 
 # ✅ Health check sunucusu
@@ -63,7 +69,8 @@ files = api.list_repo_files(repo_id=SOURCE_DATASET_ID, repo_type="dataset", toke
 csv_files = sorted([f for f in files if f.endswith(".csv")])
 selected_files = csv_files[START_CHUNK_NUMBER:START_CHUNK_NUMBER + PROCESS_CHUNK_COUNT]
 
-buffer_counter = 0
+buffer_counter_train = 0
+buffer_counter_rag = 0
 
 def tokenize(example):
     prompt = f"SORU: {example['instruction']}\nCEVAP: {example['output']}"
@@ -73,16 +80,16 @@ def tokenize(example):
     ]
     return tokenized
 
-def upload_if_ready():
-    global buffer_counter
-    if os.listdir(PARQUET_FOLDER):
-        log(f"⬆️ BUFFER doldu. Hugging Face'e yükleniyor: {TRAIN_TARGET_DATASET_ID}")
-        create_repo(TRAIN_TARGET_DATASET_ID, repo_type="dataset", token=HF_TOKEN, exist_ok=True)
-        upload_folder(repo_id=TRAIN_TARGET_DATASET_ID, folder_path=PARQUET_FOLDER, repo_type="dataset", token=HF_TOKEN)
+def upload_if_ready(folder_path, target_repo):
+    if os.listdir(folder_path):
+        log(f"⬆️ BUFFER doldu. Hugging Face'e yükleniyor: {target_repo}")
+        create_repo(target_repo, repo_type="dataset", token=HF_TOKEN, exist_ok=True)
+        upload_folder(repo_id=target_repo, folder_path=folder_path, repo_type="dataset", token=HF_TOKEN)
         log("🧹 Upload sonrası klasör temizleniyor...")
-        for f in os.listdir(PARQUET_FOLDER):
-            os.remove(os.path.join(PARQUET_FOLDER, f))
-        buffer_counter = 0
+        for f in os.listdir(folder_path):
+            os.remove(os.path.join(folder_path, f))
+        return 0
+    return 0
 
 for idx, filename in enumerate(selected_files):
     log(f"\n📄 {idx+1}/{len(selected_files)} → {filename} işleniyor...")
@@ -100,20 +107,33 @@ for idx, filename in enumerate(selected_files):
         df = df.rename(columns={"question": "instruction", "answer": "output"})
         log(f"✅ Geçerli satır sayısı: {len(df)}")
 
-        dataset = Dataset.from_pandas(df[["instruction", "output"]])
-        tokenized_dataset = dataset.map(tokenize)
-        parquet_path = os.path.join(PARQUET_FOLDER, filename.replace(".csv", ".parquet"))
-        tokenized_dataset.to_parquet(parquet_path, compression="snappy")
-        log(f"🎯 Tokenized parquet kaydedildi: {parquet_path}")
-        buffer_counter += 1
-        if buffer_counter >= BUFFER_SIZE:
-            upload_if_ready()
+        if GENERATE_RAG_DATA:
+            rag_dataset = Dataset.from_pandas(df[["instruction", "output"]])
+            rag_path = os.path.join(RAG_FOLDER, filename.replace(".csv", ".parquet"))
+            rag_dataset.to_parquet(rag_path, compression="brotli")
+            log(f"📦 RAG parquet kaydedildi: {rag_path}")
+            buffer_counter_rag += 1
+            if buffer_counter_rag >= BUFFER_SIZE:
+                buffer_counter_rag = upload_if_ready(RAG_FOLDER, RAG_TARGET_DATASET_ID)
+
+        if GENERATE_TRAIN_DATA:
+            train_dataset = Dataset.from_pandas(df[["instruction", "output"]])
+            tokenized_dataset = train_dataset.map(tokenize)
+            parquet_path = os.path.join(TRAIN_FOLDER, filename.replace(".csv", ".parquet"))
+            tokenized_dataset.to_parquet(parquet_path, compression="snappy")
+            log(f"🎯 Tokenized parquet kaydedildi: {parquet_path}")
+            buffer_counter_train += 1
+            if buffer_counter_train >= BUFFER_SIZE:
+                buffer_counter_train = upload_if_ready(TRAIN_FOLDER, TRAIN_TARGET_DATASET_ID)
     except Exception as e:
         log(f"❌ Hata oluştu: {filename} → {e}")
         traceback.print_exc()
         continue
 
-upload_if_ready()
+if GENERATE_TRAIN_DATA:
+    buffer_counter_train = upload_if_ready(TRAIN_FOLDER, TRAIN_TARGET_DATASET_ID)
+if GENERATE_RAG_DATA:
+    buffer_counter_rag = upload_if_ready(RAG_FOLDER, RAG_TARGET_DATASET_ID)
 
 log("✅ Tüm işlemler tamamlandı. Servis bekleme modunda...")
 while True: