Skip to content

vllm_omni.diffusion.models.ltx2.ltx2_runtime

Shared recipe-driven runtime for LTX pipeline variants.

logger module-attribute 

logger = init_logger(__name__)

LTXRuntime

Bases: LTXRequestMixin, LTXTextConditioningMixin, Module, CFGParallelMixin, ProgressBarMixin, SupportsComponentDiscovery, DiffusionPipelineProfilerMixin

Shared Omni runtime for recipe-driven LTX denoise phases.

component_profile class-attribute instance-attribute 

component_profile: LTXComponentProfile = (
    resolve_ltx_component_profile(
        self.pipeline_kind, self.model_version
    )
)

connector_batches_cfg class-attribute instance-attribute 

connector_batches_cfg = False

distributed_video_decode class-attribute instance-attribute 

distributed_video_decode = True

do_classifier_free_guidance property 

do_classifier_free_guidance

do_guidance property 

do_guidance

dummy_run_num_frames class-attribute 

dummy_run_num_frames: int = 0

guidance_executor class-attribute 

guidance_executor: LTXGuidanceExecutor = (
    LTX_GUIDANCE_EXECUTOR
)

guidance_scale property 

guidance_scale

interrupt property 

interrupt

model_version instance-attribute 

model_version = detect_ltx_model_version(
    od_config.model,
    revision=getattr(od_config, "revision", None),
)

pipeline_kind class-attribute 

pipeline_kind: str = 'one_stage'

pipeline_recipe class-attribute instance-attribute 

pipeline_recipe: LTXPipelineRecipe = (
    resolve_ltx_pipeline_recipe(
        self.pipeline_kind, self.model_version
    )
)

preserve_sp_padded_audio_duration class-attribute instance-attribute 

preserve_sp_padded_audio_duration = False

reports_stage_durations class-attribute instance-attribute 

reports_stage_durations = False

support_image_input class-attribute instance-attribute 

support_image_input = False

supports_request_batch class-attribute instance-attribute 

supports_request_batch = False

use_diffusion_decoder instance-attribute 

use_diffusion_decoder = _ltx2_use_diffusion_decoder(
    od_config, self.model_version
)

combine_cfg_noise 

combine_cfg_noise(
    positive_noise_pred,
    negative_noise_pred,
    true_cfg_scale,
    cfg_normalize=False,
    kwargs: dict[str, Any] | None = None,
    **context: Any,
)

decode_phase 

decode_phase(
    phase: LTXPhaseResult,
) -> DiffusionOutput | list[DiffusionOutput]

Decode one completed phase and restore per-request outputs.

eval 

eval()

load_weights 

load_weights(
    weights: Iterable[tuple[str, Tensor]],
) -> set[str]

predict_noise 

predict_noise(**kwargs)

prepare_audio_latents 

prepare_audio_latents(
    batch_size: int = 1,
    num_channels_latents: int = 8,
    audio_latent_length: int = 1,
    num_mel_bins: int = 64,
    noise_scale: float = 0.0,
    dtype: dtype | None = None,
    device: device | None = None,
    generator: Generator | list[Generator] | None = None,
    latents: Tensor | None = None,
    latents_normalized: bool = False,
) -> tuple[Tensor, int, int]

prepare_latents 

prepare_latents(
    batch_size: int = 1,
    num_channels_latents: int = 128,
    height: int = 512,
    width: int = 768,
    num_frames: int = 121,
    noise_scale: float = 0.0,
    dtype: dtype | None = None,
    device: device | None = None,
    generator: Generator | list[Generator] | None = None,
    latents: Tensor | None = None,
) -> Tensor

run_phase 

run_phase(
    req: DiffusionRequestBatch,
    request_inputs: LTXRequestInputs,
    *,
    noise_scale: float,
    sigmas: list[float] | None,
    timesteps: list[int] | None,
    attention_kwargs: dict[str, Any] | None,
    phase_recipe: LTXPhaseRecipe,
    image: Any | None = None,
    prompt_context: LTXPromptContext | None = None,
) -> LTXPhaseResult

Prepare and execute one phase without decoding its output.