File size: 6,146 Bytes
012f8b5 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 |
import os
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
from torchvision import transforms
from transformers import ViTModel, BertTokenizerFast, BertConfig, BertLMHeadModel, AdamW
from PIL import Image, ImageFile
import pandas as pd
from tqdm import tqdm
# Increase the maximum image size limit to avoid DecompressionBombWarning
Image.MAX_IMAGE_PIXELS = None
# Allow loading truncated images
ImageFile.LOAD_TRUNCATED_IMAGES = True
# Check if CUDA is available
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"Using device: {device}")
# Define constants
VIT_MODEL_NAME = "google/vit-base-patch16-224"
BERT_MODEL_NAME = "dbmdz/bert-base-turkish-cased" # Using a Turkish BERT model
model = "TeLVE_v1.0.pth"
MAX_LENGTH = 128
BATCH_SIZE = 8
EPOCHS = 5
LEARNING_RATE = 2e-5
class ImageCaptioningDataset(Dataset):
def __init__(self, dataframe, img_dir, tokenizer):
self.dataframe = dataframe
self.img_dir = img_dir
self.tokenizer = tokenizer
self.transform = transforms.Compose([
transforms.Resize((224, 224)),
transforms.ToTensor(),
transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
])
def __len__(self):
return len(self.dataframe)
def __getitem__(self, idx):
row = self.dataframe.iloc[idx]
img_path = os.path.join(self.img_dir, row['photo_id'] + ".jpg")
try:
image = Image.open(img_path).convert('RGB')
image = self.transform(image)
except (FileNotFoundError, IOError):
# Return None if the image is not found or cannot be opened
return None
caption = row['ai_description']
# Check if caption is a valid string
if not isinstance(caption, str):
return None # Skip the example if caption is not valid
encoding = self.tokenizer(
caption,
add_special_tokens=True,
max_length=MAX_LENGTH,
padding='max_length',
truncation=True,
return_attention_mask=True,
return_tensors='pt'
)
return {
'pixel_values': image,
'input_ids': encoding['input_ids'].squeeze(),
'attention_mask': encoding['attention_mask'].squeeze(),
'labels': encoding['input_ids'].squeeze() # Use input_ids as labels for calculating loss
}
class ImageCaptioningModel(nn.Module):
def __init__(self, vit_model, bert_model):
super(ImageCaptioningModel, self).__init__()
self.vit = vit_model
self.bert = bert_model
self.linear = nn.Linear(self.vit.config.hidden_size, self.bert.config.hidden_size)
def forward(self, pixel_values, input_ids, attention_mask, labels=None):
image_features = self.vit(pixel_values).last_hidden_state
image_features = self.linear(image_features)
outputs = self.bert(input_ids=input_ids,
attention_mask=attention_mask,
encoder_hidden_states=image_features,
labels=labels,
return_dict=True)
return outputs.loss, outputs.logits
def collate_fn(batch):
# Filter out None values (skipped images)
batch = list(filter(lambda x: x is not None, batch))
if len(batch) == 0:
return None
return {key: torch.stack([item[key] for item in batch]) for key in batch[0]}
def train_vlm_model():
# Load and preprocess the dataset
encodings = ['utf-8', 'iso-8859-9', 'windows-1254']
for encoding in encodings:
try:
df = pd.read_csv('./datasets/' + model + '.tsv000', sep='\t', encoding=encoding)
print(f"Successfully read the file with {encoding} encoding.")
break
except UnicodeDecodeError:
print(f"Failed to read with {encoding} encoding. Trying next...")
else:
raise ValueError("Could not read the file with any of the specified encodings.")
# Initialize the tokenizer
tokenizer = BertTokenizerFast.from_pretrained(BERT_MODEL_NAME)
# Create the dataset and dataloader
dataset = ImageCaptioningDataset(df, '../download/images', tokenizer)
dataloader = DataLoader(dataset, batch_size=BATCH_SIZE, shuffle=True, collate_fn=collate_fn)
# Initialize the model components
vit_model = ViTModel.from_pretrained(VIT_MODEL_NAME)
bert_config = BertConfig.from_pretrained(BERT_MODEL_NAME)
bert_config.is_decoder = True
bert_config.add_cross_attention = True
bert_model = BertLMHeadModel.from_pretrained(BERT_MODEL_NAME, config=bert_config)
# Create the combined model
model = ImageCaptioningModel(vit_model, bert_model)
model.to(device)
# Define optimizer
optimizer = AdamW(model.parameters(), lr=LEARNING_RATE)
# Training loop
model.train()
for epoch in range(EPOCHS):
total_loss = 0
progress_bar = tqdm(dataloader, desc=f"Epoch {epoch+1}/{EPOCHS}")
for batch in progress_bar:
if batch is None:
continue
pixel_values = batch['pixel_values'].to(device)
input_ids = batch['input_ids'].to(device)
attention_mask = batch['attention_mask'].to(device)
labels = batch['labels'].to(device)
optimizer.zero_grad()
loss, _ = model(pixel_values, input_ids, attention_mask, labels)
loss.backward()
optimizer.step()
total_loss += loss.item()
progress_bar.set_postfix({'loss': loss.item()})
print(f"Epoch {epoch+1}/{EPOCHS}, Average Loss: {total_loss/len(dataloader)}")
# Save the model
torch.save(model.state_dict(), "./models/" + model)
tokenizer.save_pretrained("./tokenizer")
if __name__ == "__main__":
train_vlm_model() |