Update app.py

app.py

@@ -1,125 +1,144 @@
-import os
-from collections.abc import Iterator
-from threading import Thread
 import gradio as gr
 import spaces
+from transformers import Qwen2_5_VLForConditionalGeneration, AutoProcessor, TextIteratorStreamer
+from qwen_vl_utils import process_vision_info
 import torch
-from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer
+from PIL import Image
+import subprocess
+import numpy as np
+import os
+from threading import Thread
+import uuid
+import io
 
-DESCRIPTION = """
-# GWQ PREV
-"""
+# Model and Processor Loading (Done once at startup)
+MODEL_ID = "Qwen/Qwen2.5-VL-3B-Instruct"
+model = Qwen2_5_VLForConditionalGeneration.from_pretrained(
+    MODEL_ID,
+    trust_remote_code=True,
+    torch_dtype=torch.float16
+).to("cuda").eval()
+processor = AutoProcessor.from_pretrained(MODEL_ID, trust_remote_code=True)
+
+DESCRIPTION = "# **Qwen2.5-VL-3B-Instruct**"
+
+image_extensions = Image.registered_extensions()
+video_extensions = ("avi", "mp4", "mov", "mkv", "flv", "wmv", "mjpeg", "wav", "gif", "webm", "m4v", "3gp")
+
+
+def identify_and_save_blob(blob_path):
+    """Identifies if the blob is an image or video and saves it accordingly."""
+    try:
+        with open(blob_path, 'rb') as file:
+            blob_content = file.read()
+            
+            # Try to identify if it's an image
+            try:
+                Image.open(io.BytesIO(blob_content)).verify()  # Check if it's a valid image
+                extension = ".png"  # Default to PNG for saving
+                media_type = "image"
+            except (IOError, SyntaxError):
+                # If it's not a valid image, assume it's a video
+                extension = ".mp4"  # Default to MP4 for saving
+                media_type = "video"
+            
+            # Create a unique filename
+            filename = f"temp_{uuid.uuid4()}_media{extension}"
+            with open(filename, "wb") as f:
+                f.write(blob_content)
+                
+            return filename, media_type
+            
+    except FileNotFoundError:
+        raise ValueError(f"The file {blob_path} was not found.")
+    except Exception as e:
+        raise ValueError(f"An error occurred while processing the file: {e}")
+
+
+@spaces.GPU
+def qwen_inference(media_input, text_input=None):
+    if isinstance(media_input, str):  # If it's a filepath
+        media_path = media_input
+        if media_path.endswith(tuple([i for i, f in image_extensions.items()])):
+            media_type = "image"
+        elif media_path.endswith(video_extensions): 
+            media_type = "video"
+        else:
+            try:
+                media_path, media_type = identify_and_save_blob(media_input)
+                print(media_path, media_type)
+            except Exception as e:
+                print(e)
+                raise ValueError(
+                    "Unsupported media type. Please upload an image or video."
+                )
+        
+
+    print(media_path)
 
-MAX_MAX_NEW_TOKENS = 2048
-DEFAULT_MAX_NEW_TOKENS = 1024
-MAX_INPUT_TOKEN_LENGTH = int(os.getenv("MAX_INPUT_TOKEN_LENGTH", "4096"))
-
-device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
-
-model_id = "prithivMLmods/GWQ2b"
-tokenizer = AutoTokenizer.from_pretrained(model_id)
-model = AutoModelForCausalLM.from_pretrained(
-    model_id,
-    device_map="auto",
-    torch_dtype=torch.bfloat16,
-)
-model.config.sliding_window = 4096
-model.eval()
-
-
-@spaces.GPU(duration=120)
-def generate(
-    message: str,
-    chat_history: list[dict],
-    max_new_tokens: int = 1024,
-    temperature: float = 0.6,
-    top_p: float = 0.9,
-    top_k: int = 50,
-    repetition_penalty: float = 1.2,
-) -> Iterator[str]:
-    conversation = chat_history.copy()
-    conversation.append({"role": "user", "content": message})
-
-    input_ids = tokenizer.apply_chat_template(conversation, add_generation_prompt=True, return_tensors="pt")
-    if input_ids.shape[1] > MAX_INPUT_TOKEN_LENGTH:
-        input_ids = input_ids[:, -MAX_INPUT_TOKEN_LENGTH:]
-        gr.Warning(f"Trimmed input from conversation as it was longer than {MAX_INPUT_TOKEN_LENGTH} tokens.")
-    input_ids = input_ids.to(model.device)
-
-    streamer = TextIteratorStreamer(tokenizer, timeout=20.0, skip_prompt=True, skip_special_tokens=True)
-    generate_kwargs = dict(
-        {"input_ids": input_ids},
-        streamer=streamer,
-        max_new_tokens=max_new_tokens,
-        do_sample=True,
-        top_p=top_p,
-        top_k=top_k,
-        temperature=temperature,
-        num_beams=1,
-        repetition_penalty=repetition_penalty,
+    messages = [
+        {
+            "role": "user",
+            "content": [
+                {
+                    "type": media_type,
+                    media_type: media_path,
+                    **({"fps": 8.0} if media_type == "video" else {}),
+                },
+                {"type": "text", "text": text_input},
+            ],
+        }
+    ]
+
+    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",
+    ).to("cuda")
+
+    streamer = TextIteratorStreamer(
+        processor, skip_prompt=True, **{"skip_special_tokens": True}
     )
-    t = Thread(target=model.generate, kwargs=generate_kwargs)
-    t.start()
-
-    outputs = []
-    for text in streamer:
-        outputs.append(text)
-        yield "".join(outputs)
-
-
-demo = gr.ChatInterface(
-    fn=generate,
-    additional_inputs=[
-        gr.Slider(
-            label="Max new tokens",
-            minimum=1,
-            maximum=MAX_MAX_NEW_TOKENS,
-            step=1,
-            value=DEFAULT_MAX_NEW_TOKENS,
-        ),
-        gr.Slider(
-            label="Temperature",
-            minimum=0.1,
-            maximum=4.0,
-            step=0.1,
-            value=0.6,
-        ),
-        gr.Slider(
-            label="Top-p (nucleus sampling)",
-            minimum=0.05,
-            maximum=1.0,
-            step=0.05,
-            value=0.9,
-        ),
-        gr.Slider(
-            label="Top-k",
-            minimum=1,
-            maximum=1000,
-            step=1,
-            value=50,
-        ),
-        gr.Slider(
-            label="Repetition penalty",
-            minimum=1.0,
-            maximum=2.0,
-            step=0.05,
-            value=1.2,
-        ),
-    ],
-    stop_btn=None,
-    examples=[
-        ["Write a Python function to reverses a string if it's length is a multiple of 4. def reverse_string(str1): if len(str1) % 4 == 0: return ''.join(reversed(str1)) return str1 print(reverse_string('abcd')) print(reverse_string('python')) "],
-        ["Rectangle $ABCD$ is the base of pyramid $PABCD$. If $AB = 10$, $BC = 5$, $\overline{PA}\perp \text{plane } ABCD$, and $PA = 8$, then what is the volume of $PABCD$?"],
-        ["Difference between List comprehension and Lambda in Python lst  =  [x ** 2  for x in range (1, 11)   if  x % 2 == 1] print(lst)"],
-        ["What happens when the sun goes down?"],
-    ],
-    cache_examples=False,
-    type="messages",
-    description=DESCRIPTION,
-    css_paths="style.css",
-    fill_height=True,
-)
-
-
-if __name__ == "__main__":
-    demo.queue(max_size=20).launch()
+    generation_kwargs = dict(inputs, streamer=streamer, max_new_tokens=1024)
+
+    thread = Thread(target=model.generate, kwargs=generation_kwargs)
+    thread.start()
+
+    buffer = ""
+    for new_text in streamer:
+        buffer += new_text
+        yield buffer
+
+css = """
+  #output {
+    height: 500px; 
+    overflow: auto; 
+    border: 1px solid #ccc; 
+  }
+"""
+
+with gr.Blocks(css=css) as demo:
+    gr.Markdown(DESCRIPTION)
+
+    with gr.Tab(label="Image/Video Input"):
+        with gr.Row():
+            with gr.Column():
+                input_media = gr.File(
+                    label="Upload Image or Video", type="filepath" 
+                )
+                text_input = gr.Textbox(label="Question")
+                submit_btn = gr.Button(value="Submit")
+            with gr.Column():
+                output_text = gr.Textbox(label="Output Text")
+
+        submit_btn.click(
+            qwen_inference, [input_media, text_input], [output_text]
+        )
+
+demo.launch(debug=True)