| 1 | from enum import Enum |
| 2 | |
| 3 | import torch |
| 4 | |
| 5 | from ltx_core.guidance.perturbations import BatchedPerturbationConfig |
| 6 | from ltx_core.model.transformer.adaln import AdaLayerNormSingle, adaln_embedding_coefficient |
| 7 | from ltx_core.model.transformer.attention import AttentionCallable, AttentionFunction |
| 8 | from ltx_core.model.transformer.modality import Modality |
| 9 | from ltx_core.model.transformer.rope import LTXRopeType |
| 10 | from ltx_core.model.transformer.transformer import BasicAVTransformerBlock, TransformerConfig |
| 11 | from ltx_core.model.transformer.transformer_args import ( |
| 12 | MultiModalTransformerArgsPreprocessor, |
| 13 | TransformerArgs, |
| 14 | TransformerArgsPreprocessor, |
| 15 | ) |
| 16 | from ltx_core.utils import to_denoised |
| 17 | |
| 18 | |
| 19 | class LTXModelType(Enum): |
| 20 | AudioVideo = "ltx av model" |
| 21 | VideoOnly = "ltx video only model" |
| 22 | AudioOnly = "ltx audio only model" |
| 23 | |
| 24 | def is_video_enabled(self) -> bool: |
| 25 | return self in (LTXModelType.AudioVideo, LTXModelType.VideoOnly) |
| 26 | |
| 27 | def is_audio_enabled(self) -> bool: |
| 28 | return self in (LTXModelType.AudioVideo, LTXModelType.AudioOnly) |
| 29 | |
| 30 | |
| 31 | class LTXModel(torch.nn.Module): |
| 32 | """ |
| 33 | LTX model transformer implementation. |
| 34 | This class implements the transformer blocks for the LTX model. |
| 35 | """ |
| 36 | |
| 37 | def __init__( # noqa: PLR0913 |
| 38 | self, |
| 39 | *, |
| 40 | model_type: LTXModelType = LTXModelType.AudioVideo, |
| 41 | num_attention_heads: int = 32, |
| 42 | attention_head_dim: int = 128, |
| 43 | in_channels: int = 128, |
| 44 | out_channels: int = 128, |
| 45 | num_layers: int = 48, |
| 46 | cross_attention_dim: int = 4096, |
| 47 | norm_eps: float = 1e-06, |
| 48 | attention_type: AttentionFunction | AttentionCallable = AttentionFunction.DEFAULT, |
| 49 | positional_embedding_theta: float = 10000.0, |
| 50 | positional_embedding_max_pos: list[int] | None = None, |
| 51 | timestep_scale_multiplier: int = 1000, |
| 52 | use_middle_indices_grid: bool = True, |
| 53 | audio_num_attention_heads: int = 32, |
| 54 | audio_attention_head_dim: int = 64, |
| 55 | audio_in_channels: int = 128, |
| 56 | audio_out_channels: int = 128, |
| 57 | audio_cross_attention_dim: int = 2048, |
| 58 | audio_positional_embedding_max_pos: list[int] | None = None, |
| 59 | av_ca_timestep_scale_multiplier: int = 1, |
| 60 | rope_type: LTXRopeType = LTXRopeType.INTERLEAVED, |
| 61 | double_precision_rope: bool = False, |
| 62 | apply_gated_attention: bool = False, |
| 63 | caption_projection: torch.nn.Module | None = None, |
| 64 | audio_caption_projection: torch.nn.Module | None = None, |
| 65 | cross_attention_adaln: bool = False, |
| 66 | ): |
| 67 | super().__init__() |
| 68 | self._enable_gradient_checkpointing = False |
| 69 | self.cross_attention_adaln = cross_attention_adaln |
| 70 | self.use_middle_indices_grid = use_middle_indices_grid |
| 71 | self.rope_type = rope_type |
| 72 | self.double_precision_rope = double_precision_rope |
| 73 | self.timestep_scale_multiplier = timestep_scale_multiplier |
| 74 | self.positional_embedding_theta = positional_embedding_theta |
| 75 | self.model_type = model_type |
| 76 | cross_pe_max_pos = None |
| 77 | if model_type.is_video_enabled(): |
| 78 | if positional_embedding_max_pos is None: |
| 79 | positional_embedding_max_pos = [20, 2048, 2048] |
| 80 | self.positional_embedding_max_pos = positional_embedding_max_pos |
| 81 | self.num_attention_heads = num_attention_heads |
| 82 | self.inner_dim = num_attention_heads * attention_head_dim |
| 83 | self._init_video( |
| 84 | in_channels=in_channels, |
| 85 | out_channels=out_channels, |
| 86 | norm_eps=norm_eps, |
| 87 | caption_projection=caption_projection, |
| 88 | ) |
| 89 | |
| 90 | if model_type.is_audio_enabled(): |
| 91 | if audio_positional_embedding_max_pos is None: |
| 92 | audio_positional_embedding_max_pos = [20] |
| 93 | self.audio_positional_embedding_max_pos = audio_positional_embedding_max_pos |
| 94 | self.audio_num_attention_heads = audio_num_attention_heads |
| 95 | self.audio_inner_dim = self.audio_num_attention_heads * audio_attention_head_dim |
| 96 | self._init_audio( |
| 97 | in_channels=audio_in_channels, |
| 98 | out_channels=audio_out_channels, |
| 99 | norm_eps=norm_eps, |
| 100 | caption_projection=audio_caption_projection, |
| 101 | ) |
| 102 | |
| 103 | if model_type.is_video_enabled() and model_type.is_audio_enabled(): |
| 104 | cross_pe_max_pos = max(self.positional_embedding_max_pos[0], self.audio_positional_embedding_max_pos[0]) |
| 105 | self.av_ca_timestep_scale_multiplier = av_ca_timestep_scale_multiplier |
| 106 | self.audio_cross_attention_dim = audio_cross_attention_dim |
| 107 | self._init_audio_video(num_scale_shift_values=4) |
| 108 | |
| 109 | self._init_preprocessors(cross_pe_max_pos) |
| 110 | # Initialize transformer blocks |
| 111 | self._init_transformer_blocks( |
| 112 | num_layers=num_layers, |
| 113 | attention_head_dim=attention_head_dim if model_type.is_video_enabled() else 0, |
| 114 | cross_attention_dim=cross_attention_dim, |
| 115 | audio_attention_head_dim=audio_attention_head_dim if model_type.is_audio_enabled() else 0, |
| 116 | audio_cross_attention_dim=audio_cross_attention_dim, |
| 117 | norm_eps=norm_eps, |
| 118 | attention_type=attention_type, |
| 119 | apply_gated_attention=apply_gated_attention, |
| 120 | ) |
| 121 | |
| 122 | @property |
| 123 | def _adaln_embedding_coefficient(self) -> int: |
| 124 | return adaln_embedding_coefficient(self.cross_attention_adaln) |
| 125 | |
| 126 | def _init_video( |
| 127 | self, |
| 128 | in_channels: int, |
| 129 | out_channels: int, |
| 130 | norm_eps: float, |
| 131 | caption_projection: torch.nn.Module | None = None, |
| 132 | ) -> None: |
| 133 | """Initialize video-specific components.""" |
| 134 | # Video input components |
| 135 | self.patchify_proj = torch.nn.Linear(in_channels, self.inner_dim, bias=True) |
| 136 | if caption_projection is not None: |
| 137 | self.caption_projection = caption_projection |
| 138 | |
| 139 | self.adaln_single = AdaLayerNormSingle(self.inner_dim, embedding_coefficient=self._adaln_embedding_coefficient) |
| 140 | |
| 141 | self.prompt_adaln_single = ( |
| 142 | AdaLayerNormSingle(self.inner_dim, embedding_coefficient=2) if self.cross_attention_adaln else None |
| 143 | ) |
| 144 | |
| 145 | # Video output components |
| 146 | self.scale_shift_table = torch.nn.Parameter(torch.empty(2, self.inner_dim)) |
| 147 | self.norm_out = torch.nn.LayerNorm(self.inner_dim, elementwise_affine=False, eps=norm_eps) |
| 148 | self.proj_out = torch.nn.Linear(self.inner_dim, out_channels) |
| 149 | |
| 150 | def _init_audio( |
| 151 | self, |
| 152 | in_channels: int, |
| 153 | out_channels: int, |
| 154 | norm_eps: float, |
| 155 | caption_projection: torch.nn.Module | None = None, |
| 156 | ) -> None: |
| 157 | """Initialize audio-specific components.""" |
| 158 | |
| 159 | # Audio input components |
| 160 | self.audio_patchify_proj = torch.nn.Linear(in_channels, self.audio_inner_dim, bias=True) |
| 161 | if caption_projection is not None: |
| 162 | self.audio_caption_projection = caption_projection |
| 163 | |
| 164 | self.audio_adaln_single = AdaLayerNormSingle( |
| 165 | self.audio_inner_dim, |
| 166 | embedding_coefficient=self._adaln_embedding_coefficient, |
| 167 | ) |
| 168 | |
| 169 | self.audio_prompt_adaln_single = ( |
| 170 | AdaLayerNormSingle(self.audio_inner_dim, embedding_coefficient=2) if self.cross_attention_adaln else None |
| 171 | ) |
| 172 | |
| 173 | # Audio output components |
| 174 | self.audio_scale_shift_table = torch.nn.Parameter(torch.empty(2, self.audio_inner_dim)) |
| 175 | self.audio_norm_out = torch.nn.LayerNorm(self.audio_inner_dim, elementwise_affine=False, eps=norm_eps) |
| 176 | self.audio_proj_out = torch.nn.Linear(self.audio_inner_dim, out_channels) |
| 177 | |
| 178 | def _init_audio_video( |
| 179 | self, |
| 180 | num_scale_shift_values: int, |
| 181 | ) -> None: |
| 182 | """Initialize audio-video cross-attention components.""" |
| 183 | self.av_ca_video_scale_shift_adaln_single = AdaLayerNormSingle( |
| 184 | self.inner_dim, |
| 185 | embedding_coefficient=num_scale_shift_values, |
| 186 | ) |
| 187 | |
| 188 | self.av_ca_audio_scale_shift_adaln_single = AdaLayerNormSingle( |
| 189 | self.audio_inner_dim, |
| 190 | embedding_coefficient=num_scale_shift_values, |
| 191 | ) |
| 192 | |
| 193 | self.av_ca_a2v_gate_adaln_single = AdaLayerNormSingle( |
| 194 | self.inner_dim, |
| 195 | embedding_coefficient=1, |
| 196 | ) |
| 197 | |
| 198 | self.av_ca_v2a_gate_adaln_single = AdaLayerNormSingle( |
| 199 | self.audio_inner_dim, |
| 200 | embedding_coefficient=1, |
| 201 | ) |
| 202 | |
| 203 | def _init_preprocessors( |
| 204 | self, |
| 205 | cross_pe_max_pos: int | None = None, |
| 206 | ) -> None: |
| 207 | """Initialize preprocessors for LTX.""" |
| 208 | |
| 209 | if self.model_type.is_video_enabled() and self.model_type.is_audio_enabled(): |
| 210 | self.video_args_preprocessor = MultiModalTransformerArgsPreprocessor( |
| 211 | patchify_proj=self.patchify_proj, |
| 212 | adaln=self.adaln_single, |
| 213 | cross_scale_shift_adaln=self.av_ca_video_scale_shift_adaln_single, |
| 214 | cross_gate_adaln=self.av_ca_a2v_gate_adaln_single, |
| 215 | inner_dim=self.inner_dim, |
| 216 | max_pos=self.positional_embedding_max_pos, |
| 217 | num_attention_heads=self.num_attention_heads, |
| 218 | cross_pe_max_pos=cross_pe_max_pos, |
| 219 | use_middle_indices_grid=self.use_middle_indices_grid, |
| 220 | audio_cross_attention_dim=self.audio_cross_attention_dim, |
| 221 | timestep_scale_multiplier=self.timestep_scale_multiplier, |
| 222 | double_precision_rope=self.double_precision_rope, |
| 223 | positional_embedding_theta=self.positional_embedding_theta, |
| 224 | rope_type=self.rope_type, |
| 225 | av_ca_timestep_scale_multiplier=self.av_ca_timestep_scale_multiplier, |
| 226 | caption_projection=getattr(self, "caption_projection", None), |
| 227 | prompt_adaln=getattr(self, "prompt_adaln_single", None), |
| 228 | ) |
| 229 | self.audio_args_preprocessor = MultiModalTransformerArgsPreprocessor( |
| 230 | patchify_proj=self.audio_patchify_proj, |
| 231 | adaln=self.audio_adaln_single, |
| 232 | cross_scale_shift_adaln=self.av_ca_audio_scale_shift_adaln_single, |
| 233 | cross_gate_adaln=self.av_ca_v2a_gate_adaln_single, |
| 234 | inner_dim=self.audio_inner_dim, |
| 235 | max_pos=self.audio_positional_embedding_max_pos, |
| 236 | num_attention_heads=self.audio_num_attention_heads, |
| 237 | cross_pe_max_pos=cross_pe_max_pos, |
| 238 | use_middle_indices_grid=self.use_middle_indices_grid, |
| 239 | audio_cross_attention_dim=self.audio_cross_attention_dim, |
| 240 | timestep_scale_multiplier=self.timestep_scale_multiplier, |
| 241 | double_precision_rope=self.double_precision_rope, |
| 242 | positional_embedding_theta=self.positional_embedding_theta, |
| 243 | rope_type=self.rope_type, |
| 244 | av_ca_timestep_scale_multiplier=self.av_ca_timestep_scale_multiplier, |
| 245 | caption_projection=getattr(self, "audio_caption_projection", None), |
| 246 | prompt_adaln=getattr(self, "audio_prompt_adaln_single", None), |
| 247 | ) |
| 248 | elif self.model_type.is_video_enabled(): |
| 249 | self.video_args_preprocessor = TransformerArgsPreprocessor( |
| 250 | patchify_proj=self.patchify_proj, |
| 251 | adaln=self.adaln_single, |
| 252 | inner_dim=self.inner_dim, |
| 253 | max_pos=self.positional_embedding_max_pos, |
| 254 | num_attention_heads=self.num_attention_heads, |
| 255 | use_middle_indices_grid=self.use_middle_indices_grid, |
| 256 | timestep_scale_multiplier=self.timestep_scale_multiplier, |
| 257 | double_precision_rope=self.double_precision_rope, |
| 258 | positional_embedding_theta=self.positional_embedding_theta, |
| 259 | rope_type=self.rope_type, |
| 260 | caption_projection=getattr(self, "caption_projection", None), |
| 261 | prompt_adaln=getattr(self, "prompt_adaln_single", None), |
| 262 | ) |
| 263 | elif self.model_type.is_audio_enabled(): |
| 264 | self.audio_args_preprocessor = TransformerArgsPreprocessor( |
| 265 | patchify_proj=self.audio_patchify_proj, |
| 266 | adaln=self.audio_adaln_single, |
| 267 | inner_dim=self.audio_inner_dim, |
| 268 | max_pos=self.audio_positional_embedding_max_pos, |
| 269 | num_attention_heads=self.audio_num_attention_heads, |
| 270 | use_middle_indices_grid=self.use_middle_indices_grid, |
| 271 | timestep_scale_multiplier=self.timestep_scale_multiplier, |
| 272 | double_precision_rope=self.double_precision_rope, |
| 273 | positional_embedding_theta=self.positional_embedding_theta, |
| 274 | rope_type=self.rope_type, |
| 275 | caption_projection=getattr(self, "audio_caption_projection", None), |
| 276 | prompt_adaln=getattr(self, "audio_prompt_adaln_single", None), |
| 277 | ) |
| 278 | |
| 279 | def _init_transformer_blocks( |
| 280 | self, |
| 281 | num_layers: int, |
| 282 | attention_head_dim: int, |
| 283 | cross_attention_dim: int, |
| 284 | audio_attention_head_dim: int, |
| 285 | audio_cross_attention_dim: int, |
| 286 | norm_eps: float, |
| 287 | attention_type: AttentionFunction | AttentionCallable, |
| 288 | apply_gated_attention: bool, |
| 289 | ) -> None: |
| 290 | """Initialize transformer blocks for LTX.""" |
| 291 | video_config = ( |
| 292 | TransformerConfig( |
| 293 | dim=self.inner_dim, |
| 294 | heads=self.num_attention_heads, |
| 295 | d_head=attention_head_dim, |
| 296 | context_dim=cross_attention_dim, |
| 297 | apply_gated_attention=apply_gated_attention, |
| 298 | cross_attention_adaln=self.cross_attention_adaln, |
| 299 | ) |
| 300 | if self.model_type.is_video_enabled() |
| 301 | else None |
| 302 | ) |
| 303 | audio_config = ( |
| 304 | TransformerConfig( |
| 305 | dim=self.audio_inner_dim, |
| 306 | heads=self.audio_num_attention_heads, |
| 307 | d_head=audio_attention_head_dim, |
| 308 | context_dim=audio_cross_attention_dim, |
| 309 | apply_gated_attention=apply_gated_attention, |
| 310 | cross_attention_adaln=self.cross_attention_adaln, |
| 311 | ) |
| 312 | if self.model_type.is_audio_enabled() |
| 313 | else None |
| 314 | ) |
| 315 | self.transformer_blocks = torch.nn.ModuleList( |
| 316 | [ |
| 317 | BasicAVTransformerBlock( |
| 318 | idx=idx, |
| 319 | num_layers=num_layers, |
| 320 | video=video_config, |
| 321 | audio=audio_config, |
| 322 | rope_type=self.rope_type, |
| 323 | norm_eps=norm_eps, |
| 324 | attention_function=attention_type, |
| 325 | ) |
| 326 | for idx in range(num_layers) |
| 327 | ] |
| 328 | ) |
| 329 | |
| 330 | def set_gradient_checkpointing(self, enable: bool) -> None: |
| 331 | """Enable or disable gradient checkpointing for transformer blocks. |
| 332 | Gradient checkpointing trades compute for memory by recomputing activations |
| 333 | during the backward pass instead of storing them. This can significantly |
| 334 | reduce memory usage at the cost of ~20-30% slower training. |
| 335 | Args: |
| 336 | enable: Whether to enable gradient checkpointing |
| 337 | """ |
| 338 | self._enable_gradient_checkpointing = enable |
| 339 | |
| 340 | def _process_transformer_blocks( |
| 341 | self, |
| 342 | video: TransformerArgs | None, |
| 343 | audio: TransformerArgs | None, |
| 344 | perturbations: BatchedPerturbationConfig, |
| 345 | ) -> tuple[TransformerArgs, TransformerArgs]: |
| 346 | """Process transformer blocks for LTXAV.""" |
| 347 | |
| 348 | # Process transformer blocks |
| 349 | for block in self.transformer_blocks: |
| 350 | if self._enable_gradient_checkpointing and self.training: |
| 351 | # Use gradient checkpointing to save memory during training. |
| 352 | # With use_reentrant=False, we can pass dataclasses directly - |
| 353 | # PyTorch will track all tensor leaves in the computation graph. |
| 354 | video, audio = torch.utils.checkpoint.checkpoint( |
| 355 | block, |
| 356 | video, |
| 357 | audio, |
| 358 | perturbations, |
| 359 | use_reentrant=False, |
| 360 | ) |
| 361 | else: |
| 362 | video, audio = block( |
| 363 | video=video, |
| 364 | audio=audio, |
| 365 | perturbations=perturbations, |
| 366 | ) |
| 367 | |
| 368 | return video, audio |
| 369 | |
| 370 | def _process_output( |
| 371 | self, |
| 372 | scale_shift_table: torch.Tensor, |
| 373 | norm_out: torch.nn.LayerNorm, |
| 374 | proj_out: torch.nn.Linear, |
| 375 | x: torch.Tensor, |
| 376 | embedded_timestep: torch.Tensor, |
| 377 | ) -> torch.Tensor: |
| 378 | """Process output for LTXV.""" |
| 379 | # Apply scale-shift modulation |
| 380 | scale_shift_values = ( |
| 381 | scale_shift_table[None, None].to(device=x.device, dtype=x.dtype) + embedded_timestep[:, :, None] |
| 382 | ) |
| 383 | shift, scale = scale_shift_values[:, :, 0], scale_shift_values[:, :, 1] |
| 384 | |
| 385 | x = norm_out(x) |
| 386 | x = x * (1 + scale) + shift |
| 387 | x = proj_out(x) |
| 388 | return x |
| 389 | |
| 390 | def forward( |
| 391 | self, video: Modality | None, audio: Modality | None, perturbations: BatchedPerturbationConfig |
| 392 | ) -> tuple[torch.Tensor, torch.Tensor]: |
| 393 | """ |
| 394 | Forward pass for LTX models. |
| 395 | Returns: |
| 396 | Processed output tensors |
| 397 | """ |
| 398 | if not self.model_type.is_video_enabled() and video is not None: |
| 399 | raise ValueError("Video is not enabled for this model") |
| 400 | if not self.model_type.is_audio_enabled() and audio is not None: |
| 401 | raise ValueError("Audio is not enabled for this model") |
| 402 | |
| 403 | video_args = self.video_args_preprocessor.prepare(video, audio) if video is not None else None |
| 404 | audio_args = self.audio_args_preprocessor.prepare(audio, video) if audio is not None else None |
| 405 | # Process transformer blocks |
| 406 | video_out, audio_out = self._process_transformer_blocks( |
| 407 | video=video_args, |
| 408 | audio=audio_args, |
| 409 | perturbations=perturbations, |
| 410 | ) |
| 411 | |
| 412 | # Process output |
| 413 | vx = ( |
| 414 | self._process_output( |
| 415 | self.scale_shift_table, self.norm_out, self.proj_out, video_out.x, video_out.embedded_timestep |
| 416 | ) |
| 417 | if video_out is not None |
| 418 | else None |
| 419 | ) |
| 420 | ax = ( |
| 421 | self._process_output( |
| 422 | self.audio_scale_shift_table, |
| 423 | self.audio_norm_out, |
| 424 | self.audio_proj_out, |
| 425 | audio_out.x, |
| 426 | audio_out.embedded_timestep, |
| 427 | ) |
| 428 | if audio_out is not None |
| 429 | else None |
| 430 | ) |
| 431 | return vx, ax |
| 432 | |
| 433 | |
| 434 | class LegacyX0Model(torch.nn.Module): |
| 435 | """ |
| 436 | Legacy X0 model implementation. |
| 437 | Returns fully denoised output based on the velocities produced by the base model. |
| 438 | """ |
| 439 | |
| 440 | def __init__(self, velocity_model: LTXModel): |
| 441 | super().__init__() |
| 442 | self.velocity_model = velocity_model |
| 443 | |
| 444 | def forward( |
| 445 | self, |
| 446 | video: Modality | None, |
| 447 | audio: Modality | None, |
| 448 | perturbations: BatchedPerturbationConfig, |
| 449 | sigma: float, |
| 450 | ) -> tuple[torch.Tensor | None, torch.Tensor | None]: |
| 451 | """ |
| 452 | Denoise the video and audio according to the sigma. |
| 453 | Returns: |
| 454 | Denoised video and audio |
| 455 | """ |
| 456 | vx, ax = self.velocity_model(video, audio, perturbations) |
| 457 | denoised_video = to_denoised(video.latent, vx, sigma) if vx is not None else None |
| 458 | denoised_audio = to_denoised(audio.latent, ax, sigma) if ax is not None else None |
| 459 | return denoised_video, denoised_audio |
| 460 | |
| 461 | |
| 462 | class X0Model(torch.nn.Module): |
| 463 | """ |
| 464 | X0 model implementation. |
| 465 | Returns fully denoised outputs based on the velocities produced by the base model. |
| 466 | Applies scaled denoising to the video and audio according to the timesteps = sigma * denoising_mask. |
| 467 | """ |
| 468 | |
| 469 | def __init__(self, velocity_model: LTXModel): |
| 470 | super().__init__() |
| 471 | self.velocity_model = velocity_model |
| 472 | |
| 473 | def forward( |
| 474 | self, |
| 475 | video: Modality | None, |
| 476 | audio: Modality | None, |
| 477 | perturbations: BatchedPerturbationConfig, |
| 478 | ) -> tuple[torch.Tensor | None, torch.Tensor | None]: |
| 479 | """ |
| 480 | Denoise the video and audio according to the sigma. |
| 481 | Returns: |
| 482 | Denoised video and audio |
| 483 | """ |
| 484 | vx, ax = self.velocity_model(video, audio, perturbations) |
| 485 | denoised_video = to_denoised(video.latent, vx, video.timesteps) if vx is not None else None |
| 486 | denoised_audio = to_denoised(audio.latent, ax, audio.timesteps) if ax is not None else None |
| 487 | return denoised_video, denoised_audio |
| 488 |