--- title: stages ---

LingBot-Video stages whose contracts differ from shared Wan behavior.

fastvideo.pipelines.basic.lingbot_video.stages.get_local_torch_device

get_local_torch_device() -> torch.device

Return the torch device for the current rank.

fastvideo.pipelines.basic.lingbot_video.stages.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.stages.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.stages.FastVideoArgs.add_cli_args

add_cli_args(parser: FlexibleArgumentParser) -> FlexibleArgumentParser

fastvideo.pipelines.basic.lingbot_video.stages.FastVideoArgs.from_cli_args

from_cli_args(args: argparse.Namespace) -> FastVideoArgs

fastvideo.pipelines.basic.lingbot_video.stages.FastVideoArgs.from_kwargs

from_kwargs(**kwargs: Any) -> FastVideoArgs

fastvideo.pipelines.basic.lingbot_video.stages.FastVideoArgs.check_fastvideo_args

check_fastvideo_args() -> None

Validate inference arguments for consistency

fastvideo.pipelines.basic.lingbot_video.stages.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.stages.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.stages.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.lingbot_video.stages.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.lingbot_video.stages.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.lingbot_video.stages.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.lingbot_video.stages.PipelineStage.device

device: torch.device

Get the device for this stage.

fastvideo.pipelines.basic.lingbot_video.stages.PipelineStage.set_logging

set_logging(enable: bool)

Enable or disable logging for this stage.

Parameters

enable: bool

Whether to enable logging.

fastvideo.pipelines.basic.lingbot_video.stages.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.lingbot_video.stages.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.lingbot_video.stages.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.lingbot_video.stages.InputValidationStage.verify_input

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

Verify input validation stage inputs.

fastvideo.pipelines.basic.lingbot_video.stages.InputValidationStage.verify_output

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

Verify input validation stage outputs.

fastvideo.pipelines.basic.lingbot_video.stages.LingBotVideoInputValidationStage

LingBotVideoInputValidationStage(refiner_enabled: bool = False)

Validate released shape constraints and construct the official CUDA RNG.

Source
class LingBotVideoInputValidationStage(InputValidationStage):
    """Validate released shape constraints and construct the official CUDA RNG."""

    def __init__(self, refiner_enabled: bool = False) -> None:
        self.refiner_enabled = refiner_enabled

    def _generate_seeds(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> None:
        """Use one device-local generator, matching the official production runner."""
        del fastvideo_args
        if batch.seed is None:
            raise ValueError("LingBot-Video requires a seed")
        if batch.num_videos_per_prompt != 1:
            raise ValueError("LingBot-Video currently supports one video per prompt")
        batch.seeds = [batch.seed]
        batch.generator = torch.Generator(device=get_local_torch_device()).manual_seed(batch.seed)

    def forward(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch:
        """Run shared validation, then enforce LingBot temporal and spatial geometry."""
        batch = super().forward(batch, fastvideo_args)
        if not isinstance(batch.num_frames, int):
            raise TypeError("LingBot-Video num_frames must be an integer")
        if batch.num_frames != 1 and (batch.num_frames - 1) % 4 != 0:
            raise ValueError(f"num_frames must be 1 or 4n+1, got {batch.num_frames}")
        if not isinstance(batch.height, int) or not isinstance(batch.width, int):
            raise TypeError("LingBot-Video height and width must be integers")
        if batch.height % 16 != 0 or batch.width % 16 != 0:
            raise ValueError(f"height and width must be divisible by 16, got {batch.height}x{batch.width}")
        if isinstance(batch.prompt, list) and len(batch.prompt) != 1:
            raise ValueError("LingBot-Video currently supports prompt batch size one")
        if self.refiner_enabled and fastvideo_args.output_type == "latent":
            raise ValueError("LingBot-Video refinement requires decoded pixel output")
        return batch

fastvideo.pipelines.basic.lingbot_video.stages.LingBotVideoInputValidationStage.forward

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

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

Source
def forward(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch:
    """Run shared validation, then enforce LingBot temporal and spatial geometry."""
    batch = super().forward(batch, fastvideo_args)
    if not isinstance(batch.num_frames, int):
        raise TypeError("LingBot-Video num_frames must be an integer")
    if batch.num_frames != 1 and (batch.num_frames - 1) % 4 != 0:
        raise ValueError(f"num_frames must be 1 or 4n+1, got {batch.num_frames}")
    if not isinstance(batch.height, int) or not isinstance(batch.width, int):
        raise TypeError("LingBot-Video height and width must be integers")
    if batch.height % 16 != 0 or batch.width % 16 != 0:
        raise ValueError(f"height and width must be divisible by 16, got {batch.height}x{batch.width}")
    if isinstance(batch.prompt, list) and len(batch.prompt) != 1:
        raise ValueError("LingBot-Video currently supports prompt batch size one")
    if self.refiner_enabled and fastvideo_args.output_type == "latent":
        raise ValueError("LingBot-Video refinement requires decoded pixel output")
    return batch

fastvideo.pipelines.basic.lingbot_video.stages.LingBotVideoLatentPreparationStage

LingBotVideoLatentPreparationStage(transformer)

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

Source
class LingBotVideoLatentPreparationStage(PipelineStage):
    """Prepare fp32 latents in the released 4x temporal and 8x spatial geometry."""

    def __init__(self, transformer) -> None:
        self.transformer = transformer

    def forward(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch:
        """Generate or validate one normalized fp32 latent video."""
        del fastvideo_args
        if not all(isinstance(value, int) for value in (batch.num_frames, batch.height, batch.width)):
            raise TypeError("latent geometry must contain integer frames, height, and width")
        shape = (
            1,
            self.transformer.num_channels_latents,
            (batch.num_frames - 1) // 4 + 1,
            batch.height // 8,
            batch.width // 8,
        )
        device = get_local_torch_device()
        if batch.latents is None:
            batch.latents = torch.randn(
                shape,
                generator=batch.generator,
                device=device,
                dtype=torch.float32,
            )
        else:
            if tuple(batch.latents.shape) != shape:
                raise ValueError(f"supplied latent shape {tuple(batch.latents.shape)} does not match {shape}")
            batch.latents = batch.latents.to(device=device, dtype=torch.float32)
        batch.raw_latent_shape = shape
        return batch

fastvideo.pipelines.basic.lingbot_video.stages.LingBotVideoLatentPreparationStage.forward

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

Generate or validate one normalized fp32 latent video.

Source
def forward(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch:
    """Generate or validate one normalized fp32 latent video."""
    del fastvideo_args
    if not all(isinstance(value, int) for value in (batch.num_frames, batch.height, batch.width)):
        raise TypeError("latent geometry must contain integer frames, height, and width")
    shape = (
        1,
        self.transformer.num_channels_latents,
        (batch.num_frames - 1) // 4 + 1,
        batch.height // 8,
        batch.width // 8,
    )
    device = get_local_torch_device()
    if batch.latents is None:
        batch.latents = torch.randn(
            shape,
            generator=batch.generator,
            device=device,
            dtype=torch.float32,
        )
    else:
        if tuple(batch.latents.shape) != shape:
            raise ValueError(f"supplied latent shape {tuple(batch.latents.shape)} does not match {shape}")
        batch.latents = batch.latents.to(device=device, dtype=torch.float32)
    batch.raw_latent_shape = shape
    return batch

fastvideo.pipelines.basic.lingbot_video.stages.LingBotVideoRefinerPreparationStage

LingBotVideoRefinerPreparationStage(vae, scheduler)

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

Source
class LingBotVideoRefinerPreparationStage(PipelineStage):
    """Resize and encode the base video, then initialize the released refiner state."""

    performance_component_metric = "vae_encode_time_s"

    def __init__(self, vae, scheduler) -> None:
        self.vae = vae
        self.scheduler = scheduler

    @staticmethod
    def _resize_video(video: torch.Tensor, height: int, width: int) -> torch.Tensor:
        """Bicubic-resize every decoded frame using the released tensor layout."""
        batch, channels, frames, source_height, source_width = video.shape
        flat = video.permute(0, 2, 1, 3, 4).reshape(batch * frames, channels, source_height, source_width)
        resized = F.interpolate(flat, size=(height, width), mode="bicubic", align_corners=False).clamp(0.0, 1.0)
        return resized.reshape(batch, frames, channels, height, width).permute(0, 2, 1, 3, 4).contiguous()

    def _encode_video(
        self,
        video: torch.Tensor,
        generator: torch.Generator,
        device: torch.device,
    ) -> torch.Tensor:
        """Encode `[0,1]` pixels and convert Wan VAE latents to normalized DiT space."""
        video = video.to(device=device, dtype=torch.float32).mul(2.0).sub(1.0)
        with torch.autocast(device_type=device.type, dtype=torch.bfloat16, enabled=device.type == "cuda"):
            encoded = self.vae.encode(video)
        if hasattr(encoded, "latent_dist"):
            latents = encoded.latent_dist.sample(generator)
        elif hasattr(encoded, "sample") and callable(encoded.sample):
            latents = encoded.sample(generator)
        elif isinstance(encoded, tuple | list):
            latents = encoded[0]
        else:
            latents = encoded
        mean = torch.tensor(self.vae.config.latents_mean, device=device, dtype=torch.float32).view(1, -1, 1, 1, 1)
        std = torch.tensor(self.vae.config.latents_std, device=device, dtype=torch.float32).view(1, -1, 1, 1, 1)
        return ((latents.float() - mean) / std).to(latents)

    def forward(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch:
        """Prepare high-resolution refiner latents and its exact truncated sigma schedule."""
        if batch.output is None or batch.output.ndim != 5:
            raise ValueError("LingBot-Video refinement requires a decoded base video")
        if not isinstance(batch.height_sr, int) or not isinstance(batch.width_sr, int):
            raise TypeError("LingBot-Video refinement requires integer height_sr and width_sr")
        if batch.height_sr % 16 != 0 or batch.width_sr % 16 != 0:
            raise ValueError("LingBot-Video refiner height_sr and width_sr must be divisible by 16")
        if batch.seed is None:
            raise ValueError("LingBot-Video refinement requires a seed")
        device = get_local_torch_device()
        if isinstance(self.vae, torch.nn.Module):
            self.vae.to(device)
        generator = torch.Generator(device=device).manual_seed(batch.seed)
        resized = self._resize_video(batch.output, batch.height_sr, batch.width_sr)
        encoded = self._encode_video(resized, generator, device)
        noise = torch.randn(encoded.shape, generator=generator, device=device, dtype=encoded.dtype)
        batch.latents = ((1.0 - batch.t_thresh) * encoded + batch.t_thresh * noise).float()
        batch.generator = generator
        batch.height = batch.height_sr
        batch.width = batch.width_sr
        batch.raw_latent_shape = tuple(batch.latents.shape)
        batch.extra["lingbot_video_base_shape"] = tuple(batch.output.shape)
        batch.output = None

        shift = fastvideo_args.pipeline_config.flow_shift
        if shift is None:
            raise ValueError("LingBot-Video refinement requires a flow shift")
        sigmas = _compute_refiner_sigmas(
            float(self.scheduler.sigma_max),
            float(self.scheduler.sigma_min),
            batch.num_inference_steps_sr,
            float(shift),
            float(batch.t_thresh),
        )
        self.scheduler.set_timesteps(len(sigmas), device=device, sigmas=sigmas, shift=1.0)
        batch.timesteps = self.scheduler.timesteps
        if getattr(fastvideo_args, "vae_cpu_offload", False):
            self.vae.to("cpu")
        return batch

fastvideo.pipelines.basic.lingbot_video.stages.LingBotVideoRefinerPreparationStage.forward

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

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

Source
def forward(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch:
    """Prepare high-resolution refiner latents and its exact truncated sigma schedule."""
    if batch.output is None or batch.output.ndim != 5:
        raise ValueError("LingBot-Video refinement requires a decoded base video")
    if not isinstance(batch.height_sr, int) or not isinstance(batch.width_sr, int):
        raise TypeError("LingBot-Video refinement requires integer height_sr and width_sr")
    if batch.height_sr % 16 != 0 or batch.width_sr % 16 != 0:
        raise ValueError("LingBot-Video refiner height_sr and width_sr must be divisible by 16")
    if batch.seed is None:
        raise ValueError("LingBot-Video refinement requires a seed")
    device = get_local_torch_device()
    if isinstance(self.vae, torch.nn.Module):
        self.vae.to(device)
    generator = torch.Generator(device=device).manual_seed(batch.seed)
    resized = self._resize_video(batch.output, batch.height_sr, batch.width_sr)
    encoded = self._encode_video(resized, generator, device)
    noise = torch.randn(encoded.shape, generator=generator, device=device, dtype=encoded.dtype)
    batch.latents = ((1.0 - batch.t_thresh) * encoded + batch.t_thresh * noise).float()
    batch.generator = generator
    batch.height = batch.height_sr
    batch.width = batch.width_sr
    batch.raw_latent_shape = tuple(batch.latents.shape)
    batch.extra["lingbot_video_base_shape"] = tuple(batch.output.shape)
    batch.output = None

    shift = fastvideo_args.pipeline_config.flow_shift
    if shift is None:
        raise ValueError("LingBot-Video refinement requires a flow shift")
    sigmas = _compute_refiner_sigmas(
        float(self.scheduler.sigma_max),
        float(self.scheduler.sigma_min),
        batch.num_inference_steps_sr,
        float(shift),
        float(batch.t_thresh),
    )
    self.scheduler.set_timesteps(len(sigmas), device=device, sigmas=sigmas, shift=1.0)
    batch.timesteps = self.scheduler.timesteps
    if getattr(fastvideo_args, "vae_cpu_offload", False):
        self.vae.to("cpu")
    return batch

fastvideo.pipelines.basic.lingbot_video.stages.LingBotVideoDenoisingStage

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

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

Source
class LingBotVideoDenoisingStage(PipelineStage):
    """Run the released batched-CFG bf16 DiT loop with fp32 scheduler state."""

    performance_component_metric = "dit_time_s"

    def __init__(self, transformer, scheduler, refiner: bool = False) -> None:
        self.transformer = transformer
        self.scheduler = scheduler
        self.refiner = refiner

    @staticmethod
    def _pad_condition(
        embeds: torch.Tensor,
        mask: torch.Tensor,
        length: int,
    ) -> tuple[torch.Tensor, torch.Tensor]:
        """Right-pad one condition stream to a shared batched-CFG length."""
        pad_length = length - embeds.shape[1]
        if pad_length < 0 or embeds.shape[:2] != mask.shape:
            raise ValueError("invalid LingBot-Video prompt embedding/mask shapes")
        if pad_length == 0:
            return embeds, mask
        embed_padding = embeds.new_zeros(embeds.shape[0], pad_length, embeds.shape[2])
        mask_padding = mask.new_zeros(mask.shape[0], pad_length)
        return (
            torch.cat((embeds, embed_padding), dim=1),
            torch.cat((mask, mask_padding), dim=1),
        )

    @staticmethod
    def _transformer_timestep(timestep: torch.Tensor, dtype: torch.dtype) -> torch.Tensor:
        """Reproduce the official divide-cast-multiply timestep rounding."""
        sigma = timestep.float() / 1000.0
        if dtype in (torch.bfloat16, torch.float16):
            sigma = sigma.to(dtype)
        return (sigma * 1000.0).float()

    def _prepare_conditions(
        self,
        batch: ForwardBatch,
        dtype: torch.dtype,
        device: torch.device,
    ) -> tuple[torch.Tensor, torch.Tensor]:
        """Pack conditional then unconditional text streams for one batched CFG call."""
        prompt = batch.prompt_embeds[0].to(device=device, dtype=dtype)
        if batch.prompt_attention_mask is None or not batch.prompt_attention_mask:
            raise ValueError("LingBot-Video requires a prompt attention mask")
        prompt_mask = batch.prompt_attention_mask[0].to(device=device)
        if not self._uses_cfg(batch) or not batch.batch_cfg:
            return prompt, prompt_mask
        negative, negative_mask = self._negative_condition(batch, prompt, prompt_mask, dtype, device)
        target_length = max(prompt.shape[1], negative.shape[1])
        prompt, prompt_mask = self._pad_condition(prompt, prompt_mask, target_length)
        negative, negative_mask = self._pad_condition(negative, negative_mask, target_length)
        return (
            torch.cat((prompt, negative), dim=0),
            torch.cat((prompt_mask, negative_mask), dim=0),
        )

    def _negative_condition(
        self,
        batch: ForwardBatch,
        prompt: torch.Tensor,
        prompt_mask: torch.Tensor,
        dtype: torch.dtype,
        device: torch.device,
    ) -> tuple[torch.Tensor, torch.Tensor]:
        """Use the refiner's zero-cloned null condition or the encoded negative prompt."""
        if self.refiner:
            return torch.zeros_like(prompt), prompt_mask.clone()
        if batch.negative_prompt_embeds is None or not batch.negative_prompt_embeds:
            raise ValueError("LingBot-Video CFG requires negative prompt embeddings")
        if batch.negative_attention_mask is None or not batch.negative_attention_mask:
            raise ValueError("LingBot-Video CFG requires a negative prompt mask")
        return (
            batch.negative_prompt_embeds[0].to(device=device, dtype=dtype),
            batch.negative_attention_mask[0].to(device=device),
        )

    def _uses_cfg(self, batch: ForwardBatch) -> bool:
        """Enable guidance independently for the base or refiner scale."""
        scale = batch.guidance_scale_2 if self.refiner else batch.guidance_scale
        return scale is not None and scale > 1.0

    def forward(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch:
        """Denoise latents while keeping scheduler samples and predictions in fp32."""
        if batch.latents is None or batch.timesteps is None:
            raise ValueError("LingBot-Video denoising requires latents and timesteps")
        device = get_local_torch_device()
        transformer_dtype = next(self.transformer.parameters()).dtype
        condition, condition_mask = self._prepare_conditions(batch, transformer_dtype, device)
        latents = batch.latents.to(device=device, dtype=torch.float32)
        do_cfg = self._uses_cfg(batch)
        negative = negative_mask = None
        if do_cfg and not batch.batch_cfg:
            negative, negative_mask = self._negative_condition(batch, condition, condition_mask, transformer_dtype,
                                                               device)
        trajectory: list[torch.Tensor] = []
        trajectory_timesteps: list[torch.Tensor] = []
        for timestep in batch.timesteps:
            # Stop if interrupted
            if getattr(self, "interrupt", False):
                break
            timestep_batch = self._transformer_timestep(timestep, transformer_dtype).expand(1).to(device)
            latent_input = latents
            if do_cfg and batch.batch_cfg:
                latent_input = torch.cat((latents, latents), dim=0)
                timestep_batch = torch.cat((timestep_batch, timestep_batch), dim=0)
            autocast_enabled = device.type == "cuda" and transformer_dtype != torch.float32
            with torch.autocast(device_type=device.type, dtype=transformer_dtype, enabled=autocast_enabled):
                prediction = self.transformer(
                    latent_input,
                    timestep_batch,
                    condition,
                    encoder_attention_mask=condition_mask,
                    return_dict=False,
                )[0].float()
            if do_cfg:
                if batch.batch_cfg:
                    conditional, unconditional = prediction.chunk(2, dim=0)
                else:
                    with torch.autocast(
                            device_type=device.type,
                            dtype=transformer_dtype,
                            enabled=autocast_enabled,
                    ):
                        unconditional = self.transformer(
                            latents,
                            timestep_batch,
                            negative,
                            encoder_attention_mask=negative_mask,
                            return_dict=False,
                        )[0].float()
                    conditional = prediction
                guidance_scale = batch.guidance_scale_2 if self.refiner else batch.guidance_scale
                if guidance_scale is None:
                    raise ValueError("LingBot-Video CFG requires a guidance scale")
                prediction = unconditional + guidance_scale * (conditional - unconditional)
            latents = self.scheduler.step(
                prediction,
                timestep,
                latents,
                return_dict=False,
                generator=batch.generator,
            )[0].float()
            if batch.return_trajectory_latents:
                trajectory.append(latents.detach().cpu())
                trajectory_timesteps.append(timestep.detach().cpu())
        batch.latents = latents
        if trajectory:
            batch.trajectory_latents = torch.stack(trajectory, dim=1)
            batch.trajectory_timesteps = trajectory_timesteps
        return batch

fastvideo.pipelines.basic.lingbot_video.stages.LingBotVideoDenoisingStage.forward

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

Denoise latents while keeping scheduler samples and predictions in fp32.

Source
def forward(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch:
    """Denoise latents while keeping scheduler samples and predictions in fp32."""
    if batch.latents is None or batch.timesteps is None:
        raise ValueError("LingBot-Video denoising requires latents and timesteps")
    device = get_local_torch_device()
    transformer_dtype = next(self.transformer.parameters()).dtype
    condition, condition_mask = self._prepare_conditions(batch, transformer_dtype, device)
    latents = batch.latents.to(device=device, dtype=torch.float32)
    do_cfg = self._uses_cfg(batch)
    negative = negative_mask = None
    if do_cfg and not batch.batch_cfg:
        negative, negative_mask = self._negative_condition(batch, condition, condition_mask, transformer_dtype,
                                                           device)
    trajectory: list[torch.Tensor] = []
    trajectory_timesteps: list[torch.Tensor] = []
    for timestep in batch.timesteps:
        # Stop if interrupted
        if getattr(self, "interrupt", False):
            break
        timestep_batch = self._transformer_timestep(timestep, transformer_dtype).expand(1).to(device)
        latent_input = latents
        if do_cfg and batch.batch_cfg:
            latent_input = torch.cat((latents, latents), dim=0)
            timestep_batch = torch.cat((timestep_batch, timestep_batch), dim=0)
        autocast_enabled = device.type == "cuda" and transformer_dtype != torch.float32
        with torch.autocast(device_type=device.type, dtype=transformer_dtype, enabled=autocast_enabled):
            prediction = self.transformer(
                latent_input,
                timestep_batch,
                condition,
                encoder_attention_mask=condition_mask,
                return_dict=False,
            )[0].float()
        if do_cfg:
            if batch.batch_cfg:
                conditional, unconditional = prediction.chunk(2, dim=0)
            else:
                with torch.autocast(
                        device_type=device.type,
                        dtype=transformer_dtype,
                        enabled=autocast_enabled,
                ):
                    unconditional = self.transformer(
                        latents,
                        timestep_batch,
                        negative,
                        encoder_attention_mask=negative_mask,
                        return_dict=False,
                    )[0].float()
                conditional = prediction
            guidance_scale = batch.guidance_scale_2 if self.refiner else batch.guidance_scale
            if guidance_scale is None:
                raise ValueError("LingBot-Video CFG requires a guidance scale")
            prediction = unconditional + guidance_scale * (conditional - unconditional)
        latents = self.scheduler.step(
            prediction,
            timestep,
            latents,
            return_dict=False,
            generator=batch.generator,
        )[0].float()
        if batch.return_trajectory_latents:
            trajectory.append(latents.detach().cpu())
            trajectory_timesteps.append(timestep.detach().cpu())
    batch.latents = latents
    if trajectory:
        batch.trajectory_latents = torch.stack(trajectory, dim=1)
        batch.trajectory_timesteps = trajectory_timesteps
    return batch