import os import pandas as pd from dotenv import load_dotenv from supabase import create_client, Client from evidently import Report from evidently.presets import DataDriftPreset import requests from transformers import AutoTokenizer, AutoModelForSequenceClassification, TrainingArguments, Trainer from datasets import Dataset import shutil # Carica le variabili d'ambiente dal file .env locale load_dotenv() def initialize_reference_dataset(): """ Scarica un campione dal dataset originale 'tweet_eval' usato per addestrare il modello RoBERTa e crea il file di baseline 'src/reference.csv' via HTTP. """ REFERENCE_FILE = "src/reference.csv" # Se il file esiste non fa nulla if os.path.exists(REFERENCE_FILE): return try: print("[MONITORING] Dynamically generate the reference file from the tweet_eval dataset.") # Link HTTP API ufficiale di Hugging Face # per lo split di validation (sentiment) url = "https://datasets-server.huggingface.co/rows?dataset=cardiffnlp/tweet_eval&config=sentiment&split=validation&offset=100&length=100" headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" } response = requests.get(url, headers=headers, timeout=15) response.raise_for_status() # Solleva un errore in caso di codice HTTP fallimentare data_json = response.json() # Estrae i record rows_data = [row['row'] for row in data_json['rows']] df = pd.json_normalize(rows_data) # Mappa le etichette numeriche (0,1,2) nelle stringhe # (negative, neutral, positive) # come quelle restituite dal modello label_mapping = {0: "negative", 1: "neutral", 2: "positive"} df["label"] = df["label"].map(label_mapping) # Verifica che la cartella di destinazione esiste os.makedirs(os.path.dirname(REFERENCE_FILE), exist_ok=True) # Salva i campi utili per Evidently AI senza l'indice df[["text", "label"]].to_csv(REFERENCE_FILE, index=False) print(f"[MONITORING SUCCESS] Created baseline file: {REFERENCE_FILE}") except Exception as e: print(f"[MONITORING ERROR] Unable to download tweet_eval via HTTP: {e}") def get_current_data_from_supabase() -> pd.DataFrame: """ Recupera i dati storici delle predizioni salvate nel database centrale Supabase. """ supabase_url = os.getenv("SUPABASE_URL") supabase_key = os.getenv("SUPABASE_KEY") if not supabase_url or not supabase_key: raise ValueError("[MONITORING CRITICAL] SUPABASE_URL or SUPABASE_KEY missing in .env file.") print("[MONITORING] Extracting current data from Supabase") # Inizializza il client Supabase supabase: Client = create_client(supabase_url, supabase_key) # Esegue la query sulla tabella # Estrae il testo e l'etichetta predetta response = supabase.table("feedback_logs").select("text, label").execute() if not response.data: print("[MONITORING WARNING] No data found on Supabase. Returning an empty set.") return pd.DataFrame(columns=["text", "label"]) return pd.DataFrame(response.data) def generate_drift_report(): """ Controlla la presenza del file di baseline, carica i dati correnti dal DB e calcola il Data Drift Report usando Evidently. """ # Assicura che il file di baseline esista prima del calcolo initialize_reference_dataset() REFERENCE_FILE = "src/reference.csv" # Verifica se procedere solo se il file esiste if not os.path.exists(REFERENCE_FILE): print("[MONITORING ERROR] Unable to calculate drift: reference.csv file missing.") return try: # Carica i due dataset quel dal file e quello da supabase reference_df = pd.read_csv(REFERENCE_FILE) current_df = get_current_data_from_supabase() if current_df.empty: print("[MONITORING SKIP] The current dataset is empty. Unable to calculate data drift.") return print("[MONITORING] Start data drift report calculation.") # Configurazione del Report per le API di Evidently text_report = Report(metrics=[DataDriftPreset(columns=["text"])]) # Esegue il calcolo del drift report_result = text_report.run(reference_data=reference_df, current_data=current_df) # Salva l'output in formato HTML nella cartella static os.makedirs("static", exist_ok=True) output_html = "static/drift_report.html" report_result.save_html(output_html) print(f"[MONITORING SUCCESS] Data Drift Report successfully saved to: '{output_html}'") dataset_drift = False try: # Legge il file HTML generato da Evidently per verificare l'esito with open(output_html, "r", encoding="utf-8") as f: html_content = f.read() # Se Evidently rileva un drift sul dataset, inserisce nel codice HTML # della dashboard la stringa di stato "Dataset drift detected" o "Drift detected" # Verifichiamo la stringa esatta del verdetto statistico di Evidently if "dataset drift is detected" in html_content.lower() and "is not detected" not in html_content.lower(): dataset_drift = True print("[MONITORING SUCCESS] Statistical result detected by HTML: Data Drift Present.") else: dataset_drift = False print("[MONITORING SUCCESS] Statistical result detected by HTML: No Data Drift.") except Exception as html_error: print(f"[MONITORING WARNING] Unable to inspect HTML: {html_error}") dataset_drift = False # Se c'e' un drift avvia il retraining automatico if dataset_drift: trigger_model_retraining(current_df) else: print("[MONITORING] No significant data drift detected. Retraining is not needed.") except Exception as e: print(f"[MONITORING CRASH] Critical error while processing report: {e}") def trigger_model_retraining(current_df: pd.DataFrame): """ Esegue il FINE-TUNING REALE del modello RoBERTa utilizzando i nuovi dati estratti storicamente da Supabase. """ print("[RETRAINING] Launch of the Fine-Tuning pipeline on the RoBERTa model.") MODEL_NAME = "cardiffnlp/twitter-roberta-base-sentiment-latest" OUTPUT_DIR = "src/models/fine_tuned_roberta/" TEMP_SAVE_DIR = "src/models/tmp_fine_tuned_roberta" try: # Preparazione dei dati: mappa le label testuali negli ID numerici richiesti dal modello id_mapping = {"negative": 0, "neutral": 1, "positive": 2} current_df = current_df.dropna(subset=["text", "label"]) current_df["label"] = current_df["label"].map(id_mapping) current_df = current_df.dropna(subset=["label"]) current_df["label"] = current_df["label"].astype(int) if len(current_df) < 5: print("[RETRAINING ABORT] Insufficient data on Supabase for fine-tuning (at least 5 samples needed).") return # Trasforma il DataFrame in Dataset di Hugging Face dataset = Dataset.from_pandas(current_df[["text", "label"]]) # Tokenizzazione print("[RETRAINING] Tokenization of the new dataset in progress") tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME) def tokenize_function(examples): return tokenizer(examples["text"], padding="max_length", truncation=True, max_length=128) tokenized_dataset = dataset.map(tokenize_function, batched=True) # Split Train/Test per monitorare l'addestramento split_dataset = tokenized_dataset.train_test_split(test_size=0.2, seed=42) if len(current_df) >= 10 else {"train": tokenized_dataset, "test": tokenized_dataset} # Caricamento modello base print(f"[RETRAINING] Loading the base model: {MODEL_NAME}") model = AutoModelForSequenceClassification.from_pretrained(MODEL_NAME, num_labels=3) # Configurazione iperparametri ottimizzati per CPU training_args = TrainingArguments( output_dir="./tmp_trainer", num_train_epochs=3, per_device_train_batch_size=2, per_device_eval_batch_size=2, warmup_steps=10, weight_decay=0.01, logging_dir="./logs", logging_steps=5, report_to="none", eval_strategy="epoch" if len(current_df) >= 10 else "no", save_strategy="no", use_cpu=True ) # Inizializzazione Trainer ed esecuzione addestramento trainer = Trainer( model=model, args=training_args, train_dataset=split_dataset["train"], eval_dataset=split_dataset.get("test") ) print("[RETRAINING] Start Computing Fine Tuning") trainer.train() # Salvataggio definitivo del modello ottimizzato os.makedirs(TEMP_SAVE_DIR, exist_ok=True) print("[RETRAINING] Writing new weights to the temporary directory.") trainer.save_model(TEMP_SAVE_DIR) tokenizer.save_pretrained(TEMP_SAVE_DIR) os.makedirs(OUTPUT_DIR, exist_ok=True) print("[RETRAINING] Synchronization of configuration files and weights.") for filename in os.listdir(TEMP_SAVE_DIR): src_file = os.path.join(TEMP_SAVE_DIR, filename) dst_file = os.path.join(OUTPUT_DIR, filename) try: # copy2 sovrascrive i file gestendo i flussi binari a basso livello shutil.copy2(src_file, dst_file) except Exception as copy_error: # Se un file specifico (es. il binario pesante) รจ bloccato da Uvicorn, # crea una versione contrassegnata con timestamp. Al riavvio l'API la legge. import time timestamp_filename = f"model_{int(time.time())}.safetensors" if "safetensors" in filename else f"patched_{filename}" shutil.copy2(src_file, os.path.join(OUTPUT_DIR, timestamp_filename)) # Pulizia della cartella temporanea shutil.rmtree(TEMP_SAVE_DIR, ignore_errors=True) REFERENCE_FILE = "src/reference.csv" current_df.to_csv(REFERENCE_FILE, index=False) print(f"[MLOps PIPELINE] Baseline updated with current data in '{REFERENCE_FILE}' to reset the drift calculation.") print(f"[RETRAINING SUCCESS] Fine-tuning completed successfully! Model saved in: {OUTPUT_DIR}") except Exception as training_error: print(f"[RETRAINING ERROR] Critical error during model fine-tuning: {training_error}") if __name__ == "__main__": # Consente l'esecuzione diretta del file test generate_drift_report()