Nagi-ovo

# image_encoder: ResNetやVision Transformerなどのモデル
# text_encoder: CBOWやText Transformerなどのモデル
# I[n, h, w, c]: 入力画像（ミニバッチ）、n枚の画像を含み、各画像の次元はh*w*c
# T[n, l]:       入力テキストのミニバッチ、nセグメントのテキストを含み、各セグメントの長さはl（通常はトークン数）
# W_i[d_i, d_e]: 学習された画像投影行列、d_i次元の画像特徴をd_e次元の埋め込み空間にマッピング
# W_t[d_t, d_e]: 学習されたテキスト投影行列、d_t次元のテキスト特徴をd_e次元の埋め込み空間にマッピング
# t:             学習された温度パラメータ（スカラー）

I_f = image_encoder(I)  # 出力形状：[n, d_i]
T_f = text_encoder(T)  # 出力形状：[n, d_t]

# --- 結合されたマルチモーダル埋め込みを計算 ---
# 画像特徴を投影行列W_iを通じてd_e次元空間にマッピング
image_embedding_projected = np.dot(I_f, W_i)
# テキスト特徴を投影行列W_tを通じてd_e次元空間にマッピング
text_embedding_projected = np.dot(T_f, W_t)
# 投影された画像とテキスト埋め込みをL2正規化（ノルムを1にする）、点積をコサイン類似度に等しくする
I_e = l2_normalize(image_embedding_projected, axis=1)  # 出力形状：[n, d_e]
T_e = l2_normalize(text_embedding_projected, axis=1)  # 出力形状：[n, d_e]

# --- スケーリングされたペアコサイン類似度を計算 ---
# n個の画像埋め込みとn個のテキスト埋め込み間の点積を計算します。
# I_eとT_eはすでにL2正規化されているため、点積はコサイン類似度に等しいです。
# T_e.TはT_eの転置で、形状は[d_e, n]です。
# np.dot(I_e, T_e.T)は[n, n]の行列を得て、(i, j)位置の値はi番目の画像とj番目のテキストのコサイン類似度です。
# np.exp(t)で類似度をスケーリングします。tは学習可能な温度パラメータです。
logits = np.dot(I_e, T_e.T) * np.exp(t)  # 出力形状：[n, n]

# --- 対称損失関数を計算 ---
# 目標ラベルを作成します。n個のペアサンプルを含むバッチに対して、正しいペアはlogits行列の対角線上にあります。各行（各画像）に対応する正しいテキストのインデックスが保存されています。
# labels = [0, 1, 2, ..., n-1]
labels = np.arange(n)

# 画像からテキストへの対比損失を計算：
# logits行列（類似度行列）の各行（画像を表す）について、すべてのテキスト類似度スコアとの交差エントロピー損失を計算します。
# 目標は正しいテキストインデックス（すなわちlabels）を予測することです。axis=0は行ごとに損失を計算することを示します（異なるフレームワークの実装は若干異なる場合がありますが、概念は同じです）。
loss_i = cross_entropy_loss(logits, labels, axis=0)

# テキストから画像への対比損失を計算：
# logits行列の各列（テキストのセグメントを表す）について、すべての画像類似度スコアとの交差エントロピー損失を計算します。
# 目標は正しい画像インデックス（すなわちlabels）を予測することです。axis=1は列ごとに損失を計算することを示します（概念的にはそうです）。
loss_t = cross_entropy_loss(logits, labels, axis=1)

# 最終的な対称損失は2つの方向の損失の平均です。
loss = (loss_i + loss_t) / 2

softmax(x_i) = exp(x_i) / sum(exp(x_j))

safe_softmax(x_i) = exp(x_i - max(x)) / sum(exp(x_j - max(x)))

class SiglipVisionConfig:

    def __init__(
        self,
        hidden_size=768,                # 埋め込みベクトルのサイズ
        intermediate_size=3072,         # FFN内の線形層のサイズ
        num_hidden_layers=12,           # ViTの層数
        num_attention_heads=12,         # MHAの注意ヘッド数
        num_channels=3,                 # 画像のチャンネル数、すなわちRGB
        image_size=224,                 # 受け取る画像特徴のサイズ
        patch_size=16,                  # 画像が分割されるパッチのサイズ
        layer_norm_eps=1e-6,
        attention_dropout=0.0,
        num_image_tokens: int = None,   # 画像がパッチに分割された後のトークンの総数
        # (image_size / patch_size)^2 = (224 // 16) ** 2 = 196
        **kwargs
    ):
        super().__init__()

        self.hidden_size = hidden_size
        ...

class SiglipVisionEmbeddings(nn.Module):
    def __init__(self, config: SiglipVisionConfig):
        super().__init__()
        self.config = config
        self.embed_dim = config.hidden_size
        self.image_size = config.image_size
        self.patch_size = config.patch_size

        self.patch_embedding = nn.Conv2d(
            in_channels=config.num_channels,
            out_channels=self.embed_dim,
            kernel_size=self.patch_size,
            stride=self.patch_size,
            padding="valid", # これはパディングが追加されないことを示します
        )

        self.num_patches = (self.image_size // self.patch_size) ** 2 # 前述の通り
        self.num_positions = self.num_patches
        self.position_embedding = nn.Embedding(self.num_positions, self.embed_dim)
        self.register_buffer(
            "position_ids",
            torch.arange(self.num_positions).expand((1, -1)),
            persistent=False, # このバッファがモデルの永続部分として保存されるかどうか
        )

    def forward(self, pixel_values: torch.FloatTensor) -> torch.Tensor:
        _, _, height, width = pixel_values.shape # [Batch_Size, Channels, Height, Width]
        # 畳み込みカーネルを画像に適用し、重ならないパッチを生成します。ストライドがカーネルサイズと等しいため、出力は[Batch_Size, Embed_Dim, Num_Patches_H, Num_Patches_W]の形状になります。
        # ここでNum_Patches_H = height // patch_sizeおよびNum_Patches_W = width // patch_sizeです。
        patch_embeds = self.patch_embedding(pixel_values)  
        # [Batch_Size, Embed_Dim, Num_Patches_H, Num_Patches_W] -> [Batch_Size, Embed_Dim, Num_Patches]
        # ここでNum_Patches = Num_Patches_H * Num_Patches_Wです。
        embeddings = patch_embeds.flatten(2)
        # [Batch_Size, Embed_Dim, Num_Patches] -> [Batch_Size, Num_Patches, Embed_Dim]
        embeddings = embeddings.transpose(1, 2)
        # 各パッチに位置埋め込みを追加します。各位置エンコーディングは[Embed_Dim]のサイズのベクトルです。
        embeddings = embeddings + self.position_embedding(self.position_ids)
        # [Batch_Size, Num_Patches, Embed_Dim]
        return embeddings

class SiglipEncoderLayer(nn.Module):
    def __init__(self, config: SiglipVisionConfig):
        super().__init__()
        self.embed_dim = config.hidden_size
        self.self_attn = SiglipAttention(config)
        self.layer_norm1 = nn.LayerNorm(self.embed_dim, eps=config.layer_norm_eps)
        self.mlp = SiglipMLP(config)
        self.layer_norm2 = nn.LayerNorm(self.embed_dim, eps=config.layer_norm_eps)

    # コピーを無視
    def forward(
        self,
        hidden_states: torch.Tensor
    ) -> torch.Tensor:
        # residual: [Batch_Size, Num_Patches, Embed_Dim]
        residual = hidden_states
        # [Batch_Size, Num_Patches, Embed_Dim] -> [Batch_Size, Num_Patches, Embed_Dim]
        hidden_states = self.layer_norm1(hidden_states)
        # [Batch_Size, Num_Patches, Embed_Dim] -> [Batch_Size, Num_Patches, Embed_Dim]
        hidden_states, _ = self.self_attn(hidden_states=hidden_states)
        # [Batch_Size, Num_Patches, Embed_Dim]
        hidden_states = residual + hidden_states
        # residual: [Batch_Size, Num_Patches, Embed_Dim] 
        residual = hidden_states
        # [Batch_Size, Num_Patches, Embed_Dim] -> [Batch_Size, Num_Patches, Embed_Dim]
        hidden_states = self.layer_norm2(hidden_states)
        # [Batch_Size, Num_Patches, Embed_Dim] -> [Batch_Size, Num_Patches, Embed_Dim]
        hidden_states = self.mlp(hidden_states)
        # [Batch_Size, Num_Patches, Embed_Dim]
        hidden_states = residual + hidden_states
        
        return hidden_states

class SiglipEncoder(nn.Module):
    def __init__(self, config: SiglipVisionConfig):
        super().__init__()
        self.config = config
        self.layers = nn.ModuleList(
            [SiglipEncoderLayer(config) for _ in range(config.num_hidden_layers)]
        )

    # コピーを無視
    def forward(
        self,
        inputs_embeds: torch.Tensor
    ) -> torch.Tensor:
        # inputs_embeds: [Batch_Size, Num_Patches, Embed_Dim]
        hidden_states = inputs_embeds

        for encoder_layer in self.layers:
            # [Batch_Size, Num_Patches, Embed_Dim] -> [Batch_Size, Num_Patches, Embed_Dim]
            hidden_states = encoder_layer(hidden_states)

        return hidden_states

class SiglipVisionTransformer(nn.Module):
    def __init__(self, config: SiglipVisionConfig):
        super().__init__()
        self.config = config
        embed_dim = config.hidden_size

        self.embeddings = SiglipVisionEmbeddings(config)
        self.encoder = SiglipEncoder(config)
        self.post_layernorm = nn.LayerNorm(embed_dim, eps=config.layer_norm_eps)

    def forward(self, pixel_values: torch.Tensor) -> torch.Tensor:
        # pixel_values: [Batch_Size, Channels, Height, Width] -> [Batch_Size, Num_Patches, Embed_Dim]
        hidden_states = self.embeddings(pixel_values)

        last_hidden_state = self.encoder(inputs_embeds=hidden_states)

        last_hidden_state = self.post_layernorm(last_hidden_state)

        return last_hidden_state

class SiglipVisionModel(nn.Module):

    def __init__(self, config: SiglipVisionConfig):
        super().__init__()
        self.config = config
        self.vision_model = SiglipVisionTransformer(config)

    def forward(self, pixel_values) -> Tuple:
        # [Batch_Size, Channels, Height, Width] -> [Batch_Size, Num_Patches, Embed_Dim]
        return self.vision_model(pixel_values=pixel_values) 

class PaliGemmaProcessor:

    IMAGE_TOKEN = "<image>"

    def __init__(self, tokenizer, num_image_tokens: int, image_size: int):
        super().__init__()

        self.image_seq_length = num_image_tokens
        self.image_size = image_size
		# tokenizer.add_tokens(EXTRA_TOKENS)
        # トークナイザーの説明はこちら：https://github.com/google-research/big_vision/blob/main/big_vision/configs/proj/paligemma/README.md#tokenizer
        tokens_to_add = {"additional_special_tokens": [self.IMAGE_TOKEN]}
        tokenizer.add_special_tokens(tokens_to_add)
        self.image_token_id = tokenizer.convert_tokens_to_ids(self.IMAGE_TOKEN)
        # BOSおよびEOSトークンは自分で追加します
        tokenizer.add_bos_token = False
        tokenizer.add_eos_token = False

        self.tokenizer = tokenizer

    def __call__(
        self,
        text: List[str],            # 入力テキストプロンプト
        images: List[Image.Image],  # 入力PIL.Image
        # 標準のトークン化とパディング戦略
        padding: str = "longest",
        truncation: bool = True,
    ) -> dict:
        assert len(images) == 1 and len(text) == 1, f"Received {len(images)} images for {len(text)} prompts." # ここでは1対の画像-テキストのみを実装しています

        pixel_values = process_images(
            images,
            size=(self.image_size, self.image_size),
            resample=Image.Resampling.BICUBIC,  # 画像リサイズ時に詳細とテクスチャを保持するためにバイキュービック補間を選択
            rescale_factor=1 / 255.0,           # ピクセル値を[0, 1]に再スケーリング
            image_mean=IMAGENET_STANDARD_MEAN,  # 標準化の平均（おおよそ[0.5, 0.5, 0.5]）
            image_std=IMAGENET_STANDARD_STD,    # 標準化の標準偏差（おおよそ[0.5, 0.5, 0.5]）
        )
        # これによりピクセル値が[-1, 1]に変換されます
        # NumPy配列のリストを形状[Batch_Size, Channel, Height, Width]の単一のNumPy配列に変換します
        pixel_values = np.stack(pixel_values, axis=0) # バッチにスタックします
        # NumPy配列をPyTorchテンソルに変換します
        pixel_values = torch.tensor(pixel_values)

   画像 (PIL.Image) → リサイズ → NumPy配列 → ピクセル値のスケーリング (0-1) → 標準化 (平均/標準偏差) → チャンネル次元の転置 → PyTorchテンソル

        # プロンプトの先頭に`self.image_seq_length`個の画像トークンを追加します
        input_strings = [
            add_image_tokens_to_prompt(
                prefix_prompt=prompt,
                bos_token=self.tokenizer.bos_token,
                image_seq_len=self.image_seq_length, # <image>プレースホルダーの数
                image_token=self.IMAGE_TOKEN,        # "<image>"文字列
            )
            for prompt in text
        ]

        # <image>プレースホルダーを含む文字列シーケンスをモデルが読み取れるinput_idsとattention_maskに変換します
        inputs = self.tokenizer(
            input_strings,
            return_tensors="pt",
            padding=padding,
            truncation=truncation,
        )
		# 処理された画像テンソルとテキスト関連のトークンシーケンスを含む辞書を返します
        return_data = {"pixel_values": pixel_values, **inputs}

        return return_data```

理解を助けるために、以下に例$$^{[2]}$$を用いて最終的な形態を可視化します：

```python
from transformers.utils.attention_visualizer import AttentionMaskVisualizer

visualizer = AttentionMaskVisualizer("google/paligemma2-3b-mix-224")
visualizer("<img> この画像には何がありますか？")

class PaliGemmaForConditionalGeneration(nn.Module):
    def __init__(self, config: PaliGemmaConfig):
        super().__init__()
        self.config = config
        self.vision_tower = SiglipVisionModel(config.vision_config)
        self.multi_modal_projector = PaliGemmaMultiModalProjector(config)
        self.vocab_size = config.vocab_size

        language_model = GemmaForCausalLM(config.text_config)
        self.language_model = language_model

        self.pad_token_id = self.config.pad_token_id if self.config.pad_token_id is not None else -1

	# 重みの結びつけ
    def tie_weights(self):
        return self.language_model.tie_weights()

def forward(
        self,
        input_ids: torch.LongTensor = None,
        pixel_values: torch.FloatTensor = None,
        attention_mask: Optional[torch.Tensor] = None,
        kv_cache: Optional[KVCache] = None,
    ) -> Tuple:

        # 入力が右パディングされていることを確認します
        assert torch.all(attention_mask == 1), "The input cannot be padded"

        # 1. 入力埋め込みを抽出します
        # 形状: (Batch_Size, Seq_Len, Hidden_Size)
        inputs_embeds = self.language_model.get_input_embeddings()(input_ids)

        # 2. テキストと画像を統合します
        # [Batch_Size, Channels, Height, Width] -> [Batch_Size, Num_Patches, Embed_Dim]
        selected_image_feature = self.vision_tower(pixel_values.to(inputs_embeds.dtype))
 

		# [Batch_Size, Num_Patches, Embed_Dim] -> [Batch_Size, Num_Patches, Hidden_Size]
		image_features = self.multi_modal_projector(selected_image_feature)

class PaliGemmaMultiModalProjector(nn.Module):
    def __init__(self, config: PaliGemmaConfig):
        super().__init__()
        self.linear = nn.Linear(config.vision_config.hidden_size, config.vision_config.projection_dim, bias=True)

    def forward(self, image_features):
        # [Batch_Size, Num_Patches, Embed_Dim] -> [Batch_Size, Num_Patches, Projection_Dim]
        hidden_states = self.linear(image_features)
        return hidden_states

        # 3. テキストトークンと画像トークンの埋め込みを統合します
        inputs_embeds, attention_mask, position_ids = self._merge_input_ids_with_image_features(image_features, inputs_embeds, input_ids, attention_mask, kv_cache)

# 形状: [Batch_Size, Seq_Len]. 画像トークンに対してTrue
image_mask = input_ids == self.config.image_token_index # "<image>": 257152

# 形状: [Batch_Size, Seq_Len]. テキストトークンに対してTrue
text_mask = (input_ids != self.config.image_token_index) & (input_ids != self.pad_token_id)

text_mask_expanded = text_mask.unsqueeze(-1).expand(-1, -1, embed_dim)
image_mask_expanded = image_mask.unsqueeze(-1).expand(-1, -1, embed_dim)

# テキスト埋め込みを配置します
final_embedding = torch.where(text_mask_expanded, inputs_embeds, final_embedding)

# 画像埋め込みを配置します
final_embedding = final_embedding.masked_scatter(image_mask_expanded, scaled_image_features)

		# 4. 言語モデルに入力します
        outputs = self.language_model(
            attention_mask=attention_mask,
            position_ids=position_ids,
            inputs_embeds=inputs_embeds,
            kv_cache=kv_cache,
        )

        return outputs

class GemmaForCausalLM(nn.Module):
    # ... (他のコード) ...

    def forward(
        self,
        attention_mask: Optional[torch.Tensor] = None,
        position_ids: Optional[torch.LongTensor] = None,
        inputs_embeds: Optional[torch.FloatTensor] = None,
        kv_cache: Optional[KVCache] = None,
    ) -> Tuple:

        # ... (モデルの前向き伝播) ...
        outputs = self.model( # self.modelはGemmaModel
            attention_mask=attention_mask,
            position_ids=position_ids,
            inputs_embeds=inputs_embeds,
            kv_cache=kv_cache,
        )

        hidden_states = outputs # これはGemmaModelからの最後の隠れ状態です
        logits = self.lm_head(hidden_states) # lm_headは線形層で、出力次元は語彙サイズです
        logits = logits.float()

        return_data = {
            "logits": logits,
        }

        if kv_cache is not None:
            # 更新されたキャッシュを返します
            return_data["kv_cache"] = kv_cache

        return return_data```
## Prospectives

- VLMが画像情報をより効果的に利用し、画像に焦点を当てずにテキストに注目する方法
- VLM + RL、画像内の情報に対する推論能力を向上させ、視覚推論タスクを完了するための解釈可能性

## Reference

- \[1] [一般化された視覚言語モデル | Lil'Log](https://lilianweng.github.io/posts/2022-06-09-vlm/)
- \[2] [PaliGemma – Googleの最先端オープンビジョン言語モデル](https://huggingface.co/blog/paligemma)
- \[3] [Transformers/PaliGemma](https://huggingface.co/docs/transformers/main/en//model_doc/paligemma)
- \[4] [Groundlight.ai: 視覚-言語モデル（VLM）はどのように機能するか？](https://www.groundlight.ai/blog/how-vlm-works-tokens)