--- title: magi_human_pipeline ---

MagiHuman base text-to-AV pipeline.

Top-level composition for the daVinci-MagiHuman base model. Wires:

InputValidationStage -> TextEncodingStage (T5-Gemma)
    -> MagiHumanLatentPreparationStage
    -> MagiHumanDenoisingStage
    -> DecodingStage (Wan 2.2 TI2V-5B VAE decode for video)
    -> MagiHumanAudioDecodingStage (Stable Audio Open 1.0 VAE decode)

The base checkpoint is a joint audio-visual generator; both the video and audio paths run in the denoising loop and both are decoded.

load_modules is overridden so the four cross-variant shared components (text_encoder, tokenizer, audio_vae, video vae) lazy-load from their canonical upstream HF repos at first build time instead of being bundled inside every converted MagiHuman variant. This keeps each variant's converted repo at ~5-30 GB (transformer + scheduler + model_index.json) instead of ~30-55 GB, and lets all variants share the same ~25 GB of cached upstream weights.

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.T5GemmaEncoderConfig

T5GemmaEncoderConfig(arch_config: TextEncoderArchConfig = T5GemmaEncoderArchConfig(), prefix: str = 't5gemma', 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)

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.OobleckVAEConfig

OobleckVAEConfig(arch_config: VAEArchConfig = OobleckVAEArchConfig(), 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 = False, use_temporal_tiling: bool = False, use_parallel_tiling: bool = False, use_temporal_scaling_frames: bool = True, pretrained_path: str = 'stabilityai/stable-audio-open-1.0', pretrained_subfolder: str = 'vae', pretrained_dtype: str = 'float16', *, _resolved_attention_backend: AttentionBackendEnum | None = None)

FastVideo VAE config wrapping the Oobleck arch.

Audio VAEs don't use the temporal/spatial tiling defaults that the base VAEConfig is shaped for (those exist for video VAEs); they are retained but irrelevant for audio.

fastvideo.pipelines.basic.magi_human.magi_human_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.magi_human.magi_human_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.magi_human.magi_human_pipeline.FastVideoArgs.add_cli_args

add_cli_args(parser: FlexibleArgumentParser) -> FlexibleArgumentParser

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.FastVideoArgs.from_cli_args

from_cli_args(args: argparse.Namespace) -> FastVideoArgs

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.FastVideoArgs.from_kwargs

from_kwargs(**kwargs: Any) -> FastVideoArgs

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.FastVideoArgs.check_fastvideo_args

check_fastvideo_args() -> None

Validate inference arguments for consistency

fastvideo.pipelines.basic.magi_human.magi_human_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.magi_human.magi_human_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.magi_human.magi_human_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.magi_human.magi_human_pipeline.T5GemmaEncoderModel

T5GemmaEncoderModel(config: TextEncoderConfig)

Thin wrapper over HuggingFace's T5GemmaEncoderModel.

On first forward, the wrapper lazily instantiates the upstream encoder from t5gemma_model_path (defaulting to google/t5gemma-9b-9b-ul2). Afterwards, forward returns a BaseEncoderOutput with last_hidden_state = [B, L, 3584] matching MagiHuman's context.half() output.

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.T5GemmaEncoderModel.named_parameters

named_parameters(prefix: str = '', recurse: bool = True)

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.T5GemmaEncoderModel.forward

forward(input_ids: torch.Tensor | None = None, position_ids: torch.Tensor | None = None, attention_mask: torch.Tensor | None = None, inputs_embeds: torch.Tensor | None = None, output_hidden_states: bool | None = None, **kwargs) -> BaseEncoderOutput

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.T5GemmaEncoderModel.load_weights

load_weights(weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.SAAudioVAEModel

SAAudioVAEModel(config: OobleckVAEConfig)

Pipeline-glue lazy loader around OobleckVAE.

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.SAAudioVAEModel.named_parameters

named_parameters(prefix: str = '', recurse: bool = True)

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.SAAudioVAEModel.decode

decode(latent: torch.Tensor) -> torch.Tensor

Decode an audio latent ([B, C_latent, L]) -> waveform ([B, audio_channels, samples]).

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.SAAudioVAEModel.encode

encode(waveform: torch.Tensor, sample_posterior: bool = False) -> torch.Tensor

Encode [B, C_audio, samples] -> latent [B, C_latent, L].

sample_posterior=False (default): deterministic mean. sample_posterior=True: stochastic sample (mean + softplus(scale) * randn).

fastvideo.pipelines.basic.magi_human.magi_human_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.magi_human.magi_human_pipeline.FlowUniPCMultistepScheduler.step_index

step_index

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

fastvideo.pipelines.basic.magi_human.magi_human_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.magi_human.magi_human_pipeline.FlowUniPCMultistepScheduler.set_shift

set_shift(shift: float) -> None

fastvideo.pipelines.basic.magi_human.magi_human_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.magi_human.magi_human_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.magi_human.magi_human_pipeline.FlowUniPCMultistepScheduler.time_shift

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

fastvideo.pipelines.basic.magi_human.magi_human_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.magi_human.magi_human_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.magi_human.magi_human_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.magi_human.magi_human_pipeline.FlowUniPCMultistepScheduler.index_for_timestep

index_for_timestep(timestep, schedule_timesteps = None) -> int

fastvideo.pipelines.basic.magi_human.magi_human_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.magi_human.magi_human_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.magi_human.magi_human_pipeline.FlowUniPCMultistepScheduler.add_noise

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

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanAudioDecodingStage

MagiHumanAudioDecodingStage(audio_vae, time_stretching: float = _UPSTREAM_AUDIO_TIME_STRETCH)

Decode batch.audio_latents to a waveform using Stable Audio's VAE.

The VAE is loaded lazily by SAAudioVAEModel.sa_audio_vae_model — the first call triggers a snapshot_download (requires HF token + accepted terms on stabilityai/stable-audio-open-1.0).

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanAudioDecodingStage.verify_input

verify_input(batch, fastvideo_args)

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanAudioDecodingStage.verify_output

verify_output(batch, fastvideo_args)

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanAudioDecodingStage.forward

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

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanDenoisingStage

MagiHumanDenoisingStage(transformer, scheduler, patch_size: tuple[int, int, int] = (1, 2, 2), video_in_channels: int = 192, audio_in_channels: int = 64, video_txt_guidance_scale: float = 5.0, audio_txt_guidance_scale: float = 5.0, cfg_number: int = 2, coords_style: str = 'v2', video_guidance_high_t_threshold: int = 500, video_guidance_low_t_value: float = 2.0)

UniPC-flow joint denoising with CFG=2 over (video, audio) latents.

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanDenoisingStage.verify_input

verify_input(batch, fastvideo_args)

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanDenoisingStage.verify_output

verify_output(batch, fastvideo_args)

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanDenoisingStage.forward

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

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanLatentPreparationStage

MagiHumanLatentPreparationStage(vae_stride: tuple[int, int, int] = (4, 16, 16), z_dim: int = 48, patch_size: tuple[int, int, int] = (1, 2, 2), fps: int = 25, t5_gemma_target_length: int = 640, coords_style: Literal['v1', 'v2'] = 'v2', text_offset: int = 0, audio_in_channels: int = 64)

Prepare latents, coords, modality maps, and padded text embed.

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanLatentPreparationStage.verify_input

verify_input(batch, fastvideo_args)

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanLatentPreparationStage.verify_output

verify_output(batch, fastvideo_args)

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanLatentPreparationStage.forward

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

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanReferenceImageStage

MagiHumanReferenceImageStage(vae: Any, vae_scale_factor: int = 16)

Encode a TI2V reference image into the first-frame video latent.

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanReferenceImageStage.verify_input

verify_input(batch, fastvideo_args)

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanReferenceImageStage.verify_output

verify_output(batch, fastvideo_args)

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanReferenceImageStage.forward

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

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanSRDenoisingStage

MagiHumanSRDenoisingStage(transformer, scheduler, patch_size: tuple[int, int, int] = (1, 2, 2), video_in_channels: int = 192, audio_in_channels: int = 64, sr_num_inference_steps: int = 5, sr_video_txt_guidance_scale: float = 3.5, use_cfg_trick: bool = True, cfg_trick_start_frame: int = 13, cfg_trick_value: float = 2.0, cfg_number: int = 2, coords_style: str = 'v1')

Denoise only the SR video latent; audio passes through unchanged.

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanSRDenoisingStage.verify_input

verify_input(batch, fastvideo_args)

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanSRDenoisingStage.verify_output

verify_output(batch, fastvideo_args)

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanSRDenoisingStage.forward

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

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanSRLatentPreparationStage

MagiHumanSRLatentPreparationStage(vae: Any, vae_stride: tuple[int, int, int] = (4, 16, 16), patch_size: tuple[int, int, int] = (1, 2, 2), noise_value: int = 220, sr_audio_noise_scale: float = 0.7, sr_height: int = 512, sr_width: int = 896, vae_scale_factor: int = 16)

Upsample base latents, add SR noise, and refresh SR conditioning.

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanSRLatentPreparationStage.verify_input

verify_input(batch, fastvideo_args)

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanSRLatentPreparationStage.verify_output

verify_output(batch, fastvideo_args)

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanSRLatentPreparationStage.forward

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

fastvideo.pipelines.basic.magi_human.magi_human_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.magi_human.magi_human_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.magi_human.magi_human_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.magi_human.magi_human_pipeline.ComposedPipelineBase.set_trainable

set_trainable() -> None

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.ComposedPipelineBase.post_init

post_init() -> None

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.ComposedPipelineBase.initialize_training_pipeline

initialize_training_pipeline(training_args: TrainingArgs)

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.ComposedPipelineBase.initialize_validation_pipeline

initialize_validation_pipeline(training_args: TrainingArgs)

fastvideo.pipelines.basic.magi_human.magi_human_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.magi_human.magi_human_pipeline.ComposedPipelineBase.get_module

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

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.ComposedPipelineBase.add_module

add_module(module_name: str, module: Any)

fastvideo.pipelines.basic.magi_human.magi_human_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.magi_human.magi_human_pipeline.ComposedPipelineBase.stages

stages: list[PipelineStage]

List of stages in the pipeline.

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.ComposedPipelineBase.create_pipeline_stages

create_pipeline_stages(fastvideo_args: FastVideoArgs)

Create the inference pipeline stages.

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.ComposedPipelineBase.create_training_stages

create_training_stages(training_args: TrainingArgs)

Create the training pipeline stages.

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.ComposedPipelineBase.initialize_pipeline

initialize_pipeline(fastvideo_args: FastVideoArgs)

Initialize the pipeline.

fastvideo.pipelines.basic.magi_human.magi_human_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.magi_human.magi_human_pipeline.ComposedPipelineBase.add_stage

add_stage(stage_name: str, stage: PipelineStage)

fastvideo.pipelines.basic.magi_human.magi_human_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.magi_human.magi_human_pipeline.ComposedPipelineBase.train

train() -> None

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.ComposedPipelineBase.close

close() -> None

fastvideo.pipelines.basic.magi_human.magi_human_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.magi_human.magi_human_pipeline.DecodingStage.verify_input

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

Verify decoding stage inputs.

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.DecodingStage.verify_output

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

Verify decoding stage outputs.

fastvideo.pipelines.basic.magi_human.magi_human_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.magi_human.magi_human_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.magi_human.magi_human_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.magi_human.magi_human_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.magi_human.magi_human_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.magi_human.magi_human_pipeline.InputValidationStage.verify_input

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

Verify input validation stage inputs.

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.InputValidationStage.verify_output

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

Verify input validation stage outputs.

fastvideo.pipelines.basic.magi_human.magi_human_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.magi_human.magi_human_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.magi_human.magi_human_pipeline.TextEncodingStage.verify_input

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

Verify text encoding stage inputs.

fastvideo.pipelines.basic.magi_human.magi_human_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.magi_human.magi_human_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.magi_human.magi_human_pipeline.TextEncodingStage.verify_output

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

Verify text encoding stage outputs.

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.maybe_download_model

maybe_download_model(model_name_or_path: str, local_dir: str | None = None, download: bool = True, revision: str | None = None, allow_patterns: list[str] | None = None) -> str

Check if the model path is a Hugging Face Hub model ID and download it if needed.

Supports an "umbrella" repo layout where a single HF repo holds multiple pipeline variants under sibling subfolders. If the input is shaped as org/repo/subfolder (i.e. a non-existent local path with 3+ slash- separated components and at least one segment that does not look like a posix-absolute path), treat the first two components as the HF repo id and the remainder as a subfolder; only the subfolder's blobs are downloaded, and the returned local path points inside that subfolder.

Parameters

model_name_or_path: str

Local path, Hugging Face Hub model ID, or org/repo/subfolder umbrella-repo reference.

local_dir: str | None

Local directory to save the model

download: bool

Whether to download the model from Hugging Face Hub

revision: str | None

Optional immutable Hub revision.

allow_patterns: list[str] | None

Optional Hub glob patterns limiting downloaded files. Local paths are returned unchanged. For umbrella references, the patterns are interpreted relative to the selected subfolder.

Returns

str

Local path to the model (or to the subfolder inside the snapshot).

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.resolve_hf_token

resolve_hf_token() -> str | None

Return the first non-empty Hugging Face token from the environment.

Reads HF_TOKEN, then HUGGING_FACE_HUB_TOKEN, then the deprecated FastVideo-specific aliases HUGGINGFACE_HUB_TOKEN and HF_API_KEY, which log a deprecation warning. Does not mutate os.environ.

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanPipeline

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

Base MagiHuman text-to-AV pipeline (no LoRA, no distill, no SR).

Source
class MagiHumanPipeline(ComposedPipelineBase):
    """Base MagiHuman text-to-AV pipeline (no LoRA, no distill, no SR)."""

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

    def load_modules(
        self,
        fastvideo_args: FastVideoArgs,
        loaded_modules: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        """Load the variant-specific transformer + scheduler from the
        converted MagiHuman repo and lazy-load the four cross-variant
        shared components from their canonical upstream HF repos:

          * text_encoder, tokenizer -> ``google/t5gemma-9b-9b-ul2``
            (gated, requires HF token with accepted terms of use)
          * audio_vae -> ``stabilityai/stable-audio-open-1.0`` (gated)
          * vae -> ``Wan-AI/Wan2.2-TI2V-5B-Diffusers``

        Backwards-compatible with bundled converted repos: if any of
        these subfolders is present locally and listed in
        ``model_index.json``, the standard component loader picks it up
        via super(). Otherwise the loader is told to skip the entry and
        we lazy-load it here.
        """
        # T5-Gemma is gated: expose `HF_API_KEY` as `HF_TOKEN` if needed.
        _ensure_hf_token_env()

        # Resolve to a local cache path so we can inspect
        # model_index.json before invoking super(). `maybe_download_model`
        # is idempotent for local paths; super() repeats the call cheaply
        # via `_load_config`.
        local_path = maybe_download_model(self.model_path)

        # Identify which cross-variant shared keys are bundled in the
        # converted repo (declared in model_index.json with a non-null
        # spec) versus absent (the umbrella scheme). Bundled keys stay
        # in `required_config_modules` and are loaded normally by super()
        # from `<model_path>/<key>/`. Absent keys are temporarily
        # dropped so super() does not fail the "every required entry
        # must appear in model_index.json" check, then lazy-loaded
        # below.
        model_index: dict[str, Any] = {}
        try:
            with open(Path(local_path) / "model_index.json") as f:
                model_index = json.load(f)
        except (FileNotFoundError, json.JSONDecodeError):
            pass

        def _is_bundled(key: str) -> bool:
            spec = model_index.get(key)
            return (isinstance(spec, list | tuple) and len(spec) >= 1 and spec[0] is not None)

        deferred = []
        for key in ("text_encoder", "tokenizer", "audio_vae", "vae"):
            if key in self.required_config_modules and not _is_bundled(key):
                self.required_config_modules.remove(key)
                deferred.append(key)

        try:
            modules = super().load_modules(fastvideo_args, loaded_modules)
        finally:
            for key in deferred:
                if key not in self.required_config_modules:
                    self.required_config_modules.append(key)

        # For each lazy-load key, prefer whatever super() already loaded
        # (a bundled subfolder, or a caller-provided override merged in
        # via `loaded_modules`). Fall back to the caller-provided
        # `loaded_modules` entry for keys absent from model_index.json
        # (super() never iterates those). Otherwise lazy-load from the
        # canonical upstream HF repo.
        def _resolve(key: str) -> bool:
            """Return True if `modules[key]` is already populated."""
            if modules.get(key) is not None:
                return True
            if loaded_modules and key in loaded_modules:
                modules[key] = loaded_modules[key]
                return True
            return False

        if not _resolve("text_encoder"):
            logger.info("Building T5-Gemma text encoder (lazy-load from %s)", _T5GEMMA_HF_ID)
            enc_config = T5GemmaEncoderConfig()
            enc_config.arch_config.t5gemma_model_path = _T5GEMMA_HF_ID
            modules["text_encoder"] = T5GemmaEncoderModel(enc_config)

        if not _resolve("tokenizer"):
            logger.info("Loading T5-Gemma tokenizer from %s", _T5GEMMA_HF_ID)
            modules["tokenizer"] = AutoTokenizer.from_pretrained(_T5GEMMA_HF_ID)

        if not _resolve("audio_vae"):
            logger.info(
                "Building Stable Audio Open 1.0 VAE (lazy-load from %s) — "
                "requires HF terms accepted for gated repo",
                _SA_AUDIO_HF_ID,
            )
            audio_config = OobleckVAEConfig()
            audio_config.pretrained_path = _SA_AUDIO_HF_ID
            modules["audio_vae"] = SAAudioVAEModel(audio_config)

        if not _resolve("vae"):
            modules["vae"] = self._load_video_vae(fastvideo_args)

        return modules

    def _load_video_vae(self, fastvideo_args: FastVideoArgs) -> Any:
        """Resolve the video VAE: prefer a bundled ``vae/`` subfolder in
        the converted repo (legacy), fall back to lazy-downloading the
        Wan 2.2 TI2V-5B VAE shards from upstream.

        Either way the load goes through FastVideo's standard
        ``VAELoader`` so the result is the same FV ``AutoencoderKLWan``
        nn.Module that the bundled path produces.
        """
        from fastvideo.models.loader.component_loader import VAELoader

        bundled = Path(self.model_path) / "vae"
        if bundled.is_dir() and (bundled / "config.json").is_file():
            logger.info("Loading bundled video VAE from %s", bundled)
            return VAELoader().load(str(bundled), fastvideo_args)

        from huggingface_hub import snapshot_download

        logger.info(
            "Bundled vae/ not found at %s; lazy-loading Wan 2.2 TI2V-5B VAE from %s",
            self.model_path,
            _WAN_VAE_HF_ID,
        )
        snapshot = snapshot_download(
            repo_id=_WAN_VAE_HF_ID,
            allow_patterns=["vae/*"],
        )
        vae_dir = os.path.join(snapshot, "vae")
        if not os.path.isdir(vae_dir):
            raise RuntimeError(
                f"snapshot_download returned {snapshot} but no vae/ "
                f"subfolder was found inside it. Check that {_WAN_VAE_HF_ID} "
                "still exposes a Diffusers-format vae/ folder.", )
        return VAELoader().load(vae_dir, fastvideo_args)

    def initialize_pipeline(self, fastvideo_args: FastVideoArgs) -> None:
        # MagiHuman applies `flow_shift` during timestep setup; keep the
        # scheduler constructor at its default no-op shift.
        self.modules["scheduler"] = FlowUniPCMultistepScheduler()

    def create_pipeline_stages(self, fastvideo_args: FastVideoArgs) -> None:
        self._add_input_and_conditioning_stages(fastvideo_args)
        self._add_base_latent_and_denoising_stages(fastvideo_args)
        self._add_decode_stages()

    def _add_input_and_conditioning_stages(self, fastvideo_args: FastVideoArgs) -> None:
        self.add_stage(
            stage_name="input_validation_stage",
            stage=InputValidationStage(),
        )

        self.add_stage(
            stage_name="prompt_encoding_stage",
            stage=TextEncodingStage(
                text_encoders=[self.get_module("text_encoder")],
                tokenizers=[self.get_module("tokenizer")],
            ),
        )

        self._add_reference_image_stage(fastvideo_args)

    def _add_base_latent_and_denoising_stages(self, fastvideo_args: FastVideoArgs) -> None:
        pc = fastvideo_args.pipeline_config
        dit_arch = pc.dit_config.arch_config

        # Data-proxy + eval knobs come from the PipelineConfig (`pc`).
        # Only DiT-architecture fields live on `dit_arch` now.
        self.add_stage(
            stage_name="latent_preparation_stage",
            stage=MagiHumanLatentPreparationStage(
                vae_stride=tuple(pc.vae_stride),
                z_dim=pc.z_dim,
                patch_size=tuple(dit_arch.patch_size),
                fps=pc.fps,
                t5_gemma_target_length=pc.t5_gemma_target_length,
                coords_style=pc.coords_style,
                text_offset=pc.text_offset,
                audio_in_channels=dit_arch.audio_in_channels,
            ),
        )

        self.add_stage(
            stage_name="denoising_stage",
            stage=MagiHumanDenoisingStage(
                transformer=self.get_module("transformer"),
                scheduler=self.get_module("scheduler"),
                patch_size=tuple(dit_arch.patch_size),
                video_in_channels=dit_arch.video_in_channels,
                audio_in_channels=dit_arch.audio_in_channels,
                video_txt_guidance_scale=pc.video_txt_guidance_scale,
                audio_txt_guidance_scale=pc.audio_txt_guidance_scale,
                cfg_number=pc.cfg_number,
                coords_style=pc.coords_style,
                video_guidance_high_t_threshold=pc.video_guidance_high_t_threshold,
                video_guidance_low_t_value=pc.video_guidance_low_t_value,
            ),
        )

    def _add_decode_stages(self) -> None:
        self.add_stage(
            stage_name="decoding_stage",
            stage=DecodingStage(vae=self.get_module("vae"), pipeline=self),
        )

        self.add_stage(
            stage_name="audio_decoding_stage",
            stage=MagiHumanAudioDecodingStage(audio_vae=self.get_module("audio_vae"), ),
        )

    def _add_reference_image_stage(self, fastvideo_args: FastVideoArgs) -> None:
        return

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanPipeline.load_modules

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

Load the variant-specific transformer + scheduler from the converted MagiHuman repo and lazy-load the four cross-variant shared components from their canonical upstream HF repos:

Backwards-compatible with bundled converted repos: if any of these subfolders is present locally and listed in model_index.json, the standard component loader picks it up via super(). Otherwise the loader is told to skip the entry and we lazy-load it here.

Source
def load_modules(
    self,
    fastvideo_args: FastVideoArgs,
    loaded_modules: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Load the variant-specific transformer + scheduler from the
    converted MagiHuman repo and lazy-load the four cross-variant
    shared components from their canonical upstream HF repos:

      * text_encoder, tokenizer -> ``google/t5gemma-9b-9b-ul2``
        (gated, requires HF token with accepted terms of use)
      * audio_vae -> ``stabilityai/stable-audio-open-1.0`` (gated)
      * vae -> ``Wan-AI/Wan2.2-TI2V-5B-Diffusers``

    Backwards-compatible with bundled converted repos: if any of
    these subfolders is present locally and listed in
    ``model_index.json``, the standard component loader picks it up
    via super(). Otherwise the loader is told to skip the entry and
    we lazy-load it here.
    """
    # T5-Gemma is gated: expose `HF_API_KEY` as `HF_TOKEN` if needed.
    _ensure_hf_token_env()

    # Resolve to a local cache path so we can inspect
    # model_index.json before invoking super(). `maybe_download_model`
    # is idempotent for local paths; super() repeats the call cheaply
    # via `_load_config`.
    local_path = maybe_download_model(self.model_path)

    # Identify which cross-variant shared keys are bundled in the
    # converted repo (declared in model_index.json with a non-null
    # spec) versus absent (the umbrella scheme). Bundled keys stay
    # in `required_config_modules` and are loaded normally by super()
    # from `<model_path>/<key>/`. Absent keys are temporarily
    # dropped so super() does not fail the "every required entry
    # must appear in model_index.json" check, then lazy-loaded
    # below.
    model_index: dict[str, Any] = {}
    try:
        with open(Path(local_path) / "model_index.json") as f:
            model_index = json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        pass

    def _is_bundled(key: str) -> bool:
        spec = model_index.get(key)
        return (isinstance(spec, list | tuple) and len(spec) >= 1 and spec[0] is not None)

    deferred = []
    for key in ("text_encoder", "tokenizer", "audio_vae", "vae"):
        if key in self.required_config_modules and not _is_bundled(key):
            self.required_config_modules.remove(key)
            deferred.append(key)

    try:
        modules = super().load_modules(fastvideo_args, loaded_modules)
    finally:
        for key in deferred:
            if key not in self.required_config_modules:
                self.required_config_modules.append(key)

    # For each lazy-load key, prefer whatever super() already loaded
    # (a bundled subfolder, or a caller-provided override merged in
    # via `loaded_modules`). Fall back to the caller-provided
    # `loaded_modules` entry for keys absent from model_index.json
    # (super() never iterates those). Otherwise lazy-load from the
    # canonical upstream HF repo.
    def _resolve(key: str) -> bool:
        """Return True if `modules[key]` is already populated."""
        if modules.get(key) is not None:
            return True
        if loaded_modules and key in loaded_modules:
            modules[key] = loaded_modules[key]
            return True
        return False

    if not _resolve("text_encoder"):
        logger.info("Building T5-Gemma text encoder (lazy-load from %s)", _T5GEMMA_HF_ID)
        enc_config = T5GemmaEncoderConfig()
        enc_config.arch_config.t5gemma_model_path = _T5GEMMA_HF_ID
        modules["text_encoder"] = T5GemmaEncoderModel(enc_config)

    if not _resolve("tokenizer"):
        logger.info("Loading T5-Gemma tokenizer from %s", _T5GEMMA_HF_ID)
        modules["tokenizer"] = AutoTokenizer.from_pretrained(_T5GEMMA_HF_ID)

    if not _resolve("audio_vae"):
        logger.info(
            "Building Stable Audio Open 1.0 VAE (lazy-load from %s) — "
            "requires HF terms accepted for gated repo",
            _SA_AUDIO_HF_ID,
        )
        audio_config = OobleckVAEConfig()
        audio_config.pretrained_path = _SA_AUDIO_HF_ID
        modules["audio_vae"] = SAAudioVAEModel(audio_config)

    if not _resolve("vae"):
        modules["vae"] = self._load_video_vae(fastvideo_args)

    return modules

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanPipeline.initialize_pipeline

initialize_pipeline(fastvideo_args: FastVideoArgs) -> None
Source
def initialize_pipeline(self, fastvideo_args: FastVideoArgs) -> None:
    # MagiHuman applies `flow_shift` during timestep setup; keep the
    # scheduler constructor at its default no-op shift.
    self.modules["scheduler"] = FlowUniPCMultistepScheduler()

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanPipeline.create_pipeline_stages

create_pipeline_stages(fastvideo_args: FastVideoArgs) -> None
Source
def create_pipeline_stages(self, fastvideo_args: FastVideoArgs) -> None:
    self._add_input_and_conditioning_stages(fastvideo_args)
    self._add_base_latent_and_denoising_stages(fastvideo_args)
    self._add_decode_stages()

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanI2VPipeline

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

MagiHuman text+image-to-AV pipeline using the T2V DiT weights.

Source
class MagiHumanI2VPipeline(MagiHumanPipeline):
    """MagiHuman text+image-to-AV pipeline using the T2V DiT weights."""

    def _add_reference_image_stage(self, fastvideo_args: FastVideoArgs) -> None:
        pc = fastvideo_args.pipeline_config
        self.add_stage(
            stage_name="reference_image_stage",
            stage=MagiHumanReferenceImageStage(
                vae=self.get_module("vae"),
                vae_scale_factor=pc.vae_stride[1],
            ),
        )

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanSRPipeline

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

Two-stage MagiHuman base + SR-540p text-to-AV pipeline.

Source
class MagiHumanSRPipeline(MagiHumanPipeline):
    """Two-stage MagiHuman base + SR-540p text-to-AV pipeline."""

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

    def create_pipeline_stages(self, fastvideo_args: FastVideoArgs) -> None:
        self._add_input_and_conditioning_stages(fastvideo_args)
        self._add_base_latent_and_denoising_stages(fastvideo_args)
        self._add_sr_latent_and_denoising_stages(fastvideo_args)
        self._add_decode_stages()

    def _add_sr_latent_and_denoising_stages(self, fastvideo_args: FastVideoArgs) -> None:
        pc = fastvideo_args.pipeline_config
        dit_arch = pc.dit_config.arch_config
        sr_transformer = self.get_module("sr_transformer")
        sr_local_attn_layers = tuple(getattr(pc, "sr_local_attn_layers", ()))
        if sr_local_attn_layers and hasattr(sr_transformer, "configure_local_attention"):
            sr_transformer.configure_local_attention(
                sr_local_attn_layers,
                frame_receptive_field=pc.frame_receptive_field,
            )

        self.add_stage(
            stage_name="sr_latent_preparation_stage",
            stage=MagiHumanSRLatentPreparationStage(
                vae=self.get_module("vae"),
                vae_stride=tuple(pc.vae_stride),
                patch_size=tuple(dit_arch.patch_size),
                noise_value=pc.noise_value,
                sr_audio_noise_scale=pc.sr_audio_noise_scale,
                sr_height=pc.sr_height,
                sr_width=pc.sr_width,
                vae_scale_factor=pc.vae_stride[1],
            ),
        )
        self.add_stage(
            stage_name="sr_denoising_stage",
            stage=MagiHumanSRDenoisingStage(
                transformer=sr_transformer,
                scheduler=self.get_module("scheduler"),
                patch_size=tuple(dit_arch.patch_size),
                video_in_channels=dit_arch.video_in_channels,
                audio_in_channels=dit_arch.audio_in_channels,
                sr_num_inference_steps=pc.sr_num_inference_steps,
                sr_video_txt_guidance_scale=pc.sr_video_txt_guidance_scale,
                use_cfg_trick=pc.use_cfg_trick,
                cfg_trick_start_frame=pc.cfg_trick_start_frame,
                cfg_trick_value=pc.cfg_trick_value,
                cfg_number=pc.cfg_number,
                coords_style="v1",
            ),
        )

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanSRPipeline.create_pipeline_stages

create_pipeline_stages(fastvideo_args: FastVideoArgs) -> None
Source
def create_pipeline_stages(self, fastvideo_args: FastVideoArgs) -> None:
    self._add_input_and_conditioning_stages(fastvideo_args)
    self._add_base_latent_and_denoising_stages(fastvideo_args)
    self._add_sr_latent_and_denoising_stages(fastvideo_args)
    self._add_decode_stages()

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanSRI2VPipeline

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

Two-stage MagiHuman base + SR-540p text+image-to-AV pipeline.

Source
class MagiHumanSRI2VPipeline(MagiHumanSRPipeline):
    """Two-stage MagiHuman base + SR-540p text+image-to-AV pipeline."""

    def _add_reference_image_stage(self, fastvideo_args: FastVideoArgs) -> None:
        pc = fastvideo_args.pipeline_config
        self.add_stage(
            stage_name="reference_image_stage",
            stage=MagiHumanReferenceImageStage(
                vae=self.get_module("vae"),
                vae_scale_factor=pc.vae_stride[1],
            ),
        )

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanSR1080pPipeline

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

Two-stage MagiHuman base + SR-1080p text-to-AV pipeline.

The stage chain is identical to SR-540p. The paired pipeline config enables block-sparse local-window attention on 32 SR-DiT layers and requests the 1080p latent target.

Source
class MagiHumanSR1080pPipeline(MagiHumanSRPipeline):
    """Two-stage MagiHuman base + SR-1080p text-to-AV pipeline.

    The stage chain is identical to SR-540p. The paired pipeline config enables
    block-sparse local-window attention on 32 SR-DiT layers and requests the
    1080p latent target.
    """

fastvideo.pipelines.basic.magi_human.magi_human_pipeline.MagiHumanSR1080pI2VPipeline

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

Two-stage MagiHuman base + SR-1080p text+image-to-AV pipeline.

Source
class MagiHumanSR1080pI2VPipeline(MagiHumanSRI2VPipeline):
    """Two-stage MagiHuman base + SR-1080p text+image-to-AV pipeline."""