AI/diffusers
1
0
Fork 0
mirror of https://github.com/huggingface/diffusers.git synced 2026-10-23 09:09:01 +03:00
Code Activity

Move LTX25AutoBlocks to new file with new model name (#14453)

Browse Source 
Refactor LTX25AutoBlocks into its own file, give it its own model name, and add tests for it
This commit is contained in:
dg845
2026-08-12 17:02:43 -07:00
committed by GitHub
parent 3a2f35d4ef
commit 614ae4bb9d
4 changed files with 597 additions and 230 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/diffusers/modular_pipelines/ltx2/__init__.py
View File

@@ -27,8 +27,8 @@ else:
"LTX2ConditionBlocks",
"LTX2ImageToVideoBlocks",
"LTX2InContextBlocks",
"LTX25AutoBlocks",
]
_import_structure["modular_blocks_ltx25"] = ["LTX25AutoBlocks"]
_import_structure["modular_pipeline"] = ["LTX2ModularPipeline", "LTX25ModularPipeline"]
if TYPE_CHECKING or DIFFUSERS_SLOW_IMPORT:
@@ -44,8 +44,8 @@ if TYPE_CHECKING or DIFFUSERS_SLOW_IMPORT:
LTX2ConditionBlocks,
LTX2ImageToVideoBlocks,
LTX2InContextBlocks,
LTX25AutoBlocks,
)
from .modular_blocks_ltx25 import LTX25AutoBlocks
from .modular_pipeline import LTX2ModularPipeline, LTX25ModularPipeline
else:
import sys

228
src/diffusers/modular_pipelines/ltx2/modular_blocks_ltx2.py
View File

@@ -33,7 +33,6 @@ from .before_denoise import (
)
from .decoders import (
LTX2AudioDecoderStep,
LTX2DiffusionVaeDecoderStep,
LTX2TrimConditionTokensStep,
LTX2VaeDecoderStep,
)
@@ -1893,230 +1892,3 @@ class LTX2AutoBlocks(SequentialPipelineBlocks):
OutputParam.template("videos"),
OutputParam("audio", type_hint=torch.Tensor, description="The generated audio waveform."),
]
# auto_docstring
class LTX25DecoderStep(SequentialPipelineBlocks):
"""
Decode stage for LTX-2.5: denoises the video latents with the diffusion decoder and vocodes the audio latents (or
returns latents).
Components:
diffusion_decoder (`LTX2VideoDiffusionDecoderModel`) video_processor (`VideoProcessor`) audio_vae
(`AutoencoderKLLTX2Audio`) vocoder (`LTX2Vocoder`)
Inputs:
latents (`Tensor`):
Pre-generated noisy latents for image generation.
output_type (`str`, *optional*, defaults to pil):
Output format: 'pil', 'np', 'pt'.
height (`int`, *optional*, defaults to 512):
The height in pixels of the generated image.
width (`int`, *optional*, defaults to 704):
The width in pixels of the generated image.
num_frames (`int`, *optional*, defaults to 121):
The number of frames in the generated video.
generator (`Generator`, *optional*):
Torch generator for deterministic generation.
dtype (`dtype`):
The dtype of the model inputs, can be generated in input step.
audio_latents (`Tensor`):
Denoised audio latents.
audio_num_frames (`int`):
Number of audio latent frames, used to unpack the audio latent sequence.
Outputs:
videos (`list`):
The generated videos.
audio (`Tensor`):
The generated audio waveform.
"""
model_name = "ltx2"
block_classes = [LTX2DiffusionVaeDecoderStep, LTX2AudioDecoderStep]
block_names = ["video_decode", "audio_decode"]
@property
def description(self):
return (
"Decode stage for LTX-2.5: denoises the video latents with the diffusion decoder and vocodes the audio "
"latents (or returns latents)."
)
@property
def outputs(self):
return [
OutputParam.template("videos"),
OutputParam("audio", type_hint=torch.Tensor, description="The generated audio waveform."),
]
# auto_docstring
class LTX25ConditionDecoderStep(SequentialPipelineBlocks):
"""
Decode stage for LTX-2.5 condition workflows: drops the appended keyframe-condition tokens, then denoises the video
latents with the diffusion decoder and vocodes the audio latents (or returns latents).
Components:
diffusion_decoder (`LTX2VideoDiffusionDecoderModel`) video_processor (`VideoProcessor`) audio_vae
(`AutoencoderKLLTX2Audio`) vocoder (`LTX2Vocoder`)
Inputs:
latents (`Tensor`):
Pre-generated noisy latents for image generation.
base_token_count (`int`):
Number of generated-video tokens, i.e. the sequence length before appended tokens.
output_type (`str`, *optional*, defaults to pil):
Output format: 'pil', 'np', 'pt'.
height (`int`, *optional*, defaults to 512):
The height in pixels of the generated image.
width (`int`, *optional*, defaults to 704):
The width in pixels of the generated image.
num_frames (`int`, *optional*, defaults to 121):
The number of frames in the generated video.
generator (`Generator`, *optional*):
Torch generator for deterministic generation.
dtype (`dtype`):
The dtype of the model inputs, can be generated in input step.
audio_latents (`Tensor`):
Denoised audio latents.
audio_num_frames (`int`):
Number of audio latent frames, used to unpack the audio latent sequence.
Outputs:
videos (`list`):
The generated videos.
audio (`Tensor`):
The generated audio waveform.
"""
model_name = "ltx2"
block_classes = [LTX2TrimConditionTokensStep, LTX2DiffusionVaeDecoderStep, LTX2AudioDecoderStep]
block_names = ["trim_condition_tokens", "video_decode", "audio_decode"]
@property
def description(self):
return (
"Decode stage for LTX-2.5 condition workflows: drops the appended keyframe-condition tokens, then "
"denoises the video latents with the diffusion decoder and vocodes the audio latents (or returns "
"latents)."
)
@property
def outputs(self):
return [
OutputParam.template("videos"),
OutputParam("audio", type_hint=torch.Tensor, description="The generated audio waveform."),
]
# auto_docstring
class LTX25AutoDecoderStep(AutoPipelineBlocks):
"""
Auto decode block for LTX-2.5 that selects the decoder based on inputs.
- `LTX25ConditionDecoderStep` when `base_token_count` is present, i.e. the denoised sequence carries appended
keyframe / reference tokens (condition, in-context).
- `LTX25DecoderStep` otherwise (text-to-video, image-to-video).
Components:
diffusion_decoder (`LTX2VideoDiffusionDecoderModel`) video_processor (`VideoProcessor`) audio_vae
(`AutoencoderKLLTX2Audio`) vocoder (`LTX2Vocoder`)
Inputs:
latents (`Tensor`):
Pre-generated noisy latents for image generation.
base_token_count (`int`, *optional*):
Number of generated-video tokens, i.e. the sequence length before appended tokens.
output_type (`str`, *optional*, defaults to pil):
Output format: 'pil', 'np', 'pt'.
height (`int`, *optional*, defaults to 512):
The height in pixels of the generated image.
width (`int`, *optional*, defaults to 704):
The width in pixels of the generated image.
num_frames (`int`, *optional*, defaults to 121):
The number of frames in the generated video.
generator (`Generator`, *optional*):
Torch generator for deterministic generation.
dtype (`dtype`):
The dtype of the model inputs, can be generated in input step.
audio_latents (`Tensor`):
Denoised audio latents.
audio_num_frames (`int`):
Number of audio latent frames, used to unpack the audio latent sequence.
Outputs:
videos (`list`):
The generated videos.
audio (`Tensor`):
The generated audio waveform.
"""
model_name = "ltx2"
# Mirrors `LTX2AutoDecoderStep` on the same `base_token_count` trigger; only the video decoder differs.
block_classes = [LTX25ConditionDecoderStep, LTX25DecoderStep]
block_names = ["condition", "default"]
block_trigger_inputs = ["base_token_count", None]
@property
def description(self):
return (
"Auto decode block for LTX-2.5 that selects the decoder based on inputs.\n"
" - `LTX25ConditionDecoderStep` when `base_token_count` is present, i.e. the denoised sequence carries "
"appended keyframe / reference tokens (condition, in-context).\n"
" - `LTX25DecoderStep` otherwise (text-to-video, image-to-video)."
)
@property
def outputs(self):
return [
OutputParam.template("videos"),
OutputParam("audio", type_hint=torch.Tensor, description="The generated audio waveform."),
]
class LTX25AutoBlocks(LTX2AutoBlocks):
"""
Auto blocks for LTX-2.5, supporting text-to-video, image-to-video, condition-to-video and in-context (IC-LoRA)
generation (joint video + audio).
Supported workflows:
- `text2video`: requires `prompt`
- `image2video`: requires `image`, `prompt`
- `condition`: requires `conditions`, `prompt`
- `in_context`: requires `reference_conditions`, `num_frames`, `prompt`
"""
model_name = "ltx2"
# Both lists are restated rather than inherited: `block_classes` and `block_names` are positional, so a subclass
# that overrides one and inherits the other silently desynchronizes when the parent's graph changes. Keep these in
# step with `LTX2AutoBlocks`; only the `decode` entry differs.
block_classes = [
LTX2AutoPromptEnhancerStep,
LTX2TextConditioningStep,
LTX2AutoDurationStep,
LTX2AutoVaeEncoderStep,
LTX2AutoConditionEncoderStep,
LTX2AutoReferenceEncoderStep,
LTX2AutoCoreDenoiseStep,
LTX25AutoDecoderStep,
]
block_names = [
"prompt_enhancer",
"text_encoder",
"duration",
"vae_encoder",
"condition_encoder",
"reference_encoder",
"denoise",
"decode",
]
@property
def description(self):
return (
"Auto blocks for LTX-2.5 supporting text-to-video, image-to-video, condition-to-video and in-context "
"(IC-LoRA) generation (joint video + audio). Identical to `LTX2AutoBlocks` except that the video decoder "
"is `LTX2DiffusionVaeDecoderStep`, since the diffusion decoder is the native default from LTX-2.5 on. To "
'decode with the convolutional VAE instead, swap the decode block: `blocks.sub_blocks["decode"] = '
"LTX2AutoDecoderStep()`."
)

389
src/diffusers/modular_pipelines/ltx2/modular_blocks_ltx25.py Normal file
View File

@@ -0,0 +1,389 @@
# Copyright 2026 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import torch
from ..modular_pipeline import AutoPipelineBlocks, SequentialPipelineBlocks
from ..modular_pipeline_utils import OutputParam
from .decoders import (
LTX2AudioDecoderStep,
LTX2DiffusionVaeDecoderStep,
LTX2TrimConditionTokensStep,
)
from .modular_blocks_ltx2 import (
LTX2AutoConditionEncoderStep,
LTX2AutoCoreDenoiseStep,
LTX2AutoDurationStep,
LTX2AutoPromptEnhancerStep,
LTX2AutoReferenceEncoderStep,
LTX2AutoVaeEncoderStep,
LTX2TextConditioningStep,
)
# auto_docstring
class LTX25DecoderStep(SequentialPipelineBlocks):
"""
Decode stage for LTX-2.5: denoises the video latents with the diffusion decoder and vocodes the audio latents (or
returns latents).
Components:
diffusion_decoder (`LTX2VideoDiffusionDecoderModel`) video_processor (`VideoProcessor`) audio_vae
(`AutoencoderKLLTX2Audio`) vocoder (`LTX2Vocoder`)
Inputs:
latents (`Tensor`):
Pre-generated noisy latents for image generation.
output_type (`str`, *optional*, defaults to pil):
Output format: 'pil', 'np', 'pt'.
height (`int`, *optional*, defaults to 512):
The height in pixels of the generated image.
width (`int`, *optional*, defaults to 704):
The width in pixels of the generated image.
num_frames (`int`, *optional*, defaults to 121):
The number of frames in the generated video.
generator (`Generator`, *optional*):
Torch generator for deterministic generation.
dtype (`dtype`):
The dtype of the model inputs, can be generated in input step.
audio_latents (`Tensor`):
Denoised audio latents.
audio_num_frames (`int`):
Number of audio latent frames, used to unpack the audio latent sequence.
Outputs:
videos (`list`):
The generated videos.
audio (`Tensor`):
The generated audio waveform.
"""
model_name = "ltx2.5"
block_classes = [LTX2DiffusionVaeDecoderStep, LTX2AudioDecoderStep]
block_names = ["video_decode", "audio_decode"]
@property
def description(self):
return (
"Decode stage for LTX-2.5: denoises the video latents with the diffusion decoder and vocodes the audio "
"latents (or returns latents)."
)
@property
def outputs(self):
return [
OutputParam.template("videos"),
OutputParam("audio", type_hint=torch.Tensor, description="The generated audio waveform."),
]
# auto_docstring
class LTX25ConditionDecoderStep(SequentialPipelineBlocks):
"""
Decode stage for LTX-2.5 condition workflows: drops the appended keyframe-condition tokens, then denoises the video
latents with the diffusion decoder and vocodes the audio latents (or returns latents).
Components:
diffusion_decoder (`LTX2VideoDiffusionDecoderModel`) video_processor (`VideoProcessor`) audio_vae
(`AutoencoderKLLTX2Audio`) vocoder (`LTX2Vocoder`)
Inputs:
latents (`Tensor`):
Pre-generated noisy latents for image generation.
base_token_count (`int`):
Number of generated-video tokens, i.e. the sequence length before appended tokens.
output_type (`str`, *optional*, defaults to pil):
Output format: 'pil', 'np', 'pt'.
height (`int`, *optional*, defaults to 512):
The height in pixels of the generated image.
width (`int`, *optional*, defaults to 704):
The width in pixels of the generated image.
num_frames (`int`, *optional*, defaults to 121):
The number of frames in the generated video.
generator (`Generator`, *optional*):
Torch generator for deterministic generation.
dtype (`dtype`):
The dtype of the model inputs, can be generated in input step.
audio_latents (`Tensor`):
Denoised audio latents.
audio_num_frames (`int`):
Number of audio latent frames, used to unpack the audio latent sequence.
Outputs:
videos (`list`):
The generated videos.
audio (`Tensor`):
The generated audio waveform.
"""
model_name = "ltx2.5"
block_classes = [LTX2TrimConditionTokensStep, LTX2DiffusionVaeDecoderStep, LTX2AudioDecoderStep]
block_names = ["trim_condition_tokens", "video_decode", "audio_decode"]
@property
def description(self):
return (
"Decode stage for LTX-2.5 condition workflows: drops the appended keyframe-condition tokens, then "
"denoises the video latents with the diffusion decoder and vocodes the audio latents (or returns "
"latents)."
)
@property
def outputs(self):
return [
OutputParam.template("videos"),
OutputParam("audio", type_hint=torch.Tensor, description="The generated audio waveform."),
]
# auto_docstring
class LTX25AutoDecoderStep(AutoPipelineBlocks):
"""
Auto decode block for LTX-2.5 that selects the decoder based on inputs.
- `LTX25ConditionDecoderStep` when `base_token_count` is present, i.e. the denoised sequence carries appended
keyframe / reference tokens (condition, in-context).
- `LTX25DecoderStep` otherwise (text-to-video, image-to-video).
Components:
diffusion_decoder (`LTX2VideoDiffusionDecoderModel`) video_processor (`VideoProcessor`) audio_vae
(`AutoencoderKLLTX2Audio`) vocoder (`LTX2Vocoder`)
Inputs:
latents (`Tensor`):
Pre-generated noisy latents for image generation.
base_token_count (`int`, *optional*):
Number of generated-video tokens, i.e. the sequence length before appended tokens.
output_type (`str`, *optional*, defaults to pil):
Output format: 'pil', 'np', 'pt'.
height (`int`, *optional*, defaults to 512):
The height in pixels of the generated image.
width (`int`, *optional*, defaults to 704):
The width in pixels of the generated image.
num_frames (`int`, *optional*, defaults to 121):
The number of frames in the generated video.
generator (`Generator`, *optional*):
Torch generator for deterministic generation.
dtype (`dtype`):
The dtype of the model inputs, can be generated in input step.
audio_latents (`Tensor`):
Denoised audio latents.
audio_num_frames (`int`):
Number of audio latent frames, used to unpack the audio latent sequence.
Outputs:
videos (`list`):
The generated videos.
audio (`Tensor`):
The generated audio waveform.
"""
model_name = "ltx2.5"
# Mirrors `LTX2AutoDecoderStep` on the same `base_token_count` trigger; only the video decoder differs.
block_classes = [LTX25ConditionDecoderStep, LTX25DecoderStep]
block_names = ["condition", "default"]
block_trigger_inputs = ["base_token_count", None]
@property
def description(self):
return (
"Auto decode block for LTX-2.5 that selects the decoder based on inputs.\n"
" - `LTX25ConditionDecoderStep` when `base_token_count` is present, i.e. the denoised sequence carries "
"appended keyframe / reference tokens (condition, in-context).\n"
" - `LTX25DecoderStep` otherwise (text-to-video, image-to-video)."
)
@property
def outputs(self):
return [
OutputParam.template("videos"),
OutputParam("audio", type_hint=torch.Tensor, description="The generated audio waveform."),
]
# auto_docstring
class LTX25AutoBlocks(SequentialPipelineBlocks):
"""
Auto blocks for LTX-2.5 supporting text-to-video, image-to-video, condition-to-video and in-context (IC-LoRA)
generation (joint video + audio). Identical to `LTX2AutoBlocks` except that the video decoder is
`LTX2DiffusionVaeDecoderStep`, since the diffusion decoder is the native default from LTX-2.5 on. To decode with
the convolutional VAE instead, swap the decode block: `blocks.sub_blocks["decode"] = LTX2AutoDecoderStep()`.
Supported workflows:
- `text2video`: requires `prompt`
- `image2video`: requires `image`, `prompt`
- `condition`: requires `conditions`, `prompt`
- `in_context`: requires `reference_conditions`, `num_frames`, `prompt`
Components:
prompt_enhancer (`PreTrainedModel`) processor (`ProcessorMixin`) text_encoder (`PreTrainedModel`) tokenizer
(`PreTrainedTokenizerBase`) connectors (`LTX2TextConnectors`) duration_head (`LTX2DurationHead`) vae
(`AutoencoderKLLTX2Video`) video_processor (`VideoProcessor`) transformer (`LTX2VideoTransformer3DModel`)
scheduler (`FlowMatchEulerDiscreteScheduler`) audio_vae (`AutoencoderKLLTX2Audio`) guider (`LTX2Guidance`)
audio_guider (`LTX2Guidance`) diffusion_decoder (`LTX2VideoDiffusionDecoderModel`) vocoder (`LTX2Vocoder`)
Inputs:
prompt (`str`, *optional*):
The prompt or prompts to guide image generation.
conditions (`list`, *optional*):
`LTX2VideoCondition` (or list of them) placing image/video conditions at latent frame indices of the
generated video.
enable_prompt_enhancement (`bool`, *optional*, defaults to False):
Whether to run the prompt enhancer. Opt-in, matching the Lightricks reference pipelines.
system_prompt (`str`, *optional*):
System prompt for enhancement. Defaults to `LTX2_5_I2V_DEFAULT_SYSTEM_PROMPT` when a `PIL.Image.Image`
condition frame is available, else `LTX2_5_T2V_DEFAULT_SYSTEM_PROMPT`.
prompt_max_new_tokens (`int`, *optional*):
Maximum number of new tokens to generate during prompt enhancement. Defaults to 600, the LTX-2.5 Gemma-4
enhancer's budget.
prompt_enhancement_kwargs (`dict`, *optional*):
Keyword arguments for the enhancer's `.generate` call. Defaults to greedy decoding.
prompt_enhancement_seed (`int`, *optional*, defaults to 10):
Random seed for prompt enhancement (inert under LTX-2.5's greedy decoding).
generator (`Generator`, *optional*):
Torch generator for deterministic generation.
image (`Image | list`, *optional*):
Reference image(s) for denoising. Can be a single image or list of images.
negative_prompt (`str`, *optional*):
The prompt or prompts not to guide the image generation.
max_sequence_length (`int`, *optional*, defaults to 1024):
Maximum sequence length for prompt encoding.
min_seconds (`float`, *optional*, defaults to 1.0):
Lower bound on the auto-predicted duration.
max_seconds (`float`, *optional*, defaults to 20.0):
Upper bound on the auto-predicted duration. Must be strictly greater than `min_seconds`.
frame_rate (`float`, *optional*, defaults to 24.0):
Frames per second of the generated video.
height (`int`, *optional*, defaults to 512):
The height in pixels of the generated image.
width (`int`, *optional*, defaults to 704):
The width in pixels of the generated image.
image_crf (`int`, *optional*):
H.264 CRF used to re-compress the conditioning `image` before VAE encode, matching the compression the
model was trained against. `None` (default) resolves from the text-encoder generation (33 through
LTX-2.3, 18 for LTX-2.5). Pass `0` to skip re-compression. Requires a `PIL.Image.Image` when
re-compression runs.
num_frames (`int`, *optional*):
The number of frames in the generated video. Omit to auto-predict via the `duration_head` (see
`LTX2AutoDurationStep`).
reference_conditions (`list`, *optional*):
`LTX2ReferenceCondition` (or list of them) whose videos are encoded into extra latent tokens the IC-LoRA
adapter attends to.
reference_downscale_factor (`int`, *optional*, defaults to 1):
Ratio between the target and reference resolutions; 2 means the reference is preprocessed at half the
target resolution. Spatial coordinates are scaled by this factor so the reference tokens land in the
target coordinate space. Must match the factor the IC-LoRA was trained with.
conditioning_attention_strength (`float`, *optional*, defaults to 1.0):
Scalar in [0, 1] controlling how strongly the noisy tokens and reference tokens attend to each other. 1.0
(default) leaves attention unmasked.
conditioning_attention_mask (`Tensor`, *optional*):
Optional pixel-space mask of shape (1, 1, F, H, W) with values in [0, 1] giving spatially varying
attention strength. Downsampled to the reference's latent grid and multiplied by
`conditioning_attention_strength`.
num_videos_per_prompt (`int`, *optional*, defaults to 1):
The number of images to generate per prompt.
condition_latents (`list`, *optional*):
Per-condition normalized VAE latents of shape [1, C, F, H, W].
condition_strengths (`list`, *optional*):
Per-condition conditioning strengths.
condition_indices (`list`, *optional*):
Per-condition latent frame index at which the condition is applied.
condition_pixel_frames (`list`, *optional*):
Per-condition trimmed pixel frame count, used to clamp single-frame keyframe coords.
reference_latents (`Tensor`, *optional*):
Packed reference tokens of shape [1, total_reference_tokens, C], or `None` when no reference conditions
were supplied (`LTX2AutoReferenceEncoderStep` is skipped).
reference_coords (`Tensor`, *optional*):
RoPE coordinates for the reference tokens.
reference_token_counts (`list`, *optional*):
Per-reference token counts, in `reference_conditions` order.
latents (`Tensor`):
Pre-generated noisy latents for image generation.
noise_scale (`float`, *optional*):
Initial noise level for the un-conditioned tokens. `None` (default) resolves to `sigmas[0]` when custom
`sigmas` are supplied, else 1.0.
sigmas (`list`, *optional*):
Custom sigmas for the denoising process.
reference_cross_mask (`Tensor`, *optional*):
Per-reference-token noisy<->reference attention strengths of shape [1, num_ref_tokens].
num_inference_steps (`int`):
The number of denoising steps.
timesteps (`Tensor`):
Timesteps for the denoising process.
audio_latents (`Tensor`):
Optional pre-encoded audio latents; random noise is used when not provided.
**denoiser_input_fields (`None`, *optional*):
conditional model inputs for the denoiser: e.g. prompt_embeds, negative_prompt_embeds, etc.
use_cross_timestep (`bool`, *optional*, defaults to True):
Whether to condition the transformer on a separate per-token cross timestep (LTX-2.3+).
attention_kwargs (`dict`, *optional*):
Additional kwargs for attention processors.
image_latents (`Tensor`, *optional*):
VAE-encoded reference-image latents used for image-to-video conditioning.
output_type (`str`, *optional*, defaults to pil):
Output format: 'pil', 'np', 'pt'.
Outputs:
videos (`list`):
The generated videos.
audio (`Tensor`):
The generated audio waveform.
"""
model_name = "ltx2.5"
block_classes = [
LTX2AutoPromptEnhancerStep,
LTX2TextConditioningStep,
LTX2AutoDurationStep,
LTX2AutoVaeEncoderStep,
LTX2AutoConditionEncoderStep,
LTX2AutoReferenceEncoderStep,
LTX2AutoCoreDenoiseStep,
LTX25AutoDecoderStep,
]
block_names = [
"prompt_enhancer",
"text_encoder",
"duration",
"vae_encoder",
"condition_encoder",
"reference_encoder",
"denoise",
"decode",
]
# `num_frames` on `in_context` is a requirement, not a trigger: the in-context checkpoints ship without a
# `duration_head`, so the workflow drops `LTX2AutoDurationStep` and `LTX2ConditionEncoderStep` raises if
# `num_frames` is still `None`. Matches `LTX2InContextBlocks`, which omits the duration step outright.
_workflow_map = {
"text2video": {"prompt": True},
"image2video": {"image": True, "prompt": True},
"condition": {"conditions": True, "prompt": True},
"in_context": {"reference_conditions": True, "num_frames": True, "prompt": True},
}
@property
def description(self):
return (
"Auto blocks for LTX-2.5 supporting text-to-video, image-to-video, condition-to-video and in-context "
"(IC-LoRA) generation (joint video + audio). Identical to `LTX2AutoBlocks` except that the video decoder "
"is `LTX2DiffusionVaeDecoderStep`, since the diffusion decoder is the native default from LTX-2.5 on. To "
'decode with the convolutional VAE instead, swap the decode block: `blocks.sub_blocks["decode"] = '
"LTX2AutoDecoderStep()`."
)
@property
def outputs(self):
return [
OutputParam.template("videos"),
OutputParam("audio", type_hint=torch.Tensor, description="The generated audio waveform."),
]

206
tests/modular_pipelines/ltx2/test_modular_pipeline_ltx25.py Normal file
View File

@@ -0,0 +1,206 @@
# coding=utf-8
# Copyright 2026 HuggingFace Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import numpy as np
import PIL.Image
import pytest
import torch
from diffusers.modular_pipelines import LTX2AutoBlocks, LTX25AutoBlocks, LTX25ModularPipeline
from diffusers.pipelines.ltx2.pipeline_ltx2_condition import LTX2VideoCondition
from diffusers.pipelines.ltx2.pipeline_ltx2_ic_lora import LTX2ReferenceCondition
from ..test_modular_pipelines_common import ModularPipelineTesterMixin
LTX25_REPO_ID = "hf-internal-testing/tiny-ltx2-5-modular-pipe"
class LTX25ModularTests(ModularPipelineTesterMixin):
pipeline_class = LTX25ModularPipeline
pipeline_blocks_class = LTX25AutoBlocks
pretrained_model_name_or_path = LTX25_REPO_ID
batch_params = frozenset(["prompt"])
optional_params = frozenset(["num_inference_steps", "num_videos_per_prompt", "latents"])
output_name = "videos"
def get_dummy_inputs(self, seed=0):
return {
"prompt": "a robot dancing",
"negative_prompt": "",
"generator": self.get_generator(seed),
"num_inference_steps": 2,
"height": 32,
"width": 32,
"num_frames": 5,
"frame_rate": 25.0,
"max_sequence_length": 16,
"output_type": "pt",
}
@pytest.mark.skip(reason="num_videos_per_prompt")
def test_num_images_per_prompt(self):
pass
def test_inference_batch_single_identical(self):
super().test_inference_batch_single_identical(expected_max_diff=1e-3)
class TestLTX25ModularPipelineFast(LTX25ModularTests):
params = frozenset(["prompt", "height", "width", "num_frames"])
expected_workflow_blocks = {
"text2video": [
("text_encoder.text_encoder", "LTX2TextEncoderStep"),
("text_encoder.connectors", "LTX2TextConnectorStep"),
("duration", "LTX2DurationStep"),
("denoise.input", "LTX2TextInputStep"),
("denoise.set_timesteps", "LTX2SetTimestepsStep"),
("denoise.prepare_latents", "LTX2PrepareLatentsStep"),
("denoise.prepare_audio_latents", "LTX2PrepareAudioLatentsStep"),
("denoise.prepare_coords", "LTX2PrepareCoordsStep"),
("denoise.denoise", "LTX2DenoiseStep"),
("decode.video_decode", "LTX2DiffusionVaeDecoderStep"),
("decode.audio_decode", "LTX2AudioDecoderStep"),
],
}
def test_diffusion_decoder_output(self):
pipe = self.get_pipeline().to("cpu")
inputs = self.get_dummy_inputs()
output = pipe(**inputs, output=["videos", "audio"])
videos, audio = output["videos"], output["audio"]
assert videos.shape == (1, 5, 3, 32, 32)
assert audio.shape[0] == 1
assert audio.shape[1] == pipe.vocoder.config.out_channels
assert torch.isnan(audio).sum() == 0
def test_graph_matches_ltx2_except_video_decode(self):
# `LTX25AutoBlocks` restates the `LTX2AutoBlocks` graph rather than subclassing it, so nothing but this
# test keeps the two in step: a stage added to one and not the other passes both blocksets' own
# `expected_workflow_blocks`.
ltx2_blocks, ltx25_blocks = LTX2AutoBlocks(), LTX25AutoBlocks()
assert ltx2_blocks.available_workflows == ltx25_blocks.available_workflows
for workflow_name in ltx25_blocks.available_workflows:
expected = [
(name, "LTX2DiffusionVaeDecoderStep" if name == "decode.video_decode" else type(block).__name__)
for name, block in ltx2_blocks.get_workflow(workflow_name).sub_blocks.items()
]
actual = [
(name, type(block).__name__)
for name, block in ltx25_blocks.get_workflow(workflow_name).sub_blocks.items()
]
assert actual == expected, f"Workflow '{workflow_name}' diverges from `LTX2AutoBlocks`"
def test_auto_duration_predicts_a_grid_valid_frame_count(self):
pipe = self.get_pipeline().to("cpu")
inputs = self.get_dummy_inputs()
inputs.pop("num_frames")
inputs["min_seconds"] = 0.5
inputs["max_seconds"] = 2.0
videos = pipe(**inputs, output="videos")
num_frames = videos.shape[1]
assert (num_frames - 1) % pipe.vae_temporal_compression_ratio == 0
assert 0 < num_frames <= round(2.0 * inputs["frame_rate"])
class TestLTX25ModularImage2VideoPipelineFast(LTX25ModularTests):
params = frozenset(["prompt", "image", "height", "width", "num_frames"])
expected_workflow_blocks = {
"image2video": [
("text_encoder.text_encoder", "LTX2TextEncoderStep"),
("text_encoder.connectors", "LTX2TextConnectorStep"),
("duration", "LTX2DurationStep"),
("vae_encoder", "LTX2VaeEncoderStep"),
("denoise.input", "LTX2TextInputStep"),
("denoise.set_timesteps", "LTX2SetTimestepsStep"),
("denoise.prepare_latents", "LTX2PrepareLatentsStep"),
("denoise.prepare_i2v_latents", "LTX2Image2VideoPrepareLatentsStep"),
("denoise.prepare_audio_latents", "LTX2PrepareAudioLatentsStep"),
("denoise.prepare_coords", "LTX2PrepareCoordsStep"),
("denoise.denoise", "LTX2Image2VideoDenoiseStep"),
("decode.video_decode", "LTX2DiffusionVaeDecoderStep"),
("decode.audio_decode", "LTX2AudioDecoderStep"),
],
}
def get_dummy_inputs(self, seed=0):
inputs = super().get_dummy_inputs(seed)
rng = np.random.default_rng(seed)
image = rng.integers(0, 256, size=(32, 32, 3), dtype=np.uint8)
inputs["image"] = PIL.Image.fromarray(image)
# Skip H.264 CRF re-compression so the test does not depend on PyAV.
inputs["image_crf"] = 0
return inputs
class TestLTX25ModularConditionPipelineFast(LTX25ModularTests):
params = frozenset(["prompt", "conditions", "height", "width", "num_frames"])
expected_workflow_blocks = {
"condition": [
("text_encoder.text_encoder", "LTX2TextEncoderStep"),
("text_encoder.connectors", "LTX2TextConnectorStep"),
("duration", "LTX2DurationStep"),
("condition_encoder", "LTX2ConditionEncoderStep"),
("denoise.input", "LTX2TextInputStep"),
("denoise.prepare_latents", "LTX2ConditionPrepareLatentsStep"),
("denoise.set_timesteps", "LTX2ConditionSetTimestepsStep"),
("denoise.prepare_audio_latents", "LTX2ConditionPrepareAudioLatentsStep"),
("denoise.prepare_coords", "LTX2ConditionPrepareCoordsStep"),
("denoise.denoise", "LTX2ConditionDenoiseStep"),
("decode.trim_condition_tokens", "LTX2TrimConditionTokensStep"),
("decode.video_decode", "LTX2DiffusionVaeDecoderStep"),
("decode.audio_decode", "LTX2AudioDecoderStep"),
],
}
def get_dummy_inputs(self, seed=0):
inputs = super().get_dummy_inputs(seed)
image = torch.rand((1, 3, 32, 32), generator=torch.Generator("cpu").manual_seed(seed))
# Synthetic float tensors skip H.264 CRF re-compression (training path uses PIL/uint8).
inputs["conditions"] = LTX2VideoCondition(frames=image, index=0, strength=1.0, crf=0)
return inputs
class TestLTX25ModularInContextPipelineFast(LTX25ModularTests):
params = frozenset(["prompt", "reference_conditions", "height", "width", "num_frames"])
expected_workflow_blocks = {
"in_context": [
("text_encoder.text_encoder", "LTX2TextEncoderStep"),
("text_encoder.connectors", "LTX2TextConnectorStep"),
("condition_encoder", "LTX2ConditionEncoderStep"),
("reference_encoder", "LTX2ReferenceEncoderStep"),
("denoise.input", "LTX2TextInputStep"),
("denoise.prepare_latents", "LTX2InContextPrepareLatentsStep"),
("denoise.set_timesteps", "LTX2ConditionSetTimestepsStep"),
("denoise.prepare_audio_latents", "LTX2ConditionPrepareAudioLatentsStep"),
("denoise.prepare_coords", "LTX2ConditionPrepareCoordsStep"),
("denoise.denoise", "LTX2ConditionDenoiseStep"),
("decode.trim_condition_tokens", "LTX2TrimConditionTokensStep"),
("decode.video_decode", "LTX2DiffusionVaeDecoderStep"),
("decode.audio_decode", "LTX2AudioDecoderStep"),
],
}
def get_dummy_inputs(self, seed=0):
inputs = super().get_dummy_inputs(seed)
video = torch.rand((1, 5, 3, 32, 32), generator=torch.Generator("cpu").manual_seed(seed))
inputs["reference_conditions"] = LTX2ReferenceCondition(frames=video, strength=1.0)
return inputs