【代碼解讀】Deepseek_vl2中具體代碼調用
文章目錄
- 【代碼解讀】Deepseek_vl2中具體代碼調用
- DeepseekVLV2Processor解讀
- DeepseekVLV2ForCausalLM - 多模態模型
- DeepSeek-VL2 Processor的輸入格式
- 單樣本格式
- 多樣本格式
- DeepSeek-VL2模型的輸出形式總結
- 主要輸出類型:DeepSeekVLV2CausalLMOutputWithPast
- 其他輸出形式
- DeepSeek-VL2 forward 方法詳細解讀
- 整體架構設計分析
DeepseekVLV2Processor解讀
DeepseekVLV2Processor.from_pretrained() 的輸入格式和輸出參數
processor = DeepseekVLV2Processor.from_pretrained(model_path)
?
支持的輸入參數
DeepseekVLV2Processor.from_pretrained(pretrained_model_name_or_path, # 必需:模型路徑 (一般選這個就夠了)**kwargs # 可選:其他參數
)具體參數說明# 必需參數
pretrained_model_name_or_path: str # 模型路徑,如 "deepseek-ai/deepseek-vl2" 或者本地下載PTH路徑# 可選參數(繼承自ProcessorMixin)
cache_dir: Optional[str] = None # 緩存目錄
force_download: bool = False # 強制下載
local_files_only: bool = False # 僅使用本地文件
token: Optional[Union[str, bool]] = None # 訪問token
revision: str = "main" # 模型版本
trust_remote_code: bool = False # 信任遠程代碼
?
返回的processor對象包含以下屬性
processor = DeepseekVLV2Processor.from_pretrained(model_path)# 主要屬性
processor.tokenizer # LlamaTokenizerFast對象
processor.candidate_resolutions # 候選分辨率 [(384, 384)]
processor.image_size # 圖像尺寸 384
processor.patch_size # patch尺寸 16
processor.downsample_ratio # 下采樣比例 2
processor.image_mean # 圖像均值 (0.5, 0.5, 0.5)
processor.image_std # 圖像標準差 (0.5, 0.5, 0.5)
processor.normalize # 是否標準化 True
processor.image_token # 圖像標記 "<image>"
processor.pad_token # 填充標記 "REDACTED_SPECIAL_TOKEN"
processor.sft_format # SFT格式 "deepseek"
processor.mask_prompt # 是否掩碼prompt True
processor.ignore_id # 忽略ID -100# 特殊token ID
processor.image_token_id # 圖像token的ID
processor.bos_id # 開始token ID
processor.eos_id # 結束token ID
processor.pad_id # 填充token ID# 圖像處理組件
processor.image_transform # ImageTransform對象
?
基本加載
# 代碼中的使用方式
processor = DeepseekVLV2Processor.from_pretrained(self.backbone.config._name_or_path
)# 等價于
processor = DeepseekVLV2Processor.from_pretrained("deepseek-ai/deepseek-vl2" # 或其他模型路徑
)
?
processor對象的主要方法的核心處理方法
# 處理單個樣本
processor.process_one(prompt=str, # 文本提示conversations=List[Dict], # 對話列表images=List[Image.Image], # 圖像列表apply_sft_format=bool, # 是否應用SFT格式inference_mode=bool, # 推理模式system_prompt=str # 系統提示
)# 批量處理
processor.__call__(prompt=str, # 文本提示conversations=List[Dict], # 對話列表images=List[Image.Image], # 圖像列表apply_sft_format=bool, # 是否應用SFT格式force_batchify=bool, # 強制批處理inference_mode=bool, # 推理模式system_prompt=str # 系統提示
)其實用的多的也就是前4組prompt=str, # 文本提示conversations=List[Dict], # 對話列表images=List[Image.Image], # 圖像列表apply_sft_format=bool, # 是否應用SFT格式
?
在語義分割任務中,目前用的形式是
processed_output = processor(prompt=full_prompt,images=pil_images,apply_sft_format=True,inference_mode=True,force_batchify=False #禁用批次處理 這樣出來的processed_output.images為[B, C, H, W] B為處理后的patch數量
)
?
輔助方法
processor.encode(text, bos=True, eos=False) # 文本編碼
processor.decode(token_ids, **kwargs) # 文本解碼
processor.format_messages(conversations, ...) # 格式化消息
processor.format_prompts(prompts, ...) # 格式化提示
processor.batchify(sample_list, ...) # 批處理
語義分割基本用不上吧,但也是工具類的東西
?
特殊token的添加
# processor會自動添加以下特殊token:
special_tokens = ["<image>", # 圖像標記"<|ref|>", "<|/ref|>", # 引用標記"<|det|>", "<|/det|>", # 檢測標記"<|grounding|>", # 定位標記"<|User|>", "<|Assistant|>" # 對話角色標記
]# 添加到tokenizer中
processor.tokenizer.add_special_tokens({"additional_special_tokens": special_tokens
})
但我們在語義分割任務中,暫時并沒有使用tokenizer
?
其中processor的處理輸出
class VLChatProcessorOutput(DictOutput):sft_format: str # 格式化后的文本input_ids: torch.LongTensor # 文本token序列target_ids: torch.LongTensor # 目標token序列(用于訓練)images: torch.Tensor # 處理后的圖像張量images_seq_mask: torch.BoolTensor # 圖像序列掩碼images_spatial_crop: torch.LongTensor # 圖像空間裁剪信息num_image_tokens: List[int] # 圖像token數量input_ids = "文本+圖像標記的完整token序列"
images = "實際的圖像像素數據"
target_ids = "用于訓練的目標token序列(圖像位置被mask)"# 你的輸入:
prompt = "Please segment this image into different regions."
image_tokens = "<image> " * 4 # 4個圖像標記
full_prompt = f"{prompt} {image_tokens}"告訴模型"請對圖像進行分割"
# 包含完整的文本指令和圖像位置標記
# 模型需要理解文本指令,然后在圖像位置生成分割結果
?
# input_ids 包含完整的token序列
input_ids = [BOS_token, # 開始標記"Please", # 文本token"segment", # 文本token"this", # 文本token"image", # 文本token"into", # 文本token"different", # 文本token"regions", # 文本token".", # 文本tokenIMAGE_token, # <image>標記IMAGE_token, # <image>標記IMAGE_token, # <image>標記IMAGE_token, # <image>標記EOS_token # 結束標記
]
設計目的:讓模型理解"在圖像位置應該做什么"# 作用:
# 1. 告訴模型"請對圖像進行分割"
# 2. 標記圖像在序列中的位置
# 3. 為多模態融合提供對齊信息
?
target_ids的生成
target_ids = [BOS_token, # 保留"Please", # 保留"segment", # 保留"this", # 保留"image", # 保留"into", # 保留"different", # 保留"regions", # 保留".", # 保留ignore_id, # 圖像位置被mask為-100ignore_id, # 圖像位置被mask為-100ignore_id, # 圖像位置被mask為-100ignore_id, # 圖像位置被mask為-100EOS_token # 保留
]也就是說這部分只保留了本文的信息,去除調了圖像
?
# images 實際的圖像像素數據
# images 是經過預處理的圖像張量
images = torch.stack(images_list, dim=0) # [N, 3, 384, 384]# 包含:
# - 全局視圖圖像
# - 局部視圖圖像(多分辨率)
# - 經過標準化、裁剪、轉換的圖像
# 在我512的輸入條件下,N的數量會變為double
?
DeepseekVLV2ForCausalLM - 多模態模型
self.backbone = DeepseekVLV2ForCausalLM.from_pretrained(model_path, torch_dtype=torch.float16)
作用:實際的神經網絡模型,執行前向推理
功能:
-
處理tokenized的輸入
-
融合文本和圖像特征
-
生成輸出(在我們的場景中是隱藏狀態)
?
與DeepseekVLV2Processor使用相同的模型配置路徑
DeepseekVLV2Processor負責將原始輸入(文本和圖像)轉換為模型可以理解的格式
?
DeepseekVLV2Processor與DeepseekVLV2ForCausalLM的對接功能:
-
文本tokenization(將文本轉換為token ID)
-
圖像預處理(調整尺寸、歸一化等)
-
添加特殊token(如、<|User|>等)
-
應用對話模板(SFT格式)
-
生成attention mask等
?
原始輸入 → Processor → 模型輸入 → Backbone → 輸出
?
具體流程如下:
(1)輸入:原始文本 + PIL圖像(2)Processor處理: processed_output = processor(prompt=full_prompt,images=pil_images,apply_sft_format=True,inference_mode=True,force_batchify=False)(3)輸出:input_ids, images, attention_mask等(4)Backbone處理:outputs = self.backbone(input_ids=input_ids,images=processed_images,attention_mask=attention_mask,# ... 其他參數)
?
DeepSeek-VL2 Processor的輸入格式
?
單樣本格式
基于 prompt 的格式
processor(prompt="Please segment this image into different regions.",images=[pil_image1, pil_image2, ...],apply_sft_format=True, # 可選:應用SFT格式force_batchify=False, # 可選:是否強制批處理inference_mode=True,system_prompt="" # 可選:系統提示
)
-
使用簡單的文本prompt
-
可以包含 token 在prompt中
-
支持 apply_sft_format=True 來應用對話模板
?
Prompt 的格式單個樣本輸出 (VLChatProcessorOutput):
{'sft_format': str,'input_ids': torch.LongTensor, # [seq_len]'target_ids': torch.LongTensor, # [seq_len] 'images': torch.Tensor, # [n_images, 3, H, W]'images_seq_mask': torch.BoolTensor, # [seq_len]'images_spatial_crop': torch.LongTensor, # [n_images, 2]'num_image_tokens': List[int]
}
?
多樣本格式
基于 conversations 的格式
processor(conversations=[{"role": "user", "content": "Please segment this image into different regions."},{"role": "assistant", "content": "I'll help you segment the image."},# 更多對話輪次...],images=[pil_image1, pil_image2, ...],force_batchify=False,inference_mode=True,system_prompt=""
)
-
使用結構化的對話格式
-
每個message有 role 和 content
-
支持的role:“user”, “assistant”, “system”
-
自動應用SFT格式
?
conversations 的格式批處理輸出 (BatchCollateOutput)
{'sft_format': List[str],'input_ids': torch.LongTensor, # [batch_size, seq_len]'labels': torch.LongTensor, # [batch_size, seq_len]'images': torch.Tensor, # [batch_size, n_images, 3, H, W]'attention_mask': torch.Tensor, # [batch_size, seq_len] - 只有批處理才有!'images_seq_mask': torch.BoolTensor, # [batch_size, seq_len]'images_spatial_crop': torch.LongTensor, # [batch_size, n_images, 2]'seq_lens': List[int]
}
?
DeepSeek-VL2模型的輸出形式總結
outputs = self.backbone 的輸出
主要輸出類型:DeepSeekVLV2CausalLMOutputWithPast
class DeepSeekVLV2CausalLMOutputWithPast(ModelOutput):loss: Optional[torch.FloatTensor] = None # 損失值(訓練時)logits: torch.FloatTensor = None # 預測logitspast_key_values: Optional[List[torch.FloatTensor]] = None # 緩存值(推理時)hidden_states: Optional[Tuple[torch.FloatTensor]] = None # 隱藏狀態attentions: Optional[Tuple[torch.FloatTensor]] = None # 注意力權重rope_deltas: Optional[torch.LongTensor] = None # RoPE增量
隱藏層狀態的形狀
?
# outputs.hidden_states 是一個tuple,包含:
# - 第0個元素:embedding層的輸出
# - 第1-32個元素:transformer層的輸出(假設有32層)
# - 每個元素的形狀:[batch_size, seq_len, hidden_dim]multimodal_features = outputs.hidden_states[-1] # 最后一層的輸出
# 形狀:[batch_size, seq_len, hidden_dim]
?
其他輸出形式
loss - 訓練損失
形狀:標量tensor
例如:tensor(2.3456, device=‘cuda:0’)
loss = outputs.loss
print(f"Loss value: {loss.item()}")outputs = self.backbone(input_ids=input_ids, labels=labels)
loss = outputs.loss
loss.backward()
optimizer.step()# 2. 損失監控
if loss.item() > 10.0:print("Warning: Loss is too high!")# 3. 梯度裁剪
loss.backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)# 4. 學習率調度
scheduler.step(loss)
?
logits - 預測logits
# 例如:[2, 512, 128000] - 2個樣本,512個token,128000詞匯表大小logits = outputs.logits
print(f"logits shape: {logits.shape}")# 1. 文本生成 - 獲取下一個token的概率
next_token_logits = logits[:, -1, :] # 最后一個位置的logits
probs = torch.softmax(next_token_logits, dim=-1)
next_token_id = torch.argmax(probs, dim=-1)# 2. 計算困惑度
loss_fct = nn.CrossEntropyLoss()
loss = loss_fct(logits.view(-1, logits.size(-1)), labels.view(-1))
perplexity = torch.exp(loss)# 3. 獲取top-k候選
top_k = 5
top_k_logits, top_k_indices = torch.topk(logits, k=top_k, dim=-1)
?
hidden_states - 隱藏狀態
# hidden_states是一個tuple,包含所有層的輸出
# 例如:33個元素(1個embedding + 32個transformer層)
# 每個元素形狀:[batch_size, seq_len, hidden_dim]hidden_states = outputs.hidden_states# 1. 獲取最后一層的特征(我們當前使用的方法)
last_layer_features = hidden_states[-1] # [2, 512, 2048]# 2. 獲取特定層的特征
layer_10_features = hidden_states[10] # 第10層
embedding_features = hidden_states[0] # embedding層# 3. 特征融合 - 多層特征加權平均
weights = torch.softmax(torch.randn(len(hidden_states)), dim=0)
fused_features = sum(w * h for w, h in zip(weights, hidden_states))# 4. 提取圖像位置的特征(我們的語義分割用法)
image_token_id = tokenizer.convert_tokens_to_ids("<image>")
image_positions = (input_ids == image_token_id)
image_features = last_layer_features[image_positions] # [num_image_tokens, 2048]
?
DeepSeek-VL2 forward 方法詳細解讀
整體架構設計分析
輸入: [input_ids, images, images_seq_mask, images_spatial_crop]↓
prepare_inputs_embeds: 多模態融合↓
輸出: inputs_embeds [batch_size, seq_len, hidden_dim]↓
language.forward: 語言模型處理↓
輸出: DeepSeekVLV2CausalLMOutputWithPast 到這步位置應該就是輸出的文本結構l
?
傳遞的參數值和參數分類
def forward(self,# ========== 文本處理參數 ==========input_ids: Optional[torch.LongTensor] = None, # 文本token序列attention_mask: Optional[torch.Tensor] = None, # 注意力掩碼position_ids: Optional[torch.LongTensor] = None, # 位置編碼past_key_values: Optional[List[torch.FloatTensor]] = None, # KV緩存inputs_embeds: Optional[torch.FloatTensor] = None, # 預計算的嵌入# ========== 圖像處理參數 ==========images: Optional[torch.FloatTensor] = None, # 圖像張量images_seq_mask: Optional[torch.LongTensor] = None, # 圖像序列掩碼images_spatial_crop: Optional[torch.LongTensor] = None, # 空間裁剪信息# ========== 控制參數 ==========labels: Optional[torch.LongTensor] = None, # 訓練標簽use_cache: Optional[bool] = None, # 是否使用緩存output_attentions: Optional[bool] = None, # 是否輸出注意力output_hidden_states: Optional[bool] = None, # 是否輸出隱藏狀態return_dict: Optional[bool] = None, # 返回格式cache_position: Optional[torch.LongTensor] = None, # 緩存位置
):
?
多模態輸入嵌入準備
if inputs_embeds is None:# 核心:調用prepare_inputs_embeds進行多模態融合inputs_embeds = self.prepare_inputs_embeds(input_ids=input_ids,images=images,images_seq_mask=images_seq_mask,images_spatial_crop=images_spatial_crop,)# 確保attention_mask在正確的設備上if attention_mask is not None:attention_mask = attention_mask.to(inputs_embeds.device)
?
調用底層語言模型進行實際的前向計算
language.forward 返回的是 CausalLMOutputWithPast 類型
CausalLMOutputWithPast {loss: Optional[torch.FloatTensor] = None, # [1] - 訓練損失(推理時為None)logits: torch.FloatTensor = [2, 512, 128000], # [batch_size, seq_len, vocab_size]past_key_values: Optional[List[torch.FloatTensor]] = None, # KV緩存(推理時可能使用)hidden_states: Optional[Tuple[torch.FloatTensor]] = None, # 隱藏狀態元組attentions: Optional[Tuple[torch.FloatTensor]] = None, # 注意力權重
}
A. loss
-
形狀: [1] 或 None
-
內容: 語言建模損失(僅在有labels時計算)
-
用途: 訓練時用于反向傳播
B. logits
-
形狀: [batch_size, seq_len, vocab_size]
-
示例: [2, 512, 128000]
-
內容: 詞匯表上每個token的預測分數(softmax前)
-
用途: 用于生成下一個token或計算損失
C. past_key_values
-
形狀: List[Tuple[torch.FloatTensor, torch.FloatTensor]]
-
內容: 每層的KV緩存,用于加速推理
-
結構: [(layer1_k, layer1_v), (layer2_k, layer2_v), …]
-
每個tensor形狀: [batch_size, num_heads, seq_len, head_dim]
D. hidden_states
-
形狀: Tuple[torch.FloatTensor]
-
內容: 每層的隱藏狀態
-
結構: (embedding_output, layer1_output, layer2_output, …, final_output)
-
每個tensor形狀: [batch_size, seq_len, hidden_size] (如 [2, 512, 2048])
E. attentions
-
形狀: Tuple[torch.FloatTensor]
-
內容: 每層的注意力權重
-
結構: (layer1_attention, layer2_attention, …)
-
每個tensor形狀: [batch_size, num_heads, seq_len, seq_len] (如 [2, 32, 512, 512])
generated_ids = self.backbone.generate(input_ids=input_ids,attention_mask=attention_mask,images=images,images_seq_mask=images_seq_mask,images_spatial_crop=images_spatial_crop,max_new_tokens=512,temperature=0.7,do_sample=True,pad_token_id=tokenizer.eos_token_id
)tokenizer = processor.tokenizer
generated_text = tokenizer.decode(generated_ids[0], skip_special_tokens=True)