diff --git a/tests/diffusion/models/t5_encoder/test_t5_encoder_tp.py b/tests/diffusion/models/t5_encoder/test_t5_encoder_tp.py new file mode 100644 index 000000000..b36acc68a --- /dev/null +++ b/tests/diffusion/models/t5_encoder/test_t5_encoder_tp.py @@ -0,0 +1,161 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +import pytest +import torch +from transformers import T5Config +from vllm.config import DeviceConfig, VllmConfig, set_current_vllm_config + +from vllm_omni.diffusion.models.t5_encoder.t5_encoder import ( + T5EncoderModel, + T5SelfAttention, +) + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + +_T5_MODULE = "vllm_omni.diffusion.models.t5_encoder.t5_encoder" + +SMALL_T5_CONFIG = dict( + d_model=64, + d_kv=8, + d_ff=128, + num_heads=8, + num_layers=2, + vocab_size=256, + relative_attention_num_buckets=32, + relative_attention_max_distance=128, + is_gated_act=True, + dense_act_fn="gelu_new", + layer_norm_epsilon=1e-6, + feed_forward_proj="gated-gelu", +) + + +@pytest.fixture(scope="module") +def t5_config() -> T5Config: + return T5Config(**SMALL_T5_CONFIG) + + +@pytest.fixture(scope="function", autouse=True) +def setup_tp_group(monkeypatch, mocker): + """Set up TP=2, rank=0, VllmConfig, and mock activation for all tests.""" + device_config = DeviceConfig(device="cpu") + + # TP world size + monkeypatch.setattr("vllm.model_executor.layers.linear.get_tensor_model_parallel_world_size", lambda: 2) + monkeypatch.setattr(f"{_T5_MODULE}.get_tensor_model_parallel_world_size", lambda: 2) + monkeypatch.setattr( + "vllm.model_executor.layers.vocab_parallel_embedding.get_tensor_model_parallel_world_size", + lambda: 2, + ) + + monkeypatch.setattr(f"{_T5_MODULE}.get_tensor_model_parallel_rank", lambda: 0) + monkeypatch.setattr( + "vllm.model_executor.layers.vocab_parallel_embedding.get_tensor_model_parallel_rank", + lambda: 0, + ) + + # TP group + mock_tp_group = mocker.MagicMock() + mock_tp_group.world_size = 2 + mocker.patch("vllm.distributed.parallel_state.get_tp_group", return_value=mock_tp_group) + + monkeypatch.setattr(f"{_T5_MODULE}.get_act_fn", lambda _: torch.nn.GELU()) + + with set_current_vllm_config(VllmConfig(device_config=device_config)): + yield + + +class TestRelativePositionBiasTPSlicing: + """Verify compute_bias slices heads correctly per TP rank.""" + + def test_compute_bias_shape(self, t5_config): + attn = T5SelfAttention(t5_config, has_relative_attention_bias=True) + + seq_len = 6 + bias = attn.compute_bias(seq_len, seq_len, device=torch.device("cpu")) + + local_heads = t5_config.num_heads // 2 + assert bias.shape == (1, local_heads, seq_len, seq_len) + + def test_all_ranks_cover_all_heads(self, t5_config, monkeypatch): + seq_len = 4 + + biases = [] + ref_weight = None + for rank in range(2): + monkeypatch.setattr(f"{_T5_MODULE}.get_tensor_model_parallel_rank", lambda r=rank: r) + attn = T5SelfAttention(t5_config, has_relative_attention_bias=True) + if rank > 0: + attn.relative_attention_bias.weight.data.copy_(ref_weight) + else: + ref_weight = attn.relative_attention_bias.weight.data.clone() + biases.append(attn.compute_bias(seq_len, seq_len, device=torch.device("cpu"))) + + full_bias = torch.cat(biases, dim=1) + assert full_bias.shape == (1, t5_config.num_heads, seq_len, seq_len) + + +class TestT5EncoderModelWeightLoading: + """Test weight loading at the top-level T5EncoderModel.""" + + def test_model_instantiation(self, t5_config): + model = T5EncoderModel(t5_config, prefix="text_encoder") + + assert model.config is t5_config + assert model.encoder is not None + assert len(model.encoder.block) == t5_config.num_layers + + def test_embedding_shape(self, t5_config): + model = T5EncoderModel(t5_config, prefix="text_encoder") + + assert model.shared.embedding_dim == t5_config.d_model + + def test_embed_input_ids(self, t5_config, monkeypatch): + # Verify method and output shape + model = T5EncoderModel(t5_config, prefix="text_encoder") + + # Mock all-reduce to be identity (no actual TP communication) + monkeypatch.setattr( + "vllm.model_executor.layers.vocab_parallel_embedding.tensor_model_parallel_all_reduce", + lambda x: x, + ) + + input_ids = torch.randint(0, t5_config.vocab_size, (2, 8)) + embeddings = model.embed_input_ids(input_ids) + + assert embeddings.shape == (2, 8, t5_config.d_model) + + def test_qkv_weights_loaded_through_blocks(self): + # Verify that HF-style separate Q/K/V weights can be loaded through + # the block hierarchy + config = T5Config(**{**SMALL_T5_CONFIG, "num_layers": 1}) + model = T5EncoderModel(config, prefix="text_encoder") + + inner_dim = config.num_heads * config.d_kv + prefix = "encoder.block.0.layer.0.SelfAttention." + + loaded = model.load_weights( + [ + (prefix + "q.weight", torch.randn(inner_dim, config.d_model)), + (prefix + "k.weight", torch.randn(inner_dim, config.d_model)), + (prefix + "v.weight", torch.randn(inner_dim, config.d_model)), + ] + ) + + assert len(loaded) > 0 + attn = model.encoder.block[0].layer[0].SelfAttention + expected_qkv_dim = 3 * (config.num_heads // 2) * config.d_kv + assert attn.qkv_proj.weight.shape == (expected_qkv_dim, config.d_model) + + +class TestTPConstraints: + """Verify that invalid TP configurations raise clear errors.""" + + def test_num_heads_not_divisible_by_tp(self): + config = T5Config(**{**SMALL_T5_CONFIG, "num_heads": 7}) + with pytest.raises(AssertionError, match=r"num_heads.*must be divisible by tp_size"): + T5SelfAttention(config) + + def test_num_heads_divisible_by_tp(self, t5_config): + attn = T5SelfAttention(t5_config) + assert attn.n_heads_per_partition == 4 diff --git a/vllm_omni/diffusion/models/flux/pipeline_flux.py b/vllm_omni/diffusion/models/flux/pipeline_flux.py index 06af7c95d..b7aa0c7b4 100644 --- a/vllm_omni/diffusion/models/flux/pipeline_flux.py +++ b/vllm_omni/diffusion/models/flux/pipeline_flux.py @@ -18,7 +18,7 @@ from diffusers.schedulers.scheduling_flow_match_euler_discrete import ( ) from diffusers.utils.torch_utils import randn_tensor from torch import nn -from transformers import CLIPTextModel, CLIPTokenizer, T5EncoderModel, T5TokenizerFast +from transformers import AutoConfig, CLIPTextModel, CLIPTokenizer, T5TokenizerFast from vllm.model_executor.models.utils import AutoWeightsLoader from vllm_omni.diffusion.data import DiffusionOutput, OmniDiffusionConfig @@ -27,6 +27,7 @@ from vllm_omni.diffusion.distributed.parallel_state import get_classifier_free_g from vllm_omni.diffusion.distributed.utils import get_local_device from vllm_omni.diffusion.model_loader.diffusers_loader import DiffusersPipelineLoader from vllm_omni.diffusion.models.flux import FluxTransformer2DModel +from vllm_omni.diffusion.models.t5_encoder import T5EncoderModel from vllm_omni.diffusion.profiler.diffusion_pipeline_profiler import DiffusionPipelineProfilerMixin from vllm_omni.diffusion.quantization import get_vllm_quant_config_for_layers from vllm_omni.diffusion.request import OmniDiffusionRequest @@ -148,7 +149,14 @@ class FluxPipeline(nn.Module, CFGParallelMixin, DiffusionPipelineProfilerMixin): revision=None, prefix="transformer.", fall_back_to_pt=True, - ) + ), + DiffusersPipelineLoader.ComponentSource( + model_or_path=od_config.model, + subfolder="text_encoder_2", + revision=None, + prefix="text_encoder_2.", + fall_back_to_pt=True, + ), ] self.device = get_local_device() @@ -162,9 +170,8 @@ class FluxPipeline(nn.Module, CFGParallelMixin, DiffusionPipelineProfilerMixin): self.text_encoder = CLIPTextModel.from_pretrained( model, subfolder="text_encoder", local_files_only=local_files_only ).to(self.device) - self.text_encoder_2 = T5EncoderModel.from_pretrained( - model, subfolder="text_encoder_2", local_files_only=local_files_only - ).to(self.device) + t5_config = AutoConfig.from_pretrained(model, subfolder="text_encoder_2", local_files_only=local_files_only) + self.text_encoder_2 = T5EncoderModel(t5_config, prefix="text_encoder_2").to(self.device) self.vae = AutoencoderKL.from_pretrained(model, subfolder="vae", local_files_only=local_files_only).to( self.device ) @@ -291,7 +298,7 @@ class FluxPipeline(nn.Module, CFGParallelMixin, DiffusionPipelineProfilerMixin): f" {max_sequence_length} tokens: {removed_text}" ) - prompt_embeds = self.text_encoder_2(text_input_ids.to(self.device), output_hidden_states=False)[0] + prompt_embeds = self.text_encoder_2(text_input_ids.to(self.device))[0] dtype = self.text_encoder_2.dtype prompt_embeds = prompt_embeds.to(dtype=dtype, device=self.device) diff --git a/vllm_omni/diffusion/models/t5_encoder/__init__.py b/vllm_omni/diffusion/models/t5_encoder/__init__.py new file mode 100644 index 000000000..f2ca52efa --- /dev/null +++ b/vllm_omni/diffusion/models/t5_encoder/__init__.py @@ -0,0 +1,7 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""Tensor-parallel T5 encoder model.""" + +from vllm_omni.diffusion.models.t5_encoder.t5_encoder import T5EncoderModel + +__all__ = ["T5EncoderModel"] diff --git a/vllm_omni/diffusion/models/t5_encoder/t5_encoder.py b/vllm_omni/diffusion/models/t5_encoder/t5_encoder.py new file mode 100644 index 000000000..7b0e842d0 --- /dev/null +++ b/vllm_omni/diffusion/models/t5_encoder/t5_encoder.py @@ -0,0 +1,389 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +from __future__ import annotations + +import math +from collections.abc import Iterable + +import torch +import torch.nn as nn +import torch.nn.functional as F +from transformers import T5Config +from vllm.distributed import get_tensor_model_parallel_rank, get_tensor_model_parallel_world_size +from vllm.model_executor.layers.activation import get_act_fn +from vllm.model_executor.layers.layernorm import RMSNorm +from vllm.model_executor.layers.linear import ( + ColumnParallelLinear, + MergedColumnParallelLinear, + QKVParallelLinear, + RowParallelLinear, +) +from vllm.model_executor.layers.vocab_parallel_embedding import VocabParallelEmbedding +from vllm.model_executor.model_loader.weight_utils import default_weight_loader + + +class T5SelfAttention(nn.Module): + def __init__( + self, + config: T5Config, + has_relative_attention_bias: bool = False, + prefix: str = "", + ): + super().__init__() + self.d_model = config.d_model + self.d_kv = config.d_kv + self.n_heads = config.num_heads + self.inner_dim = self.n_heads * self.d_kv + self.has_relative_attention_bias = has_relative_attention_bias + self.relative_attention_num_buckets = config.relative_attention_num_buckets + self.relative_attention_max_distance = config.relative_attention_max_distance + + tp_size = get_tensor_model_parallel_world_size() + assert self.n_heads % tp_size == 0, f"num_heads ({self.n_heads}) must be divisible by tp_size ({tp_size})" + self.n_heads_per_partition = self.n_heads // tp_size + + # Fused Q/K/V projection, sharded across heads. + # HF has separate q, k, v — we fuse into qkv_proj and use + # stacked_params_mapping in load_weights() to load each shard. + self.qkv_proj = QKVParallelLinear( + hidden_size=self.d_model, + head_size=self.d_kv, + total_num_heads=self.n_heads, + total_num_kv_heads=self.n_heads, # T5 uses MHA + bias=False, + ) + + # Output projection: all-reduce back to full d_model + # Named ``o`` to match HF's ``SelfAttention.o`` + self.o = RowParallelLinear( + self.inner_dim, + self.d_model, + bias=False, + input_is_parallel=True, + return_bias=False, + ) + + if has_relative_attention_bias: + # Store full embedding; slice heads per rank in forward + self.relative_attention_bias = nn.Embedding(self.relative_attention_num_buckets, self.n_heads) + + @staticmethod + def _relative_position_bucket( + relative_position: torch.Tensor, + bidirectional: bool = True, + num_buckets: int = 32, + max_distance: int = 128, + ) -> torch.Tensor: + relative_buckets = 0 + if bidirectional: + num_buckets //= 2 + relative_buckets += (relative_position > 0).to(torch.long) * num_buckets + relative_position = torch.abs(relative_position) + else: + relative_position = -torch.min(relative_position, torch.zeros_like(relative_position)) + max_exact = num_buckets // 2 + is_small = relative_position < max_exact + relative_position_if_large = max_exact + ( + torch.log(relative_position.float() / max_exact) + / math.log(max_distance / max_exact) + * (num_buckets - max_exact) + ).to(torch.long) + relative_position_if_large = torch.min( + relative_position_if_large, + torch.full_like(relative_position_if_large, num_buckets - 1), + ) + relative_buckets += torch.where(is_small, relative_position, relative_position_if_large) + return relative_buckets + + def compute_bias(self, query_length: int, key_length: int, device: torch.device) -> torch.Tensor: + """Compute relative position bias, returning only the local head shard.""" + context_position = torch.arange(query_length, dtype=torch.long, device=device)[:, None] + memory_position = torch.arange(key_length, dtype=torch.long, device=device)[None, :] + relative_position = memory_position - context_position + relative_position_bucket = self._relative_position_bucket( + relative_position, + bidirectional=True, + num_buckets=self.relative_attention_num_buckets, + max_distance=self.relative_attention_max_distance, + ) + # values: (query_length, key_length, num_heads) + values = self.relative_attention_bias(relative_position_bucket) + # Slice to local heads for this TP rank + tp_rank = get_tensor_model_parallel_rank() + head_start = tp_rank * self.n_heads_per_partition + head_end = head_start + self.n_heads_per_partition + values = values[:, :, head_start:head_end] + # (1, local_heads, query_length, key_length) + values = values.permute(2, 0, 1).unsqueeze(0) + return values + + def forward( + self, + hidden_states: torch.Tensor, + mask: torch.Tensor | None = None, + position_bias: torch.Tensor | None = None, + ) -> tuple[torch.Tensor, torch.Tensor]: + batch_size, seq_length = hidden_states.shape[:2] + + # Fused QKV projection + qkv, _ = self.qkv_proj(hidden_states) + q_size = self.n_heads_per_partition * self.d_kv + kv_size = self.n_heads_per_partition * self.d_kv + query_states, key_states, value_states = qkv.split([q_size, kv_size, kv_size], dim=-1) + + # Reshape: (batch, seq, local_heads, d_kv) -> (batch, local_heads, seq, d_kv) + query_states = query_states.view(batch_size, seq_length, self.n_heads_per_partition, self.d_kv).transpose(1, 2) + key_states = key_states.view(batch_size, seq_length, self.n_heads_per_partition, self.d_kv).transpose(1, 2) + value_states = value_states.view(batch_size, seq_length, self.n_heads_per_partition, self.d_kv).transpose(1, 2) + + # Attention scores: (batch, local_heads, seq, seq) + scores = torch.matmul(query_states, key_states.transpose(3, 2)) + + if position_bias is None: + if self.has_relative_attention_bias: + position_bias = self.compute_bias(seq_length, seq_length, device=scores.device) + else: + position_bias = torch.zeros( + (1, self.n_heads_per_partition, seq_length, seq_length), + device=scores.device, + dtype=scores.dtype, + ) + if mask is not None: + position_bias = position_bias + mask + + scores += position_bias + + attn_weights = F.softmax(scores.float(), dim=-1).type_as(scores) + attn_output = torch.matmul(attn_weights, value_states) + + # (batch, local_heads, seq, d_kv) -> (batch, seq, local_heads * d_kv) + attn_output = attn_output.transpose(1, 2).contiguous() + attn_output = attn_output.view(batch_size, seq_length, -1) + + attn_output = self.o(attn_output) + + return attn_output, position_bias + + +class T5DenseGatedActDense(nn.Module): + def __init__(self, config: T5Config, prefix: str = ""): + super().__init__() + self.wi = MergedColumnParallelLinear( + config.d_model, + [config.d_ff, config.d_ff], + bias=False, + gather_output=False, + ) + self.wo = RowParallelLinear( + config.d_ff, + config.d_model, + bias=False, + input_is_parallel=True, + return_bias=False, + ) + self.act = get_act_fn(config.dense_act_fn) + + def forward(self, hidden_states: torch.Tensor) -> torch.Tensor: + # MergedColumnParallelLinear outputs concatenated [gate, up] + gate_up, _ = self.wi(hidden_states) + gate, up = gate_up.chunk(2, dim=-1) + hidden_states = self.act(gate) * up + hidden_states = self.wo(hidden_states) + return hidden_states + + +class T5DenseActDense(nn.Module): + def __init__(self, config: T5Config, prefix: str = ""): + super().__init__() + self.wi = ColumnParallelLinear( + config.d_model, + config.d_ff, + bias=False, + gather_output=False, + return_bias=False, + ) + self.wo = RowParallelLinear( + config.d_ff, + config.d_model, + bias=False, + input_is_parallel=True, + return_bias=False, + ) + self.act = get_act_fn(config.dense_act_fn) + + def forward(self, hidden_states: torch.Tensor) -> torch.Tensor: + hidden_states = self.wi(hidden_states) + hidden_states = self.act(hidden_states) + hidden_states = self.wo(hidden_states) + return hidden_states + + +class T5LayerSelfAttention(nn.Module): + def __init__(self, config: T5Config, has_relative_attention_bias: bool = False, prefix: str = ""): + super().__init__() + self.SelfAttention = T5SelfAttention(config, has_relative_attention_bias, prefix=f"{prefix}.SelfAttention") + self.layer_norm = RMSNorm(config.d_model, eps=config.layer_norm_epsilon) + + def forward( + self, + hidden_states: torch.Tensor, + mask: torch.Tensor | None = None, + position_bias: torch.Tensor | None = None, + ) -> tuple[torch.Tensor, torch.Tensor]: + normed = self.layer_norm(hidden_states) + attn_output, position_bias = self.SelfAttention(normed, mask=mask, position_bias=position_bias) + hidden_states = hidden_states + attn_output + + # Clamp for fp16 stability (matching HF T5) + if hidden_states.dtype == torch.float16: + clamp_value = torch.finfo(hidden_states.dtype).max - 1000 + hidden_states = torch.clamp(hidden_states, min=-clamp_value, max=clamp_value) + + return hidden_states, position_bias + + +class T5LayerFF(nn.Module): + def __init__(self, config: T5Config, prefix: str = ""): + super().__init__() + if config.is_gated_act: + self.DenseReluDense = T5DenseGatedActDense(config, prefix=f"{prefix}.DenseReluDense") + else: + self.DenseReluDense = T5DenseActDense(config, prefix=f"{prefix}.DenseReluDense") + self.layer_norm = RMSNorm(config.d_model, eps=config.layer_norm_epsilon) + + def forward(self, hidden_states: torch.Tensor) -> torch.Tensor: + normed = self.layer_norm(hidden_states) + ff_output = self.DenseReluDense(normed) + hidden_states = hidden_states + ff_output + + if hidden_states.dtype == torch.float16: + clamp_value = torch.finfo(hidden_states.dtype).max - 1000 + hidden_states = torch.clamp(hidden_states, min=-clamp_value, max=clamp_value) + + return hidden_states + + +class T5Block(nn.Module): + def __init__(self, config: T5Config, has_relative_attention_bias: bool = False, prefix: str = ""): + super().__init__() + self.layer = nn.ModuleList( + [ + T5LayerSelfAttention(config, has_relative_attention_bias, prefix=f"{prefix}.layer.0"), + T5LayerFF(config, prefix=f"{prefix}.layer.1"), + ] + ) + + def forward( + self, + hidden_states: torch.Tensor, + mask: torch.Tensor | None = None, + position_bias: torch.Tensor | None = None, + ) -> tuple[torch.Tensor, torch.Tensor]: + hidden_states, position_bias = self.layer[0](hidden_states, mask=mask, position_bias=position_bias) + hidden_states = self.layer[1](hidden_states) + return hidden_states, position_bias + + +class T5Stack(nn.Module): + def __init__(self, config: T5Config, shared: nn.Embedding, prefix: str = ""): + super().__init__() + self.embed_tokens = shared + self.block = nn.ModuleList( + [ + T5Block(config, has_relative_attention_bias=(i == 0), prefix=f"{prefix}.block.{i}") + for i in range(config.num_layers) + ] + ) + self.final_layer_norm = RMSNorm(config.d_model, eps=config.layer_norm_epsilon) + + def forward( + self, + input_ids: torch.Tensor, + attention_mask: torch.Tensor | None = None, + ) -> torch.Tensor: + hidden_states = self.embed_tokens(input_ids) + + # Build attention mask: (batch, seq) -> (batch, 1, 1, seq) + if attention_mask is not None: + extended_mask = attention_mask[:, None, None, :].to(dtype=hidden_states.dtype) + extended_mask = (1.0 - extended_mask) * torch.finfo(hidden_states.dtype).min + else: + extended_mask = None + + position_bias = None + for block in self.block: + hidden_states, position_bias = block( + hidden_states, + mask=extended_mask, + position_bias=position_bias, + ) + + hidden_states = self.final_layer_norm(hidden_states) + return hidden_states + + +class T5EncoderModel(nn.Module): + """T5 encoder model applying upstream vLLM layers""" + + def __init__(self, config: T5Config, prefix: str = ""): + super().__init__() + self.config = config + self.shared = VocabParallelEmbedding(config.vocab_size, config.d_model) + self.encoder = T5Stack(config, self.shared, prefix=f"{prefix}.encoder") + + @property + def dtype(self) -> torch.dtype: + return next(self.parameters()).dtype + + @property + def device(self) -> torch.device: + return next(self.parameters()).device + + def embed_input_ids(self, input_ids: torch.Tensor) -> torch.Tensor: + return self.shared(input_ids) + + def forward( + self, + input_ids: torch.Tensor, + attention_mask: torch.Tensor | None = None, + ) -> tuple[torch.Tensor, ...]: + hidden_states = self.encoder(input_ids, attention_mask=attention_mask) + return (hidden_states,) + + def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]: + stacked_params_mapping = [ + ("qkv_proj", "q", "q"), + ("qkv_proj", "k", "k"), + ("qkv_proj", "v", "v"), + ("wi", "wi_0", 0), + ("wi", "wi_1", 1), + ] + + params_dict = dict(self.named_parameters()) + loaded_params: set[str] = set() + + for name, loaded_weight in weights: + original_name = name + lookup_name = name + + for param_name, weight_name, shard_id in stacked_params_mapping: + if f".{weight_name}." not in name: + continue + lookup_name = name.replace(weight_name, param_name) + if lookup_name not in params_dict: + continue + param = params_dict[lookup_name] + weight_loader = param.weight_loader + weight_loader(param, loaded_weight, shard_id) + break + else: + if name not in params_dict: + continue + param = params_dict[name] + weight_loader = getattr(param, "weight_loader", default_weight_loader) + weight_loader(param, loaded_weight) + + loaded_params.add(original_name) + loaded_params.add(lookup_name) + + return loaded_params