LingBot World 2 causal-fast image-to-video pipeline.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.get_local_torch_device`

get_local_torch_device() -> torch.device

Return the torch device for the current rank.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.get_sp_world_size`

get_sp_world_size() -> int

Return world size for the sequence model parallel group.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_pipeline.FastVideoArgs.add_cli_args`

add_cli_args(parser: FlexibleArgumentParser) -> FlexibleArgumentParser

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.FastVideoArgs.from_cli_args`

from_cli_args(args: argparse.Namespace) -> FastVideoArgs

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.FastVideoArgs.from_kwargs`

from_kwargs(**kwargs: Any) -> FastVideoArgs

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.FastVideoArgs.check_fastvideo_args`

check_fastvideo_args() -> None

Validate inference arguments for consistency

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_pipeline.init_logger`

init_logger(name: str) -> _FastvideoLogger

The main purpose of this function is to ensure that loggers are retrieved in such a way that we can be sure the root fastvideo logger has already been configured.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.compute_relative_poses`

compute_relative_poses(c2ws_mat: torch.Tensor, framewise: bool = False, normalize_trans: bool = True) -> torch.Tensor

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.get_Ks_transformed`

get_Ks_transformed(Ks: torch.Tensor, height_org: int, width_org: int, height_resize: int, width_resize: int, height_final: int, width_final: int)

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.get_plucker_embeddings`

get_plucker_embeddings(c2ws_mat: torch.Tensor, Ks: torch.Tensor, height: int, width: int)

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.interpolate_camera_poses`

interpolate_camera_poses(src_indices: np.ndarray, src_rot_mat: np.ndarray, src_trans_vec: np.ndarray, tgt_indices: np.ndarray) -> torch.Tensor

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.FlowUniPCMultistepScheduler`

FlowUniPCMultistepScheduler(num_train_timesteps: int = 1000, solver_order: int = 2, prediction_type: str = 'flow_prediction', shift: float | None = 1.0, use_dynamic_shifting = False, thresholding: bool = False, dynamic_thresholding_ratio: float = 0.995, sample_max_value: float = 1.0, predict_x0: bool = True, solver_type: str = 'bh2', lower_order_final: bool = True, disable_corrector: tuple = (), solver_p: SchedulerMixin = None, timestep_spacing: str = 'linspace', steps_offset: int = 0, final_sigmas_type: str | None = 'zero', **kwargs)

UniPCMultistepScheduler is a training-free framework designed for the fast sampling of diffusion models.

This model inherits from [SchedulerMixin] and [ConfigMixin]. Check the superclass documentation for the generic methods the library implements for all schedulers such as loading and saving.

Parameters

num_train_timesteps: `int`, defaults to 1000: The number of diffusion steps to train the model.
solver_order: `int`, default `2`: The UniPC order which can be any positive integer. The effective order of accuracy is solver_order + 1 due to the UniC. It is recommended to use solver_order=2 for guided sampling, and solver_order=3 for unconditional sampling.
prediction_type: `str`, defaults to "flow_prediction": Prediction type of the scheduler function; must be flow_prediction for this scheduler, which predicts the flow of the diffusion process.
thresholding: `bool`, defaults to `False`: Whether to use the "dynamic thresholding" method. This is unsuitable for latent-space diffusion models such as Stable Diffusion.
dynamic_thresholding_ratio: `float`, defaults to 0.995: The ratio for the dynamic thresholding method. Valid only when thresholding=True.
sample_max_value: `float`, defaults to 1.0: The threshold value for dynamic thresholding. Valid only when thresholding=True and predict_x0=True.
predict_x0: `bool`, defaults to `True`: Whether to use the updating algorithm on the predicted x0.
solver_type: `str`, default `bh2`: Solver type for UniPC. It is recommended to use bh1 for unconditional sampling when steps < 10, and bh2 otherwise.
lower_order_final: `bool`, default `True`: Whether to use lower-order solvers in the final steps. Only valid for < 15 inference steps. This can stabilize the sampling of DPMSolver for steps < 15, especially for steps <= 10.
disable_corrector: `list`, default `[]`: Decides which step to disable the corrector to mitigate the misalignment between epsilon_theta(x_t, c) and epsilon_theta(x_t^c, c) which can influence convergence for a large guidance scale. Corrector is usually disabled during the first few steps.
solver_p: `SchedulerMixin`, default `None`: Any other scheduler that if specified, the algorithm becomes solver_p + UniC.
use_karras_sigmas: `bool`, *optional*, defaults to `False`: Whether to use Karras sigmas for step sizes in the noise schedule during the sampling process. If True, the sigmas are determined according to a sequence of noise levels {σi}.
use_exponential_sigmas: `bool`, *optional*, defaults to `False`: Whether to use exponential sigmas for step sizes in the noise schedule during the sampling process.
timestep_spacing: `str`, defaults to `"linspace"`: The way the timesteps should be scaled. Refer to Table 2 of the Common Diffusion Noise Schedules and Sample Steps are Flawed for more information.
steps_offset: `int`, defaults to 0: An offset added to the inference steps, as required by some model families.
final_sigmas_type: `str`, defaults to `"zero"`: The final sigma value for the noise schedule during the sampling process. If "sigma_min", the final sigma is the same as the last sigma in the training schedule. If zero, the final sigma is set to 0.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.FlowUniPCMultistepScheduler.step_index`

step_index

The index counter for current timestep. It will increase 1 after each scheduler step.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.FlowUniPCMultistepScheduler.begin_index`

begin_index

The index for the first timestep. It should be set from pipeline with set_begin_index method.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.FlowUniPCMultistepScheduler.set_shift`

set_shift(shift: float) -> None

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.FlowUniPCMultistepScheduler.set_begin_index`

set_begin_index(begin_index: int = 0)

Sets the begin index for the scheduler. This function should be run from pipeline before the inference.

Parameters

begin_index: `int`: The begin index for the scheduler.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.FlowUniPCMultistepScheduler.set_timesteps`

set_timesteps(num_inference_steps: int | None = None, device: str | torch.device = None, sigmas: list[float] | None = None, mu: float | None | None = None, shift: float | None | None = None, use_karras_sigmas: bool | None = None, use_kerras_sigma: bool | None = None)

Sets the discrete timesteps used for the diffusion chain (to be run before inference). Args: num_inference_steps (int): Total number of the spacing of the time steps. device (str or torch.device, optional): The device to which the timesteps should be moved to. If None, the timesteps are not moved.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.FlowUniPCMultistepScheduler.time_shift`

time_shift(mu: float, sigma: float, t: torch.Tensor)

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.FlowUniPCMultistepScheduler.convert_model_output`

convert_model_output(model_output: torch.Tensor, *args, sample: torch.Tensor = None, **kwargs) -> torch.Tensor

Convert the model output to the corresponding type the UniPC algorithm needs.

Parameters

model_output: `torch.Tensor`: The direct output from the learned diffusion model.
timestep: `int`: The current discrete timestep in the diffusion chain.
sample: `torch.Tensor`: A current instance of a sample created by the diffusion process.

Returns

torch.Tensor: torch.Tensor: The converted model output.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.FlowUniPCMultistepScheduler.multistep_uni_p_bh_update`

multistep_uni_p_bh_update(model_output: torch.Tensor, *args, sample: torch.Tensor = None, order: int | None = None, **kwargs) -> torch.Tensor

One step for the UniP (B(h) version). Alternatively, self.solver_p is used if is specified.

Parameters

model_output: `torch.Tensor`: The direct output from the learned diffusion model at the current timestep.
prev_timestep: `int`: The previous discrete timestep in the diffusion chain.
sample: `torch.Tensor`: A current instance of a sample created by the diffusion process.
order: `int`: The order of UniP at this timestep (corresponds to the p in UniPC-p).

Returns

torch.Tensor: torch.Tensor: The sample tensor at the previous timestep.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.FlowUniPCMultistepScheduler.multistep_uni_c_bh_update`

multistep_uni_c_bh_update(this_model_output: torch.Tensor, *args, last_sample: torch.Tensor = None, this_sample: torch.Tensor = None, order: int | None = None, **kwargs) -> torch.Tensor

One step for the UniC (B(h) version).

Parameters

this_model_output: `torch.Tensor`: The model outputs at x_t.
this_timestep: `int`: The current timestep t.
last_sample: `torch.Tensor`: The generated sample before the last predictor x_{t-1}.
this_sample: `torch.Tensor`: The generated sample after the last predictor x_{t}.
order: `int`: The p of UniC-p at this step. The effective order of accuracy should be order + 1.

Returns

torch.Tensor: torch.Tensor: The corrected sample tensor at the current timestep.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.FlowUniPCMultistepScheduler.index_for_timestep`

index_for_timestep(timestep, schedule_timesteps = None) -> int

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.FlowUniPCMultistepScheduler.step`

step(model_output: torch.Tensor, timestep: int | torch.Tensor, sample: torch.Tensor, return_dict: bool = True, generator = None) -> SchedulerOutput | tuple

Predict the sample from the previous timestep by reversing the SDE. This function propagates the sample with the multistep UniPC.

Parameters

model_output: `torch.Tensor`: The direct output from learned diffusion model.
timestep: `int`: The current discrete timestep in the diffusion chain.
sample: `torch.Tensor`: A current instance of a sample created by the diffusion process.
return_dict: `bool`: Whether or not to return a [~schedulers.scheduling_utils.SchedulerOutput] or tuple.

Returns

SchedulerOutput | tuple: [~schedulers.scheduling_utils.SchedulerOutput] or tuple: If return_dict is True, [~schedulers.scheduling_utils.SchedulerOutput] is returned, otherwise a tuple is returned where the first element is the sample tensor.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.FlowUniPCMultistepScheduler.scale_model_input`

scale_model_input(sample: torch.Tensor, *args, **kwargs) -> torch.Tensor

Ensures interchangeability with schedulers that need to scale the denoising model input depending on the current timestep.

Parameters

sample: `torch.Tensor`: The input sample.

Returns

torch.Tensor: torch.Tensor: A scaled input sample.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.FlowUniPCMultistepScheduler.add_noise`

add_noise(original_samples: torch.Tensor, noise: torch.Tensor, timesteps: torch.IntTensor) -> torch.Tensor

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_pipeline.ComposedPipelineBase.set_trainable`

set_trainable() -> None

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.ComposedPipelineBase.post_init`

post_init() -> None

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.ComposedPipelineBase.initialize_training_pipeline`

initialize_training_pipeline(training_args: TrainingArgs)

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.ComposedPipelineBase.initialize_validation_pipeline`

initialize_validation_pipeline(training_args: TrainingArgs)

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_pipeline.ComposedPipelineBase.get_module`

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

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.ComposedPipelineBase.add_module`

add_module(module_name: str, module: Any)

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_pipeline.ComposedPipelineBase.stages`

stages: list[PipelineStage]

List of stages in the pipeline.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.ComposedPipelineBase.create_pipeline_stages`

create_pipeline_stages(fastvideo_args: FastVideoArgs)

Create the inference pipeline stages.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.ComposedPipelineBase.create_training_stages`

create_training_stages(training_args: TrainingArgs)

Create the training pipeline stages.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.ComposedPipelineBase.initialize_pipeline`

initialize_pipeline(fastvideo_args: FastVideoArgs)

Initialize the pipeline.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_pipeline.ComposedPipelineBase.add_stage`

add_stage(stage_name: str, stage: PipelineStage)

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_pipeline.ComposedPipelineBase.train`

train() -> None

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.ComposedPipelineBase.close`

close() -> None

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.LoRAPipeline`

LoRAPipeline(*args, **kwargs)

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

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.LoRAPipeline.add_module`

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

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.LoRAPipeline.is_target_layer`

is_target_layer(module_name: str) -> bool

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_pipeline.LoRAPipeline.set_trainable`

set_trainable() -> None

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.LoRAPipeline.convert_to_lora_layers`

convert_to_lora_layers() -> None

Unified method to convert the transformer to a LoRA transformer.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_pipeline.ForwardBatch`

ForwardBatch(data_type: str, generator: torch.Generator | list[torch.Generator] | None = None, image_path: str | None = None, image_embeds: list[torch.Tensor] = list(), pil_image: torch.Tensor | PIL.Image.Image | None = None, last_image: torch.Tensor | PIL.Image.Image | None = None, references: list[Any] | None = None, preprocessed_image: torch.Tensor | None = None, prompt: str | list[str] | None = None, negative_prompt: str | list[str] | None = None, prompt_path: str | None = None, output_path: str = 'outputs/', output_video_name: str | None = None, video_path: str | None = None, video_latent: torch.Tensor | None = None, pose_video_path: str | None = None, face_video_path: str | None = None, background_video_path: str | None = None, mask_video_path: str | None = None, animate_mode: str | None = None, pose_latents: torch.Tensor | None = None, face_pixel_values: torch.Tensor | None = None, audio_path: str | None = None, audio_embeds: torch.Tensor | None = None, t_thresh: float = 0.5, num_cond_frames: int = 0, prompt_embeds: list[torch.Tensor] = list(), negative_prompt_embeds: list[torch.Tensor] | None = None, prompt_attention_mask: list[torch.Tensor] | None = None, negative_attention_mask: list[torch.Tensor] | None = None, clip_embedding_pos: list[torch.Tensor] | None = None, clip_embedding_neg: list[torch.Tensor] | None = None, max_sequence_length: int | None = None, prompt_template: dict[str, Any] | None = None, do_classifier_free_guidance: bool = False, use_embedded_guidance: bool = False, true_cfg_scale: float = 1.0, batch_size: int | None = None, num_videos_per_prompt: int = 1, seed: int | None = None, seeds: list[int] | None = None, is_prompt_processed: bool = False, latents: torch.Tensor | None = None, audio_latents: torch.Tensor | None = None, lq_latents: torch.Tensor | None = None, raw_latent_shape: tuple[int, ...] | None = None, noise_pred: torch.Tensor | None = None, image_latent: torch.Tensor | None = None, first_frame_latent: torch.Tensor | None = None, helios_latent_chunks: list[torch.Tensor] | None = None, mouse_cond: torch.Tensor | None = None, keyboard_cond: torch.Tensor | None = None, grid_sizes: torch.Tensor | None = None, num_iterations: int | None = None, use_base_model: bool = False, pose: str | None = None, c2ws_plucker_emb: torch.Tensor | None = None, action_path: str | None = None, trajectory_type: str | None = None, movement_distance: float | None = None, camera_rotation: str | None = None, height_latents: list[int] | int | None = None, width_latents: list[int] | int | None = None, num_frames: list[int] | int = 1, height: list[int] | int | None = None, width: list[int] | int | None = None, height_sr: list[int] | int | None = None, width_sr: list[int] | int | None = None, fps: list[int] | int | None = None, timesteps: torch.Tensor | None = None, timestep: torch.Tensor | float | int | None = None, step_index: int | None = None, boundary_ratio: float | None = None, num_inference_steps: int = 50, num_inference_steps_sr: int = 50, guidance_scale: float = 1.0, batch_cfg: bool = False, guidance_scale_2: float | None = None, guidance_rescale: float = 0.0, eta: float = 0.0, sigmas: list[float] | None = None, pyramid_num_inference_steps_list: list[int] | None = None, history_sizes: list[int] | None = None, num_latent_frames_per_chunk: int = 9, keep_first_frame: bool = True, is_skip_first_chunk: bool = False, use_zero_init: bool = True, zero_steps: int = 1, is_amplify_first_chunk: bool = False, enable_teacache: bool = False, ltx2_cfg_scale_video: float = 1.0, ltx2_cfg_scale_audio: float = 1.0, ltx2_modality_scale_video: float = 1.0, ltx2_modality_scale_audio: float = 1.0, ltx2_rescale_scale: float = 0.0, ltx2_stg_scale_video: float = 0.0, ltx2_stg_scale_audio: float = 0.0, ltx2_stg_blocks_video: list[int] = list(), ltx2_stg_blocks_audio: list[int] = list(), ltx2_use_ancestral_sampler: bool = False, ltx2_images: list[tuple[str, int, float]] | None = None, ltx2_image_crf: float = 33.0, ltx2_conditioning_latent_stage1: torch.Tensor | None = None, ltx2_conditioning_latent_stage2: torch.Tensor | None = None, ltx2_video_conditions: list[tuple[list[str], int, float]] | None = None, audio_start_in_s: float | None = None, audio_end_in_s: float | None = None, init_audio: Any = None, init_audio_strength: float | None = None, init_noise_level: float | None = None, inpaint_audio: Any = None, inpaint_mask: Any = None, n_tokens: int | None = None, extra_step_kwargs: dict[str, Any] = dict(), modules: dict[str, Any] = dict(), output: torch.Tensor | None = None, return_trajectory_latents: bool = False, return_trajectory_decoded: bool = False, trajectory_timesteps: list[torch.Tensor] | None = None, trajectory_latents: torch.Tensor | None = None, trajectory_decoded: list[torch.Tensor] | None = None, continuation_state: ContinuationState | None = None, return_continuation_state: bool = False, extra: dict[str, Any] = dict(), save_video: bool = True, return_frames: bool = False, is_cfg_negative: bool = False, VSA_sparsity: float = 0.0, logging_info: PipelineLoggingInfo = PipelineLoggingInfo(), *, return_samples: bool = False)

Complete state passed through the pipeline execution.

This dataclass contains all information needed during the diffusion pipeline execution, allowing methods to update specific components without needing to manage numerous individual parameters.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_pipeline.ConditioningStage.verify_input`

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

Verify conditioning stage inputs.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.ConditioningStage.verify_output`

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

Verify conditioning stage outputs.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_pipeline.DecodingStage.verify_input`

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

Verify decoding stage inputs.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.DecodingStage.verify_output`

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

Verify decoding stage outputs.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_pipeline.InputValidationStage`

InputValidationStage()

Stage for validating and preparing inputs for diffusion pipelines.

This stage validates that all required inputs are present and properly formatted before proceeding with the diffusion process.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.InputValidationStage.forward`

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

Validate and prepare inputs.

Parameters

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

Returns

ForwardBatch: The validated batch information.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.InputValidationStage.verify_input`

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

Verify input validation stage inputs.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.InputValidationStage.verify_output`

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

Verify input validation stage outputs.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_pipeline.TextEncodingStage.verify_input`

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

Verify text encoding stage inputs.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_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.lingbotworld2.causal_fast_pipeline.TextEncodingStage.verify_output`

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

Verify text encoding stage outputs.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.PipelineStage`

PipelineStage()

Abstract base class for all pipeline stages.

A pipeline stage represents a discrete step in the diffusion process that can be composed with other stages to create a complete pipeline. Each stage is responsible for a specific part of the process, such as prompt encoding, latent preparation, etc.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.PipelineStage.verify_input`

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

Verify the input for the stage.

Example

from fastvideo.pipelines.stages.validators import V, VerificationResult

def verify_input(self, batch, fastvideo_args): result = VerificationResult() result.add_check("height", batch.height, V.positive_int_divisible(8)) result.add_check("width", batch.width, V.positive_int_divisible(8)) result.add_check("image_latent", batch.image_latent, V.is_tensor) return result

Parameters

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

Returns

VerificationResult: A VerificationResult containing the verification status.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.PipelineStage.verify_output`

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

Verify the output for the stage.

Parameters

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

Returns

VerificationResult: A VerificationResult containing the verification status.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.PipelineStage.device`

device: torch.device

Get the device for this stage.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.PipelineStage.set_logging`

set_logging(enable: bool)

Enable or disable logging for this stage.

Parameters

enable: bool: Whether to enable logging.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.PipelineStage.forward`

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

Forward pass of the stage's processing.

This method should be implemented by subclasses to provide the forward processing logic for the stage.

Parameters

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

Returns

ForwardBatch: The updated batch information after this stage's processing.

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.LingBotWorld2TextEncodingStage`

LingBotWorld2TextEncodingStage(text_encoders, tokenizers)

Keep LingBot World 2 T5 attention masks so DiT context matches the source pipeline.

Source

class LingBotWorld2TextEncodingStage(TextEncodingStage):
    """Keep LingBot World 2 T5 attention masks so DiT context matches the source pipeline."""

    def forward(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch:
        """Initialize mask storage before running the shared text-encoding stage."""
        if batch.prompt_attention_mask is None:
            batch.prompt_attention_mask = []
        return super().forward(batch, fastvideo_args)

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.LingBotWorld2TextEncodingStage.forward`

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

Initialize mask storage before running the shared text-encoding stage.

Source

def forward(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch:
    """Initialize mask storage before running the shared text-encoding stage."""
    if batch.prompt_attention_mask is None:
        batch.prompt_attention_mask = []
    return super().forward(batch, fastvideo_args)

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.LingBotWorld2CausalFastGenerationStage`

LingBotWorld2CausalFastGenerationStage(transformer, scheduler, vae)

Prepare LingBot World 2 conditions and run the released causal-fast sampling loop.

Source

class LingBotWorld2CausalFastGenerationStage(PipelineStage):
    """Prepare LingBot World 2 conditions and run the released causal-fast sampling loop."""

    def __init__(self, transformer, scheduler, vae) -> None:
        super().__init__()
        self.transformer = transformer
        self.scheduler = scheduler
        self.vae = vae
        self._cross_attn_initialized = False

    def _convert_flow_pred_to_x0(
        self,
        flow_pred: torch.Tensor,
        xt: torch.Tensor,
        timestep: torch.Tensor,
    ) -> torch.Tensor:
        """Convert LingBot World 2 flow prediction to x0 using the scheduler sigma."""
        original_dtype = flow_pred.dtype
        flow_pred, xt, sigmas, timesteps = map(
            lambda x: x.double().to(flow_pred.device),
            [flow_pred, xt, self.scheduler.sigmas, self.scheduler.timesteps],
        )
        timestep_id = torch.argmin((timesteps - timestep).abs())
        sigma_t = sigmas[timestep_id].reshape(-1, 1, 1, 1)
        return (xt - sigma_t * flow_pred).to(original_dtype)

    def _initialize_self_kv_cache(
        self,
        batch_size: int,
        kv_size: int,
        dtype: torch.dtype,
        device: torch.device,
    ) -> list[dict]:
        """Allocate per-block self-attention KV cache tensors."""
        head_dim = self.transformer.dim // self.transformer.num_heads
        num_heads = self.transformer.num_heads // get_sp_world_size()
        shape = [batch_size, kv_size, num_heads, head_dim]
        return [{
            "k": torch.zeros(shape, dtype=dtype, device=device),
            "v": torch.zeros(shape, dtype=dtype, device=device),
            "global_end_index": torch.tensor([0], dtype=torch.long, device=device),
            "local_end_index": torch.tensor([0], dtype=torch.long, device=device),
        } for _ in range(self.transformer.num_layers)]

    def _initialize_crossattn_cache(
        self,
        batch_size: int,
        max_sequence_length: int,
        dtype: torch.dtype,
        device: torch.device,
    ) -> list[dict]:
        """Allocate per-block text cross-attention KV cache tensors."""
        head_dim = self.transformer.dim // self.transformer.num_heads
        shape = [batch_size, max_sequence_length, self.transformer.num_heads, head_dim]
        return [{
            "k": torch.zeros(shape, dtype=dtype, device=device),
            "v": torch.zeros(shape, dtype=dtype, device=device),
            "is_init": torch.tensor([0], dtype=torch.bool, device=device),
        } for _ in range(self.transformer.num_layers)]

    @staticmethod
    def _prompt_context(batch: ForwardBatch, device: torch.device) -> list[torch.Tensor]:
        """Slice padded text encoder states back to LingBot World 2's unpadded context list."""
        assert batch.prompt_embeds
        context_tensor = batch.prompt_embeds[0].to(device)
        if batch.prompt_attention_mask:
            mask = batch.prompt_attention_mask[0].to(device)
            seq_lens = mask.gt(0).sum(dim=1).long()
            return [u[:v] for u, v in zip(context_tensor, seq_lens, strict=True)]
        return [u for u in context_tensor]

    def _prepare_image_tensor(self, batch: ForwardBatch, device: torch.device) -> torch.Tensor:
        """Return the source-style normalized image tensor `[C,H,W]`."""
        image = batch.pil_image
        if image is None:
            raise ValueError("LingBot World 2 causal-fast requires `image_path` or `pil_image`.")
        if isinstance(image, torch.Tensor):
            if image.ndim == 5:
                return image[0, :, 0].to(device)
            if image.ndim == 4:
                return image[0].to(device)
            return image.to(device)
        return TF.to_tensor(image).sub_(0.5).div_(0.5).to(device)

    def _prepare_camera(
        self,
        action_path: str,
        c2ws: np.ndarray,
        h: int,
        w: int,
        lat_f: int,
        lat_h: int,
        lat_w: int,
        chunk_size: int,
        dtype: torch.dtype,
        device: torch.device,
    ) -> torch.Tensor:
        """Build the LingBot World 2 camera Plucker tensor for latent chunks."""
        Ks = torch.from_numpy(np.load(os.path.join(action_path, "intrinsics.npy"))).float()
        Ks = get_Ks_transformed(
            Ks,
            height_org=480,
            width_org=832,
            height_resize=h,
            width_resize=w,
            height_final=h,
            width_final=w,
        )
        Ks = Ks[0]
        len_c2ws = len(c2ws)
        len_c2ws_ = int((len_c2ws - 1) // 4) + 1
        len_c2ws_ = int(len_c2ws_ - (len_c2ws_ % chunk_size))
        c2ws_infer = interpolate_camera_poses(
            src_indices=np.linspace(0, len_c2ws - 1, len_c2ws),
            src_rot_mat=c2ws[:, :3, :3],
            src_trans_vec=c2ws[:, :3, 3],
            tgt_indices=np.linspace(0, len_c2ws - 1, len_c2ws_),
        )
        c2ws_infer = compute_relative_poses(c2ws_infer, framewise=True)
        Ks = Ks.repeat(len(c2ws_infer), 1)
        c2ws_plucker_emb = get_plucker_embeddings(c2ws_infer.to(device), Ks.to(device), h, w)
        c2ws_plucker_emb = rearrange(
            c2ws_plucker_emb,
            "f (h c1) (w c2) c -> (f h w) (c c1 c2)",
            c1=int(h // lat_h),
            c2=int(w // lat_w),
        )
        c2ws_plucker_emb = c2ws_plucker_emb[None, ...]
        return rearrange(
            c2ws_plucker_emb,
            "b (f h w) c -> b c f h w",
            f=lat_f,
            h=lat_h,
            w=lat_w,
        ).to(device=device, dtype=dtype)

    def _encode_condition_video(
        self,
        img: torch.Tensor,
        h: int,
        w: int,
        frames: int,
        mask: torch.Tensor,
        fastvideo_args: FastVideoArgs,
    ) -> torch.Tensor:
        """Encode the first-frame conditioning video and prepend mask channels."""
        device = get_local_torch_device()
        self.vae = self.vae.to(device)
        video_condition = torch.concat(
            [
                torch.nn.functional.interpolate(
                    img[None].cpu(),
                    size=(h, w),
                    mode="bicubic",
                ).transpose(0, 1),
                torch.zeros(3, frames - 1, h, w),
            ],
            dim=1,
        ).to(device)
        vae_dtype = torch.float32
        with torch.autocast(device_type="cuda", dtype=vae_dtype, enabled=False):
            encoder_output = self.vae.encode(video_condition.unsqueeze(0).to(torch.float32))
        latent_condition = encoder_output.mean
        if not bool(getattr(self.vae, "handles_latent_denorm", False)):
            if hasattr(self.vae, "shift_factor") and self.vae.shift_factor is not None:
                latent_condition -= self.vae.shift_factor.to(latent_condition.device, latent_condition.dtype)
            latent_condition = latent_condition * self.vae.scaling_factor.to(latent_condition.device,
                                                                             latent_condition.dtype)
        if fastvideo_args.vae_cpu_offload:
            self.vae.to("cpu")
        return torch.concat([mask, latent_condition[0]], dim=0)

    @torch.no_grad()
    def forward(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch:
        """Execute LingBot World 2 causal-fast generation and store final latents on the batch."""
        device = get_local_torch_device()
        cfg = fastvideo_args.pipeline_config.dit_config.arch_config
        chunk_size = int(cfg.chunk_size)
        max_sequence_length = int(batch.max_sequence_length or cfg.text_len)
        action_path = batch.action_path
        if action_path is None:
            raise ValueError("LingBot World 2 causal-fast requires `action_path`.")

        c2ws = np.load(os.path.join(action_path, "poses.npy"))
        len_c2ws = ((len(c2ws) - 1) // 4) * 4 + 1
        frame_num = ((int(batch.num_frames) - 1) // 4) * 4 + 1
        frame_num = min(frame_num, len_c2ws)
        c2ws = c2ws[:frame_num]

        img = self._prepare_image_tensor(batch, device)
        h0, w0 = img.shape[1:]
        aspect_ratio = h0 / w0
        lat_h = round(np.sqrt(cfg.max_area * aspect_ratio) // 8 // cfg.patch_size[1] * cfg.patch_size[1])
        lat_w = round(np.sqrt(cfg.max_area / aspect_ratio) // 8 // cfg.patch_size[2] * cfg.patch_size[2])
        h = lat_h * 8
        w = lat_w * 8
        lat_f = (frame_num - 1) // 4 + 1
        lat_f = int(lat_f - (lat_f % chunk_size))
        frames = (lat_f - 1) * 4 + 1
        batch.height = h
        batch.width = w
        batch.num_frames = frames

        seed = int(batch.seed if batch.seed is not None else 42)
        seed_g = torch.Generator(device=device)
        seed_g.manual_seed(seed)
        noise = torch.randn(16, lat_f, lat_h, lat_w, dtype=torch.float32, generator=seed_g, device=device)

        mask = torch.ones(1, frames, lat_h, lat_w, device=device)
        mask[:, 1:] = 0
        mask = torch.concat([torch.repeat_interleave(mask[:, 0:1], repeats=4, dim=1), mask[:, 1:]], dim=1)
        mask = mask.view(1, mask.shape[1] // 4, 4, lat_h, lat_w).transpose(1, 2)[0]

        self.scheduler.set_timesteps(cfg.num_train_timesteps, shift=cfg.sample_shift)
        timesteps = self.scheduler.timesteps[list(cfg.timesteps_index)].to(device)
        context = self._prompt_context(batch, device)
        c2ws_plucker_emb = self._prepare_camera(
            action_path,
            c2ws,
            h,
            w,
            lat_f,
            lat_h,
            lat_w,
            chunk_size,
            torch.bfloat16,
            device,
        )
        y = self._encode_condition_video(img, h, w, frames, mask, fastvideo_args).to(device=device,
                                                                                     dtype=torch.bfloat16)

        transformer_dtype = torch.bfloat16
        frame_seqlen = int(noise.shape[-2] * noise.shape[-1] // 4)
        kv_size = frame_seqlen * cfg.local_attn_size if cfg.local_attn_size > -1 else frame_seqlen * lat_f
        self_kv_cache = self._initialize_self_kv_cache(1, kv_size, transformer_dtype, device)
        cross_kv_cache = self._initialize_crossattn_cache(1, max_sequence_length, transformer_dtype, device)

        if not fastvideo_args.use_fsdp_inference:
            self.transformer = self.transformer.to(device)
        self._cross_attn_initialized = False
        pred_latent_chunks = []
        latents_chunk = noise.split(chunk_size, dim=1)
        condition_chunk = y.split(chunk_size, dim=1)
        c2ws_plucker_emb_chunk = c2ws_plucker_emb.split(chunk_size, dim=2)
        max_seq_len = int(math.ceil(chunk_size * lat_h * lat_w // 4))

        with torch.amp.autocast("cuda", dtype=transformer_dtype):
            for chunk_id, current_latent in enumerate(latents_chunk):
                current_condition = condition_chunk[chunk_id]
                current_c2ws_plucker_emb = c2ws_plucker_emb_chunk[chunk_id]
                dit_cond_dict = {"c2ws_plucker_emb": current_c2ws_plucker_emb.chunk(1, dim=0)}
                kwargs = {
                    "context": [context[0]],
                    "seq_len": max_seq_len,
                    "y": [current_condition],
                    "dit_cond_dict": dit_cond_dict,
                    "kv_cache": self_kv_cache,
                    "crossattn_cache": cross_kv_cache,
                    "current_start": chunk_id * chunk_size * frame_seqlen,
                    "max_attention_size": kv_size,
                    "frame_seqlen": frame_seqlen,
                }
                x0 = current_latent
                for timestep_idx, timestep_value in enumerate(timesteps):
                    # Stop if interrupted
                    if getattr(self, "interrupt", False):
                        break
                    timestep = torch.stack([timestep_value]).to(device)
                    noise_pred = self.transformer(
                        x=[current_latent.to(device)],
                        t=timestep,
                        cross_attn_first_call=not self._cross_attn_initialized,
                        **kwargs,
                    )[0]
                    self._cross_attn_initialized = True
                    x0 = self._convert_flow_pred_to_x0(noise_pred, current_latent, timestep_value)
                    if timestep_idx < len(timesteps) - 1:
                        next_timestep = timesteps[timestep_idx + 1].reshape(1)
                        current_latent = self.scheduler.add_noise(
                            x0,
                            torch.randn(x0.shape, generator=seed_g, device=x0.device, dtype=x0.dtype),
                            next_timestep,
                        )
                pred_latent_chunks.append(x0)
                context_timestep = torch.stack([timesteps[-1] * 0.0]).to(device)
                self.transformer(x=[x0], t=context_timestep, cross_attn_first_call=False, **kwargs)

        batch.latents = torch.cat(pred_latent_chunks, dim=1).unsqueeze(0)
        return batch

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.LingBotWorld2CausalFastGenerationStage.forward`

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

Execute LingBot World 2 causal-fast generation and store final latents on the batch.

Source

@torch.no_grad()
def forward(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch:
    """Execute LingBot World 2 causal-fast generation and store final latents on the batch."""
    device = get_local_torch_device()
    cfg = fastvideo_args.pipeline_config.dit_config.arch_config
    chunk_size = int(cfg.chunk_size)
    max_sequence_length = int(batch.max_sequence_length or cfg.text_len)
    action_path = batch.action_path
    if action_path is None:
        raise ValueError("LingBot World 2 causal-fast requires `action_path`.")

    c2ws = np.load(os.path.join(action_path, "poses.npy"))
    len_c2ws = ((len(c2ws) - 1) // 4) * 4 + 1
    frame_num = ((int(batch.num_frames) - 1) // 4) * 4 + 1
    frame_num = min(frame_num, len_c2ws)
    c2ws = c2ws[:frame_num]

    img = self._prepare_image_tensor(batch, device)
    h0, w0 = img.shape[1:]
    aspect_ratio = h0 / w0
    lat_h = round(np.sqrt(cfg.max_area * aspect_ratio) // 8 // cfg.patch_size[1] * cfg.patch_size[1])
    lat_w = round(np.sqrt(cfg.max_area / aspect_ratio) // 8 // cfg.patch_size[2] * cfg.patch_size[2])
    h = lat_h * 8
    w = lat_w * 8
    lat_f = (frame_num - 1) // 4 + 1
    lat_f = int(lat_f - (lat_f % chunk_size))
    frames = (lat_f - 1) * 4 + 1
    batch.height = h
    batch.width = w
    batch.num_frames = frames

    seed = int(batch.seed if batch.seed is not None else 42)
    seed_g = torch.Generator(device=device)
    seed_g.manual_seed(seed)
    noise = torch.randn(16, lat_f, lat_h, lat_w, dtype=torch.float32, generator=seed_g, device=device)

    mask = torch.ones(1, frames, lat_h, lat_w, device=device)
    mask[:, 1:] = 0
    mask = torch.concat([torch.repeat_interleave(mask[:, 0:1], repeats=4, dim=1), mask[:, 1:]], dim=1)
    mask = mask.view(1, mask.shape[1] // 4, 4, lat_h, lat_w).transpose(1, 2)[0]

    self.scheduler.set_timesteps(cfg.num_train_timesteps, shift=cfg.sample_shift)
    timesteps = self.scheduler.timesteps[list(cfg.timesteps_index)].to(device)
    context = self._prompt_context(batch, device)
    c2ws_plucker_emb = self._prepare_camera(
        action_path,
        c2ws,
        h,
        w,
        lat_f,
        lat_h,
        lat_w,
        chunk_size,
        torch.bfloat16,
        device,
    )
    y = self._encode_condition_video(img, h, w, frames, mask, fastvideo_args).to(device=device,
                                                                                 dtype=torch.bfloat16)

    transformer_dtype = torch.bfloat16
    frame_seqlen = int(noise.shape[-2] * noise.shape[-1] // 4)
    kv_size = frame_seqlen * cfg.local_attn_size if cfg.local_attn_size > -1 else frame_seqlen * lat_f
    self_kv_cache = self._initialize_self_kv_cache(1, kv_size, transformer_dtype, device)
    cross_kv_cache = self._initialize_crossattn_cache(1, max_sequence_length, transformer_dtype, device)

    if not fastvideo_args.use_fsdp_inference:
        self.transformer = self.transformer.to(device)
    self._cross_attn_initialized = False
    pred_latent_chunks = []
    latents_chunk = noise.split(chunk_size, dim=1)
    condition_chunk = y.split(chunk_size, dim=1)
    c2ws_plucker_emb_chunk = c2ws_plucker_emb.split(chunk_size, dim=2)
    max_seq_len = int(math.ceil(chunk_size * lat_h * lat_w // 4))

    with torch.amp.autocast("cuda", dtype=transformer_dtype):
        for chunk_id, current_latent in enumerate(latents_chunk):
            current_condition = condition_chunk[chunk_id]
            current_c2ws_plucker_emb = c2ws_plucker_emb_chunk[chunk_id]
            dit_cond_dict = {"c2ws_plucker_emb": current_c2ws_plucker_emb.chunk(1, dim=0)}
            kwargs = {
                "context": [context[0]],
                "seq_len": max_seq_len,
                "y": [current_condition],
                "dit_cond_dict": dit_cond_dict,
                "kv_cache": self_kv_cache,
                "crossattn_cache": cross_kv_cache,
                "current_start": chunk_id * chunk_size * frame_seqlen,
                "max_attention_size": kv_size,
                "frame_seqlen": frame_seqlen,
            }
            x0 = current_latent
            for timestep_idx, timestep_value in enumerate(timesteps):
                # Stop if interrupted
                if getattr(self, "interrupt", False):
                    break
                timestep = torch.stack([timestep_value]).to(device)
                noise_pred = self.transformer(
                    x=[current_latent.to(device)],
                    t=timestep,
                    cross_attn_first_call=not self._cross_attn_initialized,
                    **kwargs,
                )[0]
                self._cross_attn_initialized = True
                x0 = self._convert_flow_pred_to_x0(noise_pred, current_latent, timestep_value)
                if timestep_idx < len(timesteps) - 1:
                    next_timestep = timesteps[timestep_idx + 1].reshape(1)
                    current_latent = self.scheduler.add_noise(
                        x0,
                        torch.randn(x0.shape, generator=seed_g, device=x0.device, dtype=x0.dtype),
                        next_timestep,
                    )
            pred_latent_chunks.append(x0)
            context_timestep = torch.stack([timesteps[-1] * 0.0]).to(device)
            self.transformer(x=[x0], t=context_timestep, cross_attn_first_call=False, **kwargs)

    batch.latents = torch.cat(pred_latent_chunks, dim=1).unsqueeze(0)
    return batch

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.LingBotWorld2CausalFastPipeline`

LingBotWorld2CausalFastPipeline(*args, **kwargs)

FastVideo pipeline for LingBot World 2 14B causal-fast I2V generation.

Source

class LingBotWorld2CausalFastPipeline(LoRAPipeline, ComposedPipelineBase):
    """FastVideo pipeline for LingBot World 2 14B causal-fast I2V generation."""

    _required_config_modules = ["text_encoder", "tokenizer", "vae", "transformer", "scheduler"]

    def initialize_pipeline(self, fastvideo_args: FastVideoArgs):
        if "scheduler" not in self.modules:
            self.modules["scheduler"] = FlowUniPCMultistepScheduler(shift=1.0, use_dynamic_shifting=False)

    def create_pipeline_stages(self, fastvideo_args: FastVideoArgs) -> None:
        """Set up the LingBot World 2 causal-fast pipeline stages."""
        self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
        self.add_stage(
            stage_name="prompt_encoding_stage",
            stage=LingBotWorld2TextEncodingStage(
                text_encoders=[self.get_module("text_encoder")],
                tokenizers=[self.get_module("tokenizer")],
            ),
        )
        self.add_stage(stage_name="conditioning_stage", stage=ConditioningStage())
        self.add_stage(
            stage_name="lingbotworld2_causal_fast_generation_stage",
            stage=LingBotWorld2CausalFastGenerationStage(
                transformer=self.get_module("transformer"),
                scheduler=self.get_module("scheduler"),
                vae=self.get_module("vae"),
            ),
        )
        self.add_stage(stage_name="decoding_stage", stage=DecodingStage(vae=self.get_module("vae")))

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.LingBotWorld2CausalFastPipeline.initialize_pipeline`

initialize_pipeline(fastvideo_args: FastVideoArgs)

Source

def initialize_pipeline(self, fastvideo_args: FastVideoArgs):
    if "scheduler" not in self.modules:
        self.modules["scheduler"] = FlowUniPCMultistepScheduler(shift=1.0, use_dynamic_shifting=False)

`fastvideo.pipelines.basic.lingbotworld2.causal_fast_pipeline.LingBotWorld2CausalFastPipeline.create_pipeline_stages`

create_pipeline_stages(fastvideo_args: FastVideoArgs) -> None

Set up the LingBot World 2 causal-fast pipeline stages.

Source

def create_pipeline_stages(self, fastvideo_args: FastVideoArgs) -> None:
    """Set up the LingBot World 2 causal-fast pipeline stages."""
    self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
    self.add_stage(
        stage_name="prompt_encoding_stage",
        stage=LingBotWorld2TextEncodingStage(
            text_encoders=[self.get_module("text_encoder")],
            tokenizers=[self.get_module("tokenizer")],
        ),
    )
    self.add_stage(stage_name="conditioning_stage", stage=ConditioningStage())
    self.add_stage(
        stage_name="lingbotworld2_causal_fast_generation_stage",
        stage=LingBotWorld2CausalFastGenerationStage(
            transformer=self.get_module("transformer"),
            scheduler=self.get_module("scheduler"),
            vae=self.get_module("vae"),
        ),
    )
    self.add_stage(stage_name="decoding_stage", stage=DecodingStage(vae=self.get_module("vae")))