USER
yesterday we were creating a image captioning model but due to time issue i left in half
i am providing half of our work
provide me code guide and step after the last line iam given
import os
import zipfile
Define paths
drive_path = '/content/drive/MyDrive'
archive_zip = os.path.join(drive_path, 'archive.zip')
extract_path = os.path.join(drive_path, 'archive_extracted')
Create a directory to extract
if not os.path.exists(extract_path):
os.makedirs(extract_path)
Extract the archive
with zipfile.ZipFile(archive_zip, 'r') as zip_ref:
zip_ref.extractall(extract_path)
print("Extraction Complete.")
Path to the extracted archive
archive_path = os.path.join(extract_path)
Paths to images and captions
image_dir = os.path.join(archive_path, 'Images')
captions_file = os.path.join(archive_path, 'captions.txt')
List files to verify
print("Images Directory:", os.listdir(image_dir)[:5]) # Display first 5 image filenames
print("Captions File Exists:", os.path.exists(captions_file))
import os
import string
def load_doc(filename):
"""Load document into memory."""
with open(filename, 'r') as file:
text = file.read()
return text
def preprocess_caption(caption):
"""
Preprocess captions:
- Lowercase
- Remove punctuation
- Add and tokens
"""
caption = caption.lower()
caption = caption.translate(str.maketrans('', '', string.punctuation))
caption = caption.strip()
caption = ' ' + caption + ' '
return caption
Path to captions file
captions_file = os.path.join(archive_path, 'captions.txt')
Load captions
captions = load_doc(captions_file)
Initialize a dictionary to hold image captions
captions_dict = {}
for line in captions.split('\n'):
if len(line) < 1:
continue # Skip empty lines
if line.startswith('image'):
continue # Skip header line
# Split only on the first comma to handle captions with commas
tokens = line.split(',', 1)
if len(tokens) != 2:
print(f"Skipping malformed line: {line}")
continue # Skip lines that don't have exactly two elements
image_id, caption = tokens
image_id = image_id.strip()
caption = caption.strip()
# Initialize list for images if not already present
if image_id not in captions_dict:
captions_dict[image_id] = []
# Preprocess and append caption
captions_dict[image_id].append(preprocess_caption(caption))
Display sample captions to verify
for key, val in list(captions_dict.items())[:5]:
print(f"Image ID: {key}")
for cap in val:
print(f"Caption: {cap}")
print('\n')
import tensorflow as tf
from tensorflow.keras.applications.inception_v3 import InceptionV3, preprocess_input
Define image directory
image_dir = os.path.join(archive_path, 'Images')
Preprocess images using InceptionV3
def load_image(image_path):
img = tf.io.read_file(image_path)
img = tf.image.decode_jpeg(img, channels=3)
img = tf.image.resize(img, (299, 299))
img = preprocess_input(img)
return img
Test the image loading function
sample_image_id = list(captions_dict.keys())[0]
sample_image_path = os.path.join(image_dir, sample_image_id)
sample_image = load_image(sample_image_path)
print(f"Sample Image Shape: {sample_image.shape}")
Load InceptionV3 model without the top classification layer
image_model = InceptionV3(include_top=False, weights='imagenet')
new_input = image_model.input
hidden_layer = image_model.layers[-1].output # Last convolutional layer
Define the feature extraction model
image_features_extract_model = tf.keras.Model(new_input, hidden_layer)
Extract features for all images and save them
import pickle
import tqdm
features_path = os.path.join(drive_path, 'image_features.pkl')
if not os.path.exists(features_path):
# Create a list of all image filenames
image_filenames = list(captions_dict.keys())
# Extract features
image_features = {}
for img_name in tqdm.tqdm(image_filenames):
img_path = os.path.join(image_dir, img_name)
img_tensor = load_image(img_path)
img_tensor = tf.expand_dims(img_tensor, 0) # Add batch dimension
img_features = image_features_extract_model(img_tensor)
img_features = tf.reshape(img_features, (img_features.shape[0], -1, img_features.shape[3]))
image_features[img_name] = img_features.numpy()
# Save features to a pickle file
with open(features_path, 'wb') as f:
pickle.dump(image_features, f)
print("Image Features Extracted and Saved.")
else:
# Load features if already extracted
with open(features_path, 'rb') as f:
image_features = pickle.load(f)
print("Image Features Loaded from Disk.")
print("Number of images with extracted features:", len(image_features))
please guide from the last code provide
from tensorflow.keras.preprocessing.text import Tokenizer
from tensorflow.keras.preprocessing.sequence import pad_sequences
Compile all captions into a list
all_captions = []
for key in captions_dict:
for cap in captions_dict[key]:
all_captions.append(cap)
print("Total Captions:", len(all_captions))
Tokenize the captions
tokenizer = Tokenizer(num_words=5000, oov_token="",
filters='!"#$%&()*+.,-/:;=?@[]^_`{|}~ ')
tokenizer.fit_on_texts(all_captions)
Create word to index mapping and add token
tokenizer.word_index[''] = 0
tokenizer.index_word[0] = ''
Save tokenizer for future use
import json
tokenizer_json = tokenizer.to_json()
with open(os.path.join(drive_path, 'tokenizer.json'), 'w') as f:
f.write(tokenizer_json)
Convert captions to sequences
train_seqs = tokenizer.texts_to_sequences(all_captions)
Pad sequences
max_length = max(len(seq) for seq in train_seqs)
print("Maximum Caption Length:", max_length)
train_seqs = pad_sequences(train_seqs, maxlen=max_length, padding='post')
print("Sample Padded Sequence:", train_seqs[0])
vocab_size = len(tokenizer.word_index) + 1
print("Vocabulary Size:", vocab_size)
Split data into training and validation (90% train, 10% val)
import numpy as np
Create image, caption pairs
image_ids = []
captions_list = []
for key in captions_dict:
for cap in captions_dict[key]:
image_ids.append(key)
captions_list.append(cap)
Convert to sequences
sequences = tokenizer.texts_to_sequences(captions_list)
sequences = pad_sequences(sequences, maxlen=max_length, padding='post')
Convert to numpy arrays
image_ids = np.array(image_ids)
sequences = np.array(sequences)
Shuffle the data
dataset_size = len(image_ids)
indices = np.arange(dataset_size)
np.random.shuffle(indices)
image_ids = image_ids[indices]
sequences = sequences[indices]
Split into training and validation sets
split_index = int(0.9 * dataset_size)
train_image_ids = image_ids[:split_index]
train_sequences = sequences[:split_index]
val_image_ids = image_ids[split_index:]
val_sequences = sequences[split_index:]
print(f"Training Samples: {len(train_image_ids)}")
print(f"Validation Samples: {len(val_image_ids)}")
Image features shape: (batch_size, 64, 2048)
embedding_dim = 256
units = 512
from tensorflow.keras import layers
class RNN_Decoder(tf.keras.Model):
def init(self, vocab_size, embedding_dim, units):
super(RNN_Decoder, self).init()
self.units = units
self.embedding = layers.Embedding(vocab_size, embedding_dim)
self.lstm = layers.LSTM(self.units,
return_sequences=True,
return_state=True,
recurrent_initializer='glorot_uniform')
self.fc = layers.Dense(vocab_size)
# Attention layers
self.attention = layers.AdditiveAttention()
def call(self, features, captions, hidden):
# features shape: (batch_size, 64, 2048)
# captions shape: (batch_size, max_length)
caption_embeddings = self.embedding(captions)
# caption_embeddings shape: (batch_size, max_length, embedding_dim)
# Apply attention
context_vector = self.attention([caption_embeddings, features])
# context_vector shape: (batch_size, max_length, units)
# Concatenate context and caption embeddings
combined_input = tf.concat([context_vector, caption_embeddings], axis=-1)
# Pass through LSTM
output, state_h, state_c = self.lstm(combined_input, initial_state=hidden)
# Pass through Dense layer
output = self.fc(output)
return output, state_h, state_c
def reset_state(self, batch_size):
return [tf.zeros((batch_size, self.units)),
tf.zeros((batch_size, self.units))]
Instantiate the decoder
decoder = RNN_Decoder(vocab_size, embedding_dim, units)
Loss function
loss_object = tf.keras.losses.SparseCategoricalCrossentropy(
from_logits=True, reduction='none')
def loss_function(real, pred):
mask = tf.math.logical_not(tf.math.equal(real, 0)) # Padding mask
loss_ = loss_object(real, pred)
mask = tf.cast(mask, dtype=loss_.dtype)
loss_ *= mask
return tf.reduce_mean(loss_)
optimizer = tf.keras.optimizers.Adam()
import tensorflow as tf
train_loss = tf.keras.metrics.Mean(name='train_loss')
BATCH_SIZE = 64
BUFFER_SIZE = 1000
Function to load image features for training
def map_func_train(img_id, seq):
img_tensor = image_features[img_id.decode('utf-8')][0]
return img_tensor, seq
Create TensorFlow dataset for training
dataset = tf.data.Dataset.from_tensor_slices((train_image_ids, train_sequences))
dataset = dataset.map(lambda img, cap: tf.py_function(map_func_train, [img, cap],
[tf.float32, tf.int32]),
num_parallel_calls=tf.data.AUTOTUNE)
dataset = dataset.shuffle(BUFFER_SIZE).batch(BATCH_SIZE)
dataset = dataset.prefetch(buffer_size=tf.data.AUTOTUNE)
Create TensorFlow dataset for validation
def map_func_val(img_id, seq):
img_tensor = image_features[img_id.decode('utf-8')][0]
return img_tensor, seq
val_dataset = tf.data.Dataset.from_tensor_slices((val_image_ids, val_sequences))
val_dataset = val_dataset.map(lambda img, cap: tf.py_function(map_func_val, [img, cap],
[tf.float32, tf.int32]),
num_parallel_calls=tf.data.AUTOTUNE)
val_dataset = val_dataset.batch(BATCH_SIZE)
val_dataset = val_dataset.prefetch(buffer_size=tf.data.AUTOTUNE)
@tf.function
def train_step(img_tensor, target):
with tf.GradientTape() as tape:
# Initialize hidden state
hidden = decoder.reset_state(batch_size=target.shape[0])
# Pass image features and caption to the decoder
predictions, _, _ = decoder(img_tensor, target[:, :-1], hidden)
# Compute the loss
loss = loss_function(target[:, 1:], predictions)
# Compute gradients
trainable_variables = decoder.trainable_variables
gradients = tape.gradient(loss, trainable_variables)
# Clip gradients to prevent exploding gradients
gradients, _ = tf.clip_by_global_norm(gradients, 5.0)
# Update weights
optimizer.apply_gradients(zip(gradients, trainable_variables))
# Update the metric
train_loss_metric(loss)
@tf.function
def val_step(img_tensor, target):
loss = 0
# Initialize hidden state
hidden = decoder.reset_state(batch_size=target.shape[0])
# Pass image features and caption to the decoder
predictions, _, _ = decoder(img_tensor, target[:, :-1], hidden)
# Compute the loss
loss = loss_function(target[:, 1:], predictions)
# Update the validation loss metric
val_loss(loss)
give next step and codesASSISTANT
Great progress so far! You've successfully prepared your dataset, extracted image features, tokenized captions, and built the decoder model with attention. The next steps involve:
1. **Completing the Training Loop**: Train the model over multiple epochs, track training and validation loss, and save checkpoints.
2. **Saving the Model**: Save the trained model and tokenizer for future use.
3. **Inference**: Implement a function to generate captions for new images.
4. **Evaluation (Optional)**: Evaluate your model's performance using metrics like BLEU scores.
Let's proceed step-by-step.
---
## 1. Complete the Training Loop
### a. Initialize Validation Loss Metric
First, ensure that you have a metric to track validation loss.
```python
val_loss = tf.keras.metrics.Mean(name='val_loss')
```
### b. Define the Number of Epochs and Checkpointing
Set the number of epochs and configure checkpoints to save the best model based on validation loss.
```python
EPOCHS = 20 # You can adjust this based on your requirements
# Define checkpoint path
checkpoint_path = os.path.join(drive_path, "checkpoints/train")
# Create a checkpoint directory
if not os.path.exists(checkpoint_path):
os.makedirs(checkpoint_path)
# Initialize the checkpoint
ckpt = tf.train.Checkpoint(encoder=image_features_extract_model,
decoder=decoder,
optimizer=optimizer)
# Restore the latest checkpoint (if any)
ckpt_manager = tf.train.CheckpointManager(ckpt, checkpoint_path, max_to_keep=5)
if ckpt_manager.latest_checkpoint:
ckpt.restore(ckpt_manager.latest_checkpoint)
print('Latest checkpoint restored!')
```
### c. Define the Training Loop
Implement the training loop to iterate over epochs and batches, perform training and validation steps, and track losses.
```python
for epoch in range(EPOCHS):
start = tf.timestamp()
train_loss.reset_states()
val_loss.reset_states()
# Training
for (batch, (img_tensor, target)) in enumerate(dataset):
train_step(img_tensor, target)
if batch % 100 == 0:
print(f'Epoch {epoch+1} Batch {batch} Loss {train_loss.result():.4f}')
# Validation
for (batch, (img_tensor, target)) in enumerate(val_dataset):
val_step(img_tensor, target)
# Save the checkpoint if the validation loss is the best so far
if ckpt_manager.latest_checkpoint:
ckpt_manager.save()
print(f'Epoch {epoch+1} Loss {train_loss.result():.4f} Validation Loss {val_loss.result():.4f}')
print(f'Time taken for 1 epoch {tf.timestamp() - start} sec\n')
```
**Explanation:**
- **Epoch Loop**: Iterate over the number of epochs.
- **Training Phase**: For each batch in the training dataset, perform a training step and occasionally print the loss.
- **Validation Phase**: After training, iterate over the validation dataset to compute validation loss.
- **Checkpointing**: Save the model checkpoint after each epoch.
- **Logging**: Print training and validation loss for each epoch along with the time taken.
---
## 2. Saving the Model and Tokenizer
After training, it's essential to save both the model and the tokenizer for future use.
### a. Save the Trained Decoder
```python
# Save the decoder model
decoder.save(os.path.join(drive_path, 'decoder_model.h5'))
print("Decoder model saved.")
```
### b. Save the Tokenizer
You've already saved the tokenizer earlier, but ensure it's correctly saved.
```python
# Ensure tokenizer is saved
with open(os.path.join(drive_path, 'tokenizer.json'), 'w') as f:
f.write(tokenizer_json)
print("Tokenizer saved.")
```
---
## 3. Inference: Generating Captions for New Images
To generate captions for new images, you'll need to define an encoder (image feature extractor) and the trained decoder. Here's how you can implement the inference pipeline:
### a. Load the Tokenizer and Models
```python
import tensorflow as tf
from tensorflow.keras.models import load_model
from tensorflow.keras.preprocessing.sequence import pad_sequences
import json
# Load the tokenizer
with open(os.path.join(drive_path, 'tokenizer.json'), 'r') as f:
tokenizer_json = f.read()
tokenizer = tf.keras.preprocessing.text.tokenizer_from_json(tokenizer_json)
# Load the decoder
decoder = load_model(os.path.join(drive_path, 'decoder_model.h5'), compile=False)
print("Decoder model loaded.")
# Load the image feature extractor (encoder)
image_features_extract_model = tf.keras.models.load_model(
os.path.join(drive_path, 'image_features_extract_model.h5'),
compile=False
)
print("Image feature extractor loaded.")
```
> **Note**: If you saved the `image_features_extract_model` as a `tf.keras.Model`, ensure you save and load it appropriately. You might need to adjust the loading mechanism based on how you saved the model.
### b. Define the Caption Generation Function
```python
def load_image_for_inference(image_path):
img = tf.io.read_file(image_path)
img = tf.image.decode_jpeg(img, channels=3)
img = tf.image.resize(img, (299, 299))
img = preprocess_input(img)
img = tf.expand_dims(img, 0) # Add batch dimension
return img
def generate_caption(image_path, max_length=50):
# Extract features
img_tensor = load_image_for_inference(image_path)
img_features = image_features_extract_model(img_tensor)
img_features = tf.reshape(img_features, (img_features.shape[0], -1, img_features.shape[3]))
# Initialize the decoder's hidden state
hidden = decoder.reset_state(batch_size=1)
# Start token
input_seq = tf.constant([tokenizer.word_index['']]) # Assuming '' is the start token
result = []
for i in range(max_length):
# Pass through the decoder
predictions, hidden_h, hidden_c = decoder(img_features, input_seq, hidden)
# Get the predicted id
predicted_id = tf.argmax(predictions[0, -1, :]).numpy()
# Map the id to the word
predicted_word = tokenizer.index_word.get(predicted_id, '')
if predicted_word == '':
break
result.append(predicted_word)
# Update the input sequence
input_seq = tf.expand_dims([predicted_id], 0)
return ' '.join(result)
```
**Explanation:**
- **Feature Extraction**: Preprocess the input image and extract its features using the image feature extractor.
- **Caption Generation**: Iteratively generate words by feeding the predicted word back into the decoder until the maximum length is reached or the end token is generated.
- **Start Token**: Replace `''` with your actual start token if different.
### c. Test the Caption Generation
```python
# Path to a new image
new_image_path = os.path.join(image_dir, 'example.jpg') # Replace with your image path
# Generate caption
caption = generate_caption(new_image_path)
print("Generated Caption:", caption)
```
> **Note**: Replace `'example.jpg'` with the actual image filename you want to caption.
---
## 4. Optional: Evaluating the Model
To evaluate your model's performance, consider using metrics like BLEU, METEOR, or CIDEr. Here's an example using the BLEU score with the `nltk` library.
### a. Install and Import NLTK
```python
!pip install nltk
import nltk
from nltk.translate.bleu_score import corpus_bleu
# Download necessary NLTK data
nltk.download('punkt')
```
### b. Prepare References and Hypotheses
```python
# Prepare references (ground truth) and hypotheses (generated captions)
references = []
hypotheses = []
for img_id, caps in captions_dict.items():
img_path = os.path.join(image_dir, img_id)
for cap in caps:
# Generate caption
generated_caption = generate_caption(img_path)
# Tokenize reference and hypothesis
references.append([cap.split()])
hypotheses.append(generated_caption.split())
# Calculate BLEU score
bleu_score = corpus_bleu(references, hypotheses)
print(f'BLEU Score: {bleu_score:.4f}')
```
**Explanation:**
- **References**: Ground truth captions for each image.
- **Hypotheses**: Captions generated by your model.
- **BLEU Score**: Measures how closely the generated captions match the references.
> **Caution**: Generating captions for the entire dataset during evaluation can be time-consuming. Consider using a subset for quicker evaluation.
---
## Full Code Overview
Here's a summary of the complete workflow, including the newly added sections:
```python
import os
import zipfile
import tensorflow as tf
from tensorflow.keras.applications.inception_v3 import InceptionV3, preprocess_input
from tensorflow.keras.preprocessing.text import Tokenizer
from tensorflow.keras.preprocessing.sequence import pad_sequences
from tensorflow.keras import layers
import pickle
import tqdm
import numpy as np
import json
import string
import nltk
from nltk.translate.bleu_score import corpus_bleu
# Set paths
drive_path = '/content/drive/MyDrive'
archive_zip = os.path.join(drive_path, 'archive.zip')
extract_path = os.path.join(drive_path, 'archive_extracted')
# Extract archive
if not os.path.exists(extract_path):
os.makedirs(extract_path)
with zipfile.ZipFile(archive_zip, 'r') as zip_ref:
zip_ref.extractall(extract_path)
print("Extraction Complete.")
# Define paths
archive_path = os.path.join(extract_path)
image_dir = os.path.join(archive_path, 'Images')
captions_file = os.path.join(archive_path, 'captions.txt')
# Verify files
print("Images Directory:", os.listdir(image_dir)[:5])
print("Captions File Exists:", os.path.exists(captions_file))
# Load and preprocess captions
def load_doc(filename):
with open(filename, 'r') as file:
text = file.read()
return text
def preprocess_caption(caption):
caption = caption.lower()
caption = caption.translate(str.maketrans('', '', string.punctuation))
caption = caption.strip()
caption = ' ' + caption + ' '
return caption
captions = load_doc(captions_file)
captions_dict = {}
for line in captions.split('\n'):
if len(line) < 1:
continue
if line.startswith('image'):
continue
tokens = line.split(',', 1)
if len(tokens) != 2:
print(f"Skipping malformed line: {line}")
continue
image_id, caption = tokens
image_id = image_id.strip()
caption = caption.strip()
if image_id not in captions_dict:
captions_dict[image_id] = []
captions_dict[image_id].append(preprocess_caption(caption))
# Display sample captions
for key, val in list(captions_dict.items())[:5]:
print(f"Image ID: {key}")
for cap in val:
print(f"Caption: {cap}")
print('\n')
# Preprocess images using InceptionV3
def load_image(image_path):
img = tf.io.read_file(image_path)
img = tf.image.decode_jpeg(img, channels=3)
img = tf.image.resize(img, (299, 299))
img = preprocess_input(img)
return img
# Test image loading
sample_image_id = list(captions_dict.keys())[0]
sample_image_path = os.path.join(image_dir, sample_image_id)
sample_image = load_image(sample_image_path)
print(f"Sample Image Shape: {sample_image.shape}")
# Load InceptionV3 model
image_model = InceptionV3(include_top=False, weights='imagenet')
new_input = image_model.input
hidden_layer = image_model.layers[-1].output
image_features_extract_model = tf.keras.Model(new_input, hidden_layer)
# Extract features and save
features_path = os.path.join(drive_path, 'image_features.pkl')
if not os.path.exists(features_path):
image_filenames = list(captions_dict.keys())
image_features = {}
for img_name in tqdm.tqdm(image_filenames):
img_path = os.path.join(image_dir, img_name)
img_tensor = load_image(img_path)
img_tensor = tf.expand_dims(img_tensor, 0)
img_features = image_features_extract_model(img_tensor)
img_features = tf.reshape(img_features, (img_features.shape[0], -1, img_features.shape[3]))
image_features[img_name] = img_features.numpy()
with open(features_path, 'wb') as f:
pickle.dump(image_features, f)
print("Image Features Extracted and Saved.")
else:
with open(features_path, 'rb') as f:
image_features = pickle.load(f)
print("Image Features Loaded from Disk.")
print("Number of images with extracted features:", len(image_features))
# Tokenize captions
all_captions = []
for key in captions_dict:
for cap in captions_dict[key]:
all_captions.append(cap)
print("Total Captions:", len(all_captions))
tokenizer = Tokenizer(num_words=5000, oov_token="", filters='!"#$%&()*+.,-/:;=?@[]^_`{|}~ ')
tokenizer.fit_on_texts(all_captions)
# Add token to word index
tokenizer.word_index[''] = 0
tokenizer.index_word[0] = ''
# Save tokenizer
tokenizer_json = tokenizer.to_json()
with open(os.path.join(drive_path, 'tokenizer.json'), 'w') as f:
f.write(tokenizer_json)
# Convert captions to sequences
train_seqs = tokenizer.texts_to_sequences(all_captions)
# Pad sequences
max_length = max(len(seq) for seq in train_seqs)
print("Maximum Caption Length:", max_length)
train_seqs = pad_sequences(train_seqs, maxlen=max_length, padding='post')
print("Sample Padded Sequence:", train_seqs[0])
vocab_size = len(tokenizer.word_index) + 1
print("Vocabulary Size:", vocab_size)
# Split data
image_ids = []
captions_list = []
for key in captions_dict:
for cap in captions_dict[key]:
image_ids.append(key)
captions_list.append(cap)
sequences = tokenizer.texts_to_sequences(captions_list)
sequences = pad_sequences(sequences, maxlen=max_length, padding='post')
image_ids = np.array(image_ids)
sequences = np.array(sequences)
# Shuffle data
dataset_size = len(image_ids)
indices = np.arange(dataset_size)
np.random.shuffle(indices)
image_ids = image_ids[indices]
sequences = sequences[indices]
# Split into training and validation
split_index = int(0.9 * dataset_size)
train_image_ids = image_ids[:split_index]
train_sequences = sequences[:split_index]
val_image_ids = image_ids[split_index:]
val_sequences = sequences[split_index:]
print(f"Training Samples: {len(train_image_ids)}")
print(f"Validation Samples: {len(val_image_ids)}")
# Define the RNN Decoder with Attention
class RNN_Decoder(tf.keras.Model):
def __init__(self, vocab_size, embedding_dim, units):
super(RNN_Decoder, self).__init__()
self.units = units
self.embedding = layers.Embedding(vocab_size, embedding_dim)
self.lstm = layers.LSTM(self.units,
return_sequences=True,
return_state=True,
recurrent_initializer='glorot_uniform')
self.fc = layers.Dense(vocab_size)
# Attention layers
self.attention = layers.AdditiveAttention()
def call(self, features, captions, hidden):
# features shape: (batch_size, 64, 2048)
# captions shape: (batch_size, max_length)
caption_embeddings = self.embedding(captions)
# caption_embeddings shape: (batch_size, max_length, embedding_dim)
# Apply attention
context_vector = self.attention([caption_embeddings, features])
# context_vector shape: (batch_size, max_length, units)
# Concatenate context and caption embeddings
combined_input = tf.concat([context_vector, caption_embeddings], axis=-1)
# Pass through LSTM
output, state_h, state_c = self.lstm(combined_input, initial_state=hidden)
# Pass through Dense layer
output = self.fc(output)
return output, state_h, state_c
def reset_state(self, batch_size):
return [tf.zeros((batch_size, self.units)),
tf.zeros((batch_size, self.units))]
# Instantiate the decoder
embedding_dim = 256
units = 512
decoder = RNN_Decoder(vocab_size, embedding_dim, units)
# Define the loss and optimizer
loss_object = tf.keras.losses.SparseCategoricalCrossentropy(
from_logits=True, reduction='none')
def loss_function(real, pred):
mask = tf.math.logical_not(tf.math.equal(real, 0)) # Padding mask
loss_ = loss_object(real, pred)
mask = tf.cast(mask, dtype=loss_.dtype)
loss_ *= mask
return tf.reduce_mean(loss_)
optimizer = tf.keras.optimizers.Adam()
# Define metrics
train_loss_metric = tf.keras.metrics.Mean(name='train_loss')
val_loss_metric = tf.keras.metrics.Mean(name='val_loss')
# Batch and buffer size
BATCH_SIZE = 64
BUFFER_SIZE = 1000
# Function to load image features for training
def map_func_train(img_id, seq):
img_tensor = image_features[img_id.decode('utf-8')][0]
return img_tensor, seq
# Create TensorFlow dataset for training
dataset = tf.data.Dataset.from_tensor_slices((train_image_ids, train_sequences))
dataset = dataset.map(lambda img, cap: tf.py_function(map_func_train, [img, cap],
[tf.float32, tf.int32]),
num_parallel_calls=tf.data.AUTOTUNE)
dataset = dataset.shuffle(BUFFER_SIZE).batch(BATCH_SIZE)
dataset = dataset.prefetch(buffer_size=tf.data.AUTOTUNE)
# Create TensorFlow dataset for validation
def map_func_val(img_id, seq):
img_tensor = image_features[img_id.decode('utf-8')][0]
return img_tensor, seq
val_dataset = tf.data.Dataset.from_tensor_slices((val_image_ids, val_sequences))
val_dataset = val_dataset.map(lambda img, cap: tf.py_function(map_func_val, [img, cap],
[tf.float32, tf.int32]),
num_parallel_calls=tf.data.AUTOTUNE)
val_dataset = val_dataset.batch(BATCH_SIZE)
val_dataset = val_dataset.prefetch(buffer_size=tf.data.AUTOTUNE)
# Define the train and validation steps
@tf.function
def train_step(img_tensor, target):
with tf.GradientTape() as tape:
# Initialize hidden state
hidden = decoder.reset_state(batch_size=target.shape[0])
# Pass image features and caption to the decoder
predictions, _, _ = decoder(img_tensor, target[:, :-1], hidden)
# Compute the loss
loss = loss_function(target[:, 1:], predictions)
# Compute gradients
trainable_variables = decoder.trainable_variables
gradients = tape.gradient(loss, trainable_variables)
# Clip gradients to prevent exploding gradients
gradients, _ = tf.clip_by_global_norm(gradients, 5.0)
# Update weights
optimizer.apply_gradients(zip(gradients, trainable_variables))
# Update the metric
train_loss_metric(loss)
@tf.function
def val_step(img_tensor, target):
# Initialize hidden state
hidden = decoder.reset_state(batch_size=target.shape[0])
# Pass image features and caption to the decoder
predictions, _, _ = decoder(img_tensor, target[:, :-1], hidden)
# Compute the loss
loss = loss_function(target[:, 1:], predictions)
# Update the validation loss metric
val_loss_metric(loss)
# Define training loop with checkpointing
EPOCHS = 20
# Define checkpoint path
checkpoint_path = os.path.join(drive_path, "checkpoints/train")
# Create a checkpoint directory
if not os.path.exists(checkpoint_path):
os.makedirs(checkpoint_path)
# Initialize the checkpoint
ckpt = tf.train.Checkpoint(encoder=image_features_extract_model,
decoder=decoder,
optimizer=optimizer)
# Restore the latest checkpoint (if any)
ckpt_manager = tf.train.CheckpointManager(ckpt, checkpoint_path, max_to_keep=5)
if ckpt_manager.latest_checkpoint:
ckpt.restore(ckpt_manager.latest_checkpoint)
print('Latest checkpoint restored!')
for epoch in range(EPOCHS):
start = tf.timestamp()
train_loss_metric.reset_states()
val_loss_metric.reset_states()
# Training
for (batch, (img_tensor, target)) in enumerate(dataset):
train_step(img_tensor, target)
if batch % 100 == 0:
print(f'Epoch {epoch+1} Batch {batch} Loss {train_loss_metric.result():.4f}')
# Validation
for (batch, (img_tensor, target)) in enumerate(val_dataset):
val_step(img_tensor, target)
# Save the checkpoint
ckpt_manager.save()
print(f'Epoch {epoch+1} Loss {train_loss_metric.result():.4f} Validation Loss {val_loss_metric.result():.4f}')
print(f'Time taken for 1 epoch {tf.timestamp() - start} sec\n')
# After training, save the decoder and tokenizer
decoder.save(os.path.join(drive_path, 'decoder_model.h5'))
print("Decoder model saved.")
with open(os.path.join(drive_path, 'tokenizer.json'), 'w') as f:
f.write(tokenizer_json)
print("Tokenizer saved.")
# Inference functions
def load_image_for_inference(image_path):
img = tf.io.read_file(image_path)
img = tf.image.decode_jpeg(img, channels=3)
img = tf.image.resize(img, (299, 299))
img = preprocess_input(img)
img = tf.expand_dims(img, 0) # Add batch dimension
return img
def generate_caption(image_path, max_length=50):
# Extract features
img_tensor = load_image_for_inference(image_path)
img_features = image_features_extract_model(img_tensor)
img_features = tf.reshape(img_features, (img_features.shape[0], -1, img_features.shape[3]))
# Initialize the decoder's hidden state
hidden = decoder.reset_state(batch_size=1)
# Start token
input_seq = tf.constant([tokenizer.word_index['']]) # Assuming '' is the start token
result = []
for i in range(max_length):
# Pass through the decoder
predictions, hidden_h, hidden_c = decoder(img_features, input_seq, hidden)
# Get the predicted id
predicted_id = tf.argmax(predictions[0, -1, :]).numpy()
# Map the id to the word
predicted_word = tokenizer.index_word.get(predicted_id, '')
if predicted_word == '':
break
result.append(predicted_word)
# Update the input sequence
input_seq = tf.expand_dims([predicted_id], 0)
return ' '.join(result)
# Example usage
new_image_path = os.path.join(image_dir, 'example.jpg') # Replace with your image path
caption = generate_caption(new_image_path)
print("Generated Caption:", caption)
# Optional: Evaluate the model using BLEU score
nltk.download('punkt')
references = []
hypotheses = []
for img_id, caps in captions_dict.items():
img_path = os.path.join(image_dir, img_id)
for cap in caps:
# Generate caption
generated_caption = generate_caption(img_path)
# Tokenize reference and hypothesis
references.append([cap.split()])
hypotheses.append(generated_caption.split())
# Calculate BLEU score
bleu_score = corpus_bleu(references, hypotheses)
print(f'BLEU Score: {bleu_score:.4f}')
```
---
## Tips and Best Practices
1. **Start and End Tokens**: It's common to use special start (`<start>`) and end (`<end>`) tokens in captions. Ensure consistent usage during training and inference.
2. **Beam Search**: Instead of greedy search (choosing the highest probability word at each step), beam search can generate more accurate captions by considering multiple possible sequences.
3. **Early Stopping**: Monitor validation loss and implement early stopping to prevent overfitting.
4. **Fine-Tuning**: Consider fine-tuning the encoder (InceptionV3) along with the decoder for better performance.
5. **Data Augmentation**: Apply data augmentation techniques to images to increase dataset variability.
6. **Hyperparameter Tuning**: Experiment with different hyperparameters like `embedding_dim`, `units`, `batch_size`, and learning rates to optimize performance.
7. **Save Model Architecture**: Along with saving model weights, consider saving the entire model architecture for easier loading and inference.
---
Feel free to reach out if you encounter any issues or need further assistance. Happy coding!