nodes/hao-ai-lab-FastVideo
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[refactor]: read typed config paths in the remaining model pipelines

Browse Source 
The Cosmos, DreamX-World, Flux, GLM-Image, HunyuanVideo, HyWorld,
Kandinsky5, LingBot, LongCat, Matrix-Game, SD3.5, TurboDiffusion, and
Z-Image pipelines read flow_shift, precisions, vae_tiling,
dmd_denoising_steps, output_type, offload, autocast, and the LongCat BSA
settings from typed paths. LingBot's refine switch reads
pipeline.ltx2.refine.enabled, so the typed field also turns its refiner off.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012M91wnVFmPEvJ9h39r5BH7
This commit is contained in:
Davids048
2026-10-03 06:09:11 +00:00
co-authored by Claude Opus 5.5
parent 61dffbe904
commit ca7965ed28
42 changed files with 180 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fastvideo/pipelines/basic/cosmos/cosmos2_5_pipeline.py
+2 -2
View File
@@ -1,7 +1,7 @@
# SPDX-License-Identifier: Apache-2.0
"""Cosmos 2.5 pipeline entry (staged pipeline)."""
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.logger import init_logger
from fastvideo.pipelines.composed_pipeline_base import ComposedPipelineBase
from fastvideo.pipelines.stages import (ConditioningStage, Cosmos25AutoDenoisingStage,
@@ -16,7 +16,7 @@ class Cosmos2_5Pipeline(ComposedPipelineBase):
_required_config_modules = ["text_encoder", "tokenizer", "vae", "transformer", "scheduler", "safety_checker"]
def create_pipeline_stages(self, resolved_config: FastVideoArgs):
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig):
logger.info("Creating Cosmos 2.5 pipeline stages...")
self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
fastvideo/pipelines/basic/cosmos/cosmos_pipeline.py
+4 -4
View File
@@ -6,7 +6,7 @@ This module contains an implementation of the Cosmos video diffusion pipeline
using the modular pipeline architecture.
"""
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.logger import init_logger
from fastvideo.models.schedulers.scheduling_flow_match_euler_discrete import (FlowMatchEulerDiscreteScheduler)
from fastvideo.pipelines.composed_pipeline_base import ComposedPipelineBase
@@ -21,9 +21,9 @@ class Cosmos2VideoToWorldPipeline(ComposedPipelineBase):
_required_config_modules = ["text_encoder", "tokenizer", "vae", "transformer", "scheduler", "safety_checker"]
def initialize_pipeline(self, resolved_config: FastVideoArgs):
def initialize_pipeline(self, resolved_config: ResolvedGeneratorConfig):
scheduler = FlowMatchEulerDiscreteScheduler(
shift=resolved_config.pipeline_config.flow_shift,
shift=resolved_config.pipeline.flow_shift,
use_karras_sigmas=True,
)
scheduler.config.sigma_max = 80.0
@@ -35,7 +35,7 @@ class Cosmos2VideoToWorldPipeline(ComposedPipelineBase):
scheduler.sigma_data = 1.0
self.modules["scheduler"] = scheduler
def create_pipeline_stages(self, resolved_config: FastVideoArgs):
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig):
"""Set up pipeline stages with proper dependency injection."""
self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
fastvideo/pipelines/basic/dreamx_world/ar_denoising.py
+4 -4
View File
@@ -8,7 +8,7 @@ from typing import Any
import torch
from tqdm.auto import tqdm
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.pipelines.basic.dreamx_world.stages import DREAMX_Y_CAMERA_KEY
from fastvideo.pipelines.pipeline_batch_info import ForwardBatch
from fastvideo.pipelines.stages.denoising import DenoisingStage
@@ -25,18 +25,18 @@ class DreamXWorldARCausalDenoisingStage(DenoisingStage):
self.num_frame_per_block = int(getattr(self.transformer, "num_frame_per_block", 3))
self.local_attn_size = int(getattr(self.transformer, "local_attn_size", 12))
def forward(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> ForwardBatch:
def forward(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> ForwardBatch:
assert batch.latents is not None, "latents must be prepared before DreamX AR denoising"
assert batch.prompt_embeds, "prompt embeds must be prepared before DreamX AR denoising"
latents = batch.latents
device = latents.device
target_dtype = torch.bfloat16
autocast_enabled = device.type == "cuda" and not resolved_config.disable_autocast
autocast_enabled = device.type == "cuda" and not resolved_config.engine.disable_autocast
frame_seq_length = (latents.shape[-2] // self.transformer.patch_size[1]) * (latents.shape[-1] //
self.transformer.patch_size[2])
timesteps = torch.tensor(
tuple(getattr(resolved_config.pipeline_config, "dmd_denoising_steps", (1000, 750, 500, 250))),
tuple(resolved_config.pipeline.dmd_denoising_steps),
dtype=torch.long,
).cpu()
if getattr(resolved_config.pipeline_config, "warp_denoising_step", True):
fastvideo/pipelines/basic/dreamx_world/dreamx_world_ar_pipeline.py
+4 -4
View File
@@ -3,7 +3,7 @@
from fastvideo.api.sampling_param import SamplingParam
from fastvideo.configs.pipelines.dreamx_world import DreamXWorld5BARPipelineConfig
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.logger import init_logger
from fastvideo.models.schedulers.scheduling_flow_match_euler_discrete import FlowMatchEulerDiscreteScheduler
from fastvideo.pipelines import ComposedPipelineBase, LoRAPipeline
@@ -31,11 +31,11 @@ class DreamXWorldARPipeline(LoRAPipeline, ComposedPipelineBase):
pipeline_config_cls = DreamXWorld5BARPipelineConfig
sampling_params_cls = SamplingParam
def initialize_pipeline(self, resolved_config: FastVideoArgs):
self.modules["scheduler"] = FlowMatchEulerDiscreteScheduler(shift=resolved_config.pipeline_config.flow_shift)
def initialize_pipeline(self, resolved_config: ResolvedGeneratorConfig):
self.modules["scheduler"] = FlowMatchEulerDiscreteScheduler(shift=resolved_config.pipeline.flow_shift)
self.modules["scheduler"].set_timesteps(1000)
def create_pipeline_stages(self, resolved_config: FastVideoArgs) -> None:
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig) -> None:
self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
self.add_stage(stage_name="prompt_encoding_stage",
stage=TextEncodingStage(
fastvideo/pipelines/basic/dreamx_world/dreamx_world_pipeline.py
+4 -4
View File
@@ -4,7 +4,7 @@
from fastvideo.api.sampling_param import SamplingParam
from fastvideo.pipelines.basic.wan.stages.conditioning import WanFirstFrameEncodingStage
from fastvideo.pipelines.basic.wan.stages.denoising import WanDenoisingStage
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.logger import init_logger
from fastvideo.models.schedulers.scheduling_flow_match_euler_discrete import FlowMatchEulerDiscreteScheduler
from fastvideo.pipelines import ComposedPipelineBase, LoRAPipeline
@@ -29,10 +29,10 @@ class DreamXWorldPipeline(LoRAPipeline, ComposedPipelineBase):
pipeline_config_cls = DreamXWorld5BCamPipelineConfig
sampling_params_cls = SamplingParam
def initialize_pipeline(self, resolved_config: FastVideoArgs):
self.modules["scheduler"] = FlowMatchEulerDiscreteScheduler(shift=resolved_config.pipeline_config.flow_shift)
def initialize_pipeline(self, resolved_config: ResolvedGeneratorConfig):
self.modules["scheduler"] = FlowMatchEulerDiscreteScheduler(shift=resolved_config.pipeline.flow_shift)
def create_pipeline_stages(self, resolved_config: FastVideoArgs) -> None:
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig) -> None:
self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
self.add_stage(
fastvideo/pipelines/basic/dreamx_world/stages.py
+7 -7
View File
@@ -5,7 +5,7 @@ from __future__ import annotations
import torch
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.logger import init_logger
from fastvideo.pipelines.pipeline_batch_info import ForwardBatch
from fastvideo.pipelines.stages.base import PipelineStage
@@ -25,7 +25,7 @@ class DreamXWorldCameraConditioningStage(PipelineStage):
def forward(
self,
batch: ForwardBatch,
resolved_config: FastVideoArgs,
resolved_config: ResolvedGeneratorConfig,
) -> ForwardBatch:
del resolved_config
if DREAMX_Y_CAMERA_KEY in batch.extra:
@@ -67,7 +67,7 @@ class DreamXWorldCameraConditioningStage(PipelineStage):
def verify_output(
self,
batch: ForwardBatch,
resolved_config: FastVideoArgs,
resolved_config: ResolvedGeneratorConfig,
) -> VerificationResult:
del resolved_config
result = VerificationResult()
@@ -97,7 +97,7 @@ class DreamXWorldImageVAEEncodingStage(PipelineStage):
def forward(
self,
batch: ForwardBatch,
resolved_config: FastVideoArgs,
resolved_config: ResolvedGeneratorConfig,
) -> ForwardBatch:
if batch.pil_image is None:
# No conditioning image: the causal denoiser falls back to
@@ -130,8 +130,8 @@ class DreamXWorldImageVAEEncodingStage(PipelineStage):
image = image[:, :, :1]
image = image.to(device=device, dtype=torch.float32)
vae_dtype = PRECISION_TO_TYPE[resolved_config.pipeline_config.vae_precision]
vae_autocast_enabled = (vae_dtype != torch.float32) and not resolved_config.disable_autocast
vae_dtype = PRECISION_TO_TYPE[resolved_config.engine.precision.vae]
vae_autocast_enabled = (vae_dtype != torch.float32) and not resolved_config.engine.disable_autocast
self.vae = self.vae.to(device)
with torch.autocast(device_type="cuda", dtype=vae_dtype, enabled=vae_autocast_enabled):
if not vae_autocast_enabled:
@@ -151,6 +151,6 @@ class DreamXWorldImageVAEEncodingStage(PipelineStage):
batch.image_latent = latent
return batch
def verify_input(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> VerificationResult:
def verify_input(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> VerificationResult:
result = VerificationResult()
return result
fastvideo/pipelines/basic/flux/flux_pipeline.py
+2 -2
View File
@@ -2,7 +2,7 @@
from __future__ import annotations
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.pipelines.composed_pipeline_base import ComposedPipelineBase
from fastvideo.pipelines.stages.flux_stages import (
FluxConditioningStage,
@@ -28,7 +28,7 @@ class FluxPipeline(ComposedPipelineBase):
"tokenizer_2",
]
def create_pipeline_stages(self, resolved_config: FastVideoArgs) -> None:
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig) -> None:
self.add_stage(stage_name="input_validation_stage", stage=FluxInputValidationStage())
self.add_stage(
fastvideo/pipelines/basic/flux_2/flux_2_latent_preparation.py
+2 -2
View File
@@ -10,7 +10,7 @@ spatial for VAE decode. This stage prepares (B, 128, T, H//2, W//2).
import torch
from diffusers.utils.torch_utils import randn_tensor
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.pipelines.pipeline_batch_info import ForwardBatch
from fastvideo.pipelines.stages.latent_preparation import LatentPreparationStage
@@ -28,7 +28,7 @@ class Flux2LatentPreparationStage(LatentPreparationStage):
def forward(
self,
batch: ForwardBatch,
resolved_config: FastVideoArgs,
resolved_config: ResolvedGeneratorConfig,
) -> ForwardBatch:
"""Prepare latents with Flux2 packed half-spatial shape."""
from fastvideo.distributed import get_local_torch_device
fastvideo/pipelines/basic/flux_2/flux_2_pipeline.py
+2 -2
View File
@@ -7,7 +7,7 @@ This module contains an implementation of the Flux2 image diffusion pipeline
using the modular pipeline architecture.
"""
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.logger import init_logger
from fastvideo.pipelines import ComposedPipelineBase, LoRAPipeline
from fastvideo.pipelines.basic.flux_2.flux_2_latent_preparation import (
@@ -39,7 +39,7 @@ class Flux2Pipeline(LoRAPipeline, ComposedPipelineBase):
"scheduler",
]
def create_pipeline_stages(self, resolved_config: FastVideoArgs) -> None:
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig) -> None:
"""Set up pipeline stages with proper dependency injection."""
self.add_stage(
fastvideo/pipelines/basic/flux_2/flux_2_text_encoding.py
+4 -4
View File
@@ -7,7 +7,7 @@ from typing import Any
import torch
from fastvideo.distributed import get_local_torch_device
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.forward_context import set_forward_context
from fastvideo.pipelines.pipeline_batch_info import ForwardBatch, embedded_cfg_scale_for_batch
from fastvideo.pipelines.stages.text_encoding import TextEncodingStage
@@ -50,14 +50,14 @@ def _prepare_flux2_text_ids(prompt_embeds: torch.Tensor) -> torch.Tensor:
class Flux2TextEncodingStage(TextEncodingStage):
"""Text encoding for Flux2 full and Klein variants."""
def _uses_embedded_guidance(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> bool:
def _uses_embedded_guidance(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> bool:
return embedded_cfg_scale_for_batch(batch, resolved_config) is not None
@torch.no_grad()
def forward(
self,
batch: ForwardBatch,
resolved_config: FastVideoArgs,
resolved_config: ResolvedGeneratorConfig,
) -> ForwardBatch:
if self._uses_embedded_guidance(batch, resolved_config):
batch.do_classifier_free_guidance = False
@@ -87,7 +87,7 @@ class Flux2TextEncodingStage(TextEncodingStage):
def encode_flux2_full_text(
self,
text: str | list[str],
resolved_config: FastVideoArgs,
resolved_config: ResolvedGeneratorConfig,
max_length: int | None = None,
) -> tuple[torch.Tensor, torch.Tensor]:
tokenizer = self.tokenizers[0]
fastvideo/pipelines/basic/flux_2/flux_2_timestep_preparation.py
+2 -2
View File
@@ -7,7 +7,7 @@ import numpy as np
import torch
from fastvideo.distributed import get_local_torch_device
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.pipelines.pipeline_batch_info import ForwardBatch
from fastvideo.pipelines.stages.timestep_preparation import TimestepPreparationStage
@@ -36,7 +36,7 @@ class Flux2TimestepPreparationStage(TimestepPreparationStage):
def forward(
self,
batch: ForwardBatch,
resolved_config: FastVideoArgs,
resolved_config: ResolvedGeneratorConfig,
) -> ForwardBatch:
scheduler = self.scheduler
device = get_local_torch_device()
fastvideo/pipelines/basic/gamecraft/gamecraft_pipeline.py
+2 -2
View File
@@ -6,7 +6,7 @@ This module implements the HunyuanGameCraft pipeline for camera/action-condition
video generation with the modular pipeline architecture.
"""
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.logger import init_logger
from fastvideo.pipelines.composed_pipeline_base import ComposedPipelineBase
from fastvideo.pipelines.stages import (
@@ -43,7 +43,7 @@ class HunyuanGameCraftPipeline(ComposedPipelineBase):
"scheduler",
]
def create_pipeline_stages(self, resolved_config: FastVideoArgs):
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig):
"""Set up pipeline stages with proper dependency injection."""
self.add_stage(
fastvideo/pipelines/basic/gen3c/gen3c_pipeline.py
+3 -3
View File
@@ -3,7 +3,7 @@
from diffusers import EDMEulerScheduler
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.logger import init_logger
from fastvideo.pipelines.composed_pipeline_base import ComposedPipelineBase
from fastvideo.pipelines.stages import (DecodingStage, Gen3CCFGPolicyStage, Gen3CConditioningStage, Gen3CDenoisingStage,
@@ -31,7 +31,7 @@ class Gen3CPipeline(ComposedPipelineBase):
"scheduler",
]
def initialize_pipeline(self, resolved_config: FastVideoArgs):
def initialize_pipeline(self, resolved_config: ResolvedGeneratorConfig):
scheduler = self.modules.get("scheduler")
if scheduler is not None and hasattr(scheduler, "precondition_inputs"):
return
@@ -49,7 +49,7 @@ class Gen3CPipeline(ComposedPipelineBase):
sigma_data=float(getattr(resolved_config.pipeline_config, "sigma_data", 0.5)),
)
def create_pipeline_stages(self, resolved_config: FastVideoArgs):
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig):
"""Set up pipeline stages with proper dependency injection."""
self.add_stage(stage_name="cfg_policy_stage", stage=Gen3CCFGPolicyStage())
fastvideo/pipelines/basic/glm_image/glm_image_pipeline.py
+3 -3
View File
@@ -1,5 +1,5 @@
# SPDX-License-Identifier: Apache-2.0
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.logger import init_logger
from fastvideo.models.schedulers.scheduling_flow_match_euler_discrete import (
FlowMatchEulerDiscreteScheduler, )
@@ -30,10 +30,10 @@ class GlmImagePipeline(LoRAPipeline, ComposedPipelineBase):
_optional_config_modules: list[str] = []
def initialize_pipeline(self, resolved_config: FastVideoArgs) -> None:
def initialize_pipeline(self, resolved_config: ResolvedGeneratorConfig) -> None:
self.modules["scheduler"] = FlowMatchEulerDiscreteScheduler(shift=1.0)
def create_pipeline_stages(self, resolved_config: FastVideoArgs) -> None:
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig) -> None:
self.add_stage(
stage_name="input_validation_stage",
stage=InputValidationStage(),
fastvideo/pipelines/basic/glm_image/stages/before_denoising.py
+4 -4
View File
@@ -8,7 +8,7 @@ import numpy as np
import torch
from fastvideo.distributed import get_local_torch_device
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.logger import init_logger
from fastvideo.pipelines.pipeline_batch_info import ForwardBatch
from fastvideo.pipelines.stages.base import PipelineStage
@@ -123,7 +123,7 @@ class GlmImageBeforeDenoisingStage(PipelineStage):
self.vl_processor = processor
@torch.no_grad()
def forward(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> ForwardBatch:
def forward(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> ForwardBatch:
device = get_local_torch_device()
dtype = torch.bfloat16
th, tw, pth, ptw = _grid_dims(batch.height, batch.width)
@@ -258,12 +258,12 @@ class GlmImageBeforeDenoisingStage(PipelineStage):
batch.extra["glm_prior_token_image_ids"] = torch.cat(upsampled, dim=0)
batch.extra["glm_source_image_grid_thw"] = src_grids_up
def verify_input(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> VerificationResult:
def verify_input(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> VerificationResult:
result = VerificationResult()
result.add_check("prompt", batch.prompt, V.string_not_empty)
return result
def verify_output(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> VerificationResult:
def verify_output(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> VerificationResult:
result = VerificationResult()
result.add_check("prior_token_id", batch.prior_token_id, V.is_tensor)
return result
fastvideo/pipelines/basic/glm_image/stages/condition_encoding.py
+2 -2
View File
@@ -4,7 +4,7 @@ from __future__ import annotations
import torch
from fastvideo.distributed import get_local_torch_device
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.forward_context import set_forward_context
from fastvideo.image_processor import ImageProcessor
from fastvideo.models.dits.glm_image import GlmImageKVCache
@@ -23,7 +23,7 @@ class GlmImageConditionEncodingStage(PipelineStage):
self.image_processor = ImageProcessor(vae_scale_factor=_CONDITION_MULTIPLE_OF)
@torch.no_grad()
def forward(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> ForwardBatch:
def forward(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> ForwardBatch:
if batch.pil_image is None:
return batch
fastvideo/pipelines/basic/glm_image/stages/decoding.py
+5 -5
View File
@@ -4,7 +4,7 @@ from __future__ import annotations
import torch
from fastvideo.distributed import get_local_torch_device
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.pipelines.stages.decoding import DecodingStage
from fastvideo.utils import PRECISION_TO_TYPE
@@ -12,19 +12,19 @@ from fastvideo.utils import PRECISION_TO_TYPE
class GlmImageDecodingStage(DecodingStage):
@torch.no_grad()
def decode(self, latents: torch.Tensor, resolved_config: FastVideoArgs) -> torch.Tensor:
def decode(self, latents: torch.Tensor, resolved_config: ResolvedGeneratorConfig) -> torch.Tensor:
self.vae.to(get_local_torch_device())
latents = latents.to(get_local_torch_device())
vae_dtype = PRECISION_TO_TYPE[resolved_config.pipeline_config.vae_precision]
vae_autocast = (vae_dtype != torch.float32 and not resolved_config.disable_autocast)
vae_dtype = PRECISION_TO_TYPE[resolved_config.engine.precision.vae]
vae_autocast = (vae_dtype != torch.float32 and not resolved_config.engine.disable_autocast)
latents = self._denormalize_latents(latents, resolved_config)
if latents.dim() == 5:
latents = latents.squeeze(2)
with torch.autocast(device_type="cuda", dtype=vae_dtype, enabled=vae_autocast):
if resolved_config.pipeline_config.vae_tiling:
if resolved_config.pipeline.vae_tiling:
self.vae.enable_tiling()
if not vae_autocast:
latents = latents.to(vae_dtype)
fastvideo/pipelines/basic/glm_image/stages/denoising.py
+3 -3
View File
@@ -6,7 +6,7 @@ import torch
from fastvideo.attention.backends.sdpa import SDPAMetadata
from fastvideo.distributed import get_local_torch_device
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.forward_context import set_forward_context
from fastvideo.pipelines.pipeline_batch_info import ForwardBatch
from fastvideo.pipelines.stages.denoising import DenoisingStage
@@ -17,7 +17,7 @@ from fastvideo.platforms import AttentionBackendEnum
class GlmImageDenoisingStage(DenoisingStage):
def verify_input(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> VerificationResult:
def verify_input(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> VerificationResult:
result = VerificationResult()
result.add_check("timesteps", batch.timesteps, [V.is_tensor, V.min_dims(1)])
latents = getattr(batch, "latent", getattr(batch, "latents", None))
@@ -26,7 +26,7 @@ class GlmImageDenoisingStage(DenoisingStage):
result.add_check("prompt_embeds", batch.prompt_embeds, V.list_not_empty)
return result
def forward(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> ForwardBatch:
def forward(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> ForwardBatch:
device = get_local_torch_device()
dtype = torch.bfloat16
guidance_scale = batch.guidance_scale
fastvideo/pipelines/basic/hunyuan/hunyuan_pipeline.py
+2 -2
View File
@@ -6,7 +6,7 @@ This module contains an implementation of the Hunyuan video diffusion pipeline
using the modular pipeline architecture.
"""
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.logger import init_logger
from fastvideo.pipelines.composed_pipeline_base import ComposedPipelineBase
from fastvideo.pipelines.stages import (ConditioningStage, DecodingStage, DenoisingStage, InputValidationStage,
@@ -23,7 +23,7 @@ class HunyuanVideoPipeline(ComposedPipelineBase):
"text_encoder", "text_encoder_2", "tokenizer", "tokenizer_2", "vae", "transformer", "scheduler"
]
def create_pipeline_stages(self, resolved_config: FastVideoArgs):
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig):
"""Set up pipeline stages with proper dependency injection."""
self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
fastvideo/pipelines/basic/hunyuan15/hunyuan15_2sr_pipeline.py
+4 -4
View File
@@ -7,7 +7,7 @@ using the modular pipeline architecture.
"""
import torch
import time
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.pipelines.pipeline_batch_info import ForwardBatch
from fastvideo.logger import init_logger
from fastvideo.pipelines.composed_pipeline_base import ComposedPipelineBase
@@ -28,7 +28,7 @@ class HunyuanVideo152SRPipeline(ComposedPipelineBase):
"transformer_3", "scheduler", "upsampler", "upsampler_2"
]
def create_pipeline_stages(self, resolved_config: FastVideoArgs):
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig):
"""Set up pipeline stages with proper dependency injection."""
self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
@@ -81,14 +81,14 @@ class HunyuanVideo152SRPipeline(ComposedPipelineBase):
def forward(
self,
batch: ForwardBatch,
resolved_config: FastVideoArgs,
resolved_config: ResolvedGeneratorConfig,
) -> ForwardBatch:
"""
Generate a video or image using the pipeline.
Args:
batch: The batch to generate from.
fastvideo_args: The inference arguments.
resolved_config: The resolved runtime config.
Returns:
ForwardBatch: The batch with the generated video or image.
"""
fastvideo/pipelines/basic/hunyuan15/hunyuan15_i2v_pipeline.py
+2 -2
View File
@@ -6,7 +6,7 @@ This module contains an implementation of the Hunyuan video diffusion pipeline
using the modular pipeline architecture.
"""
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.logger import init_logger
from fastvideo.pipelines.composed_pipeline_base import ComposedPipelineBase
from fastvideo.pipelines.stages import (ConditioningStage, DecodingStage, DenoisingStage, InputValidationStage,
@@ -24,7 +24,7 @@ class HunyuanVideo15ImageToVideoPipeline(ComposedPipelineBase):
"text_encoder", "text_encoder_2", "tokenizer", "tokenizer_2", "vae", "transformer", "scheduler"
]
def create_pipeline_stages(self, resolved_config: FastVideoArgs):
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig):
"""Set up pipeline stages with proper dependency injection."""
self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
fastvideo/pipelines/basic/hunyuan15/hunyuan15_pipeline.py
+2 -2
View File
@@ -6,7 +6,7 @@ This module contains an implementation of the Hunyuan video diffusion pipeline
using the modular pipeline architecture.
"""
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.logger import init_logger
from fastvideo.pipelines.composed_pipeline_base import ComposedPipelineBase
from fastvideo.pipelines.stages import (ConditioningStage, DecodingStage, DenoisingStage, InputValidationStage,
@@ -24,7 +24,7 @@ class HunyuanVideo15Pipeline(ComposedPipelineBase):
"text_encoder", "text_encoder_2", "tokenizer", "tokenizer_2", "vae", "transformer", "scheduler"
]
def create_pipeline_stages(self, resolved_config: FastVideoArgs):
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig):
"""Set up pipeline stages with proper dependency injection."""
self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
fastvideo/pipelines/basic/hunyuan15/hunyuan15_sr_pipeline.py
+4 -4
View File
@@ -7,7 +7,7 @@ using the modular pipeline architecture.
"""
import torch
import time
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.pipelines.pipeline_batch_info import ForwardBatch
from fastvideo.logger import init_logger
from fastvideo.pipelines.composed_pipeline_base import ComposedPipelineBase
@@ -28,7 +28,7 @@ class HunyuanVideo15SRPipeline(ComposedPipelineBase):
"scheduler", "upsampler"
]
def create_pipeline_stages(self, resolved_config: FastVideoArgs):
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig):
"""Set up pipeline stages with proper dependency injection."""
self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
@@ -72,14 +72,14 @@ class HunyuanVideo15SRPipeline(ComposedPipelineBase):
def forward(
self,
batch: ForwardBatch,
resolved_config: FastVideoArgs,
resolved_config: ResolvedGeneratorConfig,
) -> ForwardBatch:
"""
Generate a video or image using the pipeline.
Args:
batch: The batch to generate from.
fastvideo_args: The inference arguments.
resolved_config: The resolved runtime config.
Returns:
ForwardBatch: The batch with the generated video or image.
"""
fastvideo/pipelines/basic/hyworld/hyworld_pipeline.py
+2 -2
View File
@@ -7,7 +7,7 @@ using the modular pipeline architecture with HYWorld-specific denoising stage
for chunk-based video generation with context frame selection.
"""
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.logger import init_logger
from fastvideo.pipelines.composed_pipeline_base import ComposedPipelineBase
from fastvideo.pipelines.stages import (ConditioningStage, DecodingStage, HYWorldDenoisingStage, InputValidationStage,
@@ -35,7 +35,7 @@ class HYWorldPipeline(ComposedPipelineBase):
"image_encoder", "feature_extractor"
]
def create_pipeline_stages(self, resolved_config: FastVideoArgs):
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig):
"""Set up pipeline stages with HYWorld-specific denoising stage."""
self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
fastvideo/pipelines/basic/kandinsky5/kandinsky5_dmd_pipeline.py
+2 -2
View File
@@ -10,7 +10,7 @@ validation callback.
from __future__ import annotations
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.pipelines.composed_pipeline_base import ComposedPipelineBase
from fastvideo.pipelines.stages.input_validation import InputValidationStage
from fastvideo.pipelines.stages.kandinsky5 import (
@@ -35,7 +35,7 @@ class Kandinsky5DMDPipeline(ComposedPipelineBase):
"vae",
]
def create_pipeline_stages(self, resolved_config: FastVideoArgs) -> None:
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig) -> None:
self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
self.add_stage(
fastvideo/pipelines/basic/kandinsky5/kandinsky5_i2v_pipeline.py
+2 -2
View File
@@ -2,7 +2,7 @@
from __future__ import annotations
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.pipelines.composed_pipeline_base import ComposedPipelineBase
from fastvideo.pipelines.stages.input_validation import InputValidationStage
from fastvideo.pipelines.stages.kandinsky5 import (
@@ -29,7 +29,7 @@ class Kandinsky5I2VPipeline(ComposedPipelineBase):
"vae",
]
def create_pipeline_stages(self, resolved_config: FastVideoArgs) -> None:
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig) -> None:
self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
self.add_stage(
fastvideo/pipelines/basic/kandinsky5/kandinsky5_pipeline.py
+2 -2
View File
@@ -2,7 +2,7 @@
from __future__ import annotations
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.pipelines.composed_pipeline_base import ComposedPipelineBase
from fastvideo.pipelines.stages.input_validation import InputValidationStage
from fastvideo.pipelines.stages.kandinsky5 import (
@@ -27,7 +27,7 @@ class Kandinsky5T2VPipeline(ComposedPipelineBase):
"vae",
]
def create_pipeline_stages(self, resolved_config: FastVideoArgs) -> None:
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig) -> None:
self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
self.add_stage(
fastvideo/pipelines/basic/lingbot_video/lingbot_video_pipeline.py
+8 -6
View File
@@ -3,7 +3,7 @@
from typing import Any
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.pipelines import ComposedPipelineBase, LoRAPipeline
from fastvideo.pipelines.basic.lingbot_video.stages import (
LingBotVideoDenoisingStage,
@@ -27,27 +27,29 @@ class LingBotVideoPipeline(LoRAPipeline, ComposedPipelineBase):
def load_modules(
self,
resolved_config: FastVideoArgs,
resolved_config: ResolvedGeneratorConfig,
loaded_modules: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Load the optional refiner DiT and the VAE encoder only when declared."""
model_index = self._load_config(self.model_path)
required = list(type(self)._required_config_modules)
load_refiner = "transformer_2" in model_index and getattr(resolved_config, "refine_enabled", None) is not False
# The shared stage-2 refine switch: ``pipeline.preset_overrides.refine.enabled`` and the ``refine_enabled``
# key resolve into ``pipeline.ltx2.refine.enabled``; ``None`` loads the refiner when the checkpoint has one.
load_refiner = "transformer_2" in model_index and resolved_config.pipeline.ltx2.refine.enabled is not False
if load_refiner:
required.append("transformer_2")
resolved_config.pipeline_config.vae_config.load_encoder = True
self._required_config_modules = required
return super().load_modules(resolved_config, loaded_modules)
def initialize_pipeline(self, resolved_config: FastVideoArgs) -> None:
def initialize_pipeline(self, resolved_config: ResolvedGeneratorConfig) -> None:
"""Apply the released runtime flow shift to the loaded scheduler."""
shift = resolved_config.pipeline_config.flow_shift
shift = resolved_config.pipeline.flow_shift
if shift is None:
raise ValueError("LingBot-Video requires a flow shift")
self.get_module("scheduler").set_shift(float(shift))
def create_pipeline_stages(self, resolved_config: FastVideoArgs) -> None:
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig) -> None:
"""Create base generation and the optional decoded-video refiner stages."""
refiner = self.get_module("transformer_2")
self.add_stage(
fastvideo/pipelines/basic/lingbot_video/stages.py
+9 -9
View File
@@ -8,7 +8,7 @@ import torch
import torch.nn.functional as F
from fastvideo.distributed import get_local_torch_device
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.pipelines.pipeline_batch_info import ForwardBatch
from fastvideo.pipelines.stages.base import PipelineStage
from fastvideo.pipelines.stages.input_validation import InputValidationStage
@@ -50,7 +50,7 @@ class LingBotVideoInputValidationStage(InputValidationStage):
def __init__(self, refiner_enabled: bool = False) -> None:
self.refiner_enabled = refiner_enabled
def _generate_seeds(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> None:
def _generate_seeds(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> None:
"""Use one device-local generator, matching the official production runner."""
del resolved_config
if batch.seed is None:
@@ -60,7 +60,7 @@ class LingBotVideoInputValidationStage(InputValidationStage):
batch.seeds = [batch.seed]
batch.generator = torch.Generator(device=get_local_torch_device()).manual_seed(batch.seed)
def forward(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> ForwardBatch:
def forward(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> ForwardBatch:
"""Run shared validation, then enforce LingBot temporal and spatial geometry."""
batch = super().forward(batch, resolved_config)
if not isinstance(batch.num_frames, int):
@@ -73,7 +73,7 @@ class LingBotVideoInputValidationStage(InputValidationStage):
raise ValueError(f"height and width must be divisible by 16, got {batch.height}x{batch.width}")
if isinstance(batch.prompt, list) and len(batch.prompt) != 1:
raise ValueError("LingBot-Video currently supports prompt batch size one")
if self.refiner_enabled and resolved_config.output_type == "latent":
if self.refiner_enabled and resolved_config.pipeline.output_type == "latent":
raise ValueError("LingBot-Video refinement requires decoded pixel output")
return batch
@@ -84,7 +84,7 @@ class LingBotVideoLatentPreparationStage(PipelineStage):
def __init__(self, transformer) -> None:
self.transformer = transformer
def forward(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> ForwardBatch:
def forward(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> ForwardBatch:
"""Generate or validate one normalized fp32 latent video."""
del resolved_config
if not all(isinstance(value, int) for value in (batch.num_frames, batch.height, batch.width)):
@@ -151,7 +151,7 @@ class LingBotVideoRefinerPreparationStage(PipelineStage):
std = torch.tensor(self.vae.config.latents_std, device=device, dtype=torch.float32).view(1, -1, 1, 1, 1)
return ((latents.float() - mean) / std).to(latents)
def forward(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> ForwardBatch:
def forward(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> ForwardBatch:
"""Prepare high-resolution refiner latents and its exact truncated sigma schedule."""
if batch.output is None or batch.output.ndim != 5:
raise ValueError("LingBot-Video refinement requires a decoded base video")
@@ -176,7 +176,7 @@ class LingBotVideoRefinerPreparationStage(PipelineStage):
batch.extra["lingbot_video_base_shape"] = tuple(batch.output.shape)
batch.output = None
shift = resolved_config.pipeline_config.flow_shift
shift = resolved_config.pipeline.flow_shift
if shift is None:
raise ValueError("LingBot-Video refinement requires a flow shift")
sigmas = _compute_refiner_sigmas(
@@ -188,7 +188,7 @@ class LingBotVideoRefinerPreparationStage(PipelineStage):
)
self.scheduler.set_timesteps(len(sigmas), device=device, sigmas=sigmas, shift=1.0)
batch.timesteps = self.scheduler.timesteps
if getattr(resolved_config, "vae_cpu_offload", False):
if resolved_config.engine.offload.vae:
self.vae.to("cpu")
return batch
@@ -277,7 +277,7 @@ class LingBotVideoDenoisingStage(PipelineStage):
scale = batch.guidance_scale_2 if self.refiner else batch.guidance_scale
return scale is not None and scale > 1.0
def forward(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> ForwardBatch:
def forward(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> ForwardBatch:
"""Denoise latents while keeping scheduler samples and predictions in fp32."""
if batch.latents is None or batch.timesteps is None:
raise ValueError("LingBot-Video denoising requires latents and timesteps")
fastvideo/pipelines/basic/lingbotworld2/causal_fast_pipeline.py
+7 -7
View File
@@ -11,7 +11,7 @@ import torchvision.transforms.functional as TF
from fastvideo.distributed import get_local_torch_device
from fastvideo.distributed.parallel_state import get_sp_world_size
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.logger import init_logger
from fastvideo.models.dits.lingbotworld2.cam_utils import (
compute_relative_poses,
@@ -37,7 +37,7 @@ logger = init_logger(__name__)
class LingBotWorld2TextEncodingStage(TextEncodingStage):
"""Keep LingBot World 2 T5 attention masks so DiT context matches the source pipeline."""
def forward(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> ForwardBatch:
def forward(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> ForwardBatch:
"""Initialize mask storage before running the shared text-encoding stage."""
if batch.prompt_attention_mask is None:
batch.prompt_attention_mask = []
@@ -187,7 +187,7 @@ class LingBotWorld2CausalFastGenerationStage(PipelineStage):
w: int,
frames: int,
mask: torch.Tensor,
resolved_config: FastVideoArgs,
resolved_config: ResolvedGeneratorConfig,
) -> torch.Tensor:
"""Encode the first-frame conditioning video and prepend mask channels."""
device = get_local_torch_device()
@@ -212,12 +212,12 @@ class LingBotWorld2CausalFastGenerationStage(PipelineStage):
latent_condition -= self.vae.shift_factor.to(latent_condition.device, latent_condition.dtype)
latent_condition = latent_condition * self.vae.scaling_factor.to(latent_condition.device,
latent_condition.dtype)
if resolved_config.vae_cpu_offload:
if resolved_config.engine.offload.vae:
self.vae.to("cpu")
return torch.concat([mask, latent_condition[0]], dim=0)
@torch.no_grad()
def forward(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> ForwardBatch:
def forward(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> ForwardBatch:
"""Execute LingBot World 2 causal-fast generation and store final latents on the batch."""
device = get_local_torch_device()
cfg = resolved_config.pipeline_config.dit_config.arch_config
@@ -336,11 +336,11 @@ class LingBotWorld2CausalFastPipeline(LoRAPipeline, ComposedPipelineBase):
_required_config_modules = ["text_encoder", "tokenizer", "vae", "transformer", "scheduler"]
def initialize_pipeline(self, resolved_config: FastVideoArgs):
def initialize_pipeline(self, resolved_config: ResolvedGeneratorConfig):
if "scheduler" not in self.modules:
self.modules["scheduler"] = FlowUniPCMultistepScheduler(shift=1.0, use_dynamic_shifting=False)
def create_pipeline_stages(self, resolved_config: FastVideoArgs) -> None:
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig) -> None:
"""Set up the LingBot World 2 causal-fast pipeline stages."""
self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
self.add_stage(
fastvideo/pipelines/basic/longcat/longcat_i2v_pipeline.py
+9 -8
View File
@@ -11,7 +11,7 @@ Supports:
- Refinement I2V for 720p upscaling (with refinement LoRA + BSA)
"""
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig, thaw
from fastvideo.logger import init_logger
from fastvideo.pipelines import ComposedPipelineBase, LoRAPipeline
from fastvideo.pipelines.stages import (
@@ -42,21 +42,22 @@ class LongCatImageToVideoPipeline(LoRAPipeline, ComposedPipelineBase):
_required_config_modules = ["text_encoder", "tokenizer", "vae", "transformer", "scheduler"]
def initialize_pipeline(self, resolved_config: FastVideoArgs):
def initialize_pipeline(self, resolved_config: ResolvedGeneratorConfig):
"""Initialize LongCat-specific components."""
# Same BSA initialization as base LongCat pipeline
pipeline_config = resolved_config.pipeline_config
bsa_options = resolved_config.pipeline.longcat
transformer = self.get_module("transformer", None)
if transformer is None:
return
# Enable BSA if configured
if pipeline_config.enable_bsa:
if bsa_options.enable_bsa:
bsa_params_cfg = getattr(pipeline_config, 'bsa_params', None) or {}
sparsity = getattr(pipeline_config, 'bsa_sparsity', None)
cdf_threshold = getattr(pipeline_config, 'bsa_cdf_threshold', None)
chunk_q = getattr(pipeline_config, 'bsa_chunk_q', None)
chunk_k = getattr(pipeline_config, 'bsa_chunk_k', None)
sparsity = bsa_options.bsa_sparsity
cdf_threshold = bsa_options.bsa_cdf_threshold
chunk_q = thaw(bsa_options.bsa_chunk_q)
chunk_k = thaw(bsa_options.bsa_chunk_k)
effective_bsa_params = dict(bsa_params_cfg) if isinstance(bsa_params_cfg, dict) else {}
if sparsity is not None:
@@ -88,7 +89,7 @@ class LongCatImageToVideoPipeline(LoRAPipeline, ComposedPipelineBase):
if hasattr(transformer, 'disable_bsa'):
transformer.disable_bsa()
def create_pipeline_stages(self, resolved_config: FastVideoArgs):
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig):
"""Set up I2V-specific pipeline stages."""
# 1. Input validation
fastvideo/pipelines/basic/longcat/longcat_pipeline.py
+9 -8
View File
@@ -6,7 +6,7 @@ This module implements the LongCat video diffusion pipeline using FastVideo's
modular pipeline architecture.
"""
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig, thaw
from fastvideo.logger import init_logger
from fastvideo.pipelines import ComposedPipelineBase, LoRAPipeline
from fastvideo.pipelines.stages import (
@@ -30,25 +30,26 @@ class LongCatPipeline(LoRAPipeline, ComposedPipelineBase):
_required_config_modules = ["text_encoder", "tokenizer", "vae", "transformer", "scheduler"]
def initialize_pipeline(self, resolved_config: FastVideoArgs):
def initialize_pipeline(self, resolved_config: ResolvedGeneratorConfig):
"""Initialize LongCat-specific components."""
# Enable BSA (Block Sparse Attention) if configured
pipeline_config = resolved_config.pipeline_config
bsa_options = resolved_config.pipeline.longcat
transformer = self.get_module("transformer", None)
if transformer is None:
raise RuntimeError("Transformer module not found during initializing LongCat pipeline.")
# If user toggles BSA via CLI/config
if pipeline_config.enable_bsa:
if bsa_options.enable_bsa:
# Build effective BSA params:
# 1) from explicit CLI overrides if provided
# 2) else from pipeline_config.bsa_params
# 3) else fall back to reasonable defaults
bsa_params_cfg = pipeline_config.bsa_params
sparsity = pipeline_config.bsa_sparsity
cdf_threshold = pipeline_config.bsa_cdf_threshold
chunk_q = pipeline_config.bsa_chunk_q
chunk_k = pipeline_config.bsa_chunk_k
sparsity = bsa_options.bsa_sparsity
cdf_threshold = bsa_options.bsa_cdf_threshold
chunk_q = thaw(bsa_options.bsa_chunk_q)
chunk_k = thaw(bsa_options.bsa_chunk_k)
effective_bsa_params = dict(bsa_params_cfg) if isinstance(bsa_params_cfg, dict) else {}
if sparsity is not None:
@@ -83,7 +84,7 @@ class LongCatPipeline(LoRAPipeline, ComposedPipelineBase):
if hasattr(transformer, 'disable_bsa'):
transformer.disable_bsa()
def create_pipeline_stages(self, resolved_config: FastVideoArgs) -> None:
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig) -> None:
"""Set up pipeline stages with proper dependency injection."""
self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
fastvideo/pipelines/basic/longcat/longcat_vc_pipeline.py
+9 -8
View File
@@ -11,7 +11,7 @@ Supports:
- KV cache for conditioning frames
"""
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig, thaw
from fastvideo.logger import init_logger
from fastvideo.pipelines import ComposedPipelineBase, LoRAPipeline
from fastvideo.pipelines.stages import (
@@ -45,20 +45,21 @@ class LongCatVideoContinuationPipeline(LoRAPipeline, ComposedPipelineBase):
_required_config_modules = ["text_encoder", "tokenizer", "vae", "transformer", "scheduler"]
def initialize_pipeline(self, resolved_config: FastVideoArgs):
def initialize_pipeline(self, resolved_config: ResolvedGeneratorConfig):
"""Initialize LongCat-specific components."""
pipeline_config = resolved_config.pipeline_config
bsa_options = resolved_config.pipeline.longcat
transformer = self.get_module("transformer", None)
if transformer is None:
return
# Enable BSA if configured (for VC, BSA may not be needed)
if getattr(pipeline_config, 'enable_bsa', False):
if bsa_options.enable_bsa:
bsa_params_cfg = getattr(pipeline_config, 'bsa_params', None) or {}
sparsity = getattr(pipeline_config, 'bsa_sparsity', None)
cdf_threshold = getattr(pipeline_config, 'bsa_cdf_threshold', None)
chunk_q = getattr(pipeline_config, 'bsa_chunk_q', None)
chunk_k = getattr(pipeline_config, 'bsa_chunk_k', None)
sparsity = bsa_options.bsa_sparsity
cdf_threshold = bsa_options.bsa_cdf_threshold
chunk_q = thaw(bsa_options.bsa_chunk_q)
chunk_k = thaw(bsa_options.bsa_chunk_k)
effective_bsa_params = dict(bsa_params_cfg) if isinstance(bsa_params_cfg, dict) else {}
if sparsity is not None:
@@ -90,7 +91,7 @@ class LongCatVideoContinuationPipeline(LoRAPipeline, ComposedPipelineBase):
if hasattr(transformer, 'disable_bsa'):
transformer.disable_bsa()
def create_pipeline_stages(self, resolved_config: FastVideoArgs):
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig):
"""Set up VC-specific pipeline stages."""
# 1. Input validation
fastvideo/pipelines/basic/matrixgame2/matrixgame2_causal_dmd_pipeline.py
+3 -3
View File
@@ -1,7 +1,7 @@
# SPDX-License-Identifier: Apache-2.0
"""Matrix-Game causal DMD pipeline implementation."""
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
import torch
from fastvideo.logger import init_logger
from fastvideo.pipelines import ComposedPipelineBase, ForwardBatch, LoRAPipeline
@@ -17,7 +17,7 @@ logger = init_logger(__name__)
class MatrixGame2CausalDMDPipeline(LoRAPipeline, ComposedPipelineBase):
_required_config_modules = ["vae", "transformer", "scheduler", "image_encoder", "image_processor"]
def create_pipeline_stages(self, resolved_config: FastVideoArgs) -> None:
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig) -> None:
self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
if (self.get_module("text_encoder", None) is not None and self.get_module("tokenizer", None) is not None):
@@ -56,7 +56,7 @@ class MatrixGame2CausalDMDPipeline(LoRAPipeline, ComposedPipelineBase):
logger.info("MatrixGame2CausalDMDPipeline initialized with action support")
@torch.no_grad()
def streaming_reset(self, batch: ForwardBatch, resolved_config: FastVideoArgs):
def streaming_reset(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig):
if not self.post_init_called:
self.post_init()
fastvideo/pipelines/basic/matrixgame2/matrixgame2_i2v_pipeline.py
+4 -4
View File
@@ -1,7 +1,7 @@
# SPDX-License-Identifier: Apache-2.0
"""Matrix-Game I2V pipeline implementation."""
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.logger import init_logger
from fastvideo.pipelines.composed_pipeline_base import ComposedPipelineBase
from fastvideo.pipelines.lora_pipeline import LoRAPipeline
@@ -18,10 +18,10 @@ logger = init_logger(__name__)
class MatrixGame2I2VPipeline(LoRAPipeline, ComposedPipelineBase):
_required_config_modules = ["vae", "transformer", "scheduler", "image_encoder", "image_processor"]
def initialize_pipeline(self, resolved_config: FastVideoArgs):
self.modules["scheduler"] = FlowUniPCMultistepScheduler(shift=resolved_config.pipeline_config.flow_shift)
def initialize_pipeline(self, resolved_config: ResolvedGeneratorConfig):
self.modules["scheduler"] = FlowUniPCMultistepScheduler(shift=resolved_config.pipeline.flow_shift)
def create_pipeline_stages(self, resolved_config: FastVideoArgs):
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig):
self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
if (self.get_module("text_encoder", None) is not None and self.get_module("tokenizer", None) is not None):
fastvideo/pipelines/basic/matrixgame3/matrixgame3_i2v_pipeline.py
+4 -4
View File
@@ -1,7 +1,7 @@
# SPDX-License-Identifier: Apache-2.0
"""Matrix-Game 3.0 I2V pipeline implementation."""
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.logger import init_logger
from fastvideo.models.schedulers.scheduling_flow_unipc_multistep import FlowUniPCMultistepScheduler
from fastvideo.pipelines.composed_pipeline_base import ComposedPipelineBase
@@ -17,10 +17,10 @@ class MatrixGame3I2VPipeline(LoRAPipeline, ComposedPipelineBase):
_required_config_modules = ["vae", "transformer", "scheduler", "text_encoder", "tokenizer"]
_extra_config_module_map = {"vae": "light_vae"}
def initialize_pipeline(self, resolved_config: FastVideoArgs):
self.modules["scheduler"] = FlowUniPCMultistepScheduler(shift=resolved_config.pipeline_config.flow_shift)
def initialize_pipeline(self, resolved_config: ResolvedGeneratorConfig):
self.modules["scheduler"] = FlowUniPCMultistepScheduler(shift=resolved_config.pipeline.flow_shift)
def create_pipeline_stages(self, resolved_config: FastVideoArgs):
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig):
self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
self.add_stage(stage_name="prompt_encoding_stage",
fastvideo/pipelines/basic/sd35/sd35_pipeline.py
+3 -3
View File
@@ -2,7 +2,7 @@
from __future__ import annotations
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.logger import init_logger
from fastvideo.pipelines.composed_pipeline_base import ComposedPipelineBase
from fastvideo.pipelines.stages.input_validation import InputValidationStage
@@ -34,7 +34,7 @@ class SD35Pipeline(ComposedPipelineBase):
"tokenizer_3",
]
def initialize_pipeline(self, resolved_config: FastVideoArgs) -> None:
def initialize_pipeline(self, resolved_config: ResolvedGeneratorConfig) -> None:
te_cfgs = list(resolved_config.pipeline_config.text_encoder_configs)
if len(te_cfgs) >= 2:
for i in (0, 1):
@@ -49,7 +49,7 @@ class SD35Pipeline(ComposedPipelineBase):
te_cfgs[2].tokenizer_kwargs.setdefault("truncation", True)
te_cfgs[2].tokenizer_kwargs.setdefault("return_tensors", "pt")
def create_pipeline_stages(self, resolved_config: FastVideoArgs) -> None:
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig) -> None:
self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
self.add_stage(
fastvideo/pipelines/basic/turbodiffusion/turbodiffusion_i2v_pipeline.py
+3 -3
View File
@@ -13,7 +13,7 @@ Key differences from T2V:
"""
from fastvideo.pipelines.basic.wan.stages.denoising import WanDenoisingStage
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.logger import init_logger
from fastvideo.models.schedulers.scheduling_rcm import RCMScheduler
from fastvideo.pipelines import ComposedPipelineBase, LoRAPipeline
@@ -33,12 +33,12 @@ class TurboDiffusionI2VPipeline(LoRAPipeline, ComposedPipelineBase):
_required_config_modules = ["text_encoder", "tokenizer", "vae", "transformer", "transformer_2", "scheduler"]
def initialize_pipeline(self, resolved_config: FastVideoArgs):
def initialize_pipeline(self, resolved_config: ResolvedGeneratorConfig):
# Use RCM scheduler with higher sigma_max for I2V
logger.info("Initializing RCM scheduler for TurboDiffusion I2V (sigma_max=200)")
self.modules["scheduler"] = RCMScheduler(sigma_max=200.0)
def create_pipeline_stages(self, resolved_config: FastVideoArgs) -> None:
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig) -> None:
"""Set up pipeline stages with proper dependency injection."""
self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
fastvideo/pipelines/basic/turbodiffusion/turbodiffusion_pipeline.py
+3 -3
View File
@@ -8,7 +8,7 @@ with SLA (Sparse-Linear Attention).
"""
from fastvideo.pipelines.basic.wan.stages.denoising import WanDenoisingStage
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.logger import init_logger
from fastvideo.models.schedulers.scheduling_rcm import RCMScheduler
from fastvideo.pipelines import ComposedPipelineBase, LoRAPipeline
@@ -27,12 +27,12 @@ class TurboDiffusionPipeline(LoRAPipeline, ComposedPipelineBase):
_required_config_modules = ["text_encoder", "tokenizer", "vae", "transformer", "scheduler"]
def initialize_pipeline(self, resolved_config: FastVideoArgs):
def initialize_pipeline(self, resolved_config: ResolvedGeneratorConfig):
# Use RCM scheduler for TurboDiffusion
logger.info("Initializing RCM scheduler for TurboDiffusion")
self.modules["scheduler"] = RCMScheduler(sigma_max=80.0)
def create_pipeline_stages(self, resolved_config: FastVideoArgs) -> None:
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig) -> None:
"""Set up pipeline stages with proper dependency injection."""
self.add_stage(stage_name="input_validation_stage", stage=InputValidationStage())
fastvideo/pipelines/basic/zimage/stages.py
+11 -11
View File
@@ -8,7 +8,7 @@ import inspect
import torch
from fastvideo.distributed import get_local_torch_device
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.forward_context import set_forward_context
from fastvideo.hooks.activation_trace import trace_step
from fastvideo.pipelines.pipeline_batch_info import ForwardBatch
@@ -20,14 +20,14 @@ from fastvideo.utils import PRECISION_TO_TYPE
class ZImageInputValidationStage(InputValidationStage):
"""Validate the image geometry and reproduce the official device RNG."""
def _generate_seeds(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> None:
def _generate_seeds(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> None:
del resolved_config
assert batch.seed is not None
batch.seeds = [batch.seed]
device = get_local_torch_device()
batch.generator = torch.Generator(device=device).manual_seed(batch.seed)
def forward(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> ForwardBatch:
def forward(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> ForwardBatch:
if batch.do_classifier_free_guidance and batch.negative_prompt is None and not batch.negative_prompt_embeds:
batch.negative_prompt = ""
batch = super().forward(batch, resolved_config)
@@ -61,7 +61,7 @@ class ZImageConditioningStage(PipelineStage):
return [item for item in items for _ in range(count)]
@torch.no_grad()
def forward(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> ForwardBatch:
def forward(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> ForwardBatch:
del resolved_config
if len(batch.prompt_embeds) != 1:
raise ValueError(f"Z-Image expects one text encoder, got {len(batch.prompt_embeds)}")
@@ -107,7 +107,7 @@ class ZImageLatentPreparationStage(PipelineStage):
return torch.randn(shape, generator=generators, device=device, dtype=torch.float32)
@torch.no_grad()
def forward(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> ForwardBatch:
def forward(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> ForwardBatch:
if batch.height is None or batch.width is None:
raise ValueError("Z-Image requires height and width before latent preparation")
@@ -157,7 +157,7 @@ class ZImageTimestepPreparationStage(PipelineStage):
slope = (max_shift - base_shift) / (max_seq_len - base_seq_len)
return image_seq_len * slope + base_shift - slope * base_seq_len
def forward(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> ForwardBatch:
def forward(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> ForwardBatch:
if batch.latents is None:
raise ValueError("Z-Image latents must be prepared before timesteps")
if batch.timesteps is not None and batch.sigmas is not None:
@@ -208,7 +208,7 @@ class ZImageDenoisingStage(PipelineStage):
self.scheduler = scheduler
@torch.no_grad()
def forward(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> ForwardBatch:
def forward(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> ForwardBatch:
if batch.latents is None or batch.timesteps is None:
raise ValueError("Z-Image denoising requires latents and timesteps")
@@ -219,7 +219,7 @@ class ZImageDenoisingStage(PipelineStage):
raise ValueError("Z-Image denoising requires prompt embeddings")
device = get_local_torch_device()
target_dtype = PRECISION_TO_TYPE[resolved_config.pipeline_config.dit_precision]
target_dtype = PRECISION_TO_TYPE[resolved_config.engine.precision.dit]
batch_size = latents.shape[0]
for index, timestep_value in enumerate(batch.timesteps):
@@ -305,10 +305,10 @@ class ZImageDecodingStage(PipelineStage):
self.vae = vae
@torch.no_grad()
def forward(self, batch: ForwardBatch, resolved_config: FastVideoArgs) -> ForwardBatch:
def forward(self, batch: ForwardBatch, resolved_config: ResolvedGeneratorConfig) -> ForwardBatch:
if batch.latents is None:
raise ValueError("Z-Image decoding requires latents")
if resolved_config.output_type == "latent":
if resolved_config.pipeline.output_type == "latent":
# FastVideo standardizes image and video latents as [B,C,T,H,W].
# Tongyi's image-only API returns the equivalent tensor with T squeezed.
batch.output = batch.latents
@@ -332,6 +332,6 @@ class ZImageDecodingStage(PipelineStage):
decoded = (decoded / 2 + 0.5).clamp(0, 1)
batch.output = decoded.unsqueeze(2).float()
if resolved_config.vae_cpu_offload:
if resolved_config.engine.offload.vae:
self.vae.to("cpu")
return batch
fastvideo/pipelines/basic/zimage/zimage_pipeline.py
+2 -2
View File
@@ -3,7 +3,7 @@
from __future__ import annotations
from fastvideo.configs.pipelines.zimage import ZImagePipelineConfig
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.api.resolution import ResolvedGeneratorConfig
from fastvideo.pipelines.composed_pipeline_base import ComposedPipelineBase
from fastvideo.pipelines.stages.text_encoding import TextEncodingStage
@@ -29,7 +29,7 @@ class ZImagePipeline(ComposedPipelineBase):
"vae",
]
def create_pipeline_stages(self, resolved_config: FastVideoArgs) -> None:
def create_pipeline_stages(self, resolved_config: ResolvedGeneratorConfig) -> None:
scheduler = self.get_module("scheduler")
transformer = self.get_module("transformer")
fastvideo/tests/pipelines/test_direct_pipeline_unified_memory.py
+15 -7
View File
@@ -8,8 +8,10 @@ from types import SimpleNamespace
import torch
import fastvideo.pipelines.composed_pipeline_base as composed_pipeline_base
from fastvideo.fastvideo_args import UNIFIED_MEMORY_OFFLOAD_FLAGS, FastVideoArgs
from fastvideo.api.device_policy import UNIFIED_MEMORY_OFFLOAD_PATHS
from fastvideo.api.inference_resolution import resolve_inference_config
from fastvideo.pipelines.composed_pipeline_base import ComposedPipelineBase
from fastvideo.tests.api.config_snapshot import isolated_environment
class _Profiler:
@@ -24,8 +26,8 @@ class _Pipeline(ComposedPipelineBase):
def load_modules(self, resolved_config, loaded_modules=None):
del loaded_modules
policy_state = {flag: getattr(resolved_config, flag) for flag in UNIFIED_MEMORY_OFFLOAD_FLAGS}
policy_state["use_fsdp_inference"] = resolved_config.use_fsdp_inference
policy_state = {path: resolved_config.provenance(path).value for path in UNIFIED_MEMORY_OFFLOAD_PATHS.values()}
policy_state["engine.use_fsdp_inference"] = resolved_config.engine.use_fsdp_inference
self.events.append(("load_modules", policy_state))
return {}
@@ -36,7 +38,13 @@ class _Pipeline(ComposedPipelineBase):
def test_direct_pipeline_applies_policy_after_device_initialization(monkeypatch) -> None:
events = []
monkeypatch.setattr(_Pipeline, "events", events)
args = FastVideoArgs(model_path="unused", use_fsdp_inference=True)
with isolated_environment():
resolved_config = resolve_inference_config({
"model_path": "Wan-AI/Wan2.1-T2V-1.3B-Diffusers",
"engine": {
"use_fsdp_inference": True
},
})
def classify_device(device_id):
events.append(("offload_policy", device_id))
@@ -54,7 +62,7 @@ def test_direct_pipeline_applies_policy_after_device_initialization(monkeypatch)
monkeypatch.setattr("fastvideo.platforms.current_platform.get_device_name", lambda device_id: "NVIDIA GB10")
monkeypatch.setattr("fastvideo.platforms.current_platform.is_mps", lambda: False)
pipeline = _Pipeline("unused", args, required_config_modules=[])
pipeline = _Pipeline("unused", resolved_config, required_config_modules=[])
assert pipeline.modules == {}
assert events == [
@@ -63,8 +71,8 @@ def test_direct_pipeline_applies_policy_after_device_initialization(monkeypatch)
(
"load_modules",
{
**{flag: False for flag in UNIFIED_MEMORY_OFFLOAD_FLAGS},
"use_fsdp_inference": True,
**{path: False for path in UNIFIED_MEMORY_OFFLOAD_PATHS.values()},
"engine.use_fsdp_inference": True,
},
),
]