Update app.py

app.py

@@ -9,158 +9,61 @@ from datetime import datetime
 import numpy as np
 import os
 
+# Install flash-attn
+subprocess.run('pip install flash-attn --no-build-isolation', env={'FLASH_ATTENTION_SKIP_CUDA_BUILD': "TRUE"}, shell=True)
 
-# subprocess.run('pip install flash-attn --no-build-isolation', env={'FLASH_ATTENTION_SKIP_CUDA_BUILD': "TRUE"}, shell=True)
-
-# models = {
-#     "Qwen/Qwen2-VL-7B-Instruct": AutoModelForCausalLM.from_pretrained("Qwen/Qwen2-VL-7B-Instruct", trust_remote_code=True, torch_dtype="auto", _attn_implementation="flash_attention_2").cuda().eval()
-
-# }
-def array_to_image_path(image):
-    if image is None:
-        gr.Warning("No video provided. Please upload an video before submitting.")
-        raise ValueError("No image provided. Please upload an image before submitting.")
-    
-    # Generate a unique filename using timestamp
-    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-    filename = f"image_{timestamp}.png"
-    
-    # Save the image
-    image.save(filename)
-    
-    # Get the full path of the saved image
-    full_path = os.path.abspath(filename)
-    
-    return full_path
-
-def array_to_video_path(video):
-    if video is None:
-        gr.Warning("No video provided. Please upload an video before submitting.")
-        raise ValueError("No video provided. Please upload an video before submitting.")
-    
-    # Generate a unique filename using timestamp
-    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-    filename = f"video_{timestamp}.mp4"
-    
-    # Save the image
-    video.save(filename)
-    
-    # Get the full path of the saved image
-    full_path = os.path.abspath(filename)
-    
-    return full_path
-    
-models = {
-    "Qwen/Qwen2-VL-7B-Instruct": Qwen2VLForConditionalGeneration.from_pretrained("Qwen/Qwen2-VL-7B-Instruct", trust_remote_code=True, torch_dtype="auto").cuda().eval()
-
-}
-
-processors = {
-    "Qwen/Qwen2-VL-7B-Instruct": AutoProcessor.from_pretrained("Qwen/Qwen2-VL-7B-Instruct", trust_remote_code=True)
-}
+# Model and Processor Loading (Done once at startup)
+MODEL_ID = "Qwen/Qwen2-VL-7B-Instruct"
+model = Qwen2VLForConditionalGeneration.from_pretrained(MODEL_ID, trust_remote_code=True, attn_implementation="flash_attention_2", torch_dtype="auto").cuda().eval()
+processor = AutoProcessor.from_pretrained(MODEL_ID, trust_remote_code=True)
 
 DESCRIPTION = "[Qwen2-VL-7B Demo](https://huggingface.co/Qwen/Qwen2-VL-7B-Instruct)"
 
-kwargs = {}
-kwargs['torch_dtype'] = torch.bfloat16
+# Helper function to save media and return path 
+def save_media_and_get_path(media, media_type):
+    if media is None:
+        gr.Warning(f"No {media_type} provided. Please upload a {media_type} before submitting.")
+        raise ValueError(f"No {media_type} provided.")
 
-user_prompt = '<|user|>\n'
-assistant_prompt = '<|assistant|>\n'
-prompt_suffix = "<|end|>\n"
+    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+    filename = f"{media_type}_{timestamp}.{'png' if media_type == 'image' else 'mp4'}" 
+    media.save(filename)
+    return os.path.abspath(filename)
 
 @spaces.GPU
-def qwen_image(image, text_input=None, model_id="Qwen/Qwen2-VL-7B-Instruct"):
-    image_path = array_to_image_path(image)
+def qwen_inference(media, media_type, text_input=None):
+    media_path = save_media_and_get_path(media, media_type)
     
-    print(image_path)
-    model = models[model_id]
-    processor = processors[model_id]
     messages = [
-    {
+        {
             "role": "user",
             "content": [
                 {
-                    "type": "image",
-                    "image": image_path,
+                    "type": media_type,
+                    media_type: media_path,
+                    **({"max_pixels": 360 * 420, "fps": 6.0} if media_type == "video" else {}), 
                 },
                 {"type": "text", "text": text_input},
             ],
         }
     ]
-    
-    # Preparation for inference
-    text = processor.apply_chat_template(
-        messages, tokenize=False, add_generation_prompt=True
-    )
-    image_inputs, video_inputs = process_vision_info(messages)
-    inputs = processor(
-        text=[text],
-        images=image_inputs,
-        videos=video_inputs,
-        padding=True,
-        return_tensors="pt",
-    )
-    inputs = inputs.to("cuda")
-    
-    # Inference: Generation of the output
-    generated_ids = model.generate(**inputs, max_new_tokens=1024)
-    generated_ids_trimmed = [
-        out_ids[len(in_ids) :] for in_ids, out_ids in zip(inputs.input_ids, generated_ids)
-    ]
-    output_text = processor.batch_decode(
-        generated_ids_trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False
-    )
-    
-    return output_text[0]
 
-@spaces.GPU(duration=125)
-def qwen_video(video, text_input=None, model_id="Qwen/Qwen2-VL-7B-Instruct"):
-    video_path = array_to_video_path(video)
-    
-    print(video_path)
-    model = models[model_id]
-    processor = processors[model_id]
-    
-    messages = [
-    {
-            "role": "user",
-            "content": [
-                {
-                    "type": "video",
-                    "video": video_path,
-                    "max_pixels": 360 * 420,
-                    "fps": 6.0,
-                },
-                {"type": "text", "text": text_input},
-            ],
-        }
-    ]
-    
-    # Preparation for inference
-    text = processor.apply_chat_template(
-        messages, tokenize=False, add_generation_prompt=True
-    )
-    image_inputs, video_inputs = process_vision_info(messages)
+    text = processor.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
+    image_inputs, video_inputs = process_vision_info(messages) 
     inputs = processor(
         text=[text],
         images=image_inputs,
         videos=video_inputs,
         padding=True,
         return_tensors="pt",
-    )
-    inputs = inputs.to("cuda")
-    
-    # Inference: Generation of the output
+    ).to("cuda")
+
     generated_ids = model.generate(**inputs, max_new_tokens=1024)
-    generated_ids_trimmed = [
-        out_ids[len(in_ids) :] for in_ids, out_ids in zip(inputs.input_ids, generated_ids)
-    ]
-    output_text = processor.batch_decode(
-        generated_ids_trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False
-    )
-    
-    return output_text[0]
+    generated_ids_trimmed = [out_ids[len(in_ids):] for in_ids, out_ids in zip(inputs.input_ids, generated_ids)]
+    output_text = processor.batch_decode(generated_ids_trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False)[0]
     
+    return output_text
+
 css = """
   #output {
     height: 500px; 
@@ -171,28 +74,28 @@ css = """
 
 with gr.Blocks(css=css) as demo:
     gr.Markdown(DESCRIPTION)
-    with gr.Tab(label="Qwen2-VL-7B Input"):
+
+    with gr.Tab(label="Image Input"):
         with gr.Row():
             with gr.Column():
                 input_img = gr.Image(label="Input Picture", type="pil")
-                model_selector = gr.Dropdown(choices=list(models.keys()), label="Model", value="Qwen/Qwen2-VL-7B-Instruct")
-                text_input = gr.Textbox(label="Question")
-                submit_btn = gr.Button(value="Submit")
+                text_input_image = gr.Textbox(label="Question")
+                submit_btn_image = gr.Button(value="Submit")
             with gr.Column():
-                output_text = gr.Textbox(label="Output Text")
+                output_text_image = gr.Textbox(label="Output Text")
+
+        submit_btn_image.click(qwen_inference, [input_img, "image", text_input_image], [output_text_image])
 
-        submit_btn.click(qwen_image, [input_img, text_input, model_selector], [output_text])
-    with gr.Tab(label="Qwen2-VL-7B Input"):
+    with gr.Tab(label="Video Input"):
         with gr.Row():
             with gr.Column():
-                input_img = gr.Video(label="Input Picture")
-                model_selector = gr.Dropdown(choices=list(models.keys()), label="Model", value="Qwen/Qwen2-VL-7B-Instruct")
-                text_input = gr.Textbox(label="Question")
-                submit_btn = gr.Button(value="Submit")
+                input_video = gr.Video(label="Input Video")
+                text_input_video = gr.Textbox(label="Question")
+                submit_btn_video = gr.Button(value="Submit")
             with gr.Column():
-                output_text = gr.Textbox(label="Output Text")
+                output_text_video = gr.Textbox(label="Output Text")
 
-        submit_btn.click(qwen_video, [input_img, text_input, model_selector], [output_text])
+        submit_btn_video.click(qwen_inference, [input_video, "video", text_input_video], [output_text_video])
 
 demo.queue(api_open=False)
 demo.launch(debug=True)