Skip to content

minimax_h3_pipeline ¶

FastVideo composed pipelines for MiniMax H3.

Classes¶

fastvideo.pipelines.basic.minimax_h3.minimax_h3_pipeline.MiniMaxH3BasePipeline ¶

MiniMaxH3BasePipeline(*args: Any, **kwargs: Any)

Bases: LoRAPipeline, ComposedPipelineBase

Shared loading and target-generation path for MiniMax H3.

Inherits LoRAPipeline so acceleration and distillation adapters can be merged in; without it every adapter is rejected with "pipeline is not a LoRAPipeline".

Source code in fastvideo/pipelines/basic/minimax_h3/minimax_h3_pipeline.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    self._ref2va = getattr(self, "_ref2va_default", False)
    self._denoise_stages_ready = False
    super().__init__(*args, **kwargs)

Methods:¶

fastvideo.pipelines.basic.minimax_h3.minimax_h3_pipeline.MiniMaxH3BasePipeline.load_modules ¶

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

Load the Qwen3-VL conditioner first; defer DiT and VAEs until after encode.

Source code in fastvideo/pipelines/basic/minimax_h3/minimax_h3_pipeline.py

def load_modules(self,
                 fastvideo_args: FastVideoArgs,
                 loaded_modules: dict[str, torch.nn.Module] | None = None) -> dict[str, Any]:
    """Load the Qwen3-VL conditioner first; defer DiT and VAEs until after encode."""
    # _load_config checks the transformer against the run's settings before any component loads.
    self._check_transformer_before_loading = loaded_modules is None or "transformer" not in loaded_modules
    if h3_encoder_split_enabled(fastvideo_args):
        # Component-level pipeline parallel: the role decides the module set
        # outright, so no deferral/lazy machinery is involved on either side.
        if h3_is_encoder_worker(fastvideo_args):
            keep = H3_ENCODER_MODULE_NAMES
        else:
            keep = H3_DENOISE_MODULE_NAMES
            if _use_taeh3_t2va(fastvideo_args, ref2va=self._ref2va):
                # Mirrors `_denoise_module_names`: T2VA with the TAEH3 preview
                # decoder never touches the full video VAE.
                keep = keep - {"vae"}
        saved = list(self.required_config_modules)
        self._required_config_modules = [name for name in saved if name in keep]
        try:
            logger.info("MiniMax-H3 encoder split: %s worker loading modules %s",
                        "encoder" if h3_is_encoder_worker(fastvideo_args) else "denoise",
                        self._required_config_modules)
            return super().load_modules(fastvideo_args, loaded_modules)
        finally:
            self._required_config_modules = saved
    if not self._defer_denoise_modules(fastvideo_args):
        if _use_taeh3_t2va(fastvideo_args, ref2va=self._ref2va):
            saved = list(self.required_config_modules)
            self._required_config_modules = [name for name in saved if name != "vae"]
            try:
                return super().load_modules(fastvideo_args, loaded_modules)
            finally:
                self._required_config_modules = saved
        return super().load_modules(fastvideo_args, loaded_modules)
    if loaded_modules is not None and all(name in loaded_modules
                                          for name in self._denoise_module_names(fastvideo_args)):
        return super().load_modules(fastvideo_args, loaded_modules)

    saved = list(self.required_config_modules)
    # Always defer the full denoise set on the first load. TAEH3 T2VA then
    # omits the video VAE from the second load via `_denoise_module_names`.
    self._required_config_modules = [name for name in saved if name not in _DENOISE_MODULE_NAMES]
    try:
        logger.info("Loading MiniMax-H3 condition modules first: %s", self._required_config_modules)
        return super().load_modules(fastvideo_args, loaded_modules)
    finally:
        self._required_config_modules = saved

fastvideo.pipelines.basic.minimax_h3.minimax_h3_pipeline.MiniMaxH3ModularPipeline ¶

MiniMaxH3ModularPipeline(*args: Any, **kwargs: Any)

Bases: MiniMaxH3Pipeline

Public T2VA/FL2VA entry matching the official manifest class name.

Source code in fastvideo/pipelines/basic/minimax_h3/minimax_h3_pipeline.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    self._ref2va = getattr(self, "_ref2va_default", False)
    self._denoise_stages_ready = False
    super().__init__(*args, **kwargs)

fastvideo.pipelines.basic.minimax_h3.minimax_h3_pipeline.MiniMaxH3Pipeline ¶

MiniMaxH3Pipeline(*args: Any, **kwargs: Any)

Bases: MiniMaxH3BasePipeline

One-request joint video/stereo-audio pipeline for T2VA and FL2VA.

Source code in fastvideo/pipelines/basic/minimax_h3/minimax_h3_pipeline.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    self._ref2va = getattr(self, "_ref2va_default", False)
    self._denoise_stages_ready = False
    super().__init__(*args, **kwargs)

fastvideo.pipelines.basic.minimax_h3.minimax_h3_pipeline.MiniMaxH3Ref2VAModularPipeline ¶

MiniMaxH3Ref2VAModularPipeline(*args: Any, **kwargs: Any)

Bases: MiniMaxH3RefPipeline

Public Ref2VA entry using the checkpoint's transformer_ref partition.

Source code in fastvideo/pipelines/basic/minimax_h3/minimax_h3_pipeline.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    self._ref2va = getattr(self, "_ref2va_default", False)
    self._denoise_stages_ready = False
    super().__init__(*args, **kwargs)

fastvideo.pipelines.basic.minimax_h3.minimax_h3_pipeline.MiniMaxH3RefPipeline ¶

MiniMaxH3RefPipeline(*args: Any, **kwargs: Any)

Bases: MiniMaxH3BasePipeline

Ordered-reference joint video/stereo-audio pipeline for Ref2VA.

Source code in fastvideo/pipelines/basic/minimax_h3/minimax_h3_pipeline.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    self._ref2va = getattr(self, "_ref2va_default", False)
    self._denoise_stages_ready = False
    super().__init__(*args, **kwargs)

Functions:¶