add

source/services/predicting_effective_arguments/train/model.py

@@ -0,0 +1,105 @@
+from datasets import load_dataset
+from transformers import AutoTokenizer, AutoModelForSequenceClassification, Trainer, TrainingArguments, pipeline, DataCollatorWithPadding
+from sklearn.metrics import accuracy_score, f1_score
+import torch
+import numpy as np
+import torch.nn.functional as F
+import matplotlib.pyplot as plt
+from sklearn.metrics import ConfusionMatrixDisplay, confusion_matrix
+
+device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+
+
+
+class TransformersSequenceClassifier:
+    def __init__(self,
+                 model_output_dir,
+                 num_labels,
+                 tokenizer : AutoTokenizer,
+                 model_checkpoint="distilbert-base-uncased"
+                 ):
+        self.model_output_dir = model_output_dir
+        self.tokenizer = AutoTokenizer.from_pretrained(model_checkpoint)
+        self.model = AutoModelForSequenceClassification.from_pretrained(model_checkpoint, num_labels=num_labels).to(device)
+
+    def tokenizer_func(self, batch):
+        return self.tokenizer(batch["text"], truncation=True, max_len=386)
+
+    def train(self, train_dataset, eval_dataset, epochs=2, batch_size=64):
+        
+        train_tok_dataset = train_dataset.map(self.tokenizer_func, batched=True, remove_columns=('inputs', '__index_level_0__'))
+        val_tok_dataset = eval_dataset.map(self.tokenizer_func, batched=True, remove_columns=('inputs', '__index_level_0__'))
+        data_collator = DataCollatorWithPadding(tokenizer=self.tokenizer, padding='longest')
+        training_args = TrainingArguments(output_dir=self.model_output_dir,
+                                          num_train_epochs=epochs,
+                                          learning_rate=2e-5,
+                                          per_device_train_batch_size=batch_size,
+                                          per_device_eval_batch_size=batch_size,
+                                          weight_decay=0.01,
+                                          evaluation_strategy="epoch",
+                                          disable_tqdm=False,
+                                          logging_steps=len(train_dataset) // batch_size,
+                                          push_to_hub=True,
+                                          log_level="error")
+        self.trainer = Trainer(
+            model=self.model,
+            args=training_args,
+            compute_metrics=self._compute_metrics,
+            train_dataset=train_tok_dataset,
+            eval_dataset=val_tok_dataset,
+            tokenizer=self.tokenizer,
+            data_collator=data_collator
+        )
+        self.trainer.train()
+
+    @staticmethod
+    def _compute_metrics(pred):
+        labels = pred.label_ids
+        preds = pred.predictions.argmax(-1)
+        f1 = f1_score(labels, preds, average="weighted")
+        acc = accuracy_score(labels, preds)
+        return {"accuracy": acc, "f1": f1}
+
+    def forward_pass_with_label(self, batch):
+        # Place all input tensors on the same device as the model
+        inputs = {k:v.to(device) for k,v in batch.items()
+                if k in self.tokenizer.model_input_names}
+
+        with torch.no_grad():
+            output = self.model(**inputs)
+            pred_label = torch.argmax(output.logits, axis=-1)
+            loss = F.cross_entropy(output.logits, batch["label"].to(device), 
+                                reduction="none")
+
+        # Place outputs on CPU for compatibility with other dataset columns
+        return {"loss": loss.cpu().numpy(), 
+                "predicted_label": pred_label.cpu().numpy()}
+
+    def compute_loss_per_pred(self, valid_dataset):
+        # Compute loss values
+        return valid_dataset.map(self.forward_pass_with_label, batched=True, batch_size=16)
+
+    @staticmethod
+    def plot_confusion_matrix(y_preds, y_true, labels):
+        cm = confusion_matrix(y_true, y_preds, normalize="true")
+        fig, ax = plt.subplots(figsize=(6, 6))
+        disp = ConfusionMatrixDisplay(confusion_matrix=cm, display_labels=labels)
+        disp.plot(cmap="Blues", values_format=".2f", ax=ax, colorbar=False)
+        plt.title("Normalized confusion matrix")
+        plt.show()
+
+    def predict_valid_data(self, valid_dataset):
+        #trainer = Trainer(model=self.model)
+        preds_output = self.trainer.predict(valid_dataset)
+        print(preds_output.metrics)
+        y_preds = np.argmax(preds_output.predictions, axis=1)
+        return y_preds
+    
+    @staticmethod
+    def predict_test_data(model_checkpoint, test_data):
+        pipe_classifier = pipeline("text-classification", model=model_checkpoint)
+        preds = pipe_classifier(test_data, return_all_scores=True)
+        return preds
+
+
+

source/services/predicting_effective_arguments/train/seq_classification.py

@@ -0,0 +1,85 @@
+import pandas as pd
+import numpy as np
+import matplotlib.pyplot as plt
+from datasets import load_dataset
+from transformers import AutoTokenizer
+from datasets import Dataset, load_metric
+from sklearn.model_selection import train_test_split
+from source.services.predicting_effective_arguments.train.model import TransformersSequenceClassifier
+
+TARGET = 'discourse_effectiveness'
+TEXT = "discourse_text"
+MODEL_CHECKPOINT = "distilbert-base-uncased"
+MODEL_OUTPUT_DIR ='source/services/predicting_effective_arguments/model/hf_textclassification'
+class CFG:
+    TARGET = 'discourse_effectiveness'
+    TEXT = "discourse_text"
+    MODEL_CHECKPOINT = "distilbert-base-uncased"
+    MODEL_OUTPUT_DIR ='source/services/predicting_effective_arguments/model/hf_textclassification'
+    model_name="debertav3base"
+    learning_rate=1.5e-5
+    weight_decay=0.02
+    hidden_dropout_prob=0.007
+    attention_probs_dropout_prob=0.007
+    num_train_epochs=10
+    n_splits=4
+    batch_size=12
+    random_seed=42
+    save_steps=100
+    max_length=512
+
+tokenizer = AutoTokenizer.from_pretrained(MODEL_CHECKPOINT)
+
+def seed_everything(seed: int):
+    import random, os
+    import numpy as np
+    import torch
+    
+    random.seed(seed)
+    os.environ['PYTHONHASHSEED'] = str(seed)
+    np.random.seed(seed)
+    torch.manual_seed(seed)
+    torch.cuda.manual_seed(seed)
+    torch.backends.cudnn.deterministic = True
+    torch.backends.cudnn.benchmark = True
+
+
+def prepare_input_text(df, sep_token):
+    df['inputs'] = df.discourse_type.str.lower() + ' ' + sep_token + ' ' + df.discourse_text.str.lower()
+    return df
+
+
+if __name__ == '__main__':
+
+    config = CFG()
+    seqClassifer = TransformersSequenceClassifier(model_output_dir=config.MODEL_OUTPUT_DIR, tokenizer=tokenizer, model_checkpoint="distilbert-base-uncased", num_labels=3)
+    data = pd.read_csv("data/raw_data/train.csv")
+    test_df = pd.read_csv("data/raw_data/test.csv")
+    train_df, valid_df = train_test_split(data, test_size=0.30, random_state=42)
+
+    train_df = prepare_input_text(train_df, sep_token=tokenizer.sep_token)
+    valid_df = prepare_input_text(valid_df, sep_token=tokenizer.sep_token)
+    
+    train_dataset = Dataset.from_pandas(train_df[['inputs', TARGET]]).rename_column(TARGET, 'label').class_encode_column("label")
+    val_dataset = Dataset.from_pandas(valid_df[['inputs', TARGET]]).rename_column(TARGET, 'label').class_encode_column("label")
+    seqClassifer.train(train_dataset=train_dataset, eval_dataset=val_dataset, epochs=2, batch_size=64)
+
+    """
+    train_df[TARGET].value_counts(ascending=True).plot.barh()
+    plt.title("Frequency of Classes")
+    plt.show()
+
+    train_df['discourse_type'].value_counts(ascending=True).plot.barh()
+    plt.title("Frequency of discourse_type")
+    plt.show()
+
+    train_df["Words Per text"] = train_df[TEXT].str.split().apply(len)
+    train_df.boxplot("Words Per text", by=TARGET, grid=False, showfliers=False,
+            color="black")
+    plt.suptitle("")
+    plt.xlabel("")
+    plt.show()
+    """
+
+
+    pass