lingbot_video_pipeline

Stage-composed LingBot-Video Dense and MoE/refiner T2V pipeline.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.FastVideoArgs`

FastVideoArgs(model_path: str, mode: ExecutionMode = ExecutionMode.INFERENCE, workload_type: WorkloadType = WorkloadType.T2V, distributed_executor_backend: str = 'mp', enable_streaming_ipc_queues: bool = False, ray_placement_group: PlacementGroup | None = None, ray_runtime_env: RuntimeEnv | None = None, inference_mode: bool = True, trust_remote_code: bool = False, revision: str | None = None, num_gpus: int = 1, tp_size: int = -1, sp_size: int = -1, hsdp_replicate_dim: int = 1, hsdp_shard_dim: int = -1, dist_timeout: int | None = None, pipeline_config: PipelineConfig = PipelineConfig(), preprocess_config: PreprocessConfig | None = None, lora_path: str | None = None, lora_nickname: str = 'default', lora_strength: float = 1.0, lora_target_modules: list[str] | None = None, output_type: str = 'pil', attention_backend: str | None = None, dit_cpu_offload: bool = True, use_fsdp_inference: bool = False, dit_layerwise_offload: bool = True, text_encoder_cpu_offload: bool = True, image_encoder_cpu_offload: bool = True, vae_cpu_offload: bool = True, pin_cpu_memory: bool = True, h3_sequential_load: bool | None = None, h3_encoder_split: bool | None = None, h3_encoder_nodes: int = -1, h3_encoder_workers: int = 0, video_decode_backend: str = 'h3-vae', taeh3_checkpoint: str | None = None, taeh3_chunk_size: int = 5, h3_vae_nvfp4_decoder: str | None = None, lazy_module_load: bool | None = None, vae_parallel_decode: bool = False, vae_parallel_encode: bool = False, vae_parallel_decode_strategy: str | None = None, enable_torch_compile: bool = False, regional_compile: bool = False, enable_torch_compile_text_encoder: bool = False, enable_torch_compile_vae: bool = False, enable_torch_compile_audio_vae: bool = False, torch_compile_kwargs: dict[str, Any] = dict(), torch_compile_kwargs_dit: dict[str, Any] = dict(), torch_compile_kwargs_text_encoder: dict[str, Any] = dict(), torch_compile_kwargs_vae: dict[str, Any] = dict(), torch_compile_kwargs_audio_vae: dict[str, Any] = dict(), inference_torch_compile: bool = False, h3_vsa_fp4: dict[str, Any] = dict(), disable_autocast: bool = False, VSA_sparsity: float | None = None, VSA_tile_size: int | None = None, moba_config_path: str | None = None, moba_config: dict[str, Any] = dict(), master_port: int | None = None, enable_stage_verification: bool = True, prompt_txt: str | None = None, ltx2_vae_tiling: bool | None = None, ltx2_vae_spatial_tile_size_in_pixels: int | None = None, ltx2_vae_spatial_tile_overlap_in_pixels: int | None = None, ltx2_vae_temporal_tile_size_in_frames: int | None = None, ltx2_vae_temporal_tile_overlap_in_frames: int | None = None, ltx2_initial_latent_path: str | None = None, ltx2_audio_latent_path: str | None = None, refine_enabled: bool | None = None, refine_upsampler_path: str | None = None, refine_transformer_path: str | None = None, refine_lora_path: str | None = None, refine_num_inference_steps: int | None = None, refine_guidance_scale: float | None = None, refine_add_noise: bool | None = None, refine_noise_path: str | None = None, refine_audio_noise_path: str | None = None, ltx2_refine_enabled: bool = False, ltx2_refine_upsampler_path: str | None = None, ltx2_refine_transformer_path: str | None = None, ltx2_refine_lora_path: str | None = None, ltx2_refine_num_inference_steps: int = 3, ltx2_refine_guidance_scale: float = 1.0, ltx2_refine_add_noise: bool = True, ltx2_refine_noise_path: str | None = None, ltx2_refine_audio_noise_path: str | None = None, ltx2_legacy_native_noise_order: bool = False, ltx2_use_distilled_sigmas: bool = True, model_paths: dict[str, str] = dict(), model_loaded: dict[str, bool] = (lambda: {'transformer': True, 'vae': True, 'upsampler': True})(), override_text_encoder_safetensors: str | None = None, override_text_encoder_quant: QuantizationMethods = None, transformer_quant: Any | None = None, override_transformer_cls_name: str | None = None, init_weights_from_safetensors: str = '', init_weights_from_safetensors_2: str = '', override_pipeline_cls_name: str | None = None, boundary_ratio: float | None = 0.875)

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.FastVideoArgs.is_output_rank`

is_output_rank: bool

Whether this process may materialize user-facing outputs.

This is runtime state assigned by an SPMD executor, not a public configuration field, so dataclass serialization and typed config parsing deliberately ignore it.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.FastVideoArgs.add_cli_args`

add_cli_args(parser: FlexibleArgumentParser) -> FlexibleArgumentParser

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.FastVideoArgs.from_cli_args`

from_cli_args(args: argparse.Namespace) -> FastVideoArgs

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.FastVideoArgs.from_kwargs`

from_kwargs(**kwargs: Any) -> FastVideoArgs

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.FastVideoArgs.check_fastvideo_args`

check_fastvideo_args() -> None

Validate inference arguments for consistency

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.FastVideoArgs.finalize_device_offload_policy`

finalize_device_offload_policy(device_id: int = 0) -> bool

Apply device-local memory policy, then resolve incompatible modes.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.FastVideoArgs.disable_offload_on_unified_memory`

disable_offload_on_unified_memory(device_id: int = 0, *, offload_flag: str | None = None) -> bool

Disable host offload after a worker has selected its device.

CUDA's unified-memory probe reads runtime device properties and may initialize a CUDA context. Callers must therefore use this only inside a device-owning process, after selecting and binding device_id. Returning the classification lets direct component-loader callers apply the same policy to explicit per-call overrides. When offload_flag is given, the return value says whether this policy covers that component role.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ComposedPipelineBase`

ComposedPipelineBase(model_path: str, fastvideo_args: FastVideoArgs | TrainingArgs, required_config_modules: list[str] | None = None, loaded_modules: dict[str, torch.nn.Module] | None = None)

Base class for pipelines composed of multiple stages.

This class provides the framework for creating pipelines by composing multiple stages together. Each stage is responsible for a specific part of the diffusion process, and the pipeline orchestrates the execution of these stages.

Initialize the pipeline. After init, the pipeline should be ready to use. The pipeline should be stateless and not hold any batch state.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ComposedPipelineBase.get_hf_download_component_dirs`

get_hf_download_component_dirs() -> tuple[str, ...] | None

Return component directories for an opt-in partial Hub download.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ComposedPipelineBase.get_hf_download_allow_patterns`

get_hf_download_allow_patterns() -> list[str] | None

Return Hub patterns for the manifest and selected components.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ComposedPipelineBase.set_trainable`

set_trainable() -> None

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ComposedPipelineBase.post_init`

post_init() -> None

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ComposedPipelineBase.initialize_training_pipeline`

initialize_training_pipeline(training_args: TrainingArgs)

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ComposedPipelineBase.initialize_validation_pipeline`

initialize_validation_pipeline(training_args: TrainingArgs)

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ComposedPipelineBase.from_pretrained`

from_pretrained(model_path: str, device: str | None = None, torch_dtype: torch.dtype | None = None, pipeline_config: str | PipelineConfig | None = None, args: argparse.Namespace | None = None, required_config_modules: list[str] | None = None, loaded_modules: dict[str, torch.nn.Module] | None = None, **kwargs) -> ComposedPipelineBase

Load a pipeline from a pretrained model. loaded_modules: Optional[Dict[str, torch.nn.Module]] = None, If provided, loaded_modules will be used instead of loading from config/pretrained weights.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ComposedPipelineBase.get_module`

get_module(module_name: str, default_value: Any = None) -> Any

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ComposedPipelineBase.add_module`

add_module(module_name: str, module: Any)

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ComposedPipelineBase.required_config_modules`

required_config_modules: list[str]

List of modules that are required by the pipeline. The names should match the diffusers directory and model_index.json file. These modules will be loaded using the PipelineComponentLoader and made available in the modules dictionary. Access these modules using the get_module method.

class ConcretePipeline(ComposedPipelineBase): _required_config_modules = ["vae", "text_encoder", "transformer", "scheduler", "tokenizer"]

@property
def required_config_modules(self):
    return self._required_config_modules

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ComposedPipelineBase.stages`

stages: list[PipelineStage]

List of stages in the pipeline.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ComposedPipelineBase.create_pipeline_stages`

create_pipeline_stages(fastvideo_args: FastVideoArgs)

Create the inference pipeline stages.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ComposedPipelineBase.create_training_stages`

create_training_stages(training_args: TrainingArgs)

Create the training pipeline stages.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ComposedPipelineBase.initialize_pipeline`

initialize_pipeline(fastvideo_args: FastVideoArgs)

Initialize the pipeline.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ComposedPipelineBase.load_modules`

load_modules(fastvideo_args: FastVideoArgs, loaded_modules: dict[str, torch.nn.Module] | None = None) -> dict[str, Any]

Load the modules from the config. loaded_modules: Optional[Dict[str, torch.nn.Module]] = None, If provided, loaded_modules will be used instead of loading from config/pretrained weights.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ComposedPipelineBase.add_stage`

add_stage(stage_name: str, stage: PipelineStage)

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ComposedPipelineBase.forward`

forward(batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch

Generate a video or image using the pipeline.

Parameters

batch: ForwardBatch: The batch to generate from.
fastvideo_args: FastVideoArgs: The inference arguments.

Returns: ForwardBatch: The batch with the generated video or image.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ComposedPipelineBase.train`

train() -> None

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ComposedPipelineBase.close`

close() -> None

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.LoRAPipeline`

LoRAPipeline(*args, **kwargs)

Pipeline that supports injecting LoRA adapters into the diffusion transformer. TODO: support training.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.LoRAPipeline.add_module`

add_module(module_name: str, module: Any) -> None

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.LoRAPipeline.is_target_layer`

is_target_layer(module_name: str) -> bool

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.LoRAPipeline.convert_adapter_state_dict`

convert_adapter_state_dict(state_dict: dict[str, torch.Tensor], lora_path: str) -> dict[str, torch.Tensor]

Rewrite a foreign-layout adapter into the layout the model's name mapping speaks.

The default leaves the adapter alone. Pipelines override it for model families whose published adapters use another naming or fusion scheme, and raise when the adapter is unusable.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.LoRAPipeline.set_trainable`

set_trainable() -> None

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.LoRAPipeline.convert_to_lora_layers`

convert_to_lora_layers() -> None

Unified method to convert the transformer to a LoRA transformer.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.LoRAPipeline.set_lora_adapter`

set_lora_adapter(lora_nickname: str, lora_path: str | None = None, strength: float = 1.0, accumulate: bool = False)

Load a LoRA adapter into the pipeline and merge it into the transformer. Args: lora_nickname: The "nick name" of the adapter when referenced in the pipeline. lora_path: The path to the adapter, either a local path or a Hugging Face repo id. strength: Scale for the low-rank adapter. Hybrid adapters must set this at construction so their dense payload receives the same scale. accumulate: Add this adapter to an already merged pure low-rank adapter.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.LoRAPipeline.merge_lora_weights`

merge_lora_weights() -> None

Merge LoRA weights, then repack the packed quantized buffers for the merged state.

NVFP4 layers are only supported while their BF16 weight is retained: a purged BF16 weight cannot be unmerged and remerged.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.LoRAPipeline.unmerge_lora_weights`

unmerge_lora_weights() -> None

Unmerge LoRA weights, then repack the packed quantized buffers for the unmerged state.

NVFP4 is rejected: its FP4 buffers cannot be rebuilt from a purged BF16 weight.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.LingBotVideoDenoisingStage`

LingBotVideoDenoisingStage(transformer, scheduler, refiner: bool = False)

Run the released batched-CFG bf16 DiT loop with fp32 scheduler state.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.LingBotVideoDenoisingStage.forward`

forward(batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch

Denoise latents while keeping scheduler samples and predictions in fp32.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.LingBotVideoInputValidationStage`

LingBotVideoInputValidationStage(refiner_enabled: bool = False)

Validate released shape constraints and construct the official CUDA RNG.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.LingBotVideoInputValidationStage.forward`

forward(batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch

Run shared validation, then enforce LingBot temporal and spatial geometry.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.LingBotVideoLatentPreparationStage`

LingBotVideoLatentPreparationStage(transformer)

Prepare fp32 latents in the released 4x temporal and 8x spatial geometry.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.LingBotVideoLatentPreparationStage.forward`

forward(batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch

Generate or validate one normalized fp32 latent video.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.LingBotVideoRefinerPreparationStage`

LingBotVideoRefinerPreparationStage(vae, scheduler)

Resize and encode the base video, then initialize the released refiner state.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.LingBotVideoRefinerPreparationStage.forward`

forward(batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch

Prepare high-resolution refiner latents and its exact truncated sigma schedule.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ConditioningStage`

ConditioningStage()

Stage for applying conditioning to the diffusion process.

This stage handles the application of conditioning, such as classifier-free guidance, to the diffusion process.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ConditioningStage.forward`

forward(batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch

Apply conditioning to the diffusion process.

Parameters

batch: ForwardBatch: The current batch information.
fastvideo_args: FastVideoArgs: The inference arguments.

Returns

ForwardBatch: The batch with applied conditioning.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ConditioningStage.verify_input`

verify_input(batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> VerificationResult

Verify conditioning stage inputs.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.ConditioningStage.verify_output`

verify_output(batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> VerificationResult

Verify conditioning stage outputs.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.DecodingStage`

DecodingStage(vae, pipeline = None)

Stage for decoding latent representations into pixel space.

This stage handles the decoding of latent representations into the final output format (e.g., pixel values).

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.DecodingStage.verify_input`

verify_input(batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> VerificationResult

Verify decoding stage inputs.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.DecodingStage.verify_output`

verify_output(batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> VerificationResult

Verify decoding stage outputs.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.DecodingStage.decode`

decode(latents: torch.Tensor, fastvideo_args: FastVideoArgs) -> torch.Tensor

Decode latent representations into pixel space using VAE.

Parameters

latents: torch.Tensor: Input latent tensor with shape (batch, channels, frames, height_latents, width_latents)
fastvideo_args: FastVideoArgs: Configuration containing: - disable_autocast: Whether to disable automatic mixed precision (default: False) - pipeline_config.vae_precision: VAE computation precision ("fp32", "fp16", "bf16") - pipeline_config.vae_tiling: Whether to enable VAE tiling for memory efficiency

Returns

torch.Tensor: Decoded video tensor with shape (batch, channels, frames, height, width),
torch.Tensor: normalized to [0, 1] range and moved to CPU as float32

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.DecodingStage.streaming_decode`

streaming_decode(latents: torch.Tensor, fastvideo_args: FastVideoArgs, cache: list[torch.Tensor | None] | None = None, is_first_chunk: bool = False) -> tuple[torch.Tensor, list[torch.Tensor | None]]

Decode latent representations into pixel space using VAE with streaming cache.

Parameters

latents: torch.Tensor: Input latent tensor with shape (batch, channels, frames, height_latents, width_latents)
fastvideo_args: FastVideoArgs: Configuration object.
cache: list[torch.Tensor | None] | None: VAE cache from previous call, or None to initialize a new cache.
is_first_chunk: bool: Whether this is the first chunk.

Returns

tuple[torch.Tensor, list[torch.Tensor | None]]: A tuple of (decoded_frames, updated_cache).

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.DecodingStage.forward`

forward(batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch

Decode latent representations into pixel space.

This method processes the batch through the VAE decoder, converting latent representations to pixel-space video/images. It also optionally decodes trajectory latents for visualization purposes.

Parameters

batch: ForwardBatch: The current batch containing: - latents: Tensor to decode (batch, channels, frames, height_latents, width_latents) - return_trajectory_decoded (optional): Flag to decode trajectory latents - trajectory_latents (optional): Latents at different timesteps - trajectory_timesteps (optional): Corresponding timesteps
fastvideo_args: FastVideoArgs: Configuration containing: - output_type: "latent" to skip decoding, otherwise decode to pixels - vae_cpu_offload: Whether to offload VAE to CPU after decoding - model_loaded: Track VAE loading state - model_paths: Path to VAE model if loading needed

Returns

ForwardBatch: Modified batch with: - output: Decoded frames (batch, channels, frames, height, width) as CPU float32 - trajectory_decoded (if requested): List of decoded frames per timestep

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.TextEncodingStage`

TextEncodingStage(text_encoders, tokenizers)

Stage for encoding text prompts into embeddings for diffusion models.

This stage handles the encoding of text prompts into the embedding space expected by the diffusion model.

Initialize the prompt encoding stage.

Parameters

enable_logging: Whether to enable logging for this stage.
is_secondary: Whether this is a secondary text encoder.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.TextEncodingStage.forward`

forward(batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch

Encode the prompt into text encoder hidden states.

Parameters

batch: ForwardBatch: The current batch information.
fastvideo_args: FastVideoArgs: The inference arguments.

Returns

ForwardBatch: The batch with encoded prompt embeddings.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.TextEncodingStage.verify_input`

verify_input(batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> VerificationResult

Verify text encoding stage inputs.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.TextEncodingStage.encode_text`

encode_text(text: str | list[str], fastvideo_args: FastVideoArgs, encoder_index: int | list[int] | None = None, return_attention_mask: bool = False, return_type: str = 'list', device: torch.device | str | None = None, dtype: torch.dtype | None = None, max_length: int | Sequence[int | None] | None = None, truncation: bool | None = None, padding: bool | str | None = None)

Encode plain text using selected text encoder(s) and return embeddings.

Parameters

text: str | list[str]: A single string or a list of strings to encode.
fastvideo_args: FastVideoArgs: The inference arguments providing pipeline config, including tokenizer and encoder settings, preprocess and postprocess functions.
encoder_index: int | list[int] | None: Encoder selector by index. Accepts an int or list of ints.
return_attention_mask: bool: If True, also return attention masks for each selected encoder.
return_type: str: "list" (default) returns a list aligned with selection; "dict" returns a dict keyed by encoder index as a string; "stack" stacks along a new first dimension (requires matching shapes).
device: torch.device | str | None: Optional device override for inputs; defaults to local torch device.
dtype: torch.dtype | None: Optional dtype to cast returned embeddings to.
max_length: int | Sequence[int | None] | None: Optional per-call tokenizer override. Either one value applied to every selected encoder (existing behavior), or a sequence with one entry per encoder in encoder_index order -- None at a position falls back to that encoder's own configured default instead of the override.
truncation: bool | None: Optional per-call tokenizer override.
padding: bool | str | None: Optional per-call tokenizer override.

Returns

Depending on return_type and return_attention_mask:

list: List[Tensor] or (List[Tensor], List[Tensor])

dict: Dict[str, Tensor] or (Dict[str, Tensor], Dict[str, Tensor])

stack: Tensor of shape [num_encoders, ...] or a tuple with stacked attention masks

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.TextEncodingStage.return_embeds`

return_embeds(embeds_list: list[torch.Tensor], attn_masks_list: list[torch.Tensor], return_type: str = 'list', return_attention_mask: bool = False, indices: list[int] | None = None) -> Any

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.TextEncodingStage.verify_output`

verify_output(batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> VerificationResult

Verify text encoding stage outputs.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.TimestepPreparationStage`

TimestepPreparationStage(scheduler)

Stage for preparing timesteps for the diffusion process.

This stage handles the preparation of the timestep sequence that will be used during the diffusion process.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.TimestepPreparationStage.forward`

forward(batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch

Prepare timesteps for the diffusion process.

Parameters

batch: ForwardBatch: The current batch information.
fastvideo_args: FastVideoArgs: The inference arguments.

Returns

ForwardBatch: The batch with prepared timesteps.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.TimestepPreparationStage.verify_input`

verify_input(batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> VerificationResult

Verify timestep preparation stage inputs.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.TimestepPreparationStage.verify_output`

verify_output(batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> VerificationResult

Verify timestep preparation stage outputs.

`fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.LingBotVideoPipeline`

LingBotVideoPipeline(*args, **kwargs)

T2V pipeline with optional released MoE pixel-space refinement.

Source

class LingBotVideoPipeline(LoRAPipeline, ComposedPipelineBase):
    """T2V pipeline with optional released MoE pixel-space refinement."""
is_video_pipeline = True
_required_config_modules = &#91;&quot;text_encoder&quot;, &quot;tokenizer&quot;, &quot;vae&quot;, &quot;transformer&quot;, &quot;scheduler&quot;&#93;

def load_modules(
    self,
    fastvideo_args: FastVideoArgs,
    loaded_modules: dict&#91;str, Any&#93; | None = None,
) -&gt; dict&#91;str, Any&#93;:
    &quot;&quot;&quot;Load the optional refiner DiT and the VAE encoder only when declared.&quot;&quot;&quot;
    model_index = self._load_config(self.model_path)
    required = list(type(self)._required_config_modules)
    load_refiner = &quot;transformer_2&quot; in model_index and getattr(fastvideo_args, &quot;refine_enabled&quot;, None) is not False
    if load_refiner:
        required.append(&quot;transformer_2&quot;)
        fastvideo_args.pipeline_config.vae_config.load_encoder = True
    self._required_config_modules = required
    return super().load_modules(fastvideo_args, loaded_modules)

def initialize_pipeline(self, fastvideo_args: FastVideoArgs) -&gt; None:
    &quot;&quot;&quot;Apply the released runtime flow shift to the loaded scheduler.&quot;&quot;&quot;
    shift = fastvideo_args.pipeline_config.flow_shift
    if shift is None:
        raise ValueError(&quot;LingBot-Video requires a flow shift&quot;)
    self.get_module(&quot;scheduler&quot;).set_shift(float(shift))

def create_pipeline_stages(self, fastvideo_args: FastVideoArgs) -&gt; None:
    &quot;&quot;&quot;Create base generation and the optional decoded-video refiner stages.&quot;&quot;&quot;
    refiner = self.get_module(&quot;transformer_2&quot;)
    self.add_stage(
        &quot;input_validation_stage&quot;,
        LingBotVideoInputValidationStage(refiner_enabled=refiner is not None),
    )
    self.add_stage(
        &quot;prompt_encoding_stage&quot;,
        TextEncodingStage(
            text_encoders=&#91;self.get_module(&quot;text_encoder&quot;)&#93;,
            tokenizers=&#91;self.get_module(&quot;tokenizer&quot;)&#93;,
        ),
    )
    self.add_stage(&quot;conditioning_stage&quot;, ConditioningStage())
    self.add_stage(
        &quot;timestep_preparation_stage&quot;,
        TimestepPreparationStage(scheduler=self.get_module(&quot;scheduler&quot;)),
    )
    self.add_stage(
        &quot;latent_preparation_stage&quot;,
        LingBotVideoLatentPreparationStage(transformer=self.get_module(&quot;transformer&quot;)),
    )
    self.add_stage(
        &quot;denoising_stage&quot;,
        LingBotVideoDenoisingStage(
            transformer=self.get_module(&quot;transformer&quot;),
            scheduler=self.get_module(&quot;scheduler&quot;),
        ),
    )
    self.add_stage(
        &quot;decoding_stage&quot;,
        DecodingStage(vae=self.get_module(&quot;vae&quot;), pipeline=self),
    )
    if refiner is not None:
        self.add_stage(
            &quot;refiner_preparation_stage&quot;,
            LingBotVideoRefinerPreparationStage(
                vae=self.get_module(&quot;vae&quot;),
                scheduler=self.get_module(&quot;scheduler&quot;),
            ),
        )
        self.add_stage(
            &quot;refiner_denoising_stage&quot;,
            LingBotVideoDenoisingStage(
                transformer=refiner,
                scheduler=self.get_module(&quot;scheduler&quot;),
                refiner=True,
            ),
        )
        self.add_stage(
            &quot;refiner_decoding_stage&quot;,
            DecodingStage(vae=self.get_module(&quot;vae&quot;), pipeline=self),
        )</code></pre></details>
fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.LingBotVideoPipeline.load_modules
load_modules(fastvideo_args: FastVideoArgs, loaded_modules: dict[str, Any] | None = None) -> dict[str, Any]
Load the optional refiner DiT and the VAE encoder only when declared.
Source
def load_modules(
    self,
    fastvideo_args: FastVideoArgs,
    loaded_modules: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Load the optional refiner DiT and the VAE encoder only when declared."""
    model_index = self._load_config(self.model_path)
    required = list(type(self)._required_config_modules)
    load_refiner = "transformer_2" in model_index and getattr(fastvideo_args, "refine_enabled", None) is not False
    if load_refiner:
        required.append("transformer_2")
        fastvideo_args.pipeline_config.vae_config.load_encoder = True
    self._required_config_modules = required
    return super().load_modules(fastvideo_args, loaded_modules)
fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.LingBotVideoPipeline.initialize_pipeline
initialize_pipeline(fastvideo_args: FastVideoArgs) -> None
Apply the released runtime flow shift to the loaded scheduler.
Source
def initialize_pipeline(self, fastvideo_args: FastVideoArgs) -> None:
    """Apply the released runtime flow shift to the loaded scheduler."""
    shift = fastvideo_args.pipeline_config.flow_shift
    if shift is None:
        raise ValueError("LingBot-Video requires a flow shift")
    self.get_module("scheduler").set_shift(float(shift))
fastvideo.pipelines.basic.lingbot_video.lingbot_video_pipeline.LingBotVideoPipeline.create_pipeline_stages
create_pipeline_stages(fastvideo_args: FastVideoArgs) -> None
Create base generation and the optional decoded-video refiner stages.
Source
def create_pipeline_stages(self, fastvideo_args: FastVideoArgs) -> None:
    """Create base generation and the optional decoded-video refiner stages."""
    refiner = self.get_module("transformer_2")
    self.add_stage(
        "input_validation_stage",
        LingBotVideoInputValidationStage(refiner_enabled=refiner is not None),
    )
    self.add_stage(
        "prompt_encoding_stage",
        TextEncodingStage(
            text_encoders=[self.get_module("text_encoder")],
            tokenizers=[self.get_module("tokenizer")],
        ),
    )
    self.add_stage("conditioning_stage", ConditioningStage())
    self.add_stage(
        "timestep_preparation_stage",
        TimestepPreparationStage(scheduler=self.get_module("scheduler")),
    )
    self.add_stage(
        "latent_preparation_stage",
        LingBotVideoLatentPreparationStage(transformer=self.get_module("transformer")),
    )
    self.add_stage(
        "denoising_stage",
        LingBotVideoDenoisingStage(
            transformer=self.get_module("transformer"),
            scheduler=self.get_module("scheduler"),
        ),
    )
    self.add_stage(
        "decoding_stage",
        DecodingStage(vae=self.get_module("vae"), pipeline=self),
    )
    if refiner is not None:
        self.add_stage(
            "refiner_preparation_stage",
            LingBotVideoRefinerPreparationStage(
                vae=self.get_module("vae"),
                scheduler=self.get_module("scheduler"),
            ),
        )
        self.add_stage(
            "refiner_denoising_stage",
            LingBotVideoDenoisingStage(
                transformer=refiner,
                scheduler=self.get_module("scheduler"),
                refiner=True,
            ),
        )
        self.add_stage(
            "refiner_decoding_stage",
            DecodingStage(vae=self.get_module("vae"), pipeline=self),
        )