1
import torch
2
from torch.utils.data import DataLoader
3
from transformers import DistilBertTokenizerFast, DistilBertForSequenceClassification, Trainer, TrainingArguments
4
from datasets import load_dataset
5
import argparse
6

7
def parse_args():
8
    parser = argparse.ArgumentParser(description="Fine-tune DistilBERT for sentiment analysis.")
9
    parser.add_argument("--train_file", type=str, required=True, help="Path to training CSV.")
10
    parser.add_argument("--val_file", type=str, required=True, help="Path to validation CSV.")
11
    parser.add_argument("--epochs", type=int, default=3, help="Number of training epochs.")
12
    parser.add_argument("--batch_size", type=int, default=16, help="Batch size.")
13
    parser.add_argument("--lr", type=float, default=2e-5, help="Learning rate.")
14
    parser.add_argument("--output_dir", type=str, default="models/checkpoints", help="Directory to save model checkpoints.")
15
    return parser.parse_args()
16

17
def main():
18
    args = parse_args()
19

20
    # Load dataset using Hugging Face 'datasets' library
21
    dataset = load_dataset("csv", data_files={"train": args.train_file, "validation": args.val_file})
22

23
    # Load DistilBert tokenizer
24
    tokenizer = DistilBertTokenizerFast.from_pretrained("distilbert-base-uncased")
25

26
    def tokenize_function(example):
27
        return tokenizer(example["text"], padding="max_length", truncation=True, max_length=128)
28

29
    # Tokenize your dataset
30
    tokenized_datasets = dataset.map(tokenize_function, batched=True)
31

32
    # Rename labels column to 'labels' for HF Trainer
33
    tokenized_datasets = tokenized_datasets.rename_column("label", "labels")
34
    tokenized_datasets.set_format("torch", columns=["input_ids", "attention_mask", "labels"])
35

36
    train_dataset = tokenized_datasets["train"]
37
    val_dataset = tokenized_datasets["validation"]
38

39
    # Load model
40
    model = DistilBertForSequenceClassification.from_pretrained("distilbert-base-uncased", num_labels=2)
41

42
    training_args = TrainingArguments(
43
        output_dir=args.output_dir,
44
        evaluation_strategy="epoch",
45
        save_strategy="epoch",
46
        num_train_epochs=args.epochs,
47
        per_device_train_batch_size=args.batch_size,
48
        per_device_eval_batch_size=args.batch_size,
49
        learning_rate=args.lr,
50
        logging_steps=10,
51
        logging_dir=f"{args.output_dir}/logs",
52
        load_best_model_at_end=True,
53
    )
54

55
    trainer = Trainer(
56
        model=model,
57
        args=training_args,
58
        train_dataset=train_dataset,
59
        eval_dataset=val_dataset,
60
    )
61

62
    trainer.train()
63
    trainer.save_model(args.output_dir)
64

65
if __name__ == "__main__":
66
    main()

1
import torch
2
from transformers import DistilBertTokenizerFast, DistilBertForSequenceClassification
3

4
# Load your final model
5
MODEL_PATH = "models/checkpoints"
6
tokenizer = DistilBertTokenizerFast.from_pretrained(MODEL_PATH)
7
model = DistilBertForSequenceClassification.from_pretrained(MODEL_PATH)
8

9
def predict_sentiment(text: str):
10
    inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True)
11
    with torch.no_grad():
12
        outputs = model(**inputs)
13
    logits = outputs.logits
14
    predicted_class_id = logits.argmax().item()
15
    # 0 for negative, 1 for positive in this example
16
    return "positive" if predicted_class_id == 1 else "negative"
17

18
if __name__ == "__main__":
19
    sample_texts = [
20
        "I loved this product, would absolutely buy again!",
21
        "This was terrible, I want my money back."
22
    ]
23
    for txt in sample_texts:
24
        sentiment = predict_sentiment(txt)
25
        print(f"Text: {txt} -> Sentiment: {sentiment}")

1
from fastapi import FastAPI, Request
2
from pydantic import BaseModel
3
from scripts.predict import predict_sentiment
4

5
app = FastAPI()
6

7
class TextPayload(BaseModel):
8
    text: str
9

10
@app.post("/predict")
11
def predict(payload: TextPayload):
12
    sentiment_label = predict_sentiment(payload.text)
13
    return {"sentiment": sentiment_label}
14

15
@app.get("/")
16
def root():
17
    return {"message": "Welcome to the Zero to Hero Sentiment Analysis API!"}

1
import pytest
2
from fastapi.testclient import TestClient
3
from app.main import app
4

5
client = TestClient(app)
6

7
def test_root():
8
    response = client.get("/")
9
    assert response.status_code == 200
10
    assert response.json() == {"message": "Welcome to the Zero to Hero Sentiment Analysis API!"}
11

12
def test_predict_endpoint():
13
    response = client.post("/predict", json={"text": "I love this!"})
14
    assert response.status_code == 200
15
    assert "sentiment" in response.json()

1
from langdetect import detect
2
from transformers import XLMRobertaTokenizer, XLMRobertaForSequenceClassification
3

4
MODEL_PATH = "models/multilingual_checkpoint"
5
tokenizer = XLMRobertaTokenizer.from_pretrained(MODEL_PATH)
6
model = XLMRobertaForSequenceClassification.from_pretrained(MODEL_PATH)
7

8
def predict_sentiment_multilingual(text: str):
9
    # Detect language
10
    lang = detect(text)
11
    # Tokenize
12
    inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True)
13
    with torch.no_grad():
14
        outputs = model(**inputs)
15
    predicted_class_id = outputs.logits.argmax().item()
16
    label = "positive" if predicted_class_id == 1 else "negative"
17
    return lang, label