`fastvideo.configs.pipelines.flux.EncoderConfig`

EncoderConfig(arch_config: ArchConfig = EncoderArchConfig(), prefix: str = '', quant_config: QuantizationConfig | None = None, lora_config: Any | None = None, *, _resolved_attention_backend: AttentionBackendEnum | None = None)

`fastvideo.configs.pipelines.flux.FluxDiTConfig`

FluxDiTConfig(arch_config: DiTArchConfig = FluxTransformer2DArchConfig(), prefix: str = 'flux', quant_config: QuantizationConfig | None = None, *, _resolved_attention_backend: AttentionBackendEnum | None = None)

`fastvideo.configs.pipelines.flux.BaseEncoderOutput`

BaseEncoderOutput(last_hidden_state: torch.FloatTensor | None = None, pooler_output: torch.FloatTensor | None = None, hidden_states: tuple[torch.FloatTensor, ...] | None = None, attentions: tuple[torch.FloatTensor, ...] | None = None, attention_mask: torch.Tensor | None = None)

`fastvideo.configs.pipelines.flux.CLIPTextConfig`

CLIPTextConfig(arch_config: TextEncoderArchConfig = CLIPTextArchConfig(), prefix: str = 'clip', quant_config: QuantizationConfig | None = None, lora_config: Any | None = None, is_chat_model: bool = False, treat_empty_as_dot: bool = False, num_hidden_layers_override: int | None = None, require_post_norm: bool | None = None, enable_scale: bool = True, is_causal: bool = True, *, _resolved_attention_backend: AttentionBackendEnum | None = None, chat_template_enable_thinking: bool = False)

`fastvideo.configs.pipelines.flux.T5LargeConfig`

T5LargeConfig(arch_config: TextEncoderArchConfig = T5LargeArchConfig(), prefix: str = 't5', quant_config: QuantizationConfig | None = None, lora_config: Any | None = None, is_chat_model: bool = False, treat_empty_as_dot: bool = False, *, _resolved_attention_backend: AttentionBackendEnum | None = None, chat_template_enable_thinking: bool = False)

T5 Large configuration for your specific model.

`fastvideo.configs.pipelines.flux.AutoencoderKLVAEConfig`

AutoencoderKLVAEConfig(arch_config: VAEArchConfig = AutoencoderKLArchConfig(), load_encoder: bool = True, load_decoder: bool = True, tile_sample_min_height: int = 256, tile_sample_min_width: int = 256, tile_sample_min_num_frames: int = 16, tile_sample_stride_height: int = 192, tile_sample_stride_width: int = 192, tile_sample_stride_num_frames: int = 12, blend_num_frames: int = 0, use_tiling: bool = True, use_temporal_tiling: bool = True, use_parallel_tiling: bool = True, use_temporal_scaling_frames: bool = True, *, _resolved_attention_backend: AttentionBackendEnum | None = None)

`fastvideo.configs.pipelines.flux.PipelineConfig`

PipelineConfig(model_path: str = '', pipeline_config_path: str | None = None, embedded_cfg_scale: float = 6.0, flow_shift: float | None = None, flow_shift_sr: float | None = None, disable_autocast: bool = False, scheduler_step_in_fp32: bool = False, is_causal: bool = False, dit_config: DiTConfig = DiTConfig(), dit_precision: str = 'bf16', upsampler_config: UpsamplerConfig = UpsamplerConfig(), upsampler_precision: str = 'fp32', vae_config: VAEConfig = VAEConfig(), vae_precision: str = 'fp32', vae_decode_precision: str | None = None, vae_tiling: bool = True, vae_sp: bool = True, image_encoder_config: EncoderConfig = EncoderConfig(), image_encoder_precision: str = 'fp32', image_encoder_configs: tuple[EncoderConfig, ...] | None = None, image_encoder_precisions: tuple[str, ...] | None = None, text_encoder_configs: tuple[EncoderConfig, ...] = (lambda: (EncoderConfig(),))(), text_encoder_precisions: tuple[str, ...] = (lambda: ('fp32',))(), preprocess_text_funcs: tuple[Callable[[str], str], ...] = (lambda: (preprocess_text,))(), postprocess_text_funcs: tuple[Callable[[BaseEncoderOutput], torch.tensor], ...] = (lambda: (postprocess_text,))(), dmd_denoising_steps: list[int] | None = None, ti2v_task: bool = False, lucy_edit_task: bool = False, boundary_ratio: float | None = None)

Base configuration for all pipeline architectures.

`fastvideo.configs.pipelines.flux.PipelineConfig.add_cli_args`

add_cli_args(parser: FlexibleArgumentParser, prefix: str = '') -> FlexibleArgumentParser

`fastvideo.configs.pipelines.flux.PipelineConfig.update_config_from_dict`

update_config_from_dict(args: dict[str, Any], prefix: str = '') -> None

`fastvideo.configs.pipelines.flux.PipelineConfig.from_pretrained`

from_pretrained(model_path: str) -> PipelineConfig

use the pipeline class setting from model_path to match the pipeline config

`fastvideo.configs.pipelines.flux.PipelineConfig.from_kwargs`

from_kwargs(kwargs: dict[str, Any], config_cli_prefix: str = '') -> PipelineConfig

Load PipelineConfig from kwargs Dictionary. kwargs: dictionary of kwargs config_cli_prefix: prefix of CLI arguments for this PipelineConfig instance

`fastvideo.configs.pipelines.flux.PipelineConfig.check_pipeline_config`

check_pipeline_config() -> None

`fastvideo.configs.pipelines.flux.PipelineConfig.resolve_checkpoint_settings`

resolve_checkpoint_settings(fastvideo_args: FastVideoArgs) -> None

Apply the run settings that the checkpoint at fastvideo_args.model_path fixes.

FastVideoArgs.__post_init__ calls this once in the main process, before argument validation and before any weights load. Most checkpoints fix no run settings, so the base implementation does nothing.

`fastvideo.configs.pipelines.flux.PipelineConfig.apply_request_constraints`

apply_request_constraints(request: GenerationRequest, sampling_param: SamplingParam) -> SamplingParam

Apply this model's constraints on a generation request to its sampling parameters.

sampling_param holds the preset defaults with the request's explicit fields applied. A model sets the values it fixes when the request leaves them unset, and raises when the request sets a value it does not accept. The base PipelineConfig accepts every request unchanged.

`fastvideo.configs.pipelines.flux.PipelineConfig.dump_to_json`

dump_to_json(file_path: str)

`fastvideo.configs.pipelines.flux.PipelineConfig.load_from_json`

load_from_json(file_path: str)

`fastvideo.configs.pipelines.flux.PipelineConfig.update_pipeline_config`

update_pipeline_config(source_pipeline_dict: dict[str, Any]) -> None

`fastvideo.configs.pipelines.flux.preprocess_text`

preprocess_text(prompt: str) -> str

`fastvideo.configs.pipelines.flux.FluxPipelineConfig`

FluxPipelineConfig(model_path: str = '', pipeline_config_path: str | None = None, embedded_cfg_scale: float = 3.5, flow_shift: float | None = None, flow_shift_sr: float | None = None, disable_autocast: bool = False, scheduler_step_in_fp32: bool = False, is_causal: bool = False, dit_config: FluxDiTConfig = FluxDiTConfig(), dit_precision: str = 'bf16', upsampler_config: UpsamplerConfig = UpsamplerConfig(), upsampler_precision: str = 'fp32', vae_config: AutoencoderKLVAEConfig = AutoencoderKLVAEConfig(), vae_precision: str = 'fp32', vae_decode_precision: str | None = None, vae_tiling: bool = True, vae_sp: bool = True, image_encoder_config: EncoderConfig = EncoderConfig(), image_encoder_precision: str = 'fp32', image_encoder_configs: tuple[EncoderConfig, ...] | None = None, image_encoder_precisions: tuple[str, ...] | None = None, text_encoder_configs: tuple[EncoderConfig, ...] = (lambda: (CLIPTextConfig(), T5LargeConfig()))(), text_encoder_precisions: tuple[str, ...] = (lambda: ('fp32', 'bf16'))(), preprocess_text_funcs: tuple[Callable[[str], str], ...] = (lambda: (preprocess_text, preprocess_text))(), postprocess_text_funcs: tuple[Callable[[BaseEncoderOutput], torch.Tensor], ...] = (lambda: (_flux_clip_pooled_postprocess, _flux_t5_sequence_postprocess))(), dmd_denoising_steps: list[int] | None = None, ti2v_task: bool = False, lucy_edit_task: bool = False, boundary_ratio: float | None = None, scheduler_arch: str = 'FlowMatchEulerDiscreteScheduler', transformer_arch: str = 'FluxTransformer2DModel', vae_arch: str = 'AutoencoderKL', text_encoder_archs: tuple[str, ...] = ('CLIPTextModel', 'T5EncoderModel'), tokenizer_archs: tuple[str, ...] = ('CLIPTokenizer', 'T5TokenizerFast'))

Pipeline layout for Diffusers FLUX.1-dev (CLIP + T5 + packed DiT + FlowMatch).

Source

@dataclass
class FluxPipelineConfig(PipelineConfig):
    """Pipeline layout for Diffusers FLUX.1-dev (CLIP + T5 + packed DiT + FlowMatch)."""

    scheduler_arch: str = "FlowMatchEulerDiscreteScheduler"
    transformer_arch: str = "FluxTransformer2DModel"
    vae_arch: str = "AutoencoderKL"
    text_encoder_archs: tuple[str, ...] = ("CLIPTextModel", "T5EncoderModel")
    tokenizer_archs: tuple[str, ...] = ("CLIPTokenizer", "T5TokenizerFast")

    dit_config: FluxDiTConfig = field(default_factory=FluxDiTConfig)
    vae_config: AutoencoderKLVAEConfig = field(default_factory=AutoencoderKLVAEConfig)

    embedded_cfg_scale: float = 3.5
    flow_shift: float | None = None

    text_encoder_configs: tuple[EncoderConfig, ...] = field(default_factory=lambda: (CLIPTextConfig(), T5LargeConfig()))
    preprocess_text_funcs: tuple[Callable[[str], str],
                                 ...] = field(default_factory=lambda: (preprocess_text, preprocess_text))
    postprocess_text_funcs: tuple[Callable[[BaseEncoderOutput], torch.Tensor], ...] = field(
        default_factory=lambda: (_flux_clip_pooled_postprocess, _flux_t5_sequence_postprocess))

    dit_precision: str = "bf16"
    vae_precision: str = "fp32"
    text_encoder_precisions: tuple[str, ...] = field(default_factory=lambda: ("fp32", "bf16"))

    def __post_init__(self) -> None:
        te_cfgs = list(self.text_encoder_configs)
        if len(te_cfgs) >= 1:
            te_cfgs[0].tokenizer_kwargs.setdefault("padding", "max_length")
            te_cfgs[0].tokenizer_kwargs.setdefault("max_length", 77)
            te_cfgs[0].tokenizer_kwargs.setdefault("truncation", True)
            te_cfgs[0].tokenizer_kwargs.setdefault("return_tensors", "pt")
        if len(te_cfgs) >= 2:
            cap = 512
            te_cfgs[1].tokenizer_kwargs["max_length"] = min(int(te_cfgs[1].tokenizer_kwargs.get("max_length", cap)),
                                                            cap)
            te_cfgs[1].tokenizer_kwargs.setdefault("padding", "max_length")
            te_cfgs[1].tokenizer_kwargs.setdefault("truncation", True)
            te_cfgs[1].tokenizer_kwargs.setdefault("return_tensors", "pt")