Error: C:\Users\Dell\AI\pythonProject\.venv\Scripts\python.exe C:\Users\Dell\AI\pythonProject\!EridAI\VLM\main.py 
2024-10-06 18:09:49.813418: I tensorflow/core/util/port.cc:153] oneDNN custom operations are on. You may see slightly different numerical results due to floating-point round-off errors from different computation orders. To turn them off, set the environment variable `TF_ENABLE_ONEDNN_OPTS=0`.
2024-10-06 18:09:52.082836: I tensorflow/core/util/port.cc:153] oneDNN custom operations are on. You may see slightly different numerical results due to floating-point round-off errors from different computation orders. To turn them off, set the environment variable `TF_ENABLE_ONEDNN_OPTS=0`.
Default image_tensor shape: torch.Size([3, 224, 224])
Error: Sample 0 has 'image' of type <class 'list'>
Traceback (most recent call last):
  File "C:\Users\Dell\AI\pythonProject\!EridAI\VLM\main.py", line 254, in <module>
    main()
  File "C:\Users\Dell\AI\pythonProject\!EridAI\VLM\main.py", line 153, in main
    raise e
  File "C:\Users\Dell\AI\pythonProject\!EridAI\VLM\main.py", line 147, in main
    assert isinstance(sample['image'], torch.Tensor), f"Sample {i} has 'image' of type {type(sample['image'])}"
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AssertionError: Sample 0 has 'image' of type <class 'list'>

Process finished with exit code 1
 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

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 = images[0]

            # Ensure 'image' is not a list
            if isinstance(image, list):
                image = image[0]
                print(f"Warning: 'image' is a list, taking the first element.")

            # 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
            # Always pass a list of one image to ensure consistent processing
            encoding = processor(images=[image], return_tensors="pt")
            image_tensor = encoding['pixel_values']

            # 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}")

    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(10):
        sample = processed_dataset[i]
        assert isinstance(sample['image'], torch.Tensor), f"Sample {i} has 'image' of type {type(sample['image'])}"
        assert sample['image'].dim() == 3, 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()