From 5d7beeb5a0988979ea77e41536cbad971726899a Mon Sep 17 00:00:00 2001 From: SolitaryThinker Date: Sat, 2 Aug 2025 01:22:21 +0000 Subject: [PATCH] checkpoint --- examples/inference/basic/basic.py | 6 +- examples/inference/gradio/gradio_frontend.py | 404 ++++++++----- .../inference/gradio/ray_serve_backend.py | 565 +++++++++++++----- .../inference/gradio/start_ray_serve_app.py | 15 +- 4 files changed, 663 insertions(+), 327 deletions(-) diff --git a/examples/inference/basic/basic.py b/examples/inference/basic/basic.py index 9174c6abf..e66568a92 100644 --- a/examples/inference/basic/basic.py +++ b/examples/inference/basic/basic.py @@ -2,13 +2,17 @@ from fastvideo import VideoGenerator from fastvideo.configs.sample import SamplingParam +import os + +os.environ["FASTVIDEO_ATTENTION_BACKEND"] = "VIDEO_SPARSE_ATTN" + OUTPUT_PATH = "video_samples" def main(): # FastVideo will automatically use the optimal default arguments for the # model. # If a local path is provided, FastVideo will make a best effort # attempt to identify the optimal arguments. - model_name = "Wan-AI/Wan2.2-TI2V-5B-Diffusers" + model_name = "FastVideo/FastWan2.1-T2V-14B-Diffusers" generator = VideoGenerator.from_pretrained( model_name, # FastVideo will automatically handle distributed setup diff --git a/examples/inference/gradio/gradio_frontend.py b/examples/inference/gradio/gradio_frontend.py index 5beb4f31a..4661eb649 100644 --- a/examples/inference/gradio/gradio_frontend.py +++ b/examples/inference/gradio/gradio_frontend.py @@ -20,6 +20,7 @@ class RayServeClient: self.backend_url = backend_url.rstrip('/') self.generate_endpoint = f"{self.backend_url}/generate_video" self.health_endpoint = f"{self.backend_url}/health" + # Default request timeout in seconds. Increase if generation may run longer. self.request_timeout_s = int(os.getenv("FASTVIDEO_GENERATION_TIMEOUT", "900")) # 15 minutes default @@ -31,8 +32,14 @@ class RayServeClient: except Exception: return False + + def generate_video(self, request_data: dict) -> dict: """Send video generation request to the backend""" + import time + + request_start_time = time.time() + try: response = requests.post( self.generate_endpoint, @@ -40,16 +47,56 @@ class RayServeClient: timeout=self.request_timeout_s, ) response.raise_for_status() - return response.json() + + request_end_time = time.time() + network_time = request_end_time - request_start_time + + response_data = response.json() + response_data["network_time"] = network_time + + return response_data + except requests.exceptions.RequestException as e: return { "success": False, "error_message": f"Backend request failed: {str(e)}", - "output_path": "", - "seed": request_data.get("seed", 42) + "video_data": None, + "seed": request_data.get("seed", 42), + "network_time": time.time() - request_start_time } +def save_video_from_base64(video_data: str, output_dir: str, prompt: str) -> str: + """Save base64-encoded video data to a file""" + if not video_data: + return "No video data to save" + + try: + # Remove the data URL prefix if present + if video_data.startswith('data:video/'): + video_data = video_data.split(',')[1] + + # Decode base64 to bytes + video_bytes = base64.b64decode(video_data) + + # Create safe filename from prompt + safe_prompt = prompt[:50].replace(' ', '_').replace('/', '_').replace('\\', '_') + video_filename = f"{safe_prompt}.mp4" + video_path = os.path.join(output_dir, video_filename) + + # Ensure output directory exists + os.makedirs(output_dir, exist_ok=True) + + # Save video bytes to file + with open(video_path, 'wb') as f: + f.write(video_bytes) + + return f"Saved video to: {video_path}", video_path + + except Exception as e: + return f"Failed to save video: {str(e)}", "" + + def decode_and_save_video_from_frames(frames_b64: list, output_dir: str, prompt: str, fps: int = 24) -> str: """Decode base64 frames and save them as a video file""" if not frames_b64: @@ -115,14 +162,25 @@ def create_gradio_interface(backend_url: str, default_params: SamplingParam): num_inference_steps, randomize_seed=False, input_image=None, + model_selection="FastVideo/FastWan2.1-T2V-1.3B-Diffusers (Text-to-Video)", + progress=None, ): # Check backend health first if not client.check_health(): return None, f"Backend is not available. Please check if Ray Serve is running at {backend_url}", "" + # Update progress + if progress: + progress(0.1, desc="Checking backend health...") + + # Determine if this is I2V based on model selection + is_i2v = "I2V" in model_selection or "Image-to-Video" in model_selection + # Handle input image for I2V image_path = None - if input_image is not None: + if is_i2v and input_image is not None: + if progress: + progress(0.2, desc="Processing input image...") try: # Save the uploaded image to a temporary file import tempfile @@ -160,7 +218,10 @@ def create_gradio_interface(backend_url: str, default_params: SamplingParam): print(f"Warning: Failed to save input image: {e}") image_path = None - # Prepare request data - always request frames for video creation + # Prepare request data + if progress: + progress(0.3, desc="Preparing request...") + request_data = { "prompt": prompt, "negative_prompt": negative_prompt, @@ -172,14 +233,21 @@ def create_gradio_interface(backend_url: str, default_params: SamplingParam): "width": width, "num_inference_steps": num_inference_steps, "randomize_seed": randomize_seed, - "return_frames": False, # Always request frames + "return_frames": False, # We'll get video data directly "image_path": image_path, - "model_type": "i2v" if image_path else "t2v" # Use I2V model if image is provided, T2V otherwise + "model_type": "i2v" if is_i2v else "t2v", # Use model type selection + "model_path": model_selection.split(" (")[0] if model_selection else None # Extract model path from selection } # Send request to backend + if progress: + progress(0.4, desc="Sending request to backend...") + response = client.generate_video(request_data) + if progress: + progress(0.8, desc="Processing response...") + # Clean up temporary image file after processing if image_path and os.path.exists(image_path): try: @@ -189,32 +257,57 @@ def create_gradio_interface(backend_url: str, default_params: SamplingParam): print(f"Warning: Failed to clean up temporary image {image_path}: {e}") if response.get("success", False): - output_path = response.get("output_path", "") + video_data = response.get("video_data", "") used_seed = response.get("seed", seed) - frames_b64 = response.get("frames", []) + generation_time = response.get("generation_time", 0.0) + inference_time = response.get("inference_time", 0.0) + encoding_time = response.get("encoding_time", 0.0) + total_time = response.get("total_time", 0.0) + network_time = response.get("network_time", 0.0) print(f"Used seed: {used_seed}") - print(f"Output path: {output_path}") + print(f"Inference time: {inference_time:.2f}s") + print(f"Encoding time: {encoding_time:.2f}s") + print(f"Network transfer: {network_time:.2f}s") + print(f"Total time: {total_time:.2f}s") - # Handle frame extraction and video creation - frames_status = "" - # if frames_b64: - # try: - # Get the output directory from the video path - # output_dir = os.path.dirname(output_path) if output_path else "outputs" - # frames_status, video_path = decode_and_save_video_from_frames(frames_b64, output_dir, prompt) - # print(f"Frames: {frames_status}") - # except Exception as e: - # frames_status = f"Failed to save frames video: {str(e)}" - # print(f"Frame extraction error: {e}") - # else: - # frames_status = "No frames returned from backend" + # Create detailed timing message + timing_details = f"⏱️ **Timing Breakdown**\n\n" + timing_details += f"| Phase | Time |\n" + timing_details += f"|-------|------|\n" + timing_details += f"| 🧠 Model Inference | {inference_time:.2f}s |\n" + timing_details += f"| 🎬 Video Encoding | {encoding_time:.2f}s |\n" + timing_details += f"| 🌐 Network Transfer | {network_time:.2f}s |\n" + timing_details += f"| **📊 Total Processing** | **{total_time:.2f}s** |\n\n" - # Check if the video file exists - if os.path.exists(output_path): - return output_path, used_seed, frames_status + # Add performance insights + if inference_time > 0: + fps = num_frames / inference_time + timing_details += f"**Performance Metrics:**\n" + timing_details += f"• Generation Speed: {fps:.1f} frames/second\n" + timing_details += f"• Model Efficiency: {inference_time/total_time*100:.1f}% of total time\n" + + # Save video data to file for Gradio to display + if video_data: + try: + if progress: + progress(0.9, desc="Saving video...") + + output_dir = "outputs" + save_status, video_path = save_video_from_base64(video_data, output_dir, prompt) + print(f"Video save status: {save_status}") + + if progress: + progress(1.0, desc="Generation complete!") + + if video_path and os.path.exists(video_path): + return video_path, used_seed, timing_details + else: + return None, f"Video generated but failed to save: {save_status}", "" + except Exception as e: + return None, f"Failed to save video: {str(e)}", "" else: - return None, f"Video generated but file not found at {output_path} {frames_status}", frames_status + return None, "No video data received from backend", "" else: error_msg = response.get("error_message", "Unknown error occurred") return None, f"Generation failed: {error_msg}", "" @@ -226,17 +319,17 @@ def create_gradio_interface(backend_url: str, default_params: SamplingParam): "A crowded rooftop bar buzzes with energy, the city skyline twinkling like a field of stars in the background. Strings of fairy lights hang above, casting a warm, golden glow over the scene. Groups of people gather around high tables, their laughter blending with the soft rhythm of live jazz. The aroma of freshly mixed cocktails and charred appetizers wafts through the air, mingling with the cool night breeze.", ] - # Example I2V prompts (for when users upload images) - i2v_examples = [ - "The image comes to life with subtle movement, the scene gently animating while maintaining the original composition and mood.", - "The static image transforms into a dynamic scene with natural motion, preserving the original lighting and atmosphere.", - "The photograph animates with realistic movement, bringing the frozen moment to life while keeping the original artistic style.", + example_labels = [ + "Hand wrapping dough with plastic", + "Vintage train through mountains", + "Crowded rooftop bar at night" ] + + # Create Gradio interface with gr.Blocks() as demo: gr.Markdown("# FastVideo Inference Demo (Ray Serve Backend)") - gr.Markdown(f"**Backend URL:** {backend_url}") # Backend status indicator status_text = gr.Text( @@ -251,53 +344,75 @@ def create_gradio_interface(backend_url: str, default_params: SamplingParam): else: return "❌ Backend is not available" - with gr.Tabs(): - # Text-to-Video Tab - with gr.Tab("Text-to-Video"): - with gr.Group(): - with gr.Row(): - prompt = gr.Text( - label="Prompt", - show_label=False, - max_lines=1, - placeholder="Enter your prompt", - container=False, - ) - run_button = gr.Button("Run", scale=0) - - result = gr.Video(label="Result", show_label=False) - error_output = gr.Text(label="Error", visible=False) - frames_output = gr.Text(label="Frame Video Status", visible=False) - download_file = gr.File(visible=False) - - gr.Examples(examples=examples, inputs=prompt) + # Model selection dropdown + with gr.Row(): + model_selection = gr.Dropdown( + choices=[ + "FastVideo/FastWan2.1-T2V-1.3B-Diffusers (Text-to-Video)", + "FastVideo/FastWan2.1-T2V-14B-Diffusers (Text-to-Video)", + "Wan-AI/Wan2.1-I2V-14B-480P-Diffusers (Image-to-Video)" + ], + value="FastVideo/FastWan2.1-T2V-1.3B-Diffusers (Text-to-Video)", + label="Select Model", + interactive=True + ) + + # Main interface + with gr.Group(): + with gr.Row(): + prompt = gr.Text( + label="Prompt", + show_label=False, + max_lines=3, + placeholder="Enter your prompt", + container=False, + lines=3, + scale=4, + ) + run_button = gr.Button("Run", scale=1) - # Image-to-Video Tab - with gr.Tab("Image-to-Video"): - with gr.Group(): - with gr.Row(): - i2v_prompt = gr.Text( - label="Prompt", - show_label=False, - max_lines=1, - placeholder="Describe how the image should animate", - container=False, - ) - i2v_run_button = gr.Button("Run", scale=0) - - input_image = gr.Image( - label="Input Image", - type="pil", - show_label=True, - container=True, - ) - - i2v_result = gr.Video(label="Result", show_label=False) - i2v_error_output = gr.Text(label="Error", visible=False) - i2v_frames_output = gr.Text(label="Frame Video Status", visible=False) - i2v_download_file = gr.File(visible=False) - - gr.Examples(examples=i2v_examples, inputs=i2v_prompt) + # Input image (only visible for I2V models) + input_image = gr.Image( + label="Input Image (required for Image-to-Video models)", + type="pil", + show_label=True, + container=True, + visible=False, + ) + + # Results + result = gr.Video(label="Result", show_label=False) + error_output = gr.Text(label="Error", visible=False) + frames_output = gr.Text(label="Generation Status", visible=False) + timing_display = gr.Markdown(label="Timing Breakdown", visible=False) + download_file = gr.File(visible=False) + + # Progress tracking + progress_bar = gr.Progress() + status_updates = gr.Text(label="Status Updates", visible=False) + + # Examples dropdown + with gr.Row(): + example_dropdown = gr.Dropdown( + choices=example_labels, + label="Example Prompts", + value=None, + interactive=True, + allow_custom_value=False + ) + + # Function to update prompt when example is selected + def on_example_select(example_label): + if example_label and example_label in example_labels: + index = example_labels.index(example_label) + return examples[index] + return "" + + example_dropdown.change( + fn=on_example_select, + inputs=example_dropdown, + outputs=prompt, + ) # Shared advanced options with gr.Accordion("Advanced options", open=False): @@ -344,7 +459,8 @@ def create_gradio_interface(backend_url: str, default_params: SamplingParam): label="Use negative prompt", value=False) negative_prompt = gr.Text( label="Negative prompt", - max_lines=1, + max_lines=3, + lines=3, placeholder="Enter a negative prompt", visible=False, ) @@ -359,8 +475,6 @@ def create_gradio_interface(backend_url: str, default_params: SamplingParam): randomize_seed = gr.Checkbox(label="Randomize seed", value=False) seed_output = gr.Number(label="Used Seed") - gr.Examples(examples=examples, inputs=prompt) - # Event handlers use_negative_prompt.change( fn=lambda x: gr.update(visible=x), @@ -368,75 +482,59 @@ def create_gradio_interface(backend_url: str, default_params: SamplingParam): outputs=negative_prompt, ) - def handle_t2v_generation(*args): + # Model selection change handler + def on_model_selection_change(model_selection): + is_i2v = "I2V" in model_selection or "Image-to-Video" in model_selection + prompt_placeholder = "Describe how the image should animate" if is_i2v else "Enter your prompt" + return gr.update(visible=is_i2v), gr.update(placeholder=prompt_placeholder) + + model_selection.change( + fn=on_model_selection_change, + inputs=model_selection, + outputs=[input_image, prompt], + ) + + def handle_generation(*args, progress=None): + # Extract model selection and input image from args + model_selection, prompt, negative_prompt, use_negative_prompt, seed, guidance_scale, num_frames, height, width, num_inference_steps, randomize_seed, input_image = args + + # Determine if this is I2V based on model selection + is_i2v = "I2V" in model_selection or "Image-to-Video" in model_selection + # For T2V, we pass None as input_image - args = list(args) - args.append(None) # Add None for input_image - result_path, seed_or_error, frames_status = generate_video(*args) + if not is_i2v: + input_image = None + + # Call the generate_video function with progress tracking + result_path, seed_or_error, timing_details = generate_video( + prompt, negative_prompt, use_negative_prompt, seed, guidance_scale, + num_frames, height, width, num_inference_steps, randomize_seed, input_image, model_selection, progress + ) if result_path and os.path.exists(result_path): - # Show frame status if available - if frames_status: - return ( - result_path, - seed_or_error, - gr.update(visible=False), # error_output - gr.update(visible=True, value=frames_status), # frames_output - gr.update(visible=True, value=result_path) # download_file - ) - else: - return ( - result_path, - seed_or_error, - gr.update(visible=False), # error_output - gr.update(visible=False), # frames_output - gr.update(visible=True, value=result_path) # download_file - ) + return ( + result_path, + seed_or_error, + gr.update(visible=False), # error_output + gr.update(visible=True, value="Generation completed successfully!"), # frames_output + gr.update(visible=True, value=timing_details), # timing_display + gr.update(visible=True, value=result_path), # download_file + ) else: return ( None, seed_or_error, gr.update(visible=True, value=seed_or_error), # error_output gr.update(visible=False), # frames_output - gr.update(visible=False) # download_file + gr.update(visible=False), # timing_display + gr.update(visible=False), # download_file ) - def handle_i2v_generation(*args): - # For I2V, we need to reorder args to match generate_video signature - # args should be: [i2v_prompt, negative_prompt, use_negative_prompt, seed, guidance_scale, num_frames, height, width, num_inference_steps, randomize_seed, input_image] - result_path, seed_or_error, frames_status = generate_video(*args) - - if result_path and os.path.exists(result_path): - # Show frame status if available - if frames_status: - return ( - result_path, - seed_or_error, - gr.update(visible=False), # i2v_error_output - gr.update(visible=True, value=frames_status), # i2v_frames_output - gr.update(visible=True, value=result_path) # i2v_download_file - ) - else: - return ( - result_path, - seed_or_error, - gr.update(visible=False), # i2v_error_output - gr.update(visible=False), # i2v_frames_output - gr.update(visible=True, value=result_path) # i2v_download_file - ) - else: - return ( - None, - seed_or_error, - gr.update(visible=True, value=seed_or_error), # i2v_error_output - gr.update(visible=False), # i2v_frames_output - gr.update(visible=False) # i2v_download_file - ) - - # T2V event handler + # Unified event handler run_button.click( - fn=handle_t2v_generation, + fn=handle_generation, inputs=[ + model_selection, prompt, negative_prompt, use_negative_prompt, @@ -447,28 +545,9 @@ def create_gradio_interface(backend_url: str, default_params: SamplingParam): width, num_inference_steps, randomize_seed, - ], - outputs=[result, seed_output, error_output, frames_output, download_file], - concurrency_limit=20, - ) - - # I2V event handler - i2v_run_button.click( - fn=handle_i2v_generation, - inputs=[ - i2v_prompt, - negative_prompt, - use_negative_prompt, - seed, - guidance_scale, - num_frames, - height, - width, - num_inference_steps, - randomize_seed, input_image, ], - outputs=[i2v_result, seed_output, i2v_error_output, i2v_frames_output, i2v_download_file], + outputs=[result, seed_output, error_output, frames_output, timing_display, download_file], concurrency_limit=20, ) @@ -488,9 +567,13 @@ def main(): type=str, default="FastVideo/FastWan2.1-T2V-1.3B-Diffusers", help="Path to the T2V model (for default parameters)") + parser.add_argument("--t2v_14b_model_path", + type=str, + default="FastVideo/FastWan2.1-T2V-14B-Diffusers", + help="Path to the T2V 14B model (for default parameters)") parser.add_argument("--i2v_model_path", type=str, - default="Wan-AI/Wan2.2-TI2V-5B-Diffusers", + default="Wan-AI/Wan2.1-I2V-14B-480P-Diffusers", help="Path to the I2V model (for default parameters)") parser.add_argument("--host", type=str, @@ -523,7 +606,8 @@ def main(): print(f"Starting Gradio frontend at http://{args.host}:{args.port}") print(f"Backend URL: {args.backend_url}") - print(f"T2V Model: {args.t2v_model_path}") + print(f"T2V 1.3B Model: {args.t2v_model_path}") + print(f"T2V 14B Model: {args.t2v_14b_model_path}") print(f"I2V Model: {args.i2v_model_path}") demo.queue(max_size=20).launch( diff --git a/examples/inference/gradio/ray_serve_backend.py b/examples/inference/gradio/ray_serve_backend.py index 6263bae9f..48dfbb313 100644 --- a/examples/inference/gradio/ray_serve_backend.py +++ b/examples/inference/gradio/ray_serve_backend.py @@ -15,6 +15,8 @@ import numpy as np from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded +import imageio +from ray.serve.handle import DeploymentHandle class VideoGenerationRequest(BaseModel): @@ -31,14 +33,41 @@ class VideoGenerationRequest(BaseModel): return_frames: bool = False # Whether to return base64 encoded frames image_path: Optional[str] = None # Path to input image for I2V model_type: str = "t2v" # "t2v" or "i2v" to specify which model to use + model_path: Optional[str] = None # Specific model path to use class VideoGenerationResponse(BaseModel): - output_path: str + video_data: Optional[str] = None # Base64 encoded video data seed: int success: bool error_message: Optional[str] = None frames: Optional[List[str]] = None # Base64 encoded frames + generation_time: Optional[float] = None + # Detailed timing information + model_load_time: Optional[float] = None + inference_time: Optional[float] = None + encoding_time: Optional[float] = None + total_time: Optional[float] = None + + +def encode_video_to_base64(frames: List[np.ndarray], fps: int = 24) -> str: + """Convert numpy frames to base64-encoded MP4 video""" + if not frames: + return "" + + try: + # Save frames to bytes buffer as MP4 + buffer = io.BytesIO() + imageio.mimsave(buffer, frames, fps=fps, format="mp4") + buffer.seek(0) + + # Encode to base64 + video_base64 = base64.b64encode(buffer.getvalue()).decode('utf-8') + return f"data:video/mp4;base64,{video_base64}" + + except Exception as e: + print(f"Warning: Failed to encode video: {e}") + return "" def encode_frames_to_base64(frames: List[np.ndarray]) -> List[str]: @@ -90,6 +119,311 @@ def encode_frames_to_base64(frames: List[np.ndarray]) -> List[str]: return encoded_frames +# --------------------------------------------------------------------------- +# Model-Specific Deployments (one per model type) +# These deployments each load a single model and expose a `generate_video` +# method that can be invoked via a DeploymentHandle. Each deployment can be +# scaled independently by configuring `num_replicas` when binding. +# --------------------------------------------------------------------------- + + +@serve.deployment( # T2V 1.3B model deployment + ray_actor_options={"num_cpus": 10, "num_gpus": 1, "runtime_env": {"conda": "fv"}}, +) +class T2VModelDeployment: + """Serve deployment wrapping the 1.3 B text-to-video model.""" + + def __init__(self, t2v_model_path: str, output_path: str = "outputs"): + self.model_path = t2v_model_path + self.output_path = output_path + + # Ensure output directory exists + os.makedirs(self.output_path, exist_ok=True) + + # Delay helps avoid GPU contention when many replicas start at once + time.sleep(10) + + # Ensure correct attention backend for FastVideo + os.environ["FASTVIDEO_ATTENTION_BACKEND"] = "VIDEO_SPARSE_ATTN" + + # Lazy import to keep the deployment import-safe on head node + from fastvideo.entrypoints.video_generator import VideoGenerator + from fastvideo.configs.sample.base import SamplingParam + + print(f"Initializing T2V model: {self.model_path}") + self.generator = VideoGenerator.from_pretrained( + model_path=self.model_path, + num_gpus=1, + use_fsdp_inference=True, + text_encoder_cpu_offload=False, + dit_cpu_offload=False, + vae_cpu_offload=False, + VSA_sparsity=0.8, + ) + self.default_params = SamplingParam.from_pretrained(self.model_path) + print("✅ T2V model initialized successfully") + + def generate_video(self, video_request: "VideoGenerationRequest") -> "VideoGenerationResponse": + """Generate a video for the given request and return an encoded response.""" + import time + + total_start_time = time.time() + + # Deep-copy default sampling params and override with request values + params = deepcopy(self.default_params) + params.prompt = video_request.prompt + if video_request.use_negative_prompt: + params.negative_prompt = video_request.negative_prompt + + params.seed = video_request.seed if not video_request.randomize_seed else torch.randint(0, 1_000_000, (1,)).item() + params.guidance_scale = video_request.guidance_scale + params.num_frames = video_request.num_frames + params.height = video_request.height + params.width = video_request.width + params.num_inference_steps = video_request.num_inference_steps + + # Do not write to disk when called via API + params.save_video = False + params.return_frames = True + + # Track inference time + inference_start_time = time.time() + + # Generate video frames + result = self.generator.generate_video( + prompt=video_request.prompt, + sampling_param=params, + save_video=False, + return_frames=True, + ) + + inference_end_time = time.time() + inference_time = inference_end_time - inference_start_time + + frames = result if isinstance(result, list) else result.get("frames", []) + generation_time = result.get("generation_time", 0.0) if isinstance(result, dict) else 0.0 + + # Track encoding time + encoding_start_time = time.time() + + # Encode outputs + video_data = encode_video_to_base64(frames, fps=24) + encoded_frames = encode_frames_to_base64(frames) if video_request.return_frames and frames else None + + encoding_end_time = time.time() + encoding_time = encoding_end_time - encoding_start_time + + total_end_time = time.time() + total_time = total_end_time - total_start_time + + return VideoGenerationResponse( + video_data=video_data, + frames=encoded_frames, + seed=params.seed, + success=True, + generation_time=generation_time, + inference_time=inference_time, + encoding_time=encoding_time, + total_time=total_time, + ) + + +@serve.deployment( # I2V 14B model deployment + ray_actor_options={"num_cpus": 10, "num_gpus": 1, "runtime_env": {"conda": "fv"}}, +) +class I2VModelDeployment: + """Serve deployment wrapping the 14 B image-to-video model.""" + + def __init__(self, i2v_model_path: str, output_path: str = "outputs"): + self.model_path = i2v_model_path + self.output_path = output_path + + os.makedirs(self.output_path, exist_ok=True) + time.sleep(10) + + os.environ["FASTVIDEO_ATTENTION_BACKEND"] = "VIDEO_SPARSE_ATTN" + + from fastvideo.entrypoints.video_generator import VideoGenerator + from fastvideo.configs.sample.base import SamplingParam + + print(f"Initializing I2V model: {self.model_path}") + self.generator = VideoGenerator.from_pretrained( + model_path=self.model_path, + num_gpus=1, + use_fsdp_inference=True, + text_encoder_cpu_offload=False, + dit_cpu_offload=False, + vae_cpu_offload=False, + VSA_sparsity=0.8, + ) + self.default_params = SamplingParam.from_pretrained(self.model_path) + print("✅ I2V model initialized successfully") + + def generate_video(self, video_request: "VideoGenerationRequest") -> "VideoGenerationResponse": + import time + + total_start_time = time.time() + + params = deepcopy(self.default_params) + + params.prompt = video_request.prompt + if video_request.use_negative_prompt: + params.negative_prompt = video_request.negative_prompt + + params.seed = video_request.seed if not video_request.randomize_seed else torch.randint(0, 1_000_000, (1,)).item() + params.guidance_scale = video_request.guidance_scale + params.num_frames = video_request.num_frames + params.height = video_request.height + params.width = video_request.width + params.num_inference_steps = video_request.num_inference_steps + + if video_request.image_path: + params.image_path = video_request.image_path + + params.save_video = False + params.return_frames = True + + # Track inference time + inference_start_time = time.time() + + result = self.generator.generate_video( + prompt=video_request.prompt, + sampling_param=params, + save_video=False, + return_frames=True, + ) + + inference_end_time = time.time() + inference_time = inference_end_time - inference_start_time + + frames = result if isinstance(result, list) else result.get("frames", []) + generation_time = result.get("generation_time", 0.0) if isinstance(result, dict) else 0.0 + + # Track encoding time + encoding_start_time = time.time() + + video_data = encode_video_to_base64(frames, fps=24) + encoded_frames = encode_frames_to_base64(frames) if video_request.return_frames and frames else None + + encoding_end_time = time.time() + encoding_time = encoding_end_time - encoding_start_time + + total_end_time = time.time() + total_time = total_end_time - total_start_time + + return VideoGenerationResponse( + video_data=video_data, + frames=encoded_frames, + seed=params.seed, + success=True, + generation_time=generation_time, + inference_time=inference_time, + encoding_time=encoding_time, + total_time=total_time, + ) + + +@serve.deployment( # T2V 14B model deployment with optimized settings + ray_actor_options={"num_cpus": 16, "num_gpus": 1, "runtime_env": {"conda": "fv"}}, +) +class T2V14BModelDeployment: + """Serve deployment wrapping the 14B text-to-video model with optimized settings.""" + + def __init__(self, t2v_14b_model_path: str, output_path: str = "outputs"): + self.model_path = t2v_14b_model_path + self.output_path = output_path + + # Ensure output directory exists + os.makedirs(self.output_path, exist_ok=True) + + # Delay helps avoid GPU contention when many replicas start at once + time.sleep(10) + + # Ensure correct attention backend for FastVideo + os.environ["FASTVIDEO_ATTENTION_BACKEND"] = "VIDEO_SPARSE_ATTN" + + # Lazy import to keep the deployment import-safe on head node + from fastvideo.entrypoints.video_generator import VideoGenerator + from fastvideo.configs.sample.base import SamplingParam + + print(f"Initializing T2V 14B model: {self.model_path}") + self.generator = VideoGenerator.from_pretrained( + model_path=self.model_path, + num_gpus=1, + use_fsdp_inference=True, + text_encoder_cpu_offload=True, # Enable CPU offload for 14B model + dit_cpu_offload=True, # Enable CPU offload for 14B model + vae_cpu_offload=False, + VSA_sparsity=0.9, # Higher sparsity for 14B model + ) + self.default_params = SamplingParam.from_pretrained(self.model_path) + print("✅ T2V 14B model initialized successfully") + + def generate_video(self, video_request: "VideoGenerationRequest") -> "VideoGenerationResponse": + """Generate a video for the given request and return an encoded response.""" + import time + + total_start_time = time.time() + + # Deep-copy default sampling params and override with request values + params = deepcopy(self.default_params) + params.prompt = video_request.prompt + if video_request.use_negative_prompt: + params.negative_prompt = video_request.negative_prompt + + params.seed = video_request.seed if not video_request.randomize_seed else torch.randint(0, 1_000_000, (1,)).item() + params.guidance_scale = video_request.guidance_scale + params.num_frames = video_request.num_frames + params.height = video_request.height + params.width = video_request.width + params.num_inference_steps = video_request.num_inference_steps + + # Do not write to disk when called via API + params.save_video = False + params.return_frames = True + + # Track inference time + inference_start_time = time.time() + + # Generate video frames + result = self.generator.generate_video( + prompt=video_request.prompt, + sampling_param=params, + save_video=False, + return_frames=True, + ) + + inference_end_time = time.time() + inference_time = inference_end_time - inference_start_time + + frames = result if isinstance(result, list) else result.get("frames", []) + generation_time = result.get("generation_time", 0.0) if isinstance(result, dict) else 0.0 + + # Track encoding time + encoding_start_time = time.time() + + # Encode outputs + video_data = encode_video_to_base64(frames, fps=24) + encoded_frames = encode_frames_to_base64(frames) if video_request.return_frames and frames else None + + encoding_end_time = time.time() + encoding_time = encoding_end_time - encoding_start_time + + total_end_time = time.time() + total_time = total_end_time - total_start_time + + return VideoGenerationResponse( + video_data=video_data, + frames=encoded_frames, + seed=params.seed, + success=True, + generation_time=generation_time, + inference_time=inference_time, + encoding_time=encoding_time, + total_time=total_time, + ) + + # Create FastAPI app with rate limiting app = FastAPI() @@ -99,168 +433,43 @@ app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) -@serve.deployment( - num_replicas=8, - # ray_actor_options={"num_cpus": 10, "num_gpus": 1, "runtime_env": {"conda": "fv", "working_dir": "/mnt/fast-disks/nfs/hao_lab/FastVideo"}}, - ray_actor_options={"num_cpus": 10, "num_gpus": 1, "runtime_env": {"conda": "fv"}}, -) +@serve.deployment(ray_actor_options={"num_cpus": 2}) @serve.ingress(app) class FastVideoAPI: - def __init__(self, t2v_model_path: str, i2v_model_path: str, output_path: str): - self.t2v_model_path = t2v_model_path - self.i2v_model_path = i2v_model_path - self.output_path = output_path - - # Initialize the video generators - self.t2v_generator = None # Initialize to None - self.i2v_generator = None # Initialize to None - self.t2v_default_params = None # Initialize to None - self.i2v_default_params = None # Initialize to None - - # Ensure output directory exists - os.makedirs(output_path, exist_ok=True) - time.sleep(10) - self._initialize_models() # Ensure models are initialized - - def _initialize_models(self): - # Set VSA environment variable - os.environ["FASTVIDEO_ATTENTION_BACKEND"] = "VIDEO_SPARSE_ATTN" - # os.environ["FASTVIDEO_ATTENTION_BACKEND"] = "FLASH_ATTN" - - # Import only when needed - use direct imports to avoid module-level execution - from fastvideo.entrypoints.video_generator import VideoGenerator - from fastvideo.configs.sample.base import SamplingParam - - # Initialize T2V model - if self.t2v_generator is None: - print(f"Initializing T2V model: {self.t2v_model_path}") - self.t2v_generator = VideoGenerator.from_pretrained( - model_path=self.t2v_model_path, - num_gpus=1, - use_fsdp_inference=True, - # Adjust these offload parameters if you have < 32GB of VRAM - text_encoder_cpu_offload=False, - dit_cpu_offload=False, - vae_cpu_offload=False, - VSA_sparsity=0.8, - # master_port=port, - ) - self.t2v_default_params = SamplingParam.from_pretrained(self.t2v_model_path) - print("✅ T2V model initialized successfully") - - # Initialize I2V model - # if self.i2v_generator is None: - if False: - print(f"Initializing I2V model: {self.i2v_model_path}") - self.i2v_generator = VideoGenerator.from_pretrained( - model_path=self.i2v_model_path, - num_gpus=1, - use_fsdp_inference=True, - # Adjust these offload parameters if you have < 32GB of VRAM - text_encoder_cpu_offload=False, - dit_cpu_offload=False, - vae_cpu_offload=False, - VSA_sparsity=0.8, - # master_port=port, - ) - self.i2v_default_params = SamplingParam.from_pretrained(self.i2v_model_path) - print("✅ I2V model initialized successfully") + """Ingress deployment that routes requests to either the T2V or I2V model deployment.""" + + def __init__(self, t2v_deployment: DeploymentHandle, t2v_14b_deployment: DeploymentHandle, i2v_deployment: DeploymentHandle): + self.t2v_handle = t2v_deployment + self.t2v_14b_handle = t2v_14b_deployment + self.i2v_handle = i2v_deployment @app.post("/generate_video", response_model=VideoGenerationResponse) - @limiter.limit("50/minute") # Allow 2 requests per minute per IP + @limiter.limit("50/minute") # Allow 50 requests per minute per IP async def generate_video(self, request: Request, video_request: VideoGenerationRequest) -> VideoGenerationResponse: + """Route the request to the appropriate model deployment based on `model_type` and `model_path`.""" try: - # Select the appropriate model and parameters based on model_type if video_request.model_type.lower() == "i2v": - generator = self.i2v_generator - params = deepcopy(self.i2v_default_params) - print(f"Using I2V model for generation") + response_ref = self.i2v_handle.generate_video.remote(video_request) + elif video_request.model_type.lower() == "t2v": + # Route T2V requests based on model path + if video_request.model_path and "14b" in video_request.model_path.lower(): + response_ref = self.t2v_14b_handle.generate_video.remote(video_request) + else: + # Default to 1.3B model + response_ref = self.t2v_handle.generate_video.remote(video_request) else: - generator = self.t2v_generator - params = deepcopy(self.t2v_default_params) - print(f"Using T2V model for generation") - - # Update parameters with request values - params.prompt = video_request.prompt - # Only override negative prompt if user explicitly opts in - if video_request.use_negative_prompt: - params.negative_prompt = video_request.negative_prompt + # Default to 1.3B T2V model + response_ref = self.t2v_handle.generate_video.remote(video_request) - params.seed = video_request.seed - params.guidance_scale = video_request.guidance_scale - params.num_frames = video_request.num_frames - params.height = video_request.height - params.width = video_request.width - params.num_inference_steps = video_request.num_inference_steps - - # Handle seed randomization - if video_request.randomize_seed: - params.seed = torch.randint(0, 1000000, (1,)).item() - - # Ensure negative_prompt is a non-None string; FastVideo validation disallows None - if params.negative_prompt is None: - params.negative_prompt = "" # empty string satisfies validator - - # Set up output path and video saving - params.save_video = True - params.output_path = self.output_path - # params.return_frames = False # avoid keeping frames in memory - - # Create a clean filename from the prompt - safe_prompt = video_request.prompt[:100].replace(' ', '_').replace('/', '_').replace('\\', '_') + # Await the remote response and return it directly + return await response_ref - # Store desired video name inside the SamplingParam to avoid unknown kwarg errors - setattr(params, "output_video_name", safe_prompt) - - # Handle image_path for I2V - if video_request.image_path: - params.image_path = video_request.image_path - - # Generate the video with proper output path and filename - result = generator.generate_video( - prompt=video_request.prompt, - sampling_param=params, - save_video=True, # Match the params.save_video setting - ) - - # The actual output path where the video was saved - output_path = os.path.join(self.output_path, f"{safe_prompt}.mp4") - - # Verify the file exists - if not os.path.exists(output_path): - raise FileNotFoundError(f"Video was not saved to expected location: {output_path}") - - frames = result.get("frames", []) - - # Encode frames to base64 for web transmission only if requested - encoded_frames = None - if video_request.return_frames and frames: - try: - encoded_frames = encode_frames_to_base64(frames) - except Exception as e: - print(f"Warning: Failed to encode frames: {e}") - encoded_frames = None - - response = VideoGenerationResponse( - output_path=output_path, - frames=encoded_frames, - seed=params.seed, - success=True - ) - - # Memory cleanup to avoid OOM in repeated generations - import gc - gc.collect() - if torch.cuda.is_available(): - torch.cuda.empty_cache() - - return response except Exception as e: return VideoGenerationResponse( - output_path="", + video_data=None, seed=video_request.seed, success=False, - error_message=str(e) + error_message=str(e), ) @app.get("/health") @@ -270,24 +479,36 @@ class FastVideoAPI: def start_ray_serve( - t2v_model_path: str = "Wan-AI/Wan2.2-TI2V-5B-Diffusers", - i2v_model_path: str = "Wan-AI/Wan2.2-TI2V-5B-Diffusers", + *, + t2v_model_path: str = "FastVideo/FastWan2.1-T2V-1.3B-Diffusers", + t2v_14b_model_path: str = "FastVideo/FastWan2.1-T2V-14B-Diffusers", + i2v_model_path: str = "Wan-AI/Wan2.1-I2V-14B-480P-Diffusers", output_path: str = "outputs", host: str = "0.0.0.0", - port: int = 8000 + port: int = 8000, + t2v_replicas: int = 4, + t2v_14b_replicas: int = 4, # Reduced replicas for 14B model due to higher resource requirements + i2v_replicas: int = 4, ): - """Start the Ray Serve backend""" - # Initialize Ray + """Start Ray Serve with independently scalable deployments for each model.""" + if not ray.is_initialized(): ray.init() - - # Deploy the API - api = FastVideoAPI.bind(t2v_model_path, i2v_model_path, output_path) - serve.run(api, route_prefix="/", name="fast_video") # detach - + + # Bind model deployments with configurable replica counts + t2v_dep = T2VModelDeployment.options(num_replicas=t2v_replicas).bind(t2v_model_path, output_path) + t2v_14b_dep = T2V14BModelDeployment.options(num_replicas=t2v_14b_replicas).bind(t2v_14b_model_path, output_path) + i2v_dep = I2VModelDeployment.options(num_replicas=i2v_replicas).bind(i2v_model_path, output_path) + + # Ingress + api = FastVideoAPI.bind(t2v_dep, t2v_14b_dep, i2v_dep) + + serve.run(api, route_prefix="/", name="fast_video") + print(f"Ray Serve backend started at http://{host}:{port}") - print(f"T2V Model: {t2v_model_path}") - print(f"I2V Model: {i2v_model_path}") + print(f"T2V 1.3B Model: {t2v_model_path} | Replicas: {t2v_replicas}") + print(f"T2V 14B Model: {t2v_14b_model_path} | Replicas: {t2v_14b_replicas}") + print(f"I2V Model: {i2v_model_path} | Replicas: {i2v_replicas}") print(f"Health check: http://{host}:{port}/health") print(f"Video generation endpoint: http://{host}:{port}/generate_video") @@ -300,9 +521,13 @@ if __name__ == "__main__": type=str, default="FastVideo/FastWan2.1-T2V-1.3B-Diffusers", help="Path to the T2V model") + parser.add_argument("--t2v_14b_model_path", + type=str, + default="FastVideo/FastWan2.1-T2V-14B-Diffusers", + help="Path to the T2V 14B model") parser.add_argument("--i2v_model_path", type=str, - default="Wan-AI/Wan2.2-TI2V-5B-Diffusers", + default="Wan-AI/Wan2.1-I2V-14B-480P-Diffusers", help="Path to the I2V model") parser.add_argument("--output_path", type=str, @@ -316,15 +541,31 @@ if __name__ == "__main__": type=int, default=8000, help="Port to bind to") + parser.add_argument("--t2v_replicas", + type=int, + default=4, + help="Number of replicas for the T2V model deployment") + parser.add_argument("--t2v_14b_replicas", + type=int, + default=4, # Reduced default for 14B model due to higher resource requirements + help="Number of replicas for the T2V 14B model deployment") + parser.add_argument("--i2v_replicas", + type=int, + default=4, + help="Number of replicas for the I2V model deployment") args = parser.parse_args() start_ray_serve( t2v_model_path=args.t2v_model_path, + t2v_14b_model_path=args.t2v_14b_model_path, i2v_model_path=args.i2v_model_path, output_path=args.output_path, host=args.host, port=args.port, + t2v_replicas=args.t2v_replicas, + t2v_14b_replicas=args.t2v_14b_replicas, + i2v_replicas=args.i2v_replicas, ) # ---- keep the process alive --------------------------------- diff --git a/examples/inference/gradio/start_ray_serve_app.py b/examples/inference/gradio/start_ray_serve_app.py index 7e4fe7de7..01399ee8a 100644 --- a/examples/inference/gradio/start_ray_serve_app.py +++ b/examples/inference/gradio/start_ray_serve_app.py @@ -46,6 +46,7 @@ def start_backend(args): cmd = [ sys.executable, str(backend_script), "--t2v_model_path", args.t2v_model_path, + "--t2v_14b_model_path", args.t2v_14b_model_path, "--i2v_model_path", args.i2v_model_path, "--output_path", args.output_path, "--host", args.backend_host, @@ -84,6 +85,7 @@ def start_frontend(args): sys.executable, str(frontend_script), "--backend_url", backend_url, "--t2v_model_path", args.t2v_model_path, + "--t2v_14b_model_path", args.t2v_14b_model_path, "--i2v_model_path", args.i2v_model_path, "--host", args.frontend_host, "--port", str(args.frontend_port) @@ -118,11 +120,15 @@ def main(): # Model and output settings parser.add_argument("--t2v_model_path", type=str, - default="Wan-AI/Wan2.2-TI2V-5BDiffusers", + default="FastVideo/FastWan2.1-T2V-1.3B-Diffusers", help="Path to the T2V model") + parser.add_argument("--t2v_14b_model_path", + type=str, + default="FastVideo/FastWan2.1-T2V-14B-Diffusers", + help="Path to the T2V 14B model") parser.add_argument("--i2v_model_path", type=str, - default="Wan-AI/Wan2.2-TI2V-5B-Diffusers", + default="Wan-AI/Wan2.1-I2V-14B-480P-Diffusers", help="Path to the I2V model") parser.add_argument("--output_path", type=str, @@ -146,7 +152,7 @@ def main(): help="Frontend host to bind to") parser.add_argument("--frontend_port", type=int, - default=7860, + default=7861, help="Frontend port to bind to") # Other settings @@ -161,7 +167,8 @@ def main(): print("🎬 FastVideo Ray Serve App") print("=" * 50) - print(f"T2V Model: {args.t2v_model_path}") + print(f"T2V 1.3B Model: {args.t2v_model_path}") + print(f"T2V 14B Model: {args.t2v_14b_model_path}") print(f"I2V Model: {args.i2v_model_path}") print(f"Output: {args.output_path}") print(f"Backend: http://{args.backend_host}:{args.backend_port}")