diff --git a/src/ai_detection/classifier/classify.py b/src/ai_detection/classifier/classify.py index 074a118..3edec15 100644 --- a/src/ai_detection/classifier/classify.py +++ b/src/ai_detection/classifier/classify.py @@ -44,35 +44,42 @@ def label_encode(labels: list[str]): def split_data( x: list, y: list, train_size=0.8, valid_size=0.1, test_size=0.1 -) ->tuple[list,list,list,list,list,list,list,list,list]: # -> tuple: +) -> tuple[list, list, list, list, list, list, list, list, list]: # -> tuple: """ split_data is a function to split the data into train, valid and test sets tips: split_data is design for getting the index for train, valid and test set """ assert len(x) == len(y), "Length of x and y should be same" original_indices = pd.DataFrame(x).index - x_train, x_test, y_train, y_test, indices_train, indices_test = ( - train_test_split( - x, - y, - original_indices, - train_size=train_size, - random_state=42, - ) + x_train, x_test, y_train, y_test, indices_train, indices_test = train_test_split( + x, + y, + original_indices, + train_size=train_size, + random_state=42, + ) + x_test, x_valid, y_test, y_valid, indices_train, indices_valid = train_test_split( + x_test, + y_test, + indices_test, + test_size=test_size / (test_size + valid_size), + random_state=42, ) - x_test, x_valid, y_test, y_valid, indices_train, indices_valid = ( - train_test_split( + if not all( + isinstance(i, list) + for i in [ + x_train, + x_valid, x_test, + y_train, + y_valid, y_test, + indices_train, + indices_valid, indices_test, - test_size=test_size / (test_size + valid_size), - random_state=42, - ) - ) - if not all(isinstance(i, list) for i in [x_train, x_valid, x_test, y_train, y_valid, y_test, indices_train, indices_valid, indices_test]): - raise ValueError( - "x_train, x_valid, x_test should be list or np.array" - ) + ] + ): + raise ValueError("x_train, x_valid, x_test should be list or np.array") return ( x_train, x_valid, @@ -93,18 +100,18 @@ class TextClassifer: def __init__( self, - texts: list[str]|None = None, - labels: list[str]|None = None, + texts: list[str] | None = None, + labels: list[str] | None = None, data_name: str = "", language: str = "English", model_name: str = "bert-base-uncased", tokenizer_name: str = "bert-base-uncased", classifier_type: str = "transformers", - split_size: dict|None = None, - train_args: dict|None = None, - tokenizer_config: dict|None = None, - model_config: dict|None = None, - pipeline: list[str]|None = None, + split_size: dict | None = None, + train_args: dict | None = None, + tokenizer_config: dict | None = None, + model_config: dict | None = None, + pipeline: list[str] | None = None, ): """ Allowed and default keys for classify_config are: @@ -161,16 +168,18 @@ def __init__( self.tokenizer_config = tokenizer_config if tokenizer_config is not None else {} self.model_config = model_config if model_config is not None else {} self.pipeline = pipeline if pipeline is not None else [] - self.split_size = split_size if split_size is not None else {"train_size": 0.8, "test_size": 0.1, "valid_size": 0.1} + self.split_size = ( + split_size + if split_size is not None + else {"train_size": 0.8, "test_size": 0.1, "valid_size": 0.1} + ) self.classifier_name = ( f"{self.model_name}_{self.tokenizer_name}_{self.data_name}" ) self.model_path = ( - MODEL_PATH - / f"{self.model_name}_{self.tokenizer_name}_{self.data_name}.pkl" + MODEL_PATH / f"{self.model_name}_{self.tokenizer_name}_{self.data_name}.pkl" ) - def load_data(self, texts, labels, data_name): """ load_data is a function to load the data into the classifier @@ -181,9 +190,7 @@ def load_data(self, texts, labels, data_name): raise ValueError("Length of texts and labels should be same") self.texts = texts self.labels = labels - logging.info( - "Loading data: %s to classifier %s", data_name, self.model_name - ) + logging.info("Loading data: %s to classifier %s", data_name, self.model_name) self.data_name = data_name self.classifier_name = ( f"{self.model_name}_{self.tokenizer_name}_{self.data_name}" diff --git a/src/ai_detection/classifier/multi_feature_model/fusionBert.py b/src/ai_detection/classifier/multi_feature_model/fusionBert.py index 6a02032..adf479c 100644 --- a/src/ai_detection/classifier/multi_feature_model/fusionBert.py +++ b/src/ai_detection/classifier/multi_feature_model/fusionBert.py @@ -10,12 +10,14 @@ from sklearn.metrics import f1_score from torch.utils.data import DataLoader, Dataset from tqdm import tqdm -from transformers import BertForSequenceClassification, BertTokenizer, PreTrainedTokenizerBase +from transformers import ( + BertForSequenceClassification, + BertTokenizer, + PreTrainedTokenizerBase, +) from transformers.optimization import get_scheduler from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer -import spacy - def truncate_and_pad_single_sequence(seq, max_length): truncated_seq = seq[:max_length] @@ -26,18 +28,14 @@ def truncate_and_pad_single_sequence(seq, max_length): class FeatureFusionBertTokenizer(PreTrainedTokenizerBase): def __init__(self, feature_len, **kwargs): self.feature_len = feature_len - self.sentiment_max_length = kwargs.get( - "sentiment_max_length", feature_len - ) + self.sentiment_max_length = kwargs.get("sentiment_max_length", feature_len) self.nlp = spacy.load("en_core_web_sm") self.all_tags = self.nlp.get_pipe("tagger").labels self.sentiment_analyzer = SentimentIntensityAnalyzer() self.bert_model = BertForSequenceClassification.from_pretrained( "bert-base-uncased" ) - self.bert_tokenzier = BertTokenizer.from_pretrained( - "bert-base-uncased" - ) + self.bert_tokenzier = BertTokenizer.from_pretrained("bert-base-uncased") def semantic(self, text: str, **kwargs): tokenzied = self.bert_tokenzier( @@ -60,9 +58,9 @@ def analyze_word_level_sentiment(self, text: str) -> torch.tensor: if token.is_stop or token.is_punct: sentiment = 0.0 else: - sentiment = self.sentiment_analyzer.polarity_scores( - token.text - )["compound"] + sentiment = self.sentiment_analyzer.polarity_scores(token.text)[ + "compound" + ] word_sentiment.append(sentiment) sentiment = torch.tensor(word_sentiment, dtype=torch.float) return self._padding(sentiment) @@ -77,12 +75,8 @@ def pos_feature(self, text: str) -> torch.tensor: return self._padding(pos) def _padding(self, input_tensor): - input_tensor = truncate_and_pad_single_sequence( - input_tensor, self.feature_len - ) - attention_mask = torch.tensor( - [1] * len(input_tensor), dtype=torch.float - ) + input_tensor = truncate_and_pad_single_sequence(input_tensor, self.feature_len) + attention_mask = torch.tensor([1] * len(input_tensor), dtype=torch.float) return input_tensor.unsqueeze(0), attention_mask.unsqueeze(0) def __call__(self, text: str): @@ -114,9 +108,7 @@ def batch_encode_plus(self, batch_text: list[str]): # -> list: class HFeatureFusion(nn.Module): - def __init__( - self, feature_num, feature_len, output_dim, num_heads=4, dropout=0.1 - ): + def __init__(self, feature_num, feature_len, output_dim, num_heads=4, dropout=0.1): super().__init__() self.feature_num = feature_num self.feature_len = feature_len @@ -151,9 +143,7 @@ def forward(self, *features: torch.tensor): class CrossAttentionFeatureFusion(nn.Module): - def __init__( - self, feature_num, feature_len, output_dim, num_heads=4, dropout=0.1 - ): + def __init__(self, feature_num, feature_len, output_dim, num_heads=4, dropout=0.1): super().__init__() self.feature_len = feature_len self.num_heads = num_heads @@ -222,9 +212,7 @@ def __init__( def forward(self, *features): self.train() # Set all modules to train for feature in features: - feature["input_ids"] = F.normalize( - feature["input_ids"], p=2, dim=-1 - ) + feature["input_ids"] = F.normalize(feature["input_ids"], p=2, dim=-1) fused_features = self.fusion_module(*features) encoder_output = self.encoder(fused_features) pooler_output = self.pooler(encoder_output[0]) @@ -243,9 +231,7 @@ def test_step(self, batch, batch_idx): y_pred = torch.argmax(outputs, dim=-1) y_test = batch["labels"] acc = (y_pred == y_test).sum().float() / len(y_test) - f1 = f1_score( - y_test.cpu().numpy(), y_pred.cpu().numpy(), average="weighted" - ) + f1 = f1_score(y_test.cpu().numpy(), y_pred.cpu().numpy(), average="weighted") self.log("test_acc", acc) self.log("test_f1", f1) return {"test_acc": acc, "test_f1": f1} @@ -321,8 +307,8 @@ def prepare_data(self): pass def setup(self, stage=None): - self.tokenizer: FeatureFusionBertTokenizer = ( - FeatureFusionBertTokenizer(feature_len=768) + self.tokenizer: FeatureFusionBertTokenizer = FeatureFusionBertTokenizer( + feature_len=768 ) if not isinstance(self.tokenizer, FeatureFusionBertTokenizer): raise TypeError( diff --git a/src/ai_detection/classifier/multi_feature_model/fusionBert_backup.py b/src/ai_detection/classifier/multi_feature_model/fusionBert_backup.py index 0e7f646..d60ded5 100644 --- a/src/ai_detection/classifier/multi_feature_model/fusionBert_backup.py +++ b/src/ai_detection/classifier/multi_feature_model/fusionBert_backup.py @@ -4,7 +4,7 @@ import torch import torch.nn as nn import torch.nn.functional as F -from datasets import load_dataset,Dataset +from datasets import Dataset, load_dataset from torch.optim import AdamW from torch.utils.data import DataLoader from tqdm import tqdm @@ -24,18 +24,14 @@ def truncate_and_pad_single_sequence(seq, max_length): class FeatureFusionBertTokenizer: def __init__(self, feature_len, **kwargs): self.feature_len = feature_len - self.sentiment_max_length = kwargs.get( - "sentiment_max_length", feature_len - ) + self.sentiment_max_length = kwargs.get("sentiment_max_length", feature_len) self.nlp = spacy.load("en_core_web_sm") self.all_tags = self.nlp.get_pipe("tagger").labels self.sentiment_analyzer = SentimentIntensityAnalyzer() self.bert_model = BertForSequenceClassification.from_pretrained( "bert-base-uncased" ) - self.bert_tokenzier = BertTokenizer.from_pretrained( - "bert-base-uncased" - ) + self.bert_tokenzier = BertTokenizer.from_pretrained("bert-base-uncased") def semantic(self, text: str, **kwargs): tokenzied = self.bert_tokenzier( @@ -51,16 +47,16 @@ def semantic(self, text: str, **kwargs): embedding = torch.mean(embedding, dim=1, keepdim=False) return embedding, torch.ones_like(embedding) - def analyze_word_level_sentiment(self, text: str) ->tuple: + def analyze_word_level_sentiment(self, text: str) -> tuple: doc = self.nlp(text) word_sentiment = [] for token in doc: if token.is_stop or token.is_punct: sentiment = 0.0 else: - sentiment = self.sentiment_analyzer.polarity_scores( - token.text - )["compound"] + sentiment = self.sentiment_analyzer.polarity_scores(token.text)[ + "compound" + ] word_sentiment.append(sentiment) sentiment = torch.tensor(word_sentiment, dtype=torch.float) return self._padding(sentiment) @@ -75,14 +71,9 @@ def pos_feature(self, text: str) -> tuple: posed = self._padding(pos) return posed if isinstance(posed, tuple) else posed - def _padding(self, input_tensor): - input_tensor = truncate_and_pad_single_sequence( - input_tensor, self.feature_len - ) - attention_mask = torch.tensor( - [1] * len(input_tensor), dtype=torch.float - ) + input_tensor = truncate_and_pad_single_sequence(input_tensor, self.feature_len) + attention_mask = torch.tensor([1] * len(input_tensor), dtype=torch.float) return input_tensor.unsqueeze(0), attention_mask.unsqueeze(0) def __call__(self, text: str): @@ -113,9 +104,7 @@ def batch_encode_plus(self, batch_text: list[str], **kwargs): class HFeatureFusion(nn.Module): - def __init__( - self, feature_num, feature_len, output_dim, num_heads=4, dropout=0.1 - ): + def __init__(self, feature_num, feature_len, output_dim, num_heads=4, dropout=0.1): super().__init__() self.feature_num = feature_num self.feature_len = feature_len @@ -129,7 +118,7 @@ def __init__( self.feature_fusion_dim = (feature_num - 1) * feature_len self.fc = nn.Linear(self.feature_fusion_dim, output_dim) - def forward(self, features:list[dict[str, torch.Tensor]]): + def forward(self, features: list[dict[str, torch.Tensor]]): if len(features) != self.feature_num: raise ValueError( f"Expected {self.feature_num} features, but got {len(features)}" @@ -157,9 +146,7 @@ def forward(self, features:list[dict[str, torch.Tensor]]): class CrossAttentionFeatureFusion(nn.Module): - def __init__( - self, feature_num, feature_len, output_dim, num_heads=4, dropout=0.1 - ): + def __init__(self, feature_num, feature_len, output_dim, num_heads=4, dropout=0.1): super().__init__() self.feature_len = feature_len self.num_heads = num_heads @@ -209,7 +196,7 @@ def __init__( ) self.encoder = bert_classifier.bert.encoder self.pooler = bert_classifier.bert.pooler - + self.fusion_module = CrossAttentionFeatureFusion( feature_num=feature_num, feature_len=feature_len, @@ -266,19 +253,17 @@ def __len__(self): for epoch in range(num_epochs): for step, batch in enumerate(train_dataloader): for feature in ["semantic", "pos", "sentiment"]: - batch[feature]["input_ids"] = batch[feature]["input_ids"].to( + batch[feature]["input_ids"] = batch[feature]["input_ids"].to(device) + batch[feature]["attention_mask"] = batch[feature]["attention_mask"].to( device ) - batch[feature]["attention_mask"] = batch[feature][ - "attention_mask" - ].to(device) optimizer.zero_grad() - outputs = model( - batch["semantic"], batch["pos"], batch["sentiment"] - ) + outputs = model(batch["semantic"], batch["pos"], batch["sentiment"]) loss = F.cross_entropy(outputs, batch["labels"].to(device)) loss.backward() optimizer.step() lr_scheduler.step() if step % 100 == 0: - print(f"Epoch {epoch + 1}, step {step + 1} completed with loss: {loss.item()}") + print( + f"Epoch {epoch + 1}, step {step + 1} completed with loss: {loss.item()}" + ) diff --git a/src/ai_detection/classifier/pytorch_classier.py b/src/ai_detection/classifier/pytorch_classier.py index 0ec292a..eb530db 100644 --- a/src/ai_detection/classifier/pytorch_classier.py +++ b/src/ai_detection/classifier/pytorch_classier.py @@ -3,13 +3,19 @@ import datasets import evaluate import torch -from transformers import (AutoTokenizer, DataCollatorWithPadding, - PreTrainedTokenizerBase, Trainer, TrainingArguments) +from transformers import ( + AutoTokenizer, + DataCollatorWithPadding, + PreTrainedTokenizerBase, + Trainer, + TrainingArguments, +) -from ai_detection.classifier.classify import (TextClassifer, label_encode, - split_data) +from ai_detection.classifier.classify import TextClassifer, label_encode, split_data from ai_detection.classifier.multi_feature_model.fusionBert_backup import ( - FeatureFusionBertClassfier, FeatureFusionBertTokenizer) + FeatureFusionBertClassfier, + FeatureFusionBertTokenizer, +) from ai_detection.utils.everyai_path import MODEL_PATH