flux_pipeline

`fastvideo.pipelines.basic.flux.flux_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.flux.flux_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.flux.flux_pipeline.FastVideoArgs.add_cli_args`

add_cli_args(parser: FlexibleArgumentParser) -> FlexibleArgumentParser

`fastvideo.pipelines.basic.flux.flux_pipeline.FastVideoArgs.from_cli_args`

from_cli_args(args: argparse.Namespace) -> FastVideoArgs

`fastvideo.pipelines.basic.flux.flux_pipeline.FastVideoArgs.from_kwargs`

from_kwargs(**kwargs: Any) -> FastVideoArgs

`fastvideo.pipelines.basic.flux.flux_pipeline.FastVideoArgs.check_fastvideo_args`

check_fastvideo_args() -> None

Validate inference arguments for consistency

`fastvideo.pipelines.basic.flux.flux_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.flux.flux_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.flux.flux_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.flux.flux_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.flux.flux_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.flux.flux_pipeline.ComposedPipelineBase.set_trainable`

set_trainable() -> None

`fastvideo.pipelines.basic.flux.flux_pipeline.ComposedPipelineBase.post_init`

post_init() -> None

`fastvideo.pipelines.basic.flux.flux_pipeline.ComposedPipelineBase.initialize_training_pipeline`

initialize_training_pipeline(training_args: TrainingArgs)

`fastvideo.pipelines.basic.flux.flux_pipeline.ComposedPipelineBase.initialize_validation_pipeline`

initialize_validation_pipeline(training_args: TrainingArgs)

`fastvideo.pipelines.basic.flux.flux_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.flux.flux_pipeline.ComposedPipelineBase.get_module`

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

`fastvideo.pipelines.basic.flux.flux_pipeline.ComposedPipelineBase.add_module`

add_module(module_name: str, module: Any)

`fastvideo.pipelines.basic.flux.flux_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.flux.flux_pipeline.ComposedPipelineBase.stages`

stages: list[PipelineStage]

List of stages in the pipeline.

`fastvideo.pipelines.basic.flux.flux_pipeline.ComposedPipelineBase.create_pipeline_stages`

create_pipeline_stages(fastvideo_args: FastVideoArgs)

Create the inference pipeline stages.

`fastvideo.pipelines.basic.flux.flux_pipeline.ComposedPipelineBase.create_training_stages`

create_training_stages(training_args: TrainingArgs)

Create the training pipeline stages.

`fastvideo.pipelines.basic.flux.flux_pipeline.ComposedPipelineBase.initialize_pipeline`

initialize_pipeline(fastvideo_args: FastVideoArgs)

Initialize the pipeline.

`fastvideo.pipelines.basic.flux.flux_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.flux.flux_pipeline.ComposedPipelineBase.add_stage`

add_stage(stage_name: str, stage: PipelineStage)

`fastvideo.pipelines.basic.flux.flux_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.flux.flux_pipeline.ComposedPipelineBase.train`

train() -> None

`fastvideo.pipelines.basic.flux.flux_pipeline.ComposedPipelineBase.close`

close() -> None

`fastvideo.pipelines.basic.flux.flux_pipeline.FluxConditioningStage`

FluxConditioningStage()

Build CLIP pooled + T5 sequence + text_ids (and optional negative for true CFG).

`fastvideo.pipelines.basic.flux.flux_pipeline.FluxConditioningStage.forward`

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

`fastvideo.pipelines.basic.flux.flux_pipeline.FluxDecodingStage`

FluxDecodingStage(vae)

Unpack latents, apply VAE scaling/shift, decode to pixels (5D output B×3×1×H×W).

`fastvideo.pipelines.basic.flux.flux_pipeline.FluxDecodingStage.forward`

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

`fastvideo.pipelines.basic.flux.flux_pipeline.FluxDenoisingStage`

FluxDenoisingStage(transformer, scheduler)

`fastvideo.pipelines.basic.flux.flux_pipeline.FluxDenoisingStage.forward`

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

`fastvideo.pipelines.basic.flux.flux_pipeline.FluxInputValidationStage`

FluxInputValidationStage()

Require height/width divisible by 16 (VAE scale × 2 for FLUX packing).

`fastvideo.pipelines.basic.flux.flux_pipeline.FluxInputValidationStage.forward`

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

`fastvideo.pipelines.basic.flux.flux_pipeline.FluxLatentPreparationStage`

FluxLatentPreparationStage(scheduler)

`fastvideo.pipelines.basic.flux.flux_pipeline.FluxLatentPreparationStage.forward`

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

`fastvideo.pipelines.basic.flux.flux_pipeline.FluxTimestepPreparationStage`

FluxTimestepPreparationStage(scheduler)

Flow Match with resolution-dependent mu from packed image sequence length.

`fastvideo.pipelines.basic.flux.flux_pipeline.FluxTimestepPreparationStage.forward`

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

`fastvideo.pipelines.basic.flux.flux_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.flux.flux_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.flux.flux_pipeline.TextEncodingStage.verify_input`

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

Verify text encoding stage inputs.

`fastvideo.pipelines.basic.flux.flux_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.flux.flux_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.flux.flux_pipeline.TextEncodingStage.verify_output`

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

Verify text encoding stage outputs.

`fastvideo.pipelines.basic.flux.flux_pipeline.FluxPipeline`

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

FLUX.1-dev T2I (Diffusers module layout, packed latents, embedded guidance).

Source

class FluxPipeline(ComposedPipelineBase):
    """FLUX.1-dev T2I (Diffusers module layout, packed latents, embedded guidance)."""
_required_config_modules = &#91;
    &quot;scheduler&quot;,
    &quot;transformer&quot;,
    &quot;vae&quot;,
    &quot;text_encoder&quot;,
    &quot;text_encoder_2&quot;,
    &quot;tokenizer&quot;,
    &quot;tokenizer_2&quot;,
&#93;

def create_pipeline_stages(self, fastvideo_args: FastVideoArgs) -&gt; None:
    self.add_stage(stage_name=&quot;input_validation_stage&quot;, stage=FluxInputValidationStage())

    self.add_stage(
        stage_name=&quot;text_encoding_stage&quot;,
        stage=TextEncodingStage(
            text_encoders=&#91;
                self.get_module(&quot;text_encoder&quot;),
                self.get_module(&quot;text_encoder_2&quot;),
            &#93;,
            tokenizers=&#91;
                self.get_module(&quot;tokenizer&quot;),
                self.get_module(&quot;tokenizer_2&quot;),
            &#93;,
        ),
    )

    self.add_stage(stage_name=&quot;flux_conditioning_stage&quot;, stage=FluxConditioningStage())

    self.add_stage(
        stage_name=&quot;timestep_preparation_stage&quot;,
        stage=FluxTimestepPreparationStage(scheduler=self.get_module(&quot;scheduler&quot;)),
    )

    self.add_stage(
        stage_name=&quot;latent_preparation_stage&quot;,
        stage=FluxLatentPreparationStage(scheduler=self.get_module(&quot;scheduler&quot;)),
    )

    self.add_stage(
        stage_name=&quot;denoising_stage&quot;,
        stage=FluxDenoisingStage(
            transformer=self.get_module(&quot;transformer&quot;),
            scheduler=self.get_module(&quot;scheduler&quot;),
        ),
    )

    self.add_stage(
        stage_name=&quot;decoding_stage&quot;,
        stage=FluxDecodingStage(vae=self.get_module(&quot;vae&quot;)),
    )</code></pre></details>
fastvideo.pipelines.basic.flux.flux_pipeline.FluxPipeline.create_pipeline_stages
create_pipeline_stages(fastvideo_args: FastVideoArgs) -> None
Source
def create_pipeline_stages(self, fastvideo_args: FastVideoArgs) -> None:
    self.add_stage(stage_name="input_validation_stage", stage=FluxInputValidationStage())
self.add_stage(
    stage_name=&quot;text_encoding_stage&quot;,
    stage=TextEncodingStage(
        text_encoders=&#91;
            self.get_module(&quot;text_encoder&quot;),
            self.get_module(&quot;text_encoder_2&quot;),
        &#93;,
        tokenizers=&#91;
            self.get_module(&quot;tokenizer&quot;),
            self.get_module(&quot;tokenizer_2&quot;),
        &#93;,
    ),
)

self.add_stage(stage_name=&quot;flux_conditioning_stage&quot;, stage=FluxConditioningStage())

self.add_stage(
    stage_name=&quot;timestep_preparation_stage&quot;,
    stage=FluxTimestepPreparationStage(scheduler=self.get_module(&quot;scheduler&quot;)),
)

self.add_stage(
    stage_name=&quot;latent_preparation_stage&quot;,
    stage=FluxLatentPreparationStage(scheduler=self.get_module(&quot;scheduler&quot;)),
)

self.add_stage(
    stage_name=&quot;denoising_stage&quot;,
    stage=FluxDenoisingStage(
        transformer=self.get_module(&quot;transformer&quot;),
        scheduler=self.get_module(&quot;scheduler&quot;),
    ),
)

self.add_stage(
    stage_name=&quot;decoding_stage&quot;,
    stage=FluxDecodingStage(vae=self.get_module(&quot;vae&quot;)),
)</code></pre></details>