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

checkpoint

Browse Source
This commit is contained in:
SolitaryThinker
2025-08-02 01:22:21 +00:00
parent 7b3c4fbdd1
commit 5d7beeb5a0
4 changed files with 663 additions and 327 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/inference/basic/basic.py
+5 -1
View File
@@ -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
examples/inference/gradio/gradio_frontend.py
+244 -160
View File
@@ -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(
examples/inference/gradio/ray_serve_backend.py
+403 -162
View File
@@ -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 ---------------------------------
examples/inference/gradio/start_ray_serve_app.py
+11 -4
View File
@@ -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}")