【代码解读】Deepseek_vl2中具体代码调用

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)