From d6ef6c6ae4e8fb74e1e1ca78512bedac3140b887 Mon Sep 17 00:00:00 2001 From: Will Lin Date: Thu, 5 Jun 2025 16:08:57 -0700 Subject: [PATCH] Revert "Revert "[STA] Implement mask search for V1's Wan2.1 (#415)"" This reverts commit f657eb40dce949dff43e04ea1930bba7e3b1d068. --- examples/inference/sta_mask_search/README.md | 5 + .../sta_mask_search/inference_wan_sta.sh | 39 ++ .../inference/sta_mask_search/wan_example.py | 63 +++ fastvideo/v1/STA_configuration.py | 419 ++++++++++++++++++ .../attention/backends/sliding_tile_attn.py | 114 +++-- fastvideo/v1/configs/pipelines/base.py | 2 + fastvideo/v1/fastvideo_args.py | 18 + fastvideo/v1/pipelines/pipeline_batch_info.py | 6 + fastvideo/v1/pipelines/stages/denoising.py | 171 ++++++- 9 files changed, 803 insertions(+), 34 deletions(-) create mode 100644 examples/inference/sta_mask_search/README.md create mode 100644 examples/inference/sta_mask_search/inference_wan_sta.sh create mode 100644 examples/inference/sta_mask_search/wan_example.py create mode 100644 fastvideo/v1/STA_configuration.py diff --git a/examples/inference/sta_mask_search/README.md b/examples/inference/sta_mask_search/README.md new file mode 100644 index 000000000..d2ea6c155 --- /dev/null +++ b/examples/inference/sta_mask_search/README.md @@ -0,0 +1,5 @@ +# STA Mask Search Examples + +```bash +bash examples/inference/sta_mask_search/inference_wan_sta.sh +``` \ No newline at end of file diff --git a/examples/inference/sta_mask_search/inference_wan_sta.sh b/examples/inference/sta_mask_search/inference_wan_sta.sh new file mode 100644 index 000000000..b27218bfe --- /dev/null +++ b/examples/inference/sta_mask_search/inference_wan_sta.sh @@ -0,0 +1,39 @@ +#!/bin/bash + +export FASTVIDEO_ATTENTION_CONFIG=assets/mask_strategy_wan.json +export FASTVIDEO_ATTENTION_BACKEND=SLIDING_TILE_ATTN +export MODEL_BASE=Wan-AI/Wan2.1-T2V-14B-Diffusers + +base_port=29503 +num_gpu=$(nvidia-smi --query-gpu=gpu_name --format=csv,noheader | wc -l) +gpu_ids=$(seq 0 $((num_gpu-1))) +skip_time_steps=12 + +output_path="inference_results/sta/mask_search_full" +STA_mode="STA_searching" +for i in $gpu_ids; do + port=$((base_port+i)) + CUDA_VISIBLE_DEVICES=$i MASTER_PORT=$port python examples/inference/sta_mask_search/wan_example.py \ + --prompt_path ./assets/prompt_extend_${i}.txt \ + --output_path $output_path \ + --STA_mode $STA_mode & + sleep 1 +done +wait +echo "STA searching completed" + +output_path="inference_results/sta/mask_search_sparse" +STA_mode="STA_tuning" +for i in $gpu_ids; do + port=$((base_port+i)) + CUDA_VISIBLE_DEVICES=$i MASTER_PORT=$port python examples/inference/sta_mask_search/wan_example.py \ + --prompt_path ./assets/prompt_extend_${i}.txt \ + --output_path $output_path \ + --STA_mode $STA_mode \ + --skip_time_steps $skip_time_steps & + sleep 1 +done +wait +echo "STA tuning completed" + +echo "All jobs completed" \ No newline at end of file diff --git a/examples/inference/sta_mask_search/wan_example.py b/examples/inference/sta_mask_search/wan_example.py new file mode 100644 index 000000000..3e7e9c510 --- /dev/null +++ b/examples/inference/sta_mask_search/wan_example.py @@ -0,0 +1,63 @@ +import os +import argparse +from fastvideo import VideoGenerator, SamplingParam + +def main(args): + os.makedirs(args.output_path, exist_ok=True) + # Create a video generator with a pre-trained model + generator = VideoGenerator.from_pretrained( + "Wan-AI/Wan2.1-T2V-14B-Diffusers", + num_gpus=args.num_gpus, # Adjust based on your hardware + STA_mode=args.STA_mode, + skip_time_steps=args.skip_time_steps + ) + + # Prompts for your video + prompt = args.prompt + prompt_path = args.prompt_path + negative_prompt = "Bright tones, overexposed, static, blurred details, subtitles, style, works, paintings, images, static, overall gray, worst quality, low quality, JPEG compression residue, ugly, incomplete, extra fingers, poorly drawn hands, poorly drawn faces, deformed, disfigured, misshapen limbs, fused fingers, still picture, messy background, three legs, many people in the background, walking backwards" + + if prompt_path is not None: + with open(prompt_path, "r") as f: + prompts = f.readlines() + else: + prompts = [prompt] + + params = SamplingParam( + height=args.height, + width=args.width, + num_frames=args.num_frames, + num_inference_steps=args.num_inference_steps, + fps=args.fps, + guidance_scale=args.guidance_scale, + seed=args.seed, + return_frames=True, # Also return frames from this call (defaults to False) + output_path=args.output_path, # Controls where videos are saved + save_video=True, + negative_prompt=negative_prompt + ) + + # Generate the video + for prompt in prompts: + video = generator.generate_video( + prompt, + sampling_param=params, + ) + +if __name__ == '__main__': + parser = argparse.ArgumentParser() + parser.add_argument("--prompt", type=str, default="A man is dancing.") + parser.add_argument("--prompt_path", type=str, default=None) + parser.add_argument("--height", type=int, default=768) + parser.add_argument("--width", type=int, default=1280) + parser.add_argument("--num_frames", type=int, default=69) + parser.add_argument("--num_inference_steps", type=int, default=50) + parser.add_argument("--fps", type=int, default=16) + parser.add_argument("--guidance_scale", type=float, default=5.0) + parser.add_argument("--seed", type=int, default=12345) + parser.add_argument("--output_path", type=str, default="my_videos/") + parser.add_argument("--num_gpus", type=int, default=1) + parser.add_argument("--STA_mode", type=str, default="STA_searching") + parser.add_argument("--skip_time_steps", type=int, default=12) + args = parser.parse_args() + main(args) \ No newline at end of file diff --git a/fastvideo/v1/STA_configuration.py b/fastvideo/v1/STA_configuration.py new file mode 100644 index 000000000..88dd9494a --- /dev/null +++ b/fastvideo/v1/STA_configuration.py @@ -0,0 +1,419 @@ +import json +import os +from collections import defaultdict +from typing import Any, Dict, List, Optional, Tuple + +import numpy as np + + +def configure_sta(mode: str = 'STA_searching', + layer_num: int = 40, + time_step_num: int = 50, + head_num: int = 40, + **kwargs) -> List[List[List[Any]]]: + """ + Configure Sliding Tile Attention (STA) parameters based on the specified mode. + + Parameters: + ---------- + mode : str + The STA mode to use. Options are: + - 'STA_searching': Generate a set of mask candidates for initial search + - 'STA_tuning': Select best mask strategy based on previously saved results + - 'STA_inference': Load and use a previously tuned mask strategy + layer_num: int, number of layers + time_step_num: int, number of timesteps + head_num: int, number of heads + + **kwargs : dict + Mode-specific parameters: + + For 'STA_searching': + - mask_candidates: list of str, optional, mask candidates to use + - mask_selected: list of int, optional, indices of selected masks + + For 'STA_tuning': + - mask_search_files_path: str, required, path to mask search results + - mask_candidates: list of str, optional, mask candidates to use + - mask_selected: list of int, optional, indices of selected masks + - skip_time_steps: int, optional, number of time steps to use full attention (default 12) + - save_dir: str, optional, directory to save mask strategy (default "mask_candidates") + + For 'STA_inference': + - load_path: str, optional, path to load mask strategy (default "mask_candidates/mask_strategy.json") + """ + valid_modes = [ + 'STA_searching', 'STA_tuning', 'STA_inference', 'STA_tuning_cfg' + ] + if mode not in valid_modes: + raise ValueError(f"Mode must be one of {valid_modes}, got {mode}") + + if mode == 'STA_searching': + # Get parameters with defaults + mask_candidates: Optional[List[str]] = kwargs.get('mask_candidates') + if mask_candidates is None: + raise ValueError( + "mask_candidates is required for STA_searching mode") + mask_selected: List[int] = kwargs.get('mask_selected', + list(range(len(mask_candidates)))) + + # Parse selected masks + selected_masks: List[List[int]] = [] + for index in mask_selected: + mask = mask_candidates[index] + masks_list = [int(x) for x in mask.split(',')] + selected_masks.append(masks_list) + + # Create 3D mask structure with fixed dimensions (t=50, l=60) + masks_3d: List[List[List[List[int]]]] = [] + for i in range(time_step_num): # Fixed t dimension = 50 + row = [] + for j in range(layer_num): # Fixed l dimension = 60 + row.append(selected_masks) # Add all masks at each position + masks_3d.append(row) + + return masks_3d + + elif mode == 'STA_tuning': + # Get required parameters + mask_search_files_path: Optional[str] = kwargs.get( + 'mask_search_files_path') + if not mask_search_files_path: + raise ValueError( + "mask_search_files_path is required for STA_tuning mode") + + # Get optional parameters with defaults + mask_candidates_tuning: Optional[List[str]] = kwargs.get( + 'mask_candidates') + if mask_candidates_tuning is None: + raise ValueError("mask_candidates is required for STA_tuning mode") + mask_selected_tuning: List[int] = kwargs.get( + 'mask_selected', list(range(len(mask_candidates_tuning)))) + skip_time_steps_tuning: Optional[int] = kwargs.get('skip_time_steps') + save_dir_tuning: Optional[str] = kwargs.get('save_dir', + "mask_candidates") + + # Parse selected masks + selected_masks_tuning: List[List[int]] = [] + for index in mask_selected_tuning: + mask = mask_candidates_tuning[index] + masks_list = [int(x) for x in mask.split(',')] + selected_masks_tuning.append(masks_list) + + # Read JSON results + results = read_specific_json_files(mask_search_files_path) + averaged_results = average_head_losses(results, selected_masks_tuning) + + # Add full attention mask for specific cases + full_attention_mask_tuning: Optional[List[int]] = kwargs.get( + 'full_attention_mask') + if full_attention_mask_tuning is not None: + selected_masks_tuning.append(full_attention_mask_tuning) + + # Select best mask strategy + timesteps_tuning: int = kwargs.get('timesteps', time_step_num) + if skip_time_steps_tuning is None: + skip_time_steps_tuning = 12 + mask_strategy, sparsity, strategy_counts = select_best_mask_strategy( + averaged_results, selected_masks_tuning, skip_time_steps_tuning, + timesteps_tuning, head_num) + + # Save mask strategy + if save_dir_tuning is not None: + os.makedirs(save_dir_tuning, exist_ok=True) + file_path = os.path.join( + save_dir_tuning, + f'mask_strategy_s{skip_time_steps_tuning}.json') + with open(file_path, 'w') as f: + json.dump(mask_strategy, f, indent=4) + print(f"Successfully saved mask_strategy to {file_path}") + + # Print sparsity and strategy counts for information + print(f"Overall sparsity: {sparsity:.4f}") + print("\nStrategy usage counts:") + total_heads = time_step_num * layer_num * head_num # Fixed dimensions + for strategy, count in strategy_counts.items(): + print( + f"Strategy {strategy}: {count} heads ({count/total_heads*100:.2f}%)" + ) + + # Convert dictionary to 3D list with fixed dimensions + mask_strategy_3d = dict_to_3d_list(mask_strategy, + t_max=time_step_num, + l_max=layer_num, + h_max=head_num) + + return mask_strategy_3d + elif mode == 'STA_tuning_cfg': + # Get required parameters for both positive and negative paths + mask_search_files_path_pos: Optional[str] = kwargs.get( + 'mask_search_files_path_pos') + mask_search_files_path_neg: Optional[str] = kwargs.get( + 'mask_search_files_path_neg') + save_dir_cfg: Optional[str] = kwargs.get('save_dir') + + if not mask_search_files_path_pos or not mask_search_files_path_neg or not save_dir_cfg: + raise ValueError( + "mask_search_files_path_pos, mask_search_files_path_neg, and save_dir are required for STA_tuning_cfg mode" + ) + + # Get optional parameters with defaults + mask_candidates_cfg: Optional[List[str]] = kwargs.get('mask_candidates') + if mask_candidates_cfg is None: + raise ValueError( + "mask_candidates is required for STA_tuning_cfg mode") + mask_selected_cfg: List[int] = kwargs.get( + 'mask_selected', list(range(len(mask_candidates_cfg)))) + skip_time_steps_cfg: Optional[int] = kwargs.get('skip_time_steps') + + # Parse selected masks + selected_masks_cfg: List[List[int]] = [] + for index in mask_selected_cfg: + mask = mask_candidates_cfg[index] + masks_list = [int(x) for x in mask.split(',')] + selected_masks_cfg.append(masks_list) + + # Read JSON results for both positive and negative paths + pos_results = read_specific_json_files(mask_search_files_path_pos) + neg_results = read_specific_json_files(mask_search_files_path_neg) + # Combine positive and negative results into one list + combined_results = pos_results + neg_results + + # Average the combined results + averaged_results = average_head_losses(combined_results, + selected_masks_cfg) + + # Add full attention mask for specific cases + full_attention_mask_cfg: Optional[List[int]] = kwargs.get( + 'full_attention_mask') + if full_attention_mask_cfg is not None: + selected_masks_cfg.append(full_attention_mask_cfg) + + timesteps_cfg: int = kwargs.get('timesteps', time_step_num) + if skip_time_steps_cfg is None: + skip_time_steps_cfg = 12 + # Select best mask strategy using combined results + mask_strategy, sparsity, strategy_counts = select_best_mask_strategy( + averaged_results, selected_masks_cfg, skip_time_steps_cfg, + timesteps_cfg, head_num) + + # Save mask strategy + os.makedirs(save_dir_cfg, exist_ok=True) + file_path = os.path.join(save_dir_cfg, + f'mask_strategy_s{skip_time_steps_cfg}.json') + with open(file_path, 'w') as f: + json.dump(mask_strategy, f, indent=4) + print(f"Successfully saved mask_strategy to {file_path}") + + # Print sparsity and strategy counts for information + print(f"Overall sparsity: {sparsity:.4f}") + print("\nStrategy usage counts:") + total_heads = time_step_num * layer_num * head_num # Fixed dimensions + for strategy, count in strategy_counts.items(): + print( + f"Strategy {strategy}: {count} heads ({count/total_heads*100:.2f}%)" + ) + + # Convert dictionary to 3D list with fixed dimensions + mask_strategy_3d = dict_to_3d_list(mask_strategy, + t_max=time_step_num, + l_max=layer_num, + h_max=head_num) + + return mask_strategy_3d + + else: # STA_inference + # Get parameters with defaults + load_path: Optional[str] = kwargs.get( + 'load_path', "mask_candidates/mask_strategy.json") + if load_path is None: + raise ValueError("load_path is required for STA_inference mode") + + # Load previously saved mask strategy + with open(load_path) as f: + mask_strategy = json.load(f) + + # Convert dictionary to 3D list with fixed dimensions + mask_strategy_3d = dict_to_3d_list(mask_strategy, + t_max=time_step_num, + l_max=layer_num, + h_max=head_num) + + return mask_strategy_3d + + +# Helper functions + + +def read_specific_json_files(folder_path: str) -> List[Dict[str, Any]]: + """Read and parse JSON files containing mask search results.""" + json_contents: List[Dict[str, Any]] = [] + + # List files only in the current directory (no walk) + files = os.listdir(folder_path) + # Filter files + matching_files = [f for f in files if 'mask' in f and f.endswith('.json')] + print(f"Found {len(matching_files)} matching files: {matching_files}") + + for file_name in matching_files: + file_path = os.path.join(folder_path, file_name) + with open(file_path) as file: + data = json.load(file) + json_contents.append(data) + + return json_contents + + +def average_head_losses( + results: List[Dict[str, Any]], + selected_masks: List[List[int]]) -> Dict[str, Dict[str, np.ndarray]]: + """Average losses across all prompts for each mask strategy.""" + # Initialize a dictionary to store the averaged results + averaged_losses: Dict[str, Dict[str, np.ndarray]] = {} + loss_type = 'L2_loss' + # Get all loss types (e.g., 'L2_loss') + averaged_losses[loss_type] = {} + + for mask in selected_masks: + mask_str = str(mask) + data_shape = np.array(results[0][loss_type][mask_str]).shape + accumulated_data = np.zeros(data_shape) + + # Sum across all prompts + for prompt_result in results: + accumulated_data += np.array(prompt_result[loss_type][mask_str]) + + # Average by dividing by number of prompts + averaged_data = accumulated_data / len(results) + averaged_losses[loss_type][mask_str] = averaged_data + + return averaged_losses + + +def select_best_mask_strategy( + averaged_results: Dict[str, Dict[str, np.ndarray]], + selected_masks: List[List[int]], + skip_time_steps: int = 12, + timesteps: int = 50, + head_num: int = 40 +) -> Tuple[Dict[str, List[int]], float, Dict[str, int]]: + """Select the best mask strategy for each head based on loss minimization.""" + best_mask_strategy: Dict[str, List[int]] = {} + loss_type = 'L2_loss' + # Get the shape of time steps and layers + layers = len(averaged_results[loss_type][str(selected_masks[0])][0]) + + # Counter for sparsity calculation + total_tokens = 0 # total number of masked tokens + total_length = 0 # total sequence length + + strategy_counts: Dict[str, int] = { + str(strategy): 0 + for strategy in selected_masks + } + full_attn_strategy = selected_masks[-1] # Last strategy is full attention + print(f"Strategy {full_attn_strategy}, skip first {skip_time_steps} steps ") + + for t in range(timesteps): + for layer_idx in range(layers): + for h in range(head_num): + if t < skip_time_steps: # First steps use full attention + strategy = full_attn_strategy + else: + # Get losses for this head across all strategies + head_losses = [] + for strategy in selected_masks[: + -1]: # Exclude full attention + head_losses.append(averaged_results[loss_type][str( + strategy)][t][layer_idx][h]) + + # Find which strategy gives minimum loss + best_strategy_idx = np.argmin(head_losses) + strategy = selected_masks[best_strategy_idx] + + best_mask_strategy[f'{t}_{layer_idx}_{h}'] = strategy + + # Calculate sparsity + nums = strategy # strategy is already a list of numbers + total_tokens += nums[0] * nums[1] * nums[ + 2] # masked tokens for chosen strategy + total_length += full_attn_strategy[0] * full_attn_strategy[ + 1] * full_attn_strategy[2] + + # Count strategy usage + strategy_counts[str(strategy)] += 1 + + overall_sparsity = 1 - total_tokens / total_length + + return best_mask_strategy, overall_sparsity, strategy_counts + + +def dict_to_3d_list(mask_strategy: Optional[Dict[str, List[int]]], + t_max: int = 50, + l_max: int = 60, + h_max: int = 24) -> List[List[List[Optional[List[int]]]]]: + result: List[List[List[Optional[List[int]]]]] = [[[ + None for _ in range(h_max) + ] for _ in range(l_max)] for _ in range(t_max)] + if mask_strategy is None: + return result + for key, value in mask_strategy.items(): + t, layer_idx, h = map(int, key.split('_')) + result[t][layer_idx][h] = value + return result + + +def save_mask_search_results( + mask_search_final_result: List[Dict[str, List[float]]], + prompt: str, + mask_strategies: List[str], + output_dir: str = 'output/mask_search_result/') -> Optional[str]: + if not mask_search_final_result: + print("No mask search results to save") + return None + + # Create result dictionary with defaultdict for nested lists + mask_search_dict: Dict[str, Dict[str, List[List[float]]]] = { + "L2_loss": defaultdict(list), + "L1_loss": defaultdict(list) + } + + mask_selected = list(range(len(mask_strategies))) + selected_masks: List[List[int]] = [] + for index in mask_selected: + mask = mask_strategies[index] + masks_list = [int(x) for x in mask.split(',')] + selected_masks.append(masks_list) + + # Process each mask strategy + for i, mask_strategy in enumerate(selected_masks): + mask_strategy_str = str(mask_strategy) + # Process L2 loss + step_results: List[List[float]] = [] + for step_data in mask_search_final_result: + if isinstance(step_data, dict) and "L2_loss" in step_data: + layer_losses = [float(loss) for loss in step_data["L2_loss"]] + step_results.append(layer_losses) + mask_search_dict["L2_loss"][mask_strategy_str] = step_results + + step_results = [] + for step_data in mask_search_final_result: + if isinstance(step_data, dict) and "L1_loss" in step_data: + layer_losses = [float(loss) for loss in step_data["L1_loss"]] + step_results.append(layer_losses) + mask_search_dict["L1_loss"][mask_strategy_str] = step_results + + # Create the output directory if it doesn't exist + os.makedirs(output_dir, exist_ok=True) + + # Create a filename based on the first 20 characters of the prompt + filename = prompt[:50].replace(" ", "_") + filepath = os.path.join(output_dir, f'mask_search_{filename}.json') + + # Save the results to a JSON file + with open(filepath, 'w') as f: + json.dump(mask_search_dict, f, indent=4) + + print(f"Successfully saved mask research results to {filepath}") + + return filepath diff --git a/fastvideo/v1/attention/backends/sliding_tile_attn.py b/fastvideo/v1/attention/backends/sliding_tile_attn.py index d1a658019..719963062 100644 --- a/fastvideo/v1/attention/backends/sliding_tile_attn.py +++ b/fastvideo/v1/attention/backends/sliding_tile_attn.py @@ -1,6 +1,6 @@ import json from dataclasses import dataclass -from typing import List, Optional, Type +from typing import Any, Dict, List, Optional, Type import torch from einops import rearrange @@ -13,6 +13,7 @@ from fastvideo.v1.attention.backends.abstract import (AttentionBackend, AttentionMetadataBuilder) from fastvideo.v1.distributed import get_sp_group from fastvideo.v1.fastvideo_args import FastVideoArgs +from fastvideo.v1.forward_context import ForwardContext, get_forward_context from fastvideo.v1.logger import init_logger from fastvideo.v1.pipelines.pipeline_batch_info import ForwardBatch @@ -20,7 +21,9 @@ logger = init_logger(__name__) # TODO(will-refactor): move this to a utils file -def dict_to_3d_list(mask_strategy) -> List[List[List[Optional[torch.Tensor]]]]: +def dict_to_3d_list( + mask_strategy: Dict[str, + Any]) -> List[List[List[Optional[torch.Tensor]]]]: indices = [tuple(map(int, key.split('_'))) for key in mask_strategy] max_timesteps_idx = max( @@ -42,14 +45,14 @@ def dict_to_3d_list(mask_strategy) -> List[List[List[Optional[torch.Tensor]]]]: class RangeDict(dict): - def __getitem__(self, item): + def __getitem__(self, item: int) -> str: for key in self.keys(): if isinstance(key, tuple): low, high = key if low <= item <= high: - return super().__getitem__(key) + return str(super().__getitem__(key)) elif key == item: - return super().__getitem__(key) + return str(super().__getitem__(key)) raise KeyError(f"seq_len {item} not supported for STA") @@ -82,6 +85,8 @@ class SlidingTileAttentionBackend(AttentionBackend): @dataclass class SlidingTileAttentionMetadata(AttentionMetadata): current_timestep: int + STA_param: List[List[ + Any]] # each timestep with one metadata, shape [num_layers, num_heads] class SlidingTileAttentionMetadataBuilder(AttentionMetadataBuilder): @@ -98,8 +103,12 @@ class SlidingTileAttentionMetadataBuilder(AttentionMetadataBuilder): forward_batch: ForwardBatch, fastvideo_args: FastVideoArgs, ) -> SlidingTileAttentionMetadata: - - return SlidingTileAttentionMetadata(current_timestep=current_timestep, ) + param = forward_batch.STA_param + if param is None: + return SlidingTileAttentionMetadata( + current_timestep=current_timestep, STA_param=[]) + return SlidingTileAttentionMetadata(current_timestep=current_timestep, + STA_param=param[current_timestep]) class SlidingTileAttentionImpl(AttentionImpl): @@ -120,12 +129,12 @@ class SlidingTileAttentionImpl(AttentionImpl): if config_file is None: raise ValueError("FASTVIDEO_ATTENTION_CONFIG is not set") + # TODO(kevin): get mask strategy for different STA modes with open(config_file) as f: mask_strategy = json.load(f) - mask_strategy = dict_to_3d_list(mask_strategy) + self.mask_strategy = dict_to_3d_list(mask_strategy) self.prefix = prefix - self.mask_strategy = mask_strategy sp_group = get_sp_group() self.sp_size = sp_group.world_size # STA config @@ -205,16 +214,24 @@ class SlidingTileAttentionImpl(AttentionImpl): v: torch.Tensor, attn_metadata: SlidingTileAttentionMetadata, ) -> torch.Tensor: - - assert self.mask_strategy is not None, "mask_strategy cannot be None for SlidingTileAttention" - assert self.mask_strategy[ - 0] is not None, "mask_strategy[0] cannot be None for SlidingTileAttention" + if self.mask_strategy is None: + raise ValueError( + "mask_strategy cannot be None for SlidingTileAttention") + if self.mask_strategy[0] is None: + raise ValueError( + "mask_strategy[0] cannot be None for SlidingTileAttention") timestep = attn_metadata.current_timestep + forward_context: ForwardContext = get_forward_context() + forward_batch = forward_context.forward_batch + if forward_batch is None: + raise ValueError("forward_batch cannot be None") # pattern:'.double_blocks.0.attn.impl' or '.single_blocks.0.attn.impl' layer_idx = int(self.prefix.split('.')[-3]) - - # TODO: remove hardcode + if attn_metadata.STA_param is None or len( + attn_metadata.STA_param) <= layer_idx: + raise ValueError("Invalid STA_param") + STA_param = attn_metadata.STA_param[layer_idx] text_length = q.shape[1] - self.img_seq_length has_text = text_length > 0 @@ -227,15 +244,62 @@ class SlidingTileAttentionImpl(AttentionImpl): sp_group = get_sp_group() current_rank = sp_group.rank_in_group start_head = current_rank * head_num - windows = [ - self.mask_strategy[timestep][layer_idx][head_idx + start_head] - for head_idx in range(head_num) - ] - # if has_text is False: - # from IPython import embed - # embed() - hidden_states = sliding_tile_attention( - query, key, value, windows, text_length, has_text, - self.img_latent_shape_str).transpose(1, 2) + + # searching or tuning mode + if len(STA_param) < head_num * sp_group.world_size: + sparse_attn_hidden_states_all = [] + full_mask_window = STA_param[-1] + for window_size in STA_param[:-1]: + sparse_hidden_states = sliding_tile_attention( + query, key, value, [window_size] * head_num, text_length, + has_text, self.img_latent_shape_str).transpose(1, 2) + sparse_attn_hidden_states_all.append(sparse_hidden_states) + + hidden_states = sliding_tile_attention( + query, key, value, [full_mask_window] * head_num, text_length, + has_text, self.img_latent_shape_str).transpose(1, 2) + + attn_L2_loss = [] + attn_L1_loss = [] + # average loss across all heads + for sparse_attn_hidden_states in sparse_attn_hidden_states_all: + # L2 loss + attn_L2_loss_ = torch.mean((sparse_attn_hidden_states.float() - + hidden_states.float())**2, + dim=[0, 1, 3]).cpu().numpy() + attn_L2_loss_ = [round(float(x), 6) for x in attn_L2_loss_] + attn_L2_loss.append(attn_L2_loss_) + # L1 loss + attn_L1_loss_ = torch.mean( + torch.abs(sparse_attn_hidden_states.float() - + hidden_states.float()), + dim=[0, 1, 3]).cpu().numpy() + attn_L1_loss_ = [round(float(x), 6) for x in attn_L1_loss_] + attn_L1_loss.append(attn_L1_loss_) + + layer_loss_save = {"L2_loss": attn_L2_loss, "L1_loss": attn_L1_loss} + + if forward_batch.is_cfg_negative: + if forward_batch.mask_search_final_result_neg is not None: + forward_batch.mask_search_final_result_neg[timestep].append( + layer_loss_save) + else: + if forward_batch.mask_search_final_result_pos is not None: + forward_batch.mask_search_final_result_pos[timestep].append( + layer_loss_save) + else: + # windows = [ + # self.mask_strategy[timestep][layer_idx][head_idx + start_head] + # for head_idx in range(head_num) + # ] + windows = [ + STA_param[head_idx + start_head] for head_idx in range(head_num) + ] + # if has_text is False: + # from IPython import embed + # embed() + hidden_states = sliding_tile_attention( + query, key, value, windows, text_length, has_text, + self.img_latent_shape_str).transpose(1, 2) return hidden_states diff --git a/fastvideo/v1/configs/pipelines/base.py b/fastvideo/v1/configs/pipelines/base.py index 9a6515ed6..a0a20c02e 100644 --- a/fastvideo/v1/configs/pipelines/base.py +++ b/fastvideo/v1/configs/pipelines/base.py @@ -54,6 +54,8 @@ class PipelineConfig: # STA (Spatial-Temporal Attention) parameters mask_strategy_file_path: Optional[str] = None + STA_mode: str = "STA_inference" + skip_time_steps: int = 15 # Compilation enable_torch_compile: bool = False diff --git a/fastvideo/v1/fastvideo_args.py b/fastvideo/v1/fastvideo_args.py index f960cde23..33a9f0036 100644 --- a/fastvideo/v1/fastvideo_args.py +++ b/fastvideo/v1/fastvideo_args.py @@ -85,6 +85,9 @@ class FastVideoArgs: postprocess_text_funcs: Tuple[Callable[[Any], Any], ...] = field( default_factory=lambda: (postprocess_text, )) + # STA (Spatial-Temporal Attention) parameters + STA_mode: str = "STA_inference" + skip_time_steps: int = 15 # LoRA parameters lora_path: Optional[str] = None lora_nickname: Optional[ @@ -278,6 +281,21 @@ class FastVideoArgs: ) # STA (Spatial-Temporal Attention) parameters + parser.add_argument( + "--STA-mode", + type=str, + default=FastVideoArgs.STA_mode, + choices=[ + "STA_inference", "STA_searching", "STA_tuning", "STA_tuning_cfg" + ], + help="STA mode", + ) + parser.add_argument( + "--skip-time-steps", + type=int, + default=FastVideoArgs.skip_time_steps, + help="Number of time steps to warmup (full attention) for STA", + ) parser.add_argument( "--mask-strategy-file-path", type=str, diff --git a/fastvideo/v1/pipelines/pipeline_batch_info.py b/fastvideo/v1/pipelines/pipeline_batch_info.py index 3630837d5..c5eb2e5e0 100644 --- a/fastvideo/v1/pipelines/pipeline_batch_info.py +++ b/fastvideo/v1/pipelines/pipeline_batch_info.py @@ -114,6 +114,12 @@ class ForwardBatch: enable_teacache: bool = False teacache_params: Optional[TeaCacheParams | WanTeaCacheParams] = None + # STA parameters + STA_param: Optional[List] = None + is_cfg_negative: bool = False + mask_search_final_result_pos: Optional[List[List]] = None + mask_search_final_result_neg: Optional[List[List]] = None + def __post_init__(self): """Initialize dependent fields after dataclass initialization.""" diff --git a/fastvideo/v1/pipelines/stages/denoising.py b/fastvideo/v1/pipelines/stages/denoising.py index f7e88aade..a4afaa8a5 100644 --- a/fastvideo/v1/pipelines/stages/denoising.py +++ b/fastvideo/v1/pipelines/stages/denoising.py @@ -47,6 +47,14 @@ class DenoisingStage(PipelineStage): super().__init__() self.transformer = transformer self.scheduler = scheduler + attn_head_size = self.transformer.hidden_size // self.transformer.num_attention_heads + self.attn_backend = get_attn_backend( + head_size=attn_head_size, + dtype=torch.float16, # TODO(will): hack + supported_attention_backends=(_Backend.SLIDING_TILE_ATTN, + _Backend.FLASH_ATTN, + _Backend.TORCH_SDPA) # hack + ) def forward( self, @@ -151,6 +159,10 @@ class DenoisingStage(PipelineStage): }, ) + # Prepare STA parameters + if st_attn_available and self.attn_backend == SlidingTileAttentionBackend: + self.prepare_sta_param(batch, fastvideo_args) + # Get latents and embeddings latents = batch.latents prompt_embeds = batch.prompt_embeds @@ -192,15 +204,6 @@ class DenoisingStage(PipelineStage): dtype=target_dtype, enabled=autocast_enabled): - # TODO(will-refactor): all of this should be in the stage's init - attn_head_size = self.transformer.hidden_size // self.transformer.num_attention_heads - self.attn_backend = get_attn_backend( - head_size=attn_head_size, - dtype=torch.float16, # TODO(will): hack - supported_attention_backends=( - _Backend.SLIDING_TILE_ATTN, _Backend.FLASH_ATTN, - _Backend.TORCH_SDPA) # hack - ) if st_attn_available and self.attn_backend == SlidingTileAttentionBackend: self.attn_metadata_builder_cls = self.attn_backend.get_builder_cls( ) @@ -222,6 +225,7 @@ class DenoisingStage(PipelineStage): # support torch dynamo compilation. They pass in # attn_metadata, vllm_config, and num_tokens. We can pass in # fastvideo_args or training_args, and attn_metadata. + batch.is_cfg_negative = False with set_forward_context( current_timestep=i, attn_metadata=attn_metadata, @@ -240,6 +244,7 @@ class DenoisingStage(PipelineStage): # Apply guidance if batch.do_classifier_free_guidance: + batch.is_cfg_negative = True with set_forward_context( current_timestep=i, attn_metadata=attn_metadata, @@ -288,6 +293,15 @@ class DenoisingStage(PipelineStage): # Update batch with final latents batch.latents = latents + + # Save STA mask search results if needed + if st_attn_available and self.attn_backend == SlidingTileAttentionBackend and fastvideo_args.STA_mode == 'STA_searching': + self.save_sta_search_results(batch) + + if fastvideo_args.use_cpu_offload: + self.transformer.to('cpu') + torch.cuda.empty_cache() + return batch def prepare_extra_func_kwargs(self, func, kwargs) -> Dict[str, Any]: @@ -355,3 +369,142 @@ class DenoisingStage(PipelineStage): noise_cfg = (guidance_rescale * noise_pred_rescaled + (1 - guidance_rescale) * noise_cfg) return noise_cfg + + def prepare_sta_param(self, batch: ForwardBatch, + fastvideo_args: FastVideoArgs): + """ + Prepare Sliding Tile Attention (STA) parameters and settings. + + Args: + batch: The current batch information. + fastvideo_args: The inference arguments. + """ + # TODO(kevin): STA mask search, currently only support Wan2.1 with 69x768x1280 + from fastvideo.v1.STA_configuration import configure_sta + STA_mode = fastvideo_args.STA_mode + skip_time_steps = fastvideo_args.skip_time_steps + if batch.timesteps is None: + raise ValueError("Timesteps must be provided") + timesteps_num = batch.timesteps.shape[0] + + logger.info("STA_mode: %s", STA_mode) + if (batch.num_frames, batch.height, + batch.width) != (69, 768, 1280) and STA_mode != "STA_inference": + raise NotImplementedError( + "STA mask search/tuning is not supported for this resolution") + + if STA_mode == "STA_searching" or STA_mode == "STA_tuning" or STA_mode == "STA_tuning_cfg": + size = (batch.width, batch.height) + if size == (1280, 768): + # TODO: make it configurable + sparse_mask_candidates_searching = [ + "3, 1, 10", "1, 5, 7", "3, 3, 3", "1, 6, 5", "1, 3, 10", + "3, 6, 1" + ] + sparse_mask_candidates_tuning = [ + "3, 1, 10", "1, 5, 7", "3, 3, 3", "1, 6, 5", "1, 3, 10", + "3, 6, 1" + ] + full_mask = ["3,6,10"] + else: + raise NotImplementedError( + "STA mask search is not supported for this resolution") + layer_num = self.transformer.config.num_layers + # specific for HunyuanVideo + if hasattr(self.transformer.config, "num_single_layers"): + layer_num += self.transformer.config.num_single_layers + head_num = self.transformer.config.num_attention_heads + + if STA_mode == "STA_searching": + STA_param = configure_sta( + mode='STA_searching', + layer_num=layer_num, + head_num=head_num, + time_step_num=timesteps_num, + mask_candidates=sparse_mask_candidates_searching + + full_mask, # last is full mask; Can add more sparse masks while keep last one as full mask + ) + elif STA_mode == 'STA_tuning': + STA_param = configure_sta( + mode='STA_tuning', + layer_num=layer_num, + head_num=head_num, + time_step_num=timesteps_num, + mask_search_files_path= + f'output/mask_search_result_pos_{size[0]}x{size[1]}/', + mask_candidates=sparse_mask_candidates_tuning, + full_attention_mask=[int(x) for x in full_mask[0].split(',')], + skip_time_steps= + skip_time_steps, # Use full attention for first 12 steps + save_dir= + f'output/mask_search_strategy_{size[0]}x{size[1]}/', # Custom save directory + timesteps=timesteps_num) + elif STA_mode == 'STA_tuning_cfg': + STA_param = configure_sta( + mode='STA_tuning_cfg', + layer_num=layer_num, + head_num=head_num, + time_step_num=timesteps_num, + mask_search_files_path_pos= + f'output/mask_search_result_pos_{size[0]}x{size[1]}/', + mask_search_files_path_neg= + f'output/mask_search_result_neg_{size[0]}x{size[1]}/', + mask_candidates=sparse_mask_candidates_tuning, + full_attention_mask=[int(x) for x in full_mask[0].split(',')], + skip_time_steps=skip_time_steps, + save_dir=f'output/mask_search_strategy_{size[0]}x{size[1]}/', + timesteps=timesteps_num) + elif STA_mode == 'STA_inference': + import fastvideo.v1.envs as envs + config_file = envs.FASTVIDEO_ATTENTION_CONFIG + if config_file is None: + raise ValueError("FASTVIDEO_ATTENTION_CONFIG is not set") + STA_param = configure_sta(mode='STA_inference', + layer_num=layer_num, + head_num=head_num, + time_step_num=timesteps_num, + load_path=config_file) + + batch.STA_param = STA_param + batch.mask_search_final_result_pos = [[] for _ in range(timesteps_num)] + batch.mask_search_final_result_neg = [[] for _ in range(timesteps_num)] + + def save_sta_search_results(self, batch: ForwardBatch): + """ + Save the STA mask search results. + + Args: + batch: The current batch information. + """ + size = (batch.width, batch.height) + if size == (1280, 768): + # TODO: make it configurable + sparse_mask_candidates_searching = [ + "3, 1, 10", "1, 5, 7", "3, 3, 3", "1, 6, 5", "1, 3, 10", + "3, 6, 1" + ] + else: + raise NotImplementedError( + "STA mask search is not supported for this resolution") + + from fastvideo.v1.STA_configuration import save_mask_search_results + if batch.mask_search_final_result_pos is not None and batch.prompt is not None: + save_mask_search_results( + [ + dict(layer_data) + for layer_data in batch.mask_search_final_result_pos + ], + prompt=str(batch.prompt), + mask_strategies=sparse_mask_candidates_searching, + output_dir=f'output/mask_search_result_pos_{size[0]}x{size[1]}/' + ) + if batch.mask_search_final_result_neg is not None and batch.prompt is not None: + save_mask_search_results( + [ + dict(layer_data) + for layer_data in batch.mask_search_final_result_neg + ], + prompt=str(batch.prompt), + mask_strategies=sparse_mask_candidates_searching, + output_dir=f'output/mask_search_result_neg_{size[0]}x{size[1]}/' + )