Error: Default image_tensor shape: torch.Size([3, 224, 224])
Traceback (most recent call last):
  File "C:\Users\Dell\AI\pythonProject\!EridAI\VLM\main.py", line 270, in <module>
    main()
  File "C:\Users\Dell\AI\pythonProject\!EridAI\VLM\main.py", line 166, in main
    raise AssertionError(f"Sample {i} has 'image' of type {type(sample['image'])}")
AssertionError: Sample 0 has 'image' of type <class 'list'>
Validation Error: Sample 0 has 'image' of type <class 'list'> FULL CODE:import torch
from torch import nn
from torch.utils.data import DataLoader
from transformers import GPT2Tokenizer, GPT2LMHeadModel, CLIPProcessor, CLIPModel
from datasets import load_dataset
from PIL import Image
import numpy as np

def flatten_first_element(obj):
    while isinstance(obj, list):
        if not obj:  # Empty list
            return None
        obj = obj[0]
    return obj

class MultimodalGPT2(nn.Module):
    def __init__(self, gpt2_model_name, clip_model_name):
        super(MultimodalGPT2, self).__init__()
        self.clip = CLIPModel.from_pretrained(clip_model_name)
        self.gpt2 = GPT2LMHeadModel.from_pretrained(gpt2_model_name)
        self.adapter = nn.Linear(self.clip.config.projection_dim, self.gpt2.config.n_embd)

    def forward(self, input_ids, attention_mask, image=None):
        if image is not None:
            image_features = self.clip.get_image_features(pixel_values=image)
            adapter_features = self.adapter(image_features)
            adapter_features = adapter_features.unsqueeze(1).expand(-1, input_ids.size(1), -1)
            token_embeddings = self.gpt2.transformer.wte(input_ids)
            inputs_embeds = token_embeddings + adapter_features
            outputs = self.gpt2(
                inputs_embeds=inputs_embeds,
                attention_mask=attention_mask,
                labels=input_ids
            )
        else:
            outputs = self.gpt2(
                input_ids=input_ids,
                attention_mask=attention_mask,
                labels=input_ids
            )
        return outputs

def preprocess_data(example, processor, tokenizer, default_image):
    images = example.get('images', [])
    if images:
        try:
            image = flatten_first_element(images)

            if image is None:
                print("Warning: 'images' list is empty. Using default image.")
                image_tensor = default_image
            else:
                # Convert to PIL Image if it's a path or ensure it's in RGB
                if isinstance(image, Image.Image):
                    image = image.convert('RGB')
                else:
                    image = Image.open(image).convert('RGB')

                # Process image with CLIPProcessor
                encoding = processor(images=[image], return_tensors="pt")
                image_tensor = encoding.get('pixel_values', None)

                if image_tensor is None:
                    print("Warning: 'pixel_values' not found in encoding. Using default image.")
                    image_tensor = default_image
                else:
                    # Ensure 'pixel_values' is a tensor of shape (1, C, H, W)
                    if isinstance(image_tensor, torch.Tensor):
                        # Remove the batch dimension
                        image_tensor = image_tensor.squeeze(0)  # Shape: (C, H, W)
                        print(f"Processed image_tensor shape: {image_tensor.shape}")
                    else:
                        print(f"Warning: 'pixel_values' is not a Tensor. Type: {type(image_tensor)}. Converting to Tensor.")
                        image_tensor = torch.tensor(image_tensor)

                    # Validate the dimensions of 'image_tensor'
                    if image_tensor.dim() != 3:
                        print(f"Warning: 'image_tensor' has incorrect dimensions: {image_tensor.dim()}. Using default image.")
                        image_tensor = default_image

        except Exception as e:
            print(f"Failed to process image: {e}. Using default image.")
            image_tensor = default_image
    else:
        image_tensor = default_image

    # Combine 'user' and 'assistant' texts
    text = (example.get('user', '') + " " + example.get('assistant', '')).strip()
    encoding = tokenizer(
        text, truncation=True, padding='max_length', max_length=128
    )

    # Debugging: Check types and shapes
    if not isinstance(image_tensor, torch.Tensor):
        print(f"Warning: 'image_tensor' is not a Tensor. Type: {type(image_tensor)}")
    elif image_tensor.dim() != 3:
        print(f"Warning: 'image_tensor' has incorrect dimensions. Shape: {image_tensor.shape}")

    # Final assertion to ensure 'image' is a Tensor
    if not isinstance(image_tensor, torch.Tensor):
        print("Error: 'image_tensor' is not a Tensor even after processing. Using default image.")
        image_tensor = default_image
    elif image_tensor.dim() != 3:
        print("Error: 'image_tensor' has incorrect dimensions even after processing. Using default image.")
        image_tensor = default_image

    return {
        'input_ids': encoding['input_ids'],
        'attention_mask': encoding['attention_mask'],
        'image': image_tensor
    }

def collate_fn(batch):
    # Explicitly specify dtype for input_ids and attention_mask
    input_ids = torch.tensor([item['input_ids'] for item in batch], dtype=torch.long)
    attention_mask = torch.tensor([item['attention_mask'] for item in batch], dtype=torch.long)

    images = []
    for idx, item in enumerate(batch):
        img = item['image']
        if img is not None:
            if isinstance(img, torch.Tensor):
                images.append(img)
            else:
                print(f"Error: 'image' at index {idx} is not a Tensor. Type: {type(img)}")
                raise TypeError(f"Expected Tensor but got {type(img)} at index {idx}")
        else:
            print(f"Error: 'image' at index {idx} is None.")
            raise ValueError(f"'Image' at index {idx} is None.")

    try:
        images = torch.stack(images)  # Now shape: (batch_size, C, H, W)
    except Exception as e:
        print(f"Failed to stack images: {e}")
        raise e

    return {'input_ids': input_ids, 'attention_mask': attention_mask, 'image': images}

def main():
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    dataset = load_dataset('HuggingFaceM4/the_cauldron', 'ai2d', split='train')

    tokenizer = GPT2Tokenizer.from_pretrained('gpt2')
    tokenizer.pad_token = tokenizer.eos_token

    processor = CLIPProcessor.from_pretrained('openai/clip-vit-base-patch32')

    # Define a default image (e.g., a black image)
    default_pil_image = Image.fromarray(np.uint8(np.zeros((224, 224, 3))))
    encoding = processor(images=[default_pil_image], return_tensors="pt")
    default_image = encoding['pixel_values'].squeeze(0)  # Shape: (C, H, W)
    print(f"Default image_tensor shape: {default_image.shape}")

    # Preprocess the dataset with the default image
    processed_dataset = dataset.map(
        lambda x: preprocess_data(x, processor, tokenizer, default_image),
        remove_columns=dataset.column_names,
        num_proc=1  # Set to 1 for initial debugging
    )

    # Validate the processed dataset
    for i in range(min(100, len(processed_dataset))):
        sample = processed_dataset[i]
        if not isinstance(sample['image'], torch.Tensor):
            print(f"Validation Error: Sample {i} has 'image' of type {type(sample['image'])}")
            raise AssertionError(f"Sample {i} has 'image' of type {type(sample['image'])}")
        if sample['image'].dim() != 3:
            print(f"Validation Error: Sample {i} has 'image' with dimensions {sample['image'].shape}")
            raise AssertionError(f"Sample {i} has 'image' with dimensions {sample['image'].shape}")
    print("All checked samples have 'image' as Tensor with correct dimensions.")

    # Create DataLoader with enhanced collate_fn
    dataloader = DataLoader(
        processed_dataset,
        batch_size=8,
        shuffle=True,
        collate_fn=collate_fn,
        num_workers=4,  # Adjust based on your CPU cores
        pin_memory=True  # Speeds up data transfer to GPU
    )

    model = MultimodalGPT2('gpt2', 'openai/clip-vit-base-patch32').to(device)

    optimizer = torch.optim.AdamW(model.parameters(), lr=5e-5)  # Use PyTorch's AdamW

    model.train()
    for epoch in range(1):
        for batch_idx, batch in enumerate(dataloader):
            input_ids = batch['input_ids'].to(device)
            attention_mask = batch['attention_mask'].to(device)
            images = batch['image'].to(device)  # Ensured to have the same batch size

            outputs = model(input_ids=input_ids, attention_mask=attention_mask, image=images)
            loss = outputs.loss
            loss.backward()
            optimizer.step()
            optimizer.zero_grad()
            print(f"Epoch {epoch + 1}, Batch {batch_idx + 1}: Loss = {loss.item()}")

    torch.save(model.state_dict(), 'multimodal_gpt2.pth')
    model.eval()

    def generate(prompt_text=None, image_path=None):
        if prompt_text is not None:
            input_ids = tokenizer.encode(prompt_text, return_tensors='pt').to(device)
            attention_mask = torch.ones_like(input_ids).to(device)
        else:
            input_ids = torch.tensor([[tokenizer.eos_token_id]], dtype=torch.long).to(device)
            attention_mask = torch.ones_like(input_ids).to(device)

        if image_path is not None:
            try:
                image = Image.open(image_path).convert('RGB')
                encoding = processor(images=[image], return_tensors="pt")
                image_tensor = encoding['pixel_values']
                if isinstance(image_tensor, torch.Tensor):
                    image_tensor = image_tensor.squeeze(0)  # Shape: (C, H, W)
                else:
                    print(f"Warning: 'pixel_values' is not a Tensor. Type: {type(image_tensor)}. Converting to Tensor.")
                    image_tensor = torch.tensor(image_tensor)
                image_tensor = image_tensor.to(device)
                image_tensor = image_tensor.unsqueeze(0)  # Shape: (1, C, H, W)
                print(f"Generation using image, image_tensor shape: {image_tensor.shape}")
            except Exception as e:
                print(f"Error processing image: {e}. Using default image.")
                image_tensor = default_image.unsqueeze(0).to(device)
        else:
            # Use default image if no image_path is provided
            image_tensor = default_image.unsqueeze(0).to(device)

        with torch.no_grad():
            if image_tensor is not None:
                # Expand image to match batch size if necessary
                if image_tensor.size(0) != input_ids.size(0):
                    image_tensor = image_tensor.expand(input_ids.size(0), -1, -1, -1)
                image_features = model.clip.get_image_features(pixel_values=image_tensor)
                adapter_features = model.adapter(image_features)  # Shape: (batch_size, projection_dim)
                # Expand adapter_features to match sequence length
                adapter_features = adapter_features.unsqueeze(1).expand(-1, input_ids.size(1), -1)
                # Get token embeddings
                token_embeddings = model.gpt2.transformer.wte(input_ids)  # LongTensor
                # Combine token embeddings with adapter features
                inputs_embeds = token_embeddings + adapter_features
                # Generate with inputs_embeds
                outputs = model.gpt2.generate(
                    inputs_embeds=inputs_embeds,
                    attention_mask=attention_mask,
                    max_length=50
                )
            else:
                outputs = model.gpt2.generate(
                    input_ids=input_ids,
                    attention_mask=attention_mask,
                    max_length=50
                )

        return tokenizer.decode(outputs[0], skip_special_tokens=True)

    # Examples of generation
    print(generate(
        prompt_text="Question: What do respiration and combustion give out? Choices: A. Oxygen B. Carbon dioxide C. Nitrogen D. Heat Answer with the letter."
    ))
    print(generate(image_path='example.jpg'))
    print(generate(
        prompt_text="Question: What do respiration and combustion give out? Choices: A. Oxygen B. Carbon dioxide C. Nitrogen D. Heat Answer with the letter.",
        image_path='example.jpg'
    ))

if __name__ == "__main__":
    main()