Files
wehub-resource-sync 94057c3d3e
PR Test (NPU) / check-changes (push) Has been cancelled
PR Test (NPU) / pr-gate (push) Has been cancelled
PR Test (NPU) / set-image-config (push) Has been cancelled
PR Test (NPU) / stage-b-test-1-npu-a2 (0) (push) Has been cancelled
PR Test (NPU) / stage-b-test-1-npu-a2 (1) (push) Has been cancelled
PR Test (NPU) / stage-b-test-2-npu-a2 (0) (push) Has been cancelled
PR Test (NPU) / stage-b-test-2-npu-a2 (1) (push) Has been cancelled
PR Test (NPU) / stage-b-test-4-npu-a3 (push) Has been cancelled
PR Test (NPU) / stage-b-test-16-npu-a3 (push) Has been cancelled
PR Test (NPU) / multimodal-gen-test-1-npu-a3 (push) Has been cancelled
PR Test (NPU) / multimodal-gen-test-2-npu-a3 (push) Has been cancelled
PR Test (Arm64) / pr-gate (push) Has been cancelled
PR Test (Arm64) / check-changes (push) Has been cancelled
PR Test (Arm64) / build-test (push) Has been cancelled
PR Test (sgl-router) / gate (push) Has been cancelled
PR Test (sgl-router) / tier-1 — lint (push) Has been cancelled
PR Test (sgl-router) / tier-2 — build + test (push) Has been cancelled
PR Test (sgl-router) / tier-3 — docker (placeholder) (push) Has been cancelled
PR Test (sgl-router) / tier-3 — k8s integration (push) Has been cancelled
PR Test (sgl-router) / tier-3 — e2e (push) Has been cancelled
PR Test (sgl-router) / finish (push) Has been cancelled
PR Test (NPU) / single-node-poc (map[name:qwen3_6_27b_w8a8_1p_in64k_out1k_50ms runner:linux-aarch64-a3-2 test_case:test/registered/ascend/performance/qwen3_6_27b/test_npu_qwen3_6_27b_w8a8_1p_in64k_out1k_50ms.py test_type:perf]) (push) Has been cancelled
PR Test (NPU) / pr-test-npu-finish (push) Has been cancelled
PR Test (Xeon) / pr-gate (push) Has been cancelled
PR Test (Xeon) / check-changes (push) Has been cancelled
PR Test (Xeon) / build-test (, xeon-gnr, base-b-test-cpu) (push) Has been cancelled
PR Test (XPU) / check-changes (push) Has been cancelled
PR Test (XPU) / pr-gate (push) Has been cancelled
PR Test (XPU) / stage-a-test-1-gpu-xpu (push) Has been cancelled
PR Test (XPU) / wait-for-stage-a (push) Has been cancelled
PR Test (XPU) / stage-b-test-1-gpu-xpu (push) Has been cancelled
PR Test (XPU) / finish (push) Has been cancelled
CI Model Inventory / build-inventory (push) Has been cancelled
Lint / lint (push) Has been cancelled
PR Benchmark (SMG Components) / Benchmark Compilation Check (push) Has been cancelled
PR Benchmark (SMG Components) / Benchmark - Manual Policy (push) Has been cancelled
PR Benchmark (SMG Components) / Benchmark - Request Processing (push) Has been cancelled
PR Benchmark (SMG Components) / Benchmark Summary (push) Has been cancelled
PR Test (SMG) / build-wheel (push) Has been cancelled
Release SGLang Model Gateway to PyPI / build on windows (x86_64 - auto) (push) Has been cancelled
Release SGLang Model Gateway to PyPI / build on macos (x86_64 - auto) (push) Has been cancelled
PR Test (SMG) / python-unit-tests (push) Has been cancelled
PR Test (SMG) / unit-tests (push) Has been cancelled
PR Test (SMG) / benchmarks (push) Has been cancelled
PR Test (SMG) / chat-completions (push) Has been cancelled
PR Test (SMG) / chat-completions-4gpu (push) Has been cancelled
PR Test (SMG) / e2e (push) Has been cancelled
PR Test (SMG) / docker-build-test (push) Has been cancelled
PR Test (SMG) / k8s-integration (push) Has been cancelled
PR Test (SMG) / finish (push) Has been cancelled
PR Test (SMG) / summarize-benchmarks (push) Has been cancelled
Release SGLang Model Gateway Docker Image / publish (push) Has been cancelled
Release SGLang Model Gateway to PyPI / build on macos (aarch64 - auto) (push) Has been cancelled
Release SGLang Model Gateway to PyPI / build on linux (aarch64 - auto) (push) Has been cancelled
Release SGLang Model Gateway to PyPI / build on linux (x86_64 - auto) (push) Has been cancelled
Release SGLang Model Gateway to PyPI / build on linux (aarch64 - musllinux_1_1) (push) Has been cancelled
Release SGLang Model Gateway to PyPI / build on linux (x86_64 - musllinux_1_1) (push) Has been cancelled
Release SGLang Model Gateway to PyPI / Build SDist (push) Has been cancelled
Release SGLang Model Gateway to PyPI / Upload to PyPI (push) Has been cancelled
Release SGLang Kernels / build-cu129-matrix (aarch64, 12.9, 3.10, arm-kernel-build-node) (push) Has been cancelled
Release SGLang Kernels / build-cu129-matrix (x86_64, 12.9, 3.10, x64-kernel-build-node) (push) Has been cancelled
Release SGLang Kernels / release-cu129 (push) Has been cancelled
Release SGLang Kernels / build-cu130-matrix (aarch64, 13.0, 3.10, arm-kernel-build-node) (push) Has been cancelled
Release SGLang Kernels / build-cu130-matrix (x86_64, 13.0, 3.10, x64-kernel-build-node) (push) Has been cancelled
Release SGLang Kernels / release-cu130 (push) Has been cancelled
Release SGLang Kernels / build-rocm-matrix (3.10, 700) (push) Has been cancelled
Release SGLang Kernels / build-rocm-matrix (3.10, 720) (push) Has been cancelled
Release SGLang Kernels / release-rocm700 (push) Has been cancelled
Release SGLang Kernels / release-rocm720 (push) Has been cancelled
Release SGLang Kernels / build-musa43 (43, 3.10) (push) Has been cancelled
Release SGLang Kernels / release-musa43 (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 12:38:16 +08:00

684 lines
26 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# SPDX-License-Identifier: Apache-2.0
"""
cache-dit integration module for SGLang DiT pipelines.
This module provides helper functions to enable cache-dit acceleration
on transformer modules in SGLang's modular pipeline architecture.
"""
from dataclasses import dataclass
from typing import List, Optional
import torch
import torch.distributed as dist
from sglang.multimodal_gen.runtime.distributed.parallel_state import (
get_ring_parallel_world_size,
get_tp_world_size,
get_ulysses_parallel_world_size,
)
from sglang.multimodal_gen.runtime.utils.logging_utils import init_logger
logger = init_logger(__name__)
import cache_dit
from cache_dit import (
BlockAdapter,
DBCacheConfig,
ForwardPattern,
ParamsModifier,
TaylorSeerCalibratorConfig,
steps_mask,
)
from cache_dit.caching.block_adapters import BlockAdapterRegister
from cache_dit.parallelism import ParallelismBackend, ParallelismConfig
from sglang.multimodal_gen.runtime.distributed.parallel_state import get_dit_group
_original_similarity = None
def _patch_cache_dit_similarity():
from cache_dit.caching.cache_contexts import cache_manager
global _original_similarity
if _original_similarity is not None:
return
_original_similarity = cache_manager.CachedContextManager.similarity
def patched_similarity(self, t1, t2, *, threshold, parallelized=False, prefix="Fn"):
if not parallelized:
return _original_similarity(
self,
t1,
t2,
threshold=threshold,
parallelized=parallelized,
prefix=prefix,
)
sp_group = getattr(self, "_sglang_sp_group", None)
tp_group = getattr(self, "_sglang_tp_group", None)
tp_sp_group = getattr(self, "_sglang_tp_sp_group", None)
target_group = tp_sp_group or sp_group or tp_group
if target_group is None:
return _original_similarity(
self,
t1,
t2,
threshold=threshold,
parallelized=parallelized,
prefix=prefix,
)
# Adapted from https://github.com/vipshop/cache-dit/blob/main/src/cache_dit/caching/cache_contexts/cache_manager.py#L495-L523
condition_thresh = self.get_important_condition_threshold()
if condition_thresh > 0.0:
raw_diff = (t1 - t2).abs()
token_m_df = raw_diff.mean(dim=-1)
token_m_t1 = t1.abs().mean(dim=-1)
token_diff = token_m_df / token_m_t1
condition = token_diff > condition_thresh
if condition.sum() > 0:
condition = condition.unsqueeze(-1).expand_as(raw_diff)
mean_diff = raw_diff[condition].mean()
mean_t1 = t1[condition].abs().mean()
else:
mean_diff = (t1 - t2).abs().mean()
mean_t1 = t1.abs().mean()
else:
mean_diff = (t1 - t2).abs().mean()
mean_t1 = t1.abs().mean()
dist.all_reduce(mean_diff, op=dist.ReduceOp.AVG, group=target_group)
dist.all_reduce(mean_t1, op=dist.ReduceOp.AVG, group=target_group)
diff = (mean_diff / mean_t1).item()
self.add_residual_diff(diff)
return diff < threshold
cache_manager.CachedContextManager.similarity = patched_similarity
def _build_parallelism_config(
sp_group: Optional[torch.distributed.ProcessGroup],
tp_group: Optional[torch.distributed.ProcessGroup],
):
if sp_group is None and tp_group is None:
return None
ulysses_size = None
ring_size = None
if sp_group is not None:
ulysses_size = get_ulysses_parallel_world_size()
ring_size = get_ring_parallel_world_size()
tp_size = None
if tp_group is not None:
tp_size = get_tp_world_size()
return ParallelismConfig(
backend=ParallelismBackend.AUTO,
ulysses_size=ulysses_size,
ring_size=ring_size,
tp_size=tp_size,
)
def _mark_transformer_parallelized(transformer, config, sp_group, tp_group):
if config is None:
return
transformer._is_parallelized = True
transformer._parallelism_config = config
def get_scm_mask(
preset: str,
num_inference_steps: int,
compute_bins: Optional[List[int]] = None,
cache_bins: Optional[List[int]] = None,
) -> Optional[List[int]]:
"""
Get SCM mask using cache-dit's steps_mask().
This is a thin wrapper that delegates to cache-dit's built-in
steps_mask() function which handles all presets and scaling logic.
Args:
preset: Preset name ("none", "slow", "medium", "fast", "ultra").
compute_bins: Custom compute bins (overrides preset).
cache_bins: Custom cache bins (overrides preset).
Returns:
SCM mask list (1=compute, 0=cache), or None if disabled.
"""
if preset == "none" and not (compute_bins and cache_bins):
return None
# Use cache-dit's steps_mask() directly
mask = steps_mask(
compute_bins=compute_bins,
cache_bins=cache_bins,
total_steps=num_inference_steps,
mask_policy=preset if preset != "none" else "medium",
)
compute_count = sum(mask)
cache_count = len(mask) - compute_count
logger.info(
"SCM: generated mask with %d compute steps, %d cache steps (preset=%s)",
compute_count,
cache_count,
preset,
)
return mask
@dataclass
class CacheDitConfig:
"""Configuration for cache-dit integration.
Attributes:
enabled: Whether to enable cache-dit acceleration.
Fn_compute_blocks: Number of first blocks to always compute (DBCache F).
Bn_compute_blocks: Number of last blocks to always compute (DBCache B).
max_warmup_steps: Number of warmup steps before caching starts (DBCache W).
residual_diff_threshold: Threshold for residual difference (DBCache R).
max_continuous_cached_steps: Maximum consecutive cached steps (DBCache MC).
enable_taylorseer: Whether to enable TaylorSeer calibrator.
taylorseer_order: Order of Taylor expansion (1 or 2).
num_inference_steps: Total number of inference steps (required for transformer-only mode).
steps_computation_mask: Binary mask for step-level caching (1=compute, 0=cache).
Generated by get_scm_mask() (wrapper around cache_dit.steps_mask()).
steps_computation_policy: Caching policy for SCM ("dynamic" or "static").
"""
enabled: bool = False
Fn_compute_blocks: int = 1
Bn_compute_blocks: int = 0
# Use 4 as default warmup steps instead of 8 in cache-dit, thus making
# DBCache work for few steps distilled models, e.g., Z-Image w/ 8-steps.
max_warmup_steps: int = 4
# Use a relatively higher residual diff threshold (namely, 0.24) as default
# to allow more aggressive caching due to we have already applied max continuous
# cached steps limit, otherwise, we should use a lower threshold here like 0.12.
residual_diff_threshold: float = 0.24
max_continuous_cached_steps: int = 3
# TaylorSeer is not suitable for few steps distilled models, so, we choose
# to disable it by default. Reference:
# - From Reusing to Forecasting: Accelerating Diffusion Models with TaylorSeers,
# https://arxiv.org/pdf/2503.06923
# - FoCa: Forecast then Calibrate: Feature Caching as ODE for Efficient
# Diffusion Transformers, https://arxiv.org/pdf/2508.16211
enable_taylorseer: bool = False
taylorseer_order: int = 1
num_inference_steps: Optional[int] = None
# SCM fields (generated by _maybe_enable_cache_dit from env configuration)
steps_computation_mask: Optional[List[int]] = None
steps_computation_policy: str = "dynamic"
@dataclass(frozen=True)
class DualTransformerBlockAdapterSpec:
"""BlockAdapter metadata for dual-transformer DiT pipelines.
This describes the cache-dit-facing structure of a pair of transformers.
The denoising loop semantics live in DenoisingStage; this spec only covers
how cache-dit should find blocks and interpret each block's forward.
"""
blocks_attr: tuple[str, str]
blocks_name: Optional[List[str]]
forward_pattern: List[ForwardPattern]
check_forward_pattern: bool
check_num_outputs: bool
has_separate_cfg: bool
DUAL_TRANSFORMER_BLOCK_ADAPTER_SPECS: dict[str, DualTransformerBlockAdapterSpec] = {
"wan2.2": DualTransformerBlockAdapterSpec(
blocks_attr=("blocks", "blocks"),
blocks_name=None,
forward_pattern=[ForwardPattern.Pattern_2, ForwardPattern.Pattern_2],
check_forward_pattern=True,
check_num_outputs=False,
has_separate_cfg=True,
),
"ideogram4": DualTransformerBlockAdapterSpec(
blocks_attr=("layers", "layers"),
blocks_name=["layers", "layers"],
forward_pattern=[ForwardPattern.Pattern_3, ForwardPattern.Pattern_3],
check_forward_pattern=False,
check_num_outputs=False,
has_separate_cfg=False,
),
}
# Custom BlockAdapter for DiT models absent from cache-dit's BlockAdapterRegister.
# Value: (blocks attr, forward_pattern). forward_pattern must
# match the block's forward signature (see cache_dit.ForwardPattern; e.g., ERNIE
# uses Pattern_3). has_separate_cfg follows the run (passed by
# enable_cache_on_transformer); cache-dit auto-resolves the remaining
# fields.
_CUSTOM_BLOCK_ADAPTER_SPECS: dict[str, tuple[str, ForwardPattern]] = {
"ErnieImageTransformer2DModel": ("layers", ForwardPattern.Pattern_3),
"Krea2Transformer2DModel": ("transformer_blocks", ForwardPattern.Pattern_3),
}
def _build_custom_block_adapter(
transformer: torch.nn.Module,
has_separate_cfg: bool = False,
) -> Optional[BlockAdapter]:
"""Build a manual BlockAdapter for a model absent from cache-dit's registry,
or None if the class is unknown."""
spec = _CUSTOM_BLOCK_ADAPTER_SPECS.get(transformer.__class__.__name__)
if spec is None:
return None
blocks_attr, forward_pattern = spec
blocks = getattr(transformer, blocks_attr, None)
if blocks is None:
raise ValueError(
f"Transformer {transformer.__class__.__name__} has no attribute "
f"{blocks_attr!r} for cache-dit blocks."
)
return BlockAdapter(
transformer=transformer,
blocks=blocks,
forward_pattern=forward_pattern,
has_separate_cfg=has_separate_cfg,
)
def enable_cache_on_transformer(
transformer: torch.nn.Module,
config: CacheDitConfig,
model_name: str = "transformer",
sp_group: Optional[torch.distributed.ProcessGroup] = None,
tp_group: Optional[torch.distributed.ProcessGroup] = None,
has_separate_cfg: bool = False,
) -> torch.nn.Module:
"""Enable cache-dit on a transformer module, by wrapping the module with cache-dit
This function enables cache-dit acceleration using the BlockAdapterRegister
for pre-registered models
Args:
model_name: Name of the model for logging purposes.
sp_group: Sequence parallel process group (for Ulysses/Ring).
tp_group: Tensor parallel process group.
has_separate_cfg: Whether the run issues separate conditional/unconditional
passes per step (CFG). Used by custom adapters (ERNIE, Krea-2); a
mismatch only disables caching, never corrupts output.
"""
if not config.enabled:
return transformer
if config.num_inference_steps is None:
raise ValueError(
"num_inference_steps is required for transformer-only mode. "
"Please provide it in CacheDitConfig."
)
# Prefer the standard path (transformer pre-registered in cache-dit). For
# models absent from the registry, fall back to a manual BlockAdapter (see
# _build_custom_block_adapter).
custom_adapter = None
if not BlockAdapterRegister.is_supported(transformer):
custom_adapter = _build_custom_block_adapter(
transformer, has_separate_cfg=has_separate_cfg
)
if custom_adapter is None:
transformer_cls_name = transformer.__class__.__name__
raise ValueError(
f"{transformer_cls_name} is not officially supported by cache-dit. "
"Supported cache-dit DiT families include Flux, QwenImage, HunyuanDiT, "
"HunyuanVideo, Wan, CogVideoX, Mochi, and others. "
"Please ensure your transformer belongs to one of these families or "
"define a custom BlockAdapter."
)
# Build cache config (including SCM fields if provided)
cache_config = DBCacheConfig(
num_inference_steps=config.num_inference_steps,
Fn_compute_blocks=config.Fn_compute_blocks,
Bn_compute_blocks=config.Bn_compute_blocks,
max_warmup_steps=config.max_warmup_steps,
residual_diff_threshold=config.residual_diff_threshold,
max_continuous_cached_steps=config.max_continuous_cached_steps,
# SCM fields
steps_computation_mask=config.steps_computation_mask,
steps_computation_policy=config.steps_computation_policy,
)
# Build calibrator config if TaylorSeer is enabled
calibrator_config = None
if config.enable_taylorseer:
calibrator_config = TaylorSeerCalibratorConfig(
taylorseer_order=config.taylorseer_order,
)
# Enable cache-dit on the transformer
logger.info(
"Enabling cache-dit on %s with config: Fn=%d, Bn=%d, W=%d, R=%.2f, MC=%d, "
"TaylorSeer=%s (order=%d), steps=%d",
model_name,
config.Fn_compute_blocks,
config.Bn_compute_blocks,
config.max_warmup_steps,
config.residual_diff_threshold,
config.max_continuous_cached_steps,
config.enable_taylorseer,
config.taylorseer_order,
config.num_inference_steps,
)
# Log SCM configuration if enabled
if config.steps_computation_mask:
compute_steps = sum(config.steps_computation_mask)
cache_steps = len(config.steps_computation_mask) - compute_steps
logger.info(
"SCM enabled: %d compute steps, %d cache steps, policy=%s",
compute_steps,
cache_steps,
config.steps_computation_policy,
)
parallelism_config = _build_parallelism_config(sp_group, tp_group)
if parallelism_config is not None:
_patch_cache_dit_similarity()
_mark_transformer_parallelized(transformer, parallelism_config, sp_group, tp_group)
# Custom path: pass a pre-built BlockAdapter, bypassing the registry.
# Standard path: let enable_cache discover the registered adapter.
target = transformer
if custom_adapter is not None:
target = custom_adapter
logger.info(
"Enabling cache-dit on %s via custom BlockAdapter (%s).",
model_name,
custom_adapter.forward_pattern,
)
cache_dit.enable_cache(
target,
cache_config=cache_config,
calibrator_config=calibrator_config,
parallelism_config=None,
)
if parallelism_config is not None:
context_manager = getattr(transformer, "_context_manager", None)
if context_manager is not None:
context_manager._sglang_sp_group = sp_group
context_manager._sglang_tp_group = tp_group
# In mixed TP + SP (Ulysses/Ring) mode, cache-dit decisions must be consistent
# across the full TP×SP model-parallel slice. Prefer using SGLang's DIT group
# as a conservative superset group; fallback to None.
tp_sp_group = None
if sp_group is not None and tp_group is not None:
tp_sp_group = get_dit_group()
context_manager._sglang_tp_sp_group = tp_sp_group
return transformer
def enable_cache_on_dual_transformer(
transformer: torch.nn.Module,
transformer_2: torch.nn.Module,
primary_config: CacheDitConfig,
secondary_config: CacheDitConfig,
model_name: str = "wan2.2",
sp_group: Optional[torch.distributed.ProcessGroup] = None,
tp_group: Optional[torch.distributed.ProcessGroup] = None,
) -> tuple[torch.nn.Module, torch.nn.Module]:
"""Enable cache-dit on dual transformers using BlockAdapter.
For models with two transformers, cache-dit requires enabling cache on both
simultaneously via BlockAdapter. The two transformers may be split by denoising
range, or run as paired conditional/unconditional branches. This cannot be done
by calling enable_cache separately on each transformer.
Args:
primary_config: CacheDitConfig for primary transformer.
secondary_config: CacheDitConfig for secondary transformer.
sp_group: Sequence parallel process group (for Ulysses/Ring).
tp_group: Tensor parallel process group.
"""
adapter_spec = DUAL_TRANSFORMER_BLOCK_ADAPTER_SPECS.get(model_name)
if adapter_spec is None:
raise ValueError(
f"Dual-transformer cache-dit is only supported for "
f"{sorted(DUAL_TRANSFORMER_BLOCK_ADAPTER_SPECS)}, got {model_name}."
)
if not primary_config.enabled:
return transformer, transformer_2
if primary_config.num_inference_steps is None:
raise ValueError(
"num_inference_steps is required for dual-transformer mode. "
"Please provide it in CacheDitConfig."
)
# Build DBCacheConfig for primary transformer
primary_cache_config = DBCacheConfig(
num_inference_steps=primary_config.num_inference_steps,
Fn_compute_blocks=primary_config.Fn_compute_blocks,
Bn_compute_blocks=primary_config.Bn_compute_blocks,
max_warmup_steps=primary_config.max_warmup_steps,
residual_diff_threshold=primary_config.residual_diff_threshold,
max_continuous_cached_steps=primary_config.max_continuous_cached_steps,
steps_computation_mask=primary_config.steps_computation_mask,
steps_computation_policy=primary_config.steps_computation_policy,
)
# Build DBCacheConfig for secondary transformer
secondary_cache_config = DBCacheConfig(
num_inference_steps=secondary_config.num_inference_steps,
Fn_compute_blocks=secondary_config.Fn_compute_blocks,
Bn_compute_blocks=secondary_config.Bn_compute_blocks,
max_warmup_steps=secondary_config.max_warmup_steps,
residual_diff_threshold=secondary_config.residual_diff_threshold,
max_continuous_cached_steps=secondary_config.max_continuous_cached_steps,
steps_computation_mask=secondary_config.steps_computation_mask,
steps_computation_policy=secondary_config.steps_computation_policy,
)
# Build calibrator configs if TaylorSeer is enabled
primary_calibrator = None
if primary_config.enable_taylorseer:
primary_calibrator = TaylorSeerCalibratorConfig(
taylorseer_order=primary_config.taylorseer_order,
)
secondary_calibrator = None
if secondary_config.enable_taylorseer:
secondary_calibrator = TaylorSeerCalibratorConfig(
taylorseer_order=secondary_config.taylorseer_order,
)
# Build ParamsModifier for each transformer
primary_modifier = ParamsModifier(
cache_config=primary_cache_config,
calibrator_config=primary_calibrator,
)
secondary_modifier = ParamsModifier(
cache_config=secondary_cache_config,
calibrator_config=secondary_calibrator,
)
# Log configuration
logger.info(
"Enabling cache-dit on %s dual transformers with BlockAdapter",
model_name,
)
logger.info(
" Primary (transformer): Fn=%d, Bn=%d, W=%d, R=%.2f, MC=%d, TaylorSeer=%s",
primary_config.Fn_compute_blocks,
primary_config.Bn_compute_blocks,
primary_config.max_warmup_steps,
primary_config.residual_diff_threshold,
primary_config.max_continuous_cached_steps,
primary_config.enable_taylorseer,
)
logger.info(
" Secondary transformer: Fn=%d, Bn=%d, W=%d, R=%.2f, MC=%d, TaylorSeer=%s",
secondary_config.Fn_compute_blocks,
secondary_config.Bn_compute_blocks,
secondary_config.max_warmup_steps,
secondary_config.residual_diff_threshold,
secondary_config.max_continuous_cached_steps,
secondary_config.enable_taylorseer,
)
# Log SCM configuration if enabled
if primary_config.steps_computation_mask:
compute_steps = sum(primary_config.steps_computation_mask)
cache_steps = len(primary_config.steps_computation_mask) - compute_steps
logger.info(
" SCM enabled for primary transformer: %d compute steps, %d cache steps, policy=%s",
compute_steps,
cache_steps,
primary_config.steps_computation_policy,
)
if secondary_config.steps_computation_mask:
compute_steps = sum(secondary_config.steps_computation_mask)
cache_steps = len(secondary_config.steps_computation_mask) - compute_steps
logger.info(
" SCM enabled for secondary transformer: %d compute steps, %d cache steps, policy=%s",
compute_steps,
cache_steps,
secondary_config.steps_computation_policy,
)
parallelism_config = _build_parallelism_config(sp_group, tp_group)
if parallelism_config is not None:
_patch_cache_dit_similarity()
_mark_transformer_parallelized(transformer, parallelism_config, sp_group, tp_group)
_mark_transformer_parallelized(
transformer_2, parallelism_config, sp_group, tp_group
)
transformer_blocks_attr, transformer_2_blocks_attr = adapter_spec.blocks_attr
transformer_blocks = getattr(transformer, transformer_blocks_attr, None)
transformer_2_blocks = getattr(transformer_2, transformer_2_blocks_attr, None)
if transformer_blocks is None or transformer_2_blocks is None:
raise ValueError(
f"Dual transformers for {model_name} must expose cache-dit block "
f"attributes {adapter_spec.blocks_attr}. "
f"transformer has {transformer_blocks_attr}: "
f"{transformer_blocks is not None}, secondary transformer has "
f"{transformer_2_blocks_attr}: {transformer_2_blocks is not None}"
)
cache_dit.enable_cache(
BlockAdapter(
transformer=[transformer, transformer_2],
blocks=[transformer_blocks, transformer_2_blocks],
blocks_name=adapter_spec.blocks_name,
forward_pattern=adapter_spec.forward_pattern,
params_modifiers=[primary_modifier, secondary_modifier],
check_forward_pattern=adapter_spec.check_forward_pattern,
check_num_outputs=adapter_spec.check_num_outputs,
has_separate_cfg=adapter_spec.has_separate_cfg,
),
parallelism_config=None,
)
if parallelism_config is not None:
for t in [transformer, transformer_2]:
context_manager = getattr(t, "_context_manager", None)
if context_manager is not None:
context_manager._sglang_sp_group = sp_group
context_manager._sglang_tp_group = tp_group
tp_sp_group = None
if sp_group is not None and tp_group is not None:
try:
tp_sp_group = get_dit_group()
except Exception:
tp_sp_group = None
context_manager._sglang_tp_sp_group = tp_sp_group
return transformer, transformer_2
def refresh_context_on_transformer(
transformer: torch.nn.Module,
num_inference_steps: int,
scm_preset: str | None = None,
verbose: bool = False,
) -> None:
"""Refresh cache-dit context for transformer."""
steps_computation_mask = None
if scm_preset is not None:
steps_computation_mask = cache_dit.steps_mask(
mask_policy=scm_preset, total_steps=num_inference_steps
)
cache_dit.refresh_context(
transformer,
cache_config=DBCacheConfig().reset(
num_inference_steps=num_inference_steps,
steps_computation_mask=steps_computation_mask,
steps_computation_policy=scm_preset,
),
verbose=verbose,
)
logger.debug(f"cache-dit refreshed on transformer (steps={num_inference_steps})")
def refresh_context_on_dual_transformer(
transformer: torch.nn.Module,
transformer_2: torch.nn.Module,
num_high_noise_steps: int,
num_low_noise_steps: int,
scm_preset: str | None = None,
verbose: bool = False,
steps_computation_mask: Optional[List[int]] = None,
steps_computation_mask_2: Optional[List[int]] = None,
steps_computation_policy: str | None = None,
) -> None:
"""Refresh cache-dit context for dual transformers."""
high_noise_steps_computation_mask = steps_computation_mask
low_noise_steps_computation_mask = steps_computation_mask_2
if high_noise_steps_computation_mask is None and scm_preset is not None:
high_noise_steps_computation_mask = cache_dit.steps_mask(
mask_policy=scm_preset, total_steps=num_high_noise_steps
)
if low_noise_steps_computation_mask is None and scm_preset is not None:
low_noise_steps_computation_mask = cache_dit.steps_mask(
mask_policy=scm_preset, total_steps=num_low_noise_steps
)
policy = (
steps_computation_policy if steps_computation_policy is not None else scm_preset
)
cache_dit.refresh_context(
transformer,
cache_config=DBCacheConfig().reset(
num_inference_steps=num_high_noise_steps,
steps_computation_mask=high_noise_steps_computation_mask,
steps_computation_policy=policy,
),
verbose=verbose,
)
cache_dit.refresh_context(
transformer_2,
cache_config=DBCacheConfig().reset(
num_inference_steps=num_low_noise_steps,
steps_computation_mask=low_noise_steps_computation_mask,
steps_computation_policy=policy,
),
verbose=verbose,
)
logger.debug(
f"cache-dit refreshed on dual transformers (steps={num_high_noise_steps}, {num_low_noise_steps})"
)