BERT.py

# -*- coding: utf-8 -*-
"""Camembert_Bio_FINAL.ipynb

Automatically generated by Colab.

Original file is located at
    https://colab.research.google.com/drive/1gPrQNq4SXFxbzJFYcqJnh4v5hnLB9P08
"""

!pip install optuna
import pandas as pd
from transformers import CamembertTokenizer, CamembertForSequenceClassification, Trainer, TrainingArguments
from sklearn.preprocessing import LabelEncoder
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report, confusion_matrix, ConfusionMatrixDisplay, f1_score
import matplotlib.pyplot as plt
import torch
import re
import os
import numpy as np
import optuna
os.environ["WANDB_DISABLED"] = "true"

# Chargement des données
data_file = '/content/global_table.tsv'
df = pd.read_csv(data_file, sep="\t")
df = df.dropna(subset=['text', 'speaker'])
df.loc[(df['speaker'] == 'patient_1') | (df['speaker'] == 'patient_2'), 'speaker'] = 'patient'
df['text'] = df['text'].apply(lambda text: re.sub(r'\d+', '', re.sub(r'[^\w\s]', '', text.lower())))

# Encodage des étiquettes
label_encoder = LabelEncoder()
df['speaker_encoded'] = label_encoder.fit_transform(df['speaker'])

# Séparation des ensembles : train, validation, test (70%/15%/15%)
X_train, X_temp, y_train, y_temp = train_test_split(
    df['text'], df['speaker_encoded'], test_size=0.3, random_state=42)
X_val, X_test, y_val, y_test = train_test_split(
    X_temp, y_temp, test_size=0.5, random_state=42)

# Chargement du tokenizer
tokenizer = CamembertTokenizer.from_pretrained('medkit/CamemBert-Bio-Generalized')

# Encodage des textes
train_encodings = tokenizer(list(X_train), truncation=True, padding=True, max_length=128)
val_encodings = tokenizer(list(X_val), truncation=True, padding=True, max_length=128)
test_encodings = tokenizer(list(X_test), truncation=True, padding=True, max_length=128)

# Création de la classe Dataset
class Dataset(torch.utils.data.Dataset):
    def __init__(self, encodings, labels):
        self.encodings = encodings
        self.labels = labels

    def __getitem__(self, idx):
        item = {key: torch.tensor(val[idx]) for key, val in self.encodings.items()}
        item['labels'] = torch.tensor(self.labels[idx])
        return item

    def __len__(self):
        return len(self.labels)

# Création des datasets
train_dataset = Dataset(train_encodings, list(y_train))
val_dataset = Dataset(val_encodings, list(y_val))
test_dataset = Dataset(test_encodings, list(y_test))

# Fonction pour calculer le F1-score pondéré
def compute_metrics(pred):
    labels = pred.label_ids
    preds = np.argmax(pred.predictions, axis=1)
    f1 = f1_score(labels, preds, average="weighted")
    return {"eval_f1_weighted": f1}

# Définition de la fonction d'optimisation avec Optuna
def objective(trial):
    # Définition des hyperparamètres à optimiser
    learning_rate = trial.suggest_loguniform('learning_rate', 1e-5, 5e-4)
    num_train_epochs = trial.suggest_int('num_train_epochs', 3, 7)
    per_device_train_batch_size = trial.suggest_categorical('per_device_train_batch_size', [8, 16, 32])
    weight_decay = trial.suggest_loguniform('weight_decay', 1e-5, 0.1)

    # Chargement du modèle
    model = CamembertForSequenceClassification.from_pretrained(
        'medkit/CamemBert-Bio-Generalized',
        num_labels=len(label_encoder.classes_))

    # Définition des arguments d'entraînement
    training_args = TrainingArguments(
        output_dir='./optuna_results',
        evaluation_strategy="epoch",
        save_strategy="epoch",
        learning_rate=learning_rate,
        per_device_train_batch_size=per_device_train_batch_size,
        per_device_eval_batch_size=16,
        num_train_epochs=num_train_epochs,
        weight_decay=weight_decay,
        logging_dir='./logs',
        logging_steps=10,
        load_best_model_at_end=True,
        metric_for_best_model="eval_loss",
        greater_is_better=True
    )

    # Définition du Trainer
    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=train_dataset,
        eval_dataset=val_dataset,
        tokenizer=tokenizer,
        compute_metrics=compute_metrics
    )

    # Entraînement du modèle
    trainer.train()

    # Évaluation sur l'ensemble de validation
    eval_result = trainer.evaluate()

    return eval_result["eval_f1_weighted"]

# Exécution de l'optimisation Optuna pour maximiser le F1-score
study = optuna.create_study(direction="maximize")
study.optimize(objective, n_trials=10)

# Affichage des meilleurs hyperparamètres trouvés
best_params = study.best_params
print("\nMeilleurs hyperparamètres trouvés :")
print(best_params)

# Entraînement final avec les meilleurs hyperparamètres
final_model = CamembertForSequenceClassification.from_pretrained(
    'medkit/CamemBert-Bio-Generalized',
    num_labels=len(label_encoder.classes_)
)

final_training_args = TrainingArguments(
    output_dir='./final_results',
    evaluation_strategy="epoch",
    save_strategy="epoch",
    learning_rate=best_params['learning_rate'],
    per_device_train_batch_size=best_params['per_device_train_batch_size'],
    per_device_eval_batch_size=16,
    num_train_epochs=best_params['num_train_epochs'],
    weight_decay=best_params['weight_decay'],
    logging_dir='./logs',
    logging_steps=10,
    load_best_model_at_end=True,
    metric_for_best_model="eval_loss",
    greater_is_better=False
)

final_trainer = Trainer(
    model=final_model,
    args=final_training_args,
    train_dataset=train_dataset,
    eval_dataset=val_dataset,
    tokenizer=tokenizer,
    compute_metrics=compute_metrics
)

final_trainer.train()

# Évaluation sur l'ensemble de validation
val_predictions = final_trainer.predict(val_dataset)

# Extraction des logits et des labels réels
val_logits = val_predictions.predictions
val_labels = val_predictions.label_ids

# Conversion des logits en prédictions de classes
val_preds = np.argmax(val_logits, axis=1)

# Décodage des étiquettes pour affichage dans le rapport de classification
val_labels_decoded = label_encoder.inverse_transform(val_labels)
val_preds_decoded = label_encoder.inverse_transform(val_preds)

# Affichage du rapport de classification
print("\nRapport de classification pour l'ensemble de validation :")
print(classification_report(val_labels_decoded, val_preds_decoded, target_names=label_encoder.classes_))

# Prédictions sur l'ensemble de test
test_predictions = final_trainer.predict(test_dataset)

# Extraction des logits et des labels réels
test_logits = test_predictions.predictions
test_labels = test_predictions.label_ids

# Conversion des logits en prédictions de classes
test_preds = np.argmax(test_logits, axis=1)

# Décodage des étiquettes pour affichage dans le rapport de classification
test_labels_decoded = label_encoder.inverse_transform(test_labels)
test_preds_decoded = label_encoder.inverse_transform(test_preds)

# Affichage du rapport de classification
print("\nRapport de classification pour l'ensemble de test :")
print(classification_report(test_labels_decoded, test_preds_decoded, target_names=label_encoder.classes_))

# Matrice de confusion
cm = confusion_matrix(test_labels_decoded, test_preds_decoded, labels=label_encoder.classes_)
disp = ConfusionMatrixDisplay(confusion_matrix=cm, display_labels=label_encoder.classes_)
disp.plot(cmap=plt.cm.Blues)
plt.title("Matrice de confusion - Ensemble de test avec camembert bio")
plt.show()

import seaborn as sns
from sklearn.manifold import TSNE

# Réduction de dimension avec TSNE
features_tsne = TSNE(n_components=2, random_state=42).fit_transform(test_predictions.predictions)

# Création du DataFrame
df_tsne = pd.DataFrame(features_tsne, columns=["Dim 1", "Dim 2"])
df_tsne["Speaker"] = label_encoder.inverse_transform(test_predictions.label_ids)  # Décodage des labels

# Visualisation
plt.figure(figsize=(9, 6))
sns.scatterplot(data=df_tsne, x="Dim 1", y="Dim 2", hue="Speaker", palette="Set1", alpha=0.7, edgecolor="black")
plt.title("Visualisation TSNE des Prédictions", fontsize=14)
plt.legend(title="Speaker", bbox_to_anchor=(1.05, 1), loc="upper left")
plt.tight_layout()
plt.show()

from sklearn.discriminant_analysis import LinearDiscriminantAnalysis

# LDA (1D)
def lda_sklearn(X, y):
    return LinearDiscriminantAnalysis(n_components=1).fit_transform(X, y).flatten()

# Chargement des prédictions et labels
X_test, y_test = test_predictions.predictions, test_predictions.label_ids
labels = label_encoder.inverse_transform(y_test)

# Réduction de dimension avec LDA
features_lda = lda_sklearn(X_test, y_test)

# Visualisation LDA (1D avec stripplot)
plt.figure(figsize=(8, 3))
sns.stripplot(x=features_lda, y=[""] * len(features_lda), hue=labels, palette="Set1", jitter=True, alpha=0.7)
plt.title("Visualisation LDA")
plt.legend(title="Speaker", bbox_to_anchor=(1.05, 1), loc="upper left")
plt.tight_layout()
plt.show()

import pandas as pd

# Identification des indices des erreurs de classification
misclassified_indices = np.where(test_labels != test_preds)[0]

# Exctraction des phrases mal classées et leurs prédictions
misclassified_texts = X_test.iloc[misclassified_indices]  # Récupération des phrases originales
misclassified_true_labels = label_encoder.inverse_transform(test_labels[misclassified_indices])
misclassified_pred_labels = label_encoder.inverse_transform(test_preds[misclassified_indices])

# Création d'un DataFrame pour afficher les erreurs
df_errors = pd.DataFrame({
    "Phrase": misclassified_texts,
    "Label Réel": misclassified_true_labels,
    "Label Prédit": misclassified_pred_labels
})

df_errors.to_csv('errors.tsv',index=False,sep="\t")