diff --git a/aiak_megatron/megatron/core/transformer/transformer_config.py b/aiak_megatron/megatron/core/transformer/transformer_config.py index e41c3eeb..87f52bc7 100644 --- a/aiak_megatron/megatron/core/transformer/transformer_config.py +++ b/aiak_megatron/megatron/core/transformer/transformer_config.py @@ -3,7 +3,7 @@ """tranformer config""" import warnings -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Callable, List, Optional, Tuple, Union import torch @@ -530,6 +530,51 @@ class TransformerConfig(ModelParallelConfig): Using fp32 or fp64 can improve stability especially when the number of experts is large (e.g. finegrained-moe). None means no changes for dtype.""" + final_logit_softcapping: Optional[float] = None + """If set, apply ``cap * tanh(logits / cap)`` to the language-model output logits. + Used by Gemma4 (cap=30.0). None disables the softcap.""" + + use_layer_scalar: bool = False + """If True, each transformer layer learns a scalar gate (init 1.0) applied to the + layer output before the residual add. Used by Gemma4.""" + + partial_rotary_factor: float = 1.0 + """Fraction of the head dim that receives RoPE rotation. Gemma4 uses 0.25. + A value of 1.0 (default) keeps full RoPE and is a no-op for existing models.""" + + sliding_window: Optional[int] = None + """Window size used by sliding-window attention layers. None means no sliding mask.""" + + rotary_base_sliding: Optional[int] = None + """Optional separate RoPE base for sliding-window attention layers. + If None, sliding layers share ``rotary_base`` with global layers.""" + + layer_pattern: list = field(default_factory=list) + """Per-layer attention type as a list of strings ('sliding' / 'global'), + length == num_layers. Empty list (default) means all layers use the same + config-level head_dim / num_query_groups (no hybrid behavior).""" + + per_layer_kv_channels: dict = field(default_factory=dict) + """Optional per-layer-type override of ``kv_channels`` (a.k.a. head_dim). + Keyed by layer-type string ('sliding' / 'global'). Empty dict (default) + means use ``kv_channels`` for every layer.""" + + per_layer_num_query_groups: dict = field(default_factory=dict) + """Optional per-layer-type override of ``num_query_groups``. + Empty dict (default) means use ``num_query_groups`` for every layer.""" + + attention_k_eq_v: bool = False + """If True, Gemma4 global layers use K=V tying according to ``kv_tied_layers``. + Existing models keep the default False behavior.""" + + kv_tied_layers: list = field(default_factory=list) + """Layer indices for which K and V projections share the same weight tensor + (used by Gemma4 global layers). Empty list (default) means no tying.""" + + scale_emb_by_sqrt_hidden: bool = False + """If True, multiply token embeddings by ``sqrt(hidden_size)`` before the + transformer (Gemma family convention).""" + def __post_init__(self): """Python dataclass method that is used to modify attributes after initialization. See https://docs.python.org/3/library/dataclasses.html#post-init-processing for more diff --git a/aiak_training_llm/data/chat_templete.py b/aiak_training_llm/data/chat_templete.py index 82262cc4..8eebdefe 100644 --- a/aiak_training_llm/data/chat_templete.py +++ b/aiak_training_llm/data/chat_templete.py @@ -13,10 +13,11 @@ import re from abc import ABC, abstractmethod from dataclasses import dataclass, field -from typing import TYPE_CHECKING, Type, Dict, List, Optional, Sequence, Set, Tuple, Union +from typing import TYPE_CHECKING, Dict, List, Optional, Sequence, Set, Tuple, Type, Union from aiak_training_llm.utils.constants import DataRoles -from .mm_plugin import MMPlugin, Qwen2VLPlugin + +from .mm_plugin import Gemma4VLPlugin, MMPlugin, Qwen2VLPlugin if TYPE_CHECKING: @@ -459,3 +460,15 @@ def get_support_templates() -> List[str]: format_user=StringFormatter(slots=["<|User|>{{content}}<|Assistant|>"]), format_prefix=EmptyFormatter(slots=[{"bos_token"}]), ) + +_register_chat_template( + name="gemma4", + format_user=StringFormatter( + slots=["<|turn>user\n{{content}}\n<|turn>model\n"] + ), + format_assistant=StringFormatter(slots=["{{content}}\n"]), + format_system=StringFormatter(slots=["<|turn>system\n{{content}}\n"]), + format_separator=EmptyFormatter(slots=[""]), + format_prefix=EmptyFormatter(slots=[{"bos_token"}]), + mm_plugin=Gemma4VLPlugin(image_token="<|image|>", video_token=None), +) diff --git a/aiak_training_llm/data/mm_plugin.py b/aiak_training_llm/data/mm_plugin.py index 6438e282..09a7d57d 100644 --- a/aiak_training_llm/data/mm_plugin.py +++ b/aiak_training_llm/data/mm_plugin.py @@ -4,6 +4,7 @@ from typing import TYPE_CHECKING, Dict, List, Optional, Sequence, Tuple, Type, TypedDict, Union import numpy as np +import torch from PIL import Image from PIL.Image import Image as ImageObject from typing_extensions import override @@ -16,6 +17,7 @@ import torch from transformers.image_processing_utils import BaseImageProcessor + from transformers.processing_utils import ProcessorMixin class EncodedImage(TypedDict): """Encoded image type.""" @@ -292,3 +294,178 @@ def get_mm_inputs( ) -> Dict[str, Union[List[int], "torch.Tensor"]]: self._validate_input(images, videos) return self._get_mm_inputs(images, videos, processor) + + +class Gemma4VLPlugin(MMPlugin): + """Gemma4-VL passthrough plugin: process_messages returns (messages, mm_inputs) + where mm_inputs follows the existing OV2 flattened-patch contract. + + Cross-module contract (downstream consumers depend on these invariants): + - ``pixel_values`` shape ``[total_imgs_in_batch, P, D]`` — FLATTENED, NOT ``[B, ...]``. + Indexing by batch index will silently return wrong tensor. + - ``image_grid_thw`` is synthesized from HF ``image_position_ids`` as + ``[num_images, 3]`` rows of ``[1, H_p, W_p]``. + - Text-only batches OMIT multimodal keys entirely (no zero-shape sentinel). + All downstream consumers MUST guard with ``in mm_inputs``. + """ + + @staticmethod + def _flatten_gemma4_image_outputs( + image_outputs: dict[str, "torch.Tensor"], + ) -> dict[str, Union["torch.Tensor", list["torch.Tensor"]]]: + pixel_values = image_outputs["pixel_values"] + image_position_ids = image_outputs["image_position_ids"] + num_soft_tokens_per_image = image_outputs["num_soft_tokens_per_image"] + + valid_mask = (image_position_ids != -1).all(dim=-1) + flat_pixel_values = pixel_values[valid_mask] + + image_grid_rows: list[list[int]] = [] + patch_positions: list[torch.Tensor] = [] + for image_idx in range(image_position_ids.shape[0]): + valid_positions = image_position_ids[image_idx][valid_mask[image_idx]].to(dtype=torch.int64) + if valid_positions.numel() == 0: + raise ValueError(f"Gemma4 image {image_idx} has no valid patch positions.") + + width = int(valid_positions[:, 0].max().item()) + 1 + height = int(valid_positions[:, 1].max().item()) + 1 + patch_count = int(valid_positions.shape[0]) + if patch_count != height * width: + raise ValueError( + "Gemma4 image patch positions are not a dense single-frame grid: " + f"image_idx={image_idx}, patch_count={patch_count}, height={height}, width={width}." + ) + + image_grid_rows.append([1, height, width]) + patch_positions.append( + torch.stack( + ( + torch.zeros(patch_count, dtype=torch.int64, device=valid_positions.device), + valid_positions[:, 1], + valid_positions[:, 0], + ), + dim=-1, + ) + ) + + image_grid_thw = torch.tensor( + image_grid_rows, + dtype=torch.int32, + device=image_position_ids.device, + ) + + return { + "pixel_values": flat_pixel_values, + "image_grid_thw": image_grid_thw, + "patch_positions": patch_positions, + "image_position_ids": image_position_ids, + "num_soft_tokens_per_image": num_soft_tokens_per_image, + } + + def _build_gemma4_mm_inputs( + self, + images: Sequence["ImageInput"], + processor: Optional["ProcessorMixin"], + ) -> tuple[Optional[list["ImageObject"]], dict[str, Union["torch.Tensor", list["torch.Tensor"]]]]: + regularized_images = self._regularize_images(images) if len(images) != 0 else None + if regularized_images is None: + return None, {} + + image_outputs = processor.image_processor(regularized_images, return_tensors="pt") + return regularized_images, self._flatten_gemma4_image_outputs(dict(image_outputs)) + + def _expand_image_placeholders( + self, + messages: Sequence[dict[str, str]], + num_soft_tokens_per_image: Sequence[int], + processor: Optional["ProcessorMixin"], + ) -> list[dict[str, str]]: + messages = deepcopy(messages) + actual_num_images = len(num_soft_tokens_per_image) + + image_placeholder_count = sum(message["content"].count(Placeholder.IMAGE) for message in messages) + if actual_num_images > 0 and image_placeholder_count != actual_num_images: + for message in messages: + message["content"] = message["content"].replace(Placeholder.IMAGE, "") + + first_user_msg = None + for message in messages: + if message.get("role") == "user": + first_user_msg = message + break + + if first_user_msg is None: + raise ValueError("Cannot rebuild Gemma4 image placeholders: no user message found.") + + image_placeholders = "\n".join([Placeholder.IMAGE] * actual_num_images) + user_content = first_user_msg["content"].lstrip("\n") + first_user_msg["content"] = f"{image_placeholders}\n{user_content}" + + image_idx = 0 + for message in messages: + content = message["content"] + while Placeholder.IMAGE in content: + if image_idx >= actual_num_images: + raise ValueError( + f"The number of {Placeholder.IMAGE} tokens is greater than available images." + ) + + n_soft_tokens = int(num_soft_tokens_per_image[image_idx]) + replacement = ( + f"{processor.boi_token}{self.image_token * n_soft_tokens}{processor.eoi_token}" + ) + content = content.replace(Placeholder.IMAGE, replacement, 1) + image_idx += 1 + + if Placeholder.VIDEO in content: + raise ValueError("Gemma4-VL video placeholders are not supported in this OV2 path yet.") + + message["content"] = content + + if image_idx != actual_num_images: + raise ValueError( + f"The number of images ({actual_num_images}) does not match expanded placeholders ({image_idx})." + ) + + return messages + + @override + def _preprocess_image(self, image: "ImageObject", **kwargs) -> "ImageObject": + return super()._preprocess_image(image, **kwargs) + + @override + def process_messages( + self, + messages: Sequence[dict[str, str]], + images: Sequence["ImageInput"], + videos: Sequence["VideoInput"], + processor: Optional["ProcessorMixin"], + ) -> tuple[list[dict[str, str]], dict[str, "torch.Tensor"]]: + self._validate_input(images, videos) + _regularized_images, mm_inputs = self._build_gemma4_mm_inputs(images, processor) + + if "num_soft_tokens_per_image" in mm_inputs: + messages = self._expand_image_placeholders( + messages, + mm_inputs["num_soft_tokens_per_image"], + processor, + ) + else: + messages = list(messages) + + return messages, dict(mm_inputs) + + @override + def get_mm_inputs( + self, + images: Sequence["ImageInput"], + videos: Sequence["VideoInput"], + imglens: Sequence[int], + vidlens: Sequence[int], + seqlens: Sequence[int], + processor: Optional["ProcessorMixin"], + ) -> dict[str, Union[list[int], "torch.Tensor"]]: + self._validate_input(images, videos) + del imglens, vidlens, seqlens + _regularized_images, mm_inputs = self._build_gemma4_mm_inputs(images, processor) + return dict(mm_inputs) diff --git a/aiak_training_llm/data/multimodal/gemma4_vl_task_encoder.py b/aiak_training_llm/data/multimodal/gemma4_vl_task_encoder.py new file mode 100644 index 00000000..0f47fa13 --- /dev/null +++ b/aiak_training_llm/data/multimodal/gemma4_vl_task_encoder.py @@ -0,0 +1,934 @@ +"""Gemma4VLTaskEncoder class. + +Verbatim mirror of :mod:`qwen2vl_task_encoder` for the Gemma4-VL family. +The structural pipeline (mm_plugin → encode_multiturn → packing → batching) is +identical; only Gemma4-specific surface differs: + +- Special-token strings (Gemma4 vocab is unrelated to Qwen): + * boi (vision_start) -> ``<|image>`` (id=255999) + * image pad -> ``<|image|>`` (id=258880) + * eoi (vision_end) -> ```` (id=258882) + * video -> ``<|video|>`` (id=258884) +- Spatial merge: Gemma4 uses ``image_processor.pooling_kernel_size=3`` instead of + Qwen's ``merge_size=2``; the existing ``getattr(..., 'merge_size', 2)`` lookup + reads the wrong attribute on Gemma4Processor so we resolve it via a small + helper (``_gemma4_spatial_merge_size``) that prefers ``pooling_kernel_size``. +- ``smart_resize`` factor: Qwen uses 28 (= patch_size 14 × merge 2); Gemma4 uses + patch_size 16 × pooling 3 = 48. We pass ``size_factor=48`` through. + +We keep this as a fork rather than a subclass per user direction (avoids leaky +class-attribute overrides on the parent Qwen module-level constants). + +CAVEAT — image placeholder wrapping mismatch (deferred fix, ack'd in P4.9): + Qwen2VL inline format is ``<|vision_start|><|image_pad|><|vision_end|>`` (3 + tokens, fence + pad + fence). Gemma4 chat-template instead emits a single + ``<|image|>`` (no boi/eoi wrap). Our verbatim mirror keeps the 3-token wrap + using Gemma4 ids — correct for fence semantics but produces 1 extra image + placeholder vs the reference Gemma4 jinja path. Forward will run; lm_loss + numerics will be off until a future revision rewrites the + ````-substitution path to call ``processor.apply_chat_template`` + directly. Tracked as a Phase-P4.9 known issue. +""" + +import re +import sys +from collections.abc import Callable +from dataclasses import dataclass +from typing import Optional, TypeVar, Union + +import numpy as np +import torch +from megatron.energon import CaptioningSample, SkipSample, VQASample +from megatron.energon.flavors.base_dataset import ( + BaseCoreDatasetFactory, + SavableDataset, +) +from megatron.energon.flavors.crude import CrudeWebdataset +from megatron.energon.flavors.webdataset import VideoData +from megatron.energon.metadataset.loader_interface import DatasetBlendMode +from megatron.energon.task_encoder.base import stateless +from megatron.energon.worker import WorkerConfig +from megatron.energon.wrappers import ( + BlendDataset, + EpochizeDataset, + LogSampleDataset, + ShuffleBufferDataset, +) +from megatron.energon.wrappers.repeat_dataset import RepeatDataset +from qwen_vl_utils.vision_process import smart_nframes, smart_resize +from torchvision import transforms +from torchvision.transforms import InterpolationMode +from typing_extensions import override + +from aiak_training_llm.data.multimodal import MultiMixQASample +from aiak_training_llm.data.multimodal.length_sort_dataset import LengthPoolSortDataset +from aiak_training_llm.data.multimodal.packed_sort_dataset import PackedSeparateSortDataset +from aiak_training_llm.utils import constants, get_chat_template +from transformers import AutoProcessor + +from .task_encoder import ImageTaskBatchPacked, ImageTaskSample, ImageTaskSamplePacked, TaskEncoder + + +T = TypeVar("T") +V = TypeVar("V") +T_sample = TypeVar("T_sample") +T_encoded_sample = TypeVar("T_encoded_sample") +T_raw_batch = TypeVar("T_raw_batch") +T_batch = TypeVar("T_batch") + + +IGNORE_INDEX = -100 # ID for labels that should be ignored. +IMAGE_TOKEN = "<|image|>" +VIDEO_TOKEN = "<|video|>" +VISION_TAGS = ["<|image>", ""] +IMAGE_TOKEN_WITH_TAGS = VISION_TAGS[0] + IMAGE_TOKEN + VISION_TAGS[1] +VIDEO_TOKEN_WITH_TAGS = VISION_TAGS[0] + VIDEO_TOKEN + VISION_TAGS[1] +SKIP_LOG_LIMIT = 5 +_SKIP_LOG_COUNTS: dict[str, int] = {} + + +def _gemma4_spatial_merge_size(image_processor) -> int: + """Resolve Gemma4's spatial merge factor. + + Gemma4Processor exposes ``pooling_kernel_size`` (default 3); the parent + Qwen lookup ``merge_size`` (default 2) returns the wrong fallback on Gemma4 + and silently corrupts patch_position block layout. Prefer the Gemma4 name + when present. + """ + val = getattr(image_processor, "pooling_kernel_size", None) + if val is not None: + return int(val) + return int(getattr(image_processor, "merge_size", 2)) + + +def skip_malformed_multimodal_sample(sample_key: str, signature: str, detail: str) -> None: + count = _SKIP_LOG_COUNTS.get(signature, 0) + 1 + _SKIP_LOG_COUNTS[signature] = count + + if count <= SKIP_LOG_LIMIT: + print(f"Skipping malformed multimodal sample {sample_key}: {detail}", file=sys.stderr) + if count == SKIP_LOG_LIMIT: + print( + f"Further '{signature}' skip logs will be suppressed for this worker.", + file=sys.stderr, + ) + + raise SkipSample(f"{sample_key}: {detail}") + + +def get_stateless(fn: Callable[..., T_sample]) -> bool: + """Get whether a function is stateless.""" + return getattr(fn, "__stateless__", False) + + +def convert_positions_to_block_layout( + positions: torch.Tensor, t: int, h: int, w: int, spatial_merge_size: int = 2 +) -> torch.Tensor: + """ + Convert patch positions from row-major order to 2x2 block layout. + + This function reorders patch positions to match the 2x2 block arrangement + used by the image processor. Uses index-based reordering instead of reshape. + + Args: + positions: Patch positions in row-major order, shape [t*h*w, 3] + t: temporal dimension + h: height (unmerged patch count) + w: width (unmerged patch count) + spatial_merge_size: size of spatial merge blocks (default: 2) + + Returns: + torch.Tensor: Patch positions in 2x2 block order, same shape [t*h*w, 3] + """ + sms = spatial_merge_size + if sms == 1: + return positions + + device = positions.device + total_patches = t * h * w + + # Generate row-major indices: [0, 1, 2, ..., t*h*w-1] + # Reshape to [t, h, w] + indices = torch.arange(total_patches, device=device).view(t, h, w) + + # Calculate merged dimensions + h_merged = h // sms + w_merged = w // sms + + # Reshape to [t, h_merged, sms, w_merged, sms] + indices = indices.view(t, h_merged, sms, w_merged, sms) + + # Permute to [t, h_merged, w_merged, sms_h, sms_w] - 2x2 block order + indices = indices.permute(0, 1, 3, 2, 4).contiguous() + + # Flatten to get the reordering indices + indices = indices.view(total_patches) + + # Apply the reordering to positions + return positions[indices] + + +@dataclass +class Gemma4VLImageTaskSample(ImageTaskSample): + """An image task sample with a grid of tokens and their corresponding pixel values.""" + + image_grid_thw: torch.Tensor = None + video_grid_thw: torch.Tensor = None + + def __init__(self, image_grid_thw: str, video_grid_thw=None, **kwargs): + super().__init__(**kwargs) + self.image_grid_thw = image_grid_thw + self.video_grid_thw = video_grid_thw + + +@dataclass +class Gemma4VLImageTaskSamplePacked(ImageTaskSamplePacked): + """An image task sample with a grid of tokens and their corresponding pixel values.""" + + image_grid_thw: torch.Tensor = None + video_grid_thw: torch.Tensor = None + + def __init__(self, sample: ImageTaskSample, image_grid_thw: str, video_grid_thw=None): + super().__init__(**vars(sample)) + self.image_grid_thw = image_grid_thw + self.video_grid_thw = video_grid_thw + + +@dataclass +class Gemma4VLImageTaskBatchPacked(ImageTaskBatchPacked): + """An image task sample with a grid of tokens and their corresponding pixel values.""" + + image_grid_thw: torch.Tensor = None + video_grid_thw: torch.Tensor = None + + def __init__(self, sample: ImageTaskSample, image_grid_thw: str, video_grid_thw=None): + super().__init__(**vars(sample)) + self.image_grid_thw = image_grid_thw + self.video_grid_thw = video_grid_thw + + +class Gemma4VLTaskEncoder(TaskEncoder): + """A simple task encoder for VLMs.""" + + def __init__(self, args): + super().__init__() + if args.training_phase in ["sft"]: + self.chat_template = get_chat_template() + self.processor = AutoProcessor.from_pretrained(self.args.hf_tokenizer_path, trust_remote_code=True) + + # image + self.min_pixels = args.min_pixels + self.max_pixels = args.max_pixels + + def _normalize_image_backed_video_placeholders( + self, + messages: list[dict[str, str]], + image: list | None, + video: list | None, + ) -> list[dict[str, str]]: + """Rewrite image-backed