Sentiment Classification Using Fine-tuned BERT

import torch
from torch.utils.data import Dataset, DataLoader
from transformers import BertTokenizer, BertForSequenceClassification, Trainer, TrainingArguments
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, precision_recall_fscore_support
import pandas as pd

from huggingface_hub import login

# Log in to Hugging Face Hub using authentication token
# Required for accessing private models and pushing models to Hub
login("your_token")

from datasets import load_dataset
import pandas as pd

# Load sentiment analysis dataset from Hugging Face Hub
# Dataset contains 100k Weibo posts with sentiment labels
ds = load_dataset("dirtycomputer/weibo_senti_100k")
ds

# Convert Hugging Face dataset to pandas DataFrame
df = pd.DataFrame(ds['train'])

# Display basic information about the DataFrame
print("Dataset Overview:")
print(f"Number of samples: {len(df)}")
print(f"Columns: {df.columns.tolist()}")
print("\nLabel distribution:")
print(df['label'].value_counts())
print("\nSample reviews:")
print(df['review'].head())

# Check for any missing values
if df.isnull().sum().any():
    print("\nWarning: Dataset contains missing values!")

from sklearn.model_selection import train_test_split

# Split dataset into training and test sets
# - test_size=0.2: 80% training, 20% testing
# - shuffle=True: randomly shuffle before splitting
# - random_state=42: set seed for reproducibility
train_df, test_df = train_test_split(df, test_size=0.2, shuffle=True, random_state=42)

# Check for available CUDA device and set up GPU/CPU
# Colab typically provides a single GPU, if available
if torch.cuda.is_available():
   device = torch.device("cuda")
   # Print GPU information
   print(f"Using GPU: {torch.cuda.get_device_name(0)}")
   print(f"GPU Memory: {torch.cuda.get_device_properties(0).total_memory / 1e9:.2f} GB")
else:
   device = torch.device("cpu")
   print("No GPU available, using CPU")

# Initialize the BERT Chinese tokenizer
# Uses bert-base-chinese pre-trained model's vocabulary
tokenizer = BertTokenizer.from_pretrained('bert-base-chinese')

from torch.utils.data import Dataset
import torch
class TextDataset(Dataset):
   """
   Custom Dataset class for text data, inheriting from PyTorch's Dataset.

   Parameters:
   tokenizer (Tokenizer): Tokenizer object for text encoding
   texts (list): List of text samples
   labels (list): List of corresponding labels
   """
   def __init__(self, tokenizer, texts, labels):
       # Encode texts with padding and truncation
       self.encodings = tokenizer(
           texts,
           truncation=True,
           padding=True,
           max_length=512,  # Explicitly set max length for BERT
           return_tensors='pt'  # Return PyTorch tensors directly
       )
       # Convert labels to tensor
       self.labels = torch.tensor(labels)

   def __getitem__(self, idx):
       """
       Get a single sample by index.

       Args:
           idx (int): Sample index

       Returns:
           dict: Dictionary containing encoded text data and label
       """
       return {
           'input_ids': self.encodings['input_ids'][idx],
           'attention_mask': self.encodings['attention_mask'][idx],
           'labels': self.labels[idx]
       }

   def __len__(self):
       """
       Get dataset length.

       Returns:
           int: Number of samples in dataset
       """
       return len(self.labels)

# Print unique labels in the dataset
print("Unique labels in training data:", sorted(set(train_df['label'])))

# Create explicit label mapping
label_mapping = {
    0: 0,  # Negative
    1: 1   # Positive
}

# Map labels using explicit mapping
train_labels = [label_mapping[label] for label in train_df['label']]
test_labels = [label_mapping[label] for label in test_df['label']]

# Verify label distribution after mapping
print("\nLabel distribution after mapping:")
print("Training:", pd.Series(train_labels).value_counts())
print("Testing:", pd.Series(test_labels).value_counts())

# Explicit label mapping to ensure correct sentiment assignment
label_to_index = {
    0: 0,  # Keep negative as 0
    1: 1   # Keep positive as 1
}

# Map labels using explicit mapping
train_labels = [label_to_index[label] for label in train_df['label']]
test_labels = [label_to_index[label] for label in test_df['label']]

# Create datasets with verified labels
train_dataset = TextDataset(tokenizer, train_df['review'].tolist(), train_labels)
test_dataset = TextDataset(tokenizer, test_df['review'].tolist(), test_labels)

# Verify final mapping
print("\nFinal verification:")
print("Training set label distribution:", pd.Series(train_labels).value_counts())
print("Test set label distribution:", pd.Series(test_labels).value_counts())

# Load pre-trained Chinese BERT model and configure for binary classification
model = BertForSequenceClassification.from_pretrained(
   'bert-base-chinese',
   num_labels=2  # Binary classification (negative/positive)
)
model = model.to(device)

# Define training arguments for model fine-tuning
training_args = TrainingArguments(
   output_dir='sentiment-weibo-100k-fine-tuned-bert-test',  # Directory to save model checkpoints
   num_train_epochs=3,                                 # Number of training epochs
   per_device_train_batch_size=32,                    # Number of samples per training batch
   per_device_eval_batch_size=64,                     # Number of samples per evaluation batch
   warmup_steps=500,                                  # Steps for learning rate warmup
   weight_decay=0.01,                                 # L2 regularization factor
   logging_dir='./logs',                             # Directory for training logs
   logging_steps=100,                                # Log metrics every 100 steps
   evaluation_strategy="epoch",                      # Evaluate after each epoch
   save_strategy="epoch",                           # Save model after each epoch
   load_best_model_at_end=True,                    # Load best model after training
   push_to_hub=True,                               # Push model to Hugging Face Hub
   learning_rate=2e-5,                             # Initial learning rate
   gradient_accumulation_steps=1                   # Update model after every batch
)

def compute_metrics(pred):
   """
   Compute evaluation metrics for the model
   Args:
       pred: Contains predictions and label_ids
   Returns:
       dict: Dictionary containing accuracy, F1, precision, and recall scores
   """
   labels = pred.label_ids
   preds = pred.predictions.argmax(-1)
   precision, recall, f1, _ = precision_recall_fscore_support(
       labels,
       preds,
       average='binary',
       pos_label=1  # Define positive class for binary metrics
   )
   acc = accuracy_score(labels, preds)

   return {
       'accuracy': acc,
       'f1': f1,
       'precision': precision,
       'recall': recall
   }

# Initialize trainer with model and training configuration
trainer = Trainer(
   model=model,
   args=training_args,
   train_dataset=train_dataset,
   eval_dataset=test_dataset,
   compute_metrics=compute_metrics
)

# Print training configuration summary
print("Training Configuration:")
print(f"Model: bert-base-chinese")
print(f"Training samples: {len(train_dataset)}")
print(f"Test samples: {len(test_dataset)}")
print(f"Batch size: {training_args.per_device_train_batch_size}")
print(f"Number of epochs: {training_args.num_train_epochs}")

# Start training
trainer.train()

# Evaluate model performance
eval_results = trainer.evaluate()
print("\nEvaluation Results:", eval_results)

from transformers import pipeline
import torch

def test_sentiment(texts, yourmodel):
    """
    Test sentiment analysis model with given texts
    """
    # Create sentiment analyzer pipeline
    analyzer = pipeline(
        "sentiment-analysis",
        model=yourmodel,  # your model from HuggingFace Hub
        tokenizer="bert-base-chinese",
        device=0 if torch.cuda.is_available() else -1
    )

    # Process each text
    for text in texts:
        result = analyzer(text)[0]
        sentiment = "positive" if result['label'] == 'LABEL_1' else "negative"
        print(f"\nText: {text}")
        print(f"Sentiment: {sentiment}")
        print(f"Confidence: {result['score']:.4f}")

# Test with example texts
test_texts = [
    "这家店的菜真香，下次还来！",         # The food is delicious, will come again
    "质量有问题，不推荐购买。",           # Quality issues, not recommended
    "快递很快，包装完整。",               # Fast delivery, good packaging
    "商家态度很不好，生气。",             # Bad merchant attitude, angry
    "非常满意，超出预期。",               # Very satisfied, exceeded expectations
    "难吃到极点，太糟糕了。",             # Extremely bad taste, terrible
    "穿着很舒服，尺码合适。",             # Comfortable to wear, good size
    "卖家服务特别好！",                   # Great service from seller
    "不值这个价钱，后悔买了。",           # Not worth the price, regret buying
    "产品完全是垃圾，气死了。"            # Product is totally garbage, very angry
]

# Run the test
test_sentiment(test_texts, "BarryzZ/sentiment-weibo-100k-fine-tuned-bert-test")

# Initialize model with same configuration
model = BertForSequenceClassification.from_pretrained(
   'bert-base-chinese',
   num_labels=2
)
model = model.to(device)

# Enhanced training arguments
training_args = TrainingArguments(
   output_dir='sentiment-weibo-100k-fine-tuned-bert',
   num_train_epochs=5,                    # Increased from 3 to 5 for better learning
   per_device_train_batch_size=64,        # Doubled for faster training
   per_device_eval_batch_size=128,        # Doubled for faster evaluation
   learning_rate=2e-5,                    # Kept same learning rate
   warmup_ratio=0.1,                      # Added warmup ratio for smoother training
   weight_decay=0.01,                     # For regularization
   logging_dir='./logs',
   logging_steps=100,
   evaluation_strategy="steps",           # Changed to step-based evaluation
   eval_steps=200,                        # More frequent evaluation
   save_strategy="steps",
   save_steps=200,                        # More frequent model saving
   load_best_model_at_end=True,
   metric_for_best_model="f1_avg",        # Using average F1 score to select best model
   push_to_hub=True,
   gradient_accumulation_steps=1,
   fp16=True                              # Added mixed precision training for efficiency
)

# Enhanced metrics computation function
def compute_metrics(pred):
   """
   Compute detailed metrics including class-specific scores
   Returns metrics for both positive and negative classes
   """
   labels = pred.label_ids
   preds = pred.predictions.argmax(-1)

   precision, recall, f1, _ = precision_recall_fscore_support(
       labels,
       preds,
       average=None,
       labels=[0, 1]
   )
   acc = accuracy_score(labels, preds)
   conf_mat = confusion_matrix(labels, preds)

   return {
       'accuracy': acc,
       'f1_neg': f1[0],                   # Added separate F1 scores
       'f1_pos': f1[1],
       'f1_avg': f1.mean(),
       'precision_neg': precision[0],      # Added class-specific precision
       'precision_pos': precision[1],
       'recall_neg': recall[0],           # Added class-specific recall
       'recall_pos': recall[1],
       'confusion_matrix': conf_mat.tolist()  # Added confusion matrix
   }


# Initialize trainer
trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=train_dataset,
    eval_dataset=test_dataset,
    compute_metrics=compute_metrics,
    callbacks=[EarlyStoppingCallback(early_stopping_patience=3)]
)

# Print dataset statistics before training
print("\nDataset Statistics:")
print(f"Training samples: {len(train_dataset)}")
print(f"Test samples: {len(test_dataset)}")
print("\nLabel Distribution:")
print("Training:", pd.Series([d['labels'].item() for d in train_dataset]).value_counts())
print("Testing:", pd.Series([d['labels'].item() for d in test_dataset]).value_counts())

# Start training
trainer.train()

# Evaluate model
eval_results = trainer.evaluate()
print("\nFinal Evaluation Results:")
for metric, value in eval_results.items():
    if isinstance(value, float):
        print(f"{metric}: {value:.4f}")
    else:
        print(f"{metric}: {value}")

# Test with example texts
test_texts = [
    "这家店的菜真香，下次还来！",         # The food is delicious, will come again
    "质量有问题，不推荐。",           # Quality issues, not recommended
    "快递很快，包装完整。",               # Fast delivery, good packaging
    "商家态度不好，生气。",             # Bad merchant attitude, angry
    "非常满意，超出预期。",               # Very satisfied, exceeded expectations
    "难吃到极点，太糟糕了。",             # Extremely bad taste, terrible
    "穿着很舒服，尺码合适。",             # Comfortable to wear, good size
    "卖家服务特别好！",                   # Great service from seller
    "不值这个价钱，后悔买了。",           # Not worth the price, regret buying
    "产品完全是垃圾，气死了。"            # Product is totally garbage, very angry
]

# Run the test
test_sentiment(test_texts, "BarryzZ/sentiment-weibo-100k-fine-tuned-bert")

test_texts = [
    # Strong positive / 强烈正面
    "我考上研究生了！",  # I got accepted into graduate school!
    "今天他向我求婚了！",  # He proposed to me today!
    "终于买到梦想的房子",  # Finally bought my dream house
    "中了五百万大奖！",  # Won a 5 million prize!

    # Strong negative / 强烈负面
    "被裁员了，好绝望",  # Got laid off, feeling desperate
    "信任的人背叛我",  # Betrayed by someone I trusted
    "重要的文件全丢了",  # Lost all important documents
    "又被扣工资了，气死",  # Got my salary deducted again, so angry

    # Anger / 愤怒
    "偷我的车，混蛋！",  # Someone stole my car, bastard!
    "骗子公司，我要报警",  # Scam company, I'm calling the police
    "半夜装修，烦死了",  # Renovation at midnight, so annoying
    "商家太坑人了！",  # The merchant is such a ripoff!

    # Pleasant surprise / 惊喜
    "宝宝会走路了！",  # Baby learned to walk!
    "升职加薪啦！",  # Got promoted with a raise!
    "论文发表成功！",  # Paper got published successfully!
    "收到offer了！"  # Received a job offer!
]
# Run the test
test_sentiment(test_texts, "BarryzZ/sentiment-weibo-100k-fine-tuned-bert")