Build a Vision-Language Model (VLM) from DeepSeek and OpenAI CLIP 实例
·
本文为将一个文本大模型（LLM）转化为多模态大模型（图像-语言大模型 Vision Language Model，Multimodal Model）的教学示范，采用了业界在过去两年里的模块化神经网络的主流做法。本文提供了VLM所需运行的完整代码且经过测试，在安装完数据库后可直接运行，VLM的图像识别内容为在SCNet（国家超算互联）平台上实际训练30分钟后的使用效果，训练消耗在2元以内，使用显卡为一张RTX4090，各大算力平台均可。根据需要可自行选择LLM规模和CLIP，满足不同场景需求。
其核心为将一个Vision Language Encoder ，通常经过CLIP （Contrastive Language-Image Pretraining 语言图像对照预训练）预训练，而后将之通过一个 Projector 对齐到LLM（语言大模型）的Text Tokenizer所指向的Embedding Space里，以此，将图像内容转换为Transformer Blcok 可以处理的Embedding space vector（hidden vector）。
本文采用了代号为“deepseek-ai/DeepSeek-R1-Distill-Qwen-1.5B”的LLM，以及代号为“openai/clip-vit-base-patch32”的CLIP预训练过的Vision Language Encoder，然后Finetune（微调）在Flickr30k 数据集上，以对文本和图像进行对其，从而训练Projector。该数据集为图片文本对照样本，每张图像有一个(或五个)文本参照。
有需要的读者可以借鉴下列文章的部分核心思想，需注意，embedding space仍是目前的重要研究方向之一，对业界发展产生了深远影响，但其内容需要严苛的高等数学知识以及大量理论，实践，对照，因受众较少，通常不为人工智能的主要教授内容。
1. “Modular Network Assembly 和 Vision Transform RoPE 实验 Single Object Tracking on LaSOT”
https://blog.csdn.net/YucongCai/article/details/159987102?spm=1001.2014.3001.5501
https://blog.csdn.net/YucongCai/article/details/159987102?spm=1001.2014.3001.55012. SCNet CNN卷积神经网络 VGG19 Transfer Learning 迁移学习 CIFAR100 实例
https://blog.csdn.net/YucongCai/article/details/159773565?spm=1001.2014.3001.5501https://blog.csdn.net/YucongCai/article/details/159773565?spm=1001.2014.3001.55013. SCNet 原生训练GPT-2类模型 10-50M LLM 实例https://blog.csdn.net/YucongCai/article/details/159729082?spm=1001.2014.3001.5501https://blog.csdn.net/YucongCai/article/details/159729082?spm=1001.2014.3001.5501所有的训练参数均在Projector 内，而CLIP Encoder 和 LLM 都是 pretrained 并且其weights 都是fixed/frozen。如需更改，可以参照以下示例，通过LoRA或者逐层解冻，进行进一步地微调。
4. “SCNet 超算互联网 LLM Fine-Tuning LoRa 实例”
https://blog.csdn.net/YucongCai/article/details/159696147?spm=1001.2014.3001.5502https://blog.csdn.net/YucongCai/article/details/159696147?spm=1001.2014.3001.55025. “SCNet 超算互联网 LLM Fine-Tuning FSDP LoRA 多卡分布式微调训练 实例”
因为CLIP Encoder 处理后的图像经过Projector直接输入到了LLM的embedding space里，采用了 inputs_embeds 和 input_ids 混合的方式，所以一个attention mask需要定义给模型，与之自动的casual mask相点乘，以此来handel padding token。这会在DataLoader和训练过程中有所体现
loss = vlm(
pixel_values=images,
input_ids=input_ids,
attention_mask=attention_mask,
labels=labels
)LLM模型采用了float16，然而CLIP Encoder 和 Projector采用了 float32 格式，在训练中，训练器会auto cast 不同比特率之间的转换，但是训练完成后模型会无法使用，所以要再模型定义中增加格式的 cast 以保持参数格式的转换和一致性。本文训练时采用float32,测试时直接采用16bit提升模型速度。
同时，因为llm的任务形式，需要增加一个text的prompt来提示模型应当解答问题，具体arrangement为，[VISUAL] [PROMPT] [CAPTION] 。 如果没有这一prompt，新的VLM有可能会遇到alignment的问题，因为llm本身的定义。本文采用了一个较小的llm，所以这一现象被一定程度上削弱。
with torch.no_grad():
vision_outputs = self.vision_encoder(pixel_values=pixel_values)
vision_features = vision_outputs.last_hidden_state
visual_embeds = self.projector(vision_features)
visual_embeds = visual_embeds.to(dtype=self.llm.dtype)
text_embeds = self.llm.get_input_embeddings()(input_ids)
inputs_embeds = torch.cat([visual_embeds, text_embeds], dim=1)
然而比较利好的是，仅在一个epoch的训练后，projector就可以将 CLIP Encoder输出的在CLIP Encoder里的embedding space里的visual states转换为llm原生tokenizer映射到的embedding space 里的text states。这部分因为CLIP的训练方式，CLIP Encoder的text和image的vector在其本身的embedding space里已经对其，这极大地降低了Transfer Learning所需训练数据。
Projector为常规构架，而且，应为Projector包含全部训练参数，所以所需训练参数实际不多，实测5分钟训练内即可进行有明显感官的对齐，这也是这种方法的常规参数。
# -------------------------------
# VisionProjector (unchanged)
# -------------------------------
class VisionProjector(nn.Module):
def __init__(self, vision_hidden_size, llm_hidden_size, dropout=0.1):
super().__init__()
self.mlp = nn.Sequential(
nn.Linear(vision_hidden_size, llm_hidden_size * 2),
nn.GELU(),
nn.Dropout(dropout),
nn.Linear(llm_hidden_size * 2, llm_hidden_size),
)
def forward(self, x):
return self.mlp(x)
以下是一个实际训练场景示例，本文的训练在约为6.5个epoch后被手动停止。可见，每轮epoch耗时在5分钟以内，且一个epoch的loss已达到0.5-1.5的合理范围。
需要注意的是，CLIP Encoder仅输出visual states，它的text tokenizer并没有被使用。
本文为公益类代码，由DeepSeek辅助生成，经过实例测试。
1. 下载数据集，并预处理，增加prompt
# -------------------------------
# 1. Environment Setup
# -------------------------------
# !pip install torch torchvision transformers datasets pillow pandas tqdm accelerate
import os
os.environ['HF_ENDPOINT'] = 'https://hf-mirror.com'
# -------------------------------
# 2. Load and preprocess Flickr30k
# -------------------------------
import json
import pandas as pd
from PIL import Image
import torch
from torch.utils.data import Dataset, DataLoader
from torchvision import transforms
import random
data_root = "/root/private_data/data/flickr30k"
csv_path = os.path.join(data_root, "flickr_annotations_30k.csv")
image_folder = os.path.join(data_root, "flickr30k-images")
df = pd.read_csv(csv_path)
df['captions'] = df['raw'].apply(json.loads)
df['sentids'] = df['sentids'].apply(json.loads)
def is_valid_row(row):
if len(row['captions']) != 5:
return False
img_path = os.path.join(image_folder, row['filename'])
if not os.path.exists(img_path):
return False
return True
df_valid = df[df.apply(is_valid_row, axis=1)].copy()
train_df = df_valid[df_valid['split'] == 'train']
val_df = df_valid[df_valid['split'] == 'val']
test_df = df_valid[df_valid['split'] == 'test']
print(f"Training samples: {len(train_df)}")
print(f"Validation samples: {len(val_df)}")
print(f"Test samples: {len(test_df)}")
# -------------------------------
# 3. Dataset class WITH PROMPT (fixed padding)
# -------------------------------
class Flickr30kDataset(Dataset):
def __init__(self, dataframe, image_folder, transform=None, tokenizer=None, max_length=77, prompt="Describe this image:"):
self.dataframe = dataframe
self.image_folder = image_folder
self.transform = transform
self.tokenizer = tokenizer
self.max_length = max_length
self.prompt = prompt
def __len__(self):
return len(self.dataframe)
def __getitem__(self, idx):
row = self.dataframe.iloc[idx]
img_path = os.path.join(self.image_folder, row['filename'])
image = Image.open(img_path).convert('RGB')
if self.transform:
image = self.transform(image)
captions = row['captions']
caption = random.choice(captions)
if self.tokenizer:
full_text = f"{self.prompt} {caption}"
tokenized = self.tokenizer(
full_text,
truncation=True,
max_length=self.max_length,
padding="max_length",
return_tensors="pt"
)
input_ids = tokenized['input_ids'].squeeze(0)
attention_mask = tokenized['attention_mask'].squeeze(0)
# Create labels: mask the prompt tokens (including BOS)
labels = input_ids.clone()
prompt_tokens = self.tokenizer(self.prompt, add_special_tokens=True)['input_ids']
prompt_len = len(prompt_tokens)
labels[:prompt_len] = -100 # ignore loss on prompt
return {
'image': image,
'input_ids': input_ids,
'attention_mask': attention_mask,
'labels': labels,
'caption': caption
}
else:
return {'image': image, 'caption': caption}
# -------------------------------
# 4. Image preprocessing
# -------------------------------
from torchvision.transforms import Compose, Resize, CenterCrop, ToTensor, Normalize
clip_mean = [0.48145466, 0.4578275, 0.40821073]
clip_std = [0.26862954, 0.26130258, 0.27577711]
image_transform = Compose([
Resize(224, interpolation=Image.BICUBIC),
CenterCrop(224),
ToTensor(),
Normalize(mean=clip_mean, std=clip_std),
])
2. 导入CLIP Encoder，llm，然后定义Projector，组成一个新的VLM。
# -------------------------------
# 5. Load models and tokenizer (fix padding side)
# -------------------------------
from transformers import CLIPVisionModel, AutoModelForCausalLM, AutoTokenizer
import torch.nn as nn
vision_encoder = CLIPVisionModel.from_pretrained("openai/clip-vit-base-patch32")
for param in vision_encoder.parameters():
param.requires_grad = False
llm_model_id = "deepseek-ai/DeepSeek-R1-Distill-Qwen-1.5B"
llm = AutoModelForCausalLM.from_pretrained(
llm_model_id,
torch_dtype=torch.float16,
device_map="auto"
)
for param in llm.parameters():
param.requires_grad = False
tokenizer = AutoTokenizer.from_pretrained(llm_model_id)
if tokenizer.pad_token is None:
tokenizer.pad_token = tokenizer.eos_token
# CRITICAL: pad on the right so prompt and caption appear at the beginning
tokenizer.padding_side = "right"
# -------------------------------
# 6. VisionProjector
# -------------------------------
class VisionProjector(nn.Module):
def __init__(self, vision_hidden_size, llm_hidden_size, dropout=0.1):
super().__init__()
self.mlp = nn.Sequential(
nn.Linear(vision_hidden_size, llm_hidden_size * 2),
nn.GELU(),
nn.Dropout(dropout),
nn.Linear(llm_hidden_size * 2, llm_hidden_size),
)
def forward(self, x):
return self.mlp(x)
# -------------------------------
# 7. VisionLanguageModel (projector in float32, output cast to float16)
# -------------------------------
class VisionLanguageModel(nn.Module):
def __init__(self, vision_encoder, llm, tokenizer):
super().__init__()
self.vision_encoder = vision_encoder
self.llm = llm
self.tokenizer = tokenizer
self.vision_hidden_size = vision_encoder.config.hidden_size
self.llm_hidden_size = llm.config.hidden_size
self.projector = VisionProjector(self.vision_hidden_size, self.llm_hidden_size)
def forward(self, pixel_values, input_ids, attention_mask, labels=None):
with torch.no_grad():
vision_outputs = self.vision_encoder(pixel_values=pixel_values)
vision_features = vision_outputs.last_hidden_state
visual_embeds = self.projector(vision_features)
visual_embeds = visual_embeds.to(dtype=self.llm.dtype)
text_embeds = self.llm.get_input_embeddings()(input_ids)
inputs_embeds = torch.cat([visual_embeds, text_embeds], dim=1)
batch_size, num_visual_tokens = visual_embeds.shape[:2]
visual_attention_mask = torch.ones(batch_size, num_visual_tokens, device=inputs_embeds.device)
combined_attention_mask = torch.cat([visual_attention_mask, attention_mask], dim=1)
outputs = self.llm(
inputs_embeds=inputs_embeds,
attention_mask=combined_attention_mask,
return_dict=True
)
logits = outputs.logits
loss = None
if labels is not None:
num_vis = visual_embeds.size(1)
logits_text = logits[:, num_vis:-1, :]
shift_labels = labels[:, 1:].contiguous()
loss_fct = nn.CrossEntropyLoss(ignore_index=-100)
loss = loss_fct(logits_text.reshape(-1, logits_text.size(-1)), shift_labels.reshape(-1))
return loss if loss is not None else logits
def generate(self, pixel_values, max_new_tokens=50, **kwargs):
with torch.no_grad():
vision_outputs = self.vision_encoder(pixel_values=pixel_values)
vision_features = vision_outputs.last_hidden_state
visual_embeds = self.projector(vision_features)
visual_embeds = visual_embeds.to(dtype=self.llm.dtype)
outputs = self.llm.generate(
inputs_embeds=visual_embeds,
max_new_tokens=max_new_tokens,
pad_token_id=self.tokenizer.pad_token_id,
eos_token_id=self.tokenizer.eos_token_id,
**kwargs
)
return outputs
3. 训练Projector，通常，5-15个eopches 已足够获得初步结果。
# -------------------------------
# 8. Prepare datasets and dataloaders
# -------------------------------
train_dataset = Flickr30kDataset(
train_df, image_folder,
transform=image_transform,
tokenizer=tokenizer,
prompt="Describe this image:"
)
val_dataset = Flickr30kDataset(
val_df, image_folder,
transform=image_transform,
tokenizer=tokenizer,
prompt="Describe this image:"
)
train_loader = DataLoader(train_dataset, batch_size=16, shuffle=True, num_workers=4)
val_loader = DataLoader(val_dataset, batch_size=16, shuffle=False, num_workers=4)
# -------------------------------
# 9. Initialize model and optimizer
# -------------------------------
vlm = VisionLanguageModel(vision_encoder, llm, tokenizer)
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
vlm = vlm.to(device)
for name, param in vlm.named_parameters():
param.requires_grad = False
if 'projector' in name:
param.requires_grad = True
optimizer = torch.optim.AdamW(filter(lambda p: p.requires_grad, vlm.parameters()), lr=1e-4)
# -------------------------------
# 10. Training loop with diagnostics (first batch only)
# -------------------------------
from tqdm import tqdm
import torch.cuda.amp as amp
scaler = torch.cuda.amp.GradScaler()
epochs = 15 # train for 15 epochs
diagnostic_printed = False
for epoch in range(epochs):
vlm.train()
total_loss = 0
progress_bar = tqdm(train_loader, desc=f"Epoch {epoch+1}/{epochs}")
for batch_idx, batch in enumerate(progress_bar):
images = batch['image'].to(device)
input_ids = batch['input_ids'].to(device)
attention_mask = batch['attention_mask'].to(device)
labels = batch['labels'].to(device)
# Print diagnostic info for first batch of first epoch
if epoch == 0 and batch_idx == 0 and not diagnostic_printed:
print("\n🔍 DIAGNOSTIC: First training batch")
print(f" input_ids shape: {input_ids.shape}")
print(f" input_ids[0][:30]: {input_ids[0][:30].tolist()}")
print(f" labels[0][:30]: {labels[0][:30].tolist()}")
first_token = input_ids[0][0].item()
if first_token == tokenizer.pad_token_id:
print(" ⚠️ WARNING: First token is PAD! Check tokenizer.padding_side.")
else:
print(f" ✅ First token is {first_token} ({tokenizer.decode([first_token])}) – correct.")
prompt_tokens = tokenizer("Describe this image:", add_special_tokens=True)['input_ids']
prompt_len = len(prompt_tokens)
masked_labels = labels[0][:prompt_len]
if (masked_labels == -100).all():
print(f" ✅ Prompt tokens (first {prompt_len}) are correctly masked (-100).")
else:
print(f" ⚠️ WARNING: Prompt tokens not fully masked.")
diagnostic_printed = True
optimizer.zero_grad()
with amp.autocast():
loss = vlm(
pixel_values=images,
input_ids=input_ids,
attention_mask=attention_mask,
labels=labels
)
scaler.scale(loss).backward()
scaler.step(optimizer)
scaler.update()
total_loss += loss.item()
progress_bar.set_postfix({"loss": f"{loss.item():.4f}"})
avg_loss = total_loss / len(train_loader)
print(f"Epoch {epoch+1} - Average Training Loss: {avg_loss:.4f}")
# Validation
vlm.eval()
val_loss = 0
with torch.no_grad():
for batch in val_loader:
images = batch['image'].to(device)
input_ids = batch['input_ids'].to(device)
attention_mask = batch['attention_mask'].to(device)
labels = batch['labels'].to(device)
with amp.autocast():
loss = vlm(
pixel_values=images,
input_ids=input_ids,
attention_mask=attention_mask,
labels=labels
)
val_loss += loss.item()
avg_val_loss = val_loss / len(val_loader)
print(f"Epoch {epoch+1} - Validation Loss: {avg_val_loss:.4f}")4. 保存训练过后的模型，而后重新导入。
# -------------------------------
# 11. Save projector weights
# -------------------------------
projector_save_path = "projector_epoch15.pth"
torch.save(vlm.projector.state_dict(), projector_save_path)
print(f"Projector saved to {projector_save_path}")
# -------------------------------
# 12. Inference setup (run AFTER training)
# -------------------------------
import torch
import torch.nn as nn
from transformers import CLIPVisionModel, AutoModelForCausalLM, AutoTokenizer
from torchvision import transforms
from PIL import Image
import matplotlib.pyplot as plt
import numpy as np
import random
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
# Reload frozen components (or reuse from training)
vision_encoder = CLIPVisionModel.from_pretrained("openai/clip-vit-base-patch32").to(device)
vision_encoder.eval()
for p in vision_encoder.parameters():
p.requires_grad = False
llm = AutoModelForCausalLM.from_pretrained(
"deepseek-ai/DeepSeek-R1-Distill-Qwen-1.5B",
torch_dtype=torch.float16,
device_map="auto"
)
llm.eval()
for p in llm.parameters():
p.requires_grad = False
tokenizer = AutoTokenizer.from_pretrained("deepseek-ai/DeepSeek-R1-Distill-Qwen-1.5B")
tokenizer.pad_token = tokenizer.eos_token
tokenizer.padding_side = "right"
# Re-define projector and VLM (same as training, but projector in float16)
class VisionProjector(nn.Module):
def __init__(self, vision_hidden_size, llm_hidden_size, dropout=0.1):
super().__init__()
self.mlp = nn.Sequential(
nn.Linear(vision_hidden_size, llm_hidden_size * 2),
nn.GELU(),
nn.Dropout(dropout),
nn.Linear(llm_hidden_size * 2, llm_hidden_size),
)
def forward(self, x):
return self.mlp(x)
class VisionLanguageModel(nn.Module):
def __init__(self, vision_encoder, llm, tokenizer):
super().__init__()
self.vision_encoder = vision_encoder
self.llm = llm
self.tokenizer = tokenizer
self.vision_hidden_size = vision_encoder.config.hidden_size
self.llm_hidden_size = llm.config.hidden_size
self.projector = VisionProjector(self.vision_hidden_size, self.llm_hidden_size).to(dtype=llm.dtype)
def generate(self, pixel_values, max_new_tokens=50, **kwargs):
with torch.no_grad():
vision_outputs = self.vision_encoder(pixel_values=pixel_values)
vision_features = vision_outputs.last_hidden_state
vision_features = vision_features.to(dtype=self.llm.dtype)
projected_features = self.projector(vision_features)
outputs = self.llm.generate(
inputs_embeds=projected_features,
max_new_tokens=max_new_tokens,
pad_token_id=self.tokenizer.pad_token_id,
eos_token_id=self.tokenizer.eos_token_id,
**kwargs
)
return outputs
vlm_inference = VisionLanguageModel(vision_encoder, llm, tokenizer).to(device)
# Load projector weights
state_dict = torch.load("projector_epoch15.pth", map_location=device)
for key in state_dict:
state_dict[key] = state_dict[key].to(dtype=torch.float16)
vlm_inference.projector.load_state_dict(state_dict)
vlm_inference.projector.eval()
print(f"✅ Projector loaded, dtype={vlm_inference.projector.mlp[0].weight.dtype}")
# -------------------------------
# 13. Corrected caption generation function (no slicing)
# -------------------------------
def generate_caption(model, image_tensor, prompt="Describe this image:", max_length=50, temperature=0.7, do_sample=True):
model.eval()
with torch.no_grad():
# Visual embeddings
vision_outputs = model.vision_encoder(pixel_values=image_tensor.unsqueeze(0).to(device))
vision_features = vision_outputs.last_hidden_state
vision_features = vision_features.to(dtype=model.llm.dtype)
visual_embeds = model.projector(vision_features)
# Prompt embeddings
prompt_tokens = model.tokenizer(prompt, return_tensors="pt", add_special_tokens=True).to(device)
prompt_embeds = model.llm.get_input_embeddings()(prompt_tokens.input_ids)
# Concatenate: visual tokens first, then prompt
inputs_embeds = torch.cat([visual_embeds, prompt_embeds], dim=1)
attention_mask = torch.ones(inputs_embeds.shape[:2], dtype=torch.long, device=device)
# Generate
output_ids = model.llm.generate(
inputs_embeds=inputs_embeds,
attention_mask=attention_mask,
max_new_tokens=max_length,
temperature=temperature,
do_sample=do_sample,
pad_token_id=model.tokenizer.pad_token_id,
eos_token_id=model.tokenizer.eos_token_id,
)
# output_ids contains ONLY the newly generated tokens (no input tokens)
return model.tokenizer.decode(output_ids[0], skip_special_tokens=True)
5. 测试机个随机实例。
# -------------------------------
# 14. Test on a random validation sample
# -------------------------------
def denormalize(tensor, mean, std):
tensor = tensor.clone().detach().cpu()
for t, m, s in zip(tensor, mean, std):
t.mul_(s).add_(m)
return tensor.permute(1, 2, 0).numpy()
random_idx = random.randint(0, len(val_dataset) - 1)
sample = val_dataset[random_idx]
image_tensor = sample['image'].to(device)
ground_truth = sample['caption']
generated = generate_caption(vlm_inference, image_tensor, temperature=0.7, do_sample=True)
print(f"Random index: {random_idx}")
print("Ground truth (one of five):", ground_truth)
print("Generated caption: ", generated)
img_display = denormalize(image_tensor, clip_mean, clip_std)
img_display = np.clip(img_display, 0, 1)
plt.figure(figsize=(8, 8))
plt.imshow(img_display)
plt.axis('off')
plt.title(f"Generated: {generated}\nGround Truth: {ground_truth}", fontsize=12)
plt.tight_layout()
plt.show()至此，一个 由CLIP Encoder 和 LLM 为模块的 Vision Language Model 就完成了。
通过示例可以看出，这个简易的通过30分钟训练获得的VLM仍有不足，但它确实将图像中的关键内容进行了识别和理解。同时，因为llm本身未经过微调，对于图像的理解来自于新的VLM对于图像所看到的场景的认知，而不是来自于Flicker30k数据库。所以，虽然Generated response和ground truth有时又很大不同，但是VLM准确地理解了图像中的事件和意图，甚至可能比训练参数本身更为精确。
需要注意的是，这一新的Vision Language Model中的LLM和CLIP均为模块化引入，且未经过微调，而VLM的Transformer Block的 Embedding Space 曾很长时间被认为仅为 text （文本) 的 embedding space。
我在找工作，HR或项目合作请联系：yucongcai_business@outlook.com
与科研相关的请联系：yucongcai_research@outlook.com
欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。
更多推荐
29
0- 0
已为社区贡献2条内容
所有评论(0)