arvisioncode/Qwen2.5-VL-3B-Instruct-T4small
0
1import gradio as gr2from transformers import AutoProcessor, Qwen2_5_VLForConditionalGeneration, TextIteratorStreamer3from transformers.image_utils import load_image4from threading import Thread5import time6import torch7import spaces8import cv29import numpy as np10from PIL import Image11 12def progress_bar_html(label: str) -> str:13 """14 Returns an HTML snippet for a thin progress bar with a label.15 The progress bar is styled as a dark animated bar.16 """17 return f'''18<div style="display: flex; align-items: center;">19 <span style="margin-right: 10px; font-size: 14px;">{label}</span>20 <div style="width: 110px; height: 5px; background-color: #9370DB; border-radius: 2px; overflow: hidden;">21 <div style="width: 100%; height: 100%; background-color: #4B0082; animation: loading 1.5s linear infinite;"></div>22 </div>23</div>24<style>25@keyframes loading {{26 0% {{ transform: translateX(-100%); }}27 100% {{ transform: translateX(100%); }}28}}29</style>30 '''31 32def downsample_video(video_path):33 """34 Downsamples the video to 10 evenly spaced frames.35 Each frame is converted to a PIL Image along with its timestamp.36 """37 vidcap = cv2.VideoCapture(video_path)38 total_frames = int(vidcap.get(cv2.CAP_PROP_FRAME_COUNT))39 fps = vidcap.get(cv2.CAP_PROP_FPS)40 frames = []41 if total_frames <= 0 or fps <= 0:42 vidcap.release()43 return frames44 # Sample 10 evenly spaced frames.45 frame_indices = np.linspace(0, total_frames - 1, 10, dtype=int)46 for i in frame_indices:47 vidcap.set(cv2.CAP_PROP_POS_FRAMES, i)48 success, image = vidcap.read()49 if success:50 image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)51 pil_image = Image.fromarray(image)52 timestamp = round(i / fps, 2)53 frames.append((pil_image, timestamp))54 vidcap.release()55 return frames56 57MODEL_ID = "Qwen/Qwen2.5-VL-3B-Instruct" # Alternatively: "Qwen/Qwen2.5-VL-3B-Instruct" 58processor = AutoProcessor.from_pretrained(MODEL_ID, trust_remote_code=True)59model = Qwen2_5_VLForConditionalGeneration.from_pretrained(60 MODEL_ID,61 trust_remote_code=True,62 torch_dtype=torch.bfloat1663).to("cuda").eval()64 65@spaces.GPU66def model_inference(input_dict, history):67 text = input_dict["text"]68 files = input_dict["files"]69 70 if text.strip().lower().startswith("@video-infer"):71 # Remove the tag from the query.72 text = text[len("@video-infer"):].strip()73 if not files:74 gr.Error("Please upload a video file along with your @video-infer query.")75 return76 # Assume the first file is a video.77 video_path = files[0]78 frames = downsample_video(video_path)79 if not frames:80 gr.Error("Could not process video.")81 return82 # Build messages: start with the text prompt.83 messages = [84 {85 "role": "user",86 "content": [{"type": "text", "text": text}]87 }88 ]89 # Append each frame with a timestamp label.90 for image, timestamp in frames:91 messages[0]["content"].append({"type": "text", "text": f"Frame {timestamp}:"})92 messages[0]["content"].append({"type": "image", "image": image})93 # Collect only the images from the frames.94 video_images = [image for image, _ in frames]95 # Prepare the prompt.96 prompt = processor.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)97 inputs = processor(98 text=[prompt],99 images=video_images,100 return_tensors="pt",101 padding=True,102 ).to("cuda")103 # Set up streaming generation.104 streamer = TextIteratorStreamer(processor, skip_prompt=True, skip_special_tokens=True)105 generation_kwargs = dict(inputs, streamer=streamer, max_new_tokens=1024)106 thread = Thread(target=model.generate, kwargs=generation_kwargs)107 thread.start()108 buffer = ""109 yield progress_bar_html("Processing video with Qwen2.5VL Model")110 for new_text in streamer:111 buffer += new_text112 time.sleep(0.01)113 yield buffer114 return115 116 if len(files) > 1:117 images = [load_image(image) for image in files]118 elif len(files) == 1:119 images = [load_image(files[0])]120 else:121 images = []122 123 if text == "" and not images:124 gr.Error("Please input a query and optionally image(s).")125 return126 if text == "" and images:127 gr.Error("Please input a text query along with the image(s).")128 return129 130 messages = [131 {132 "role": "user",133 "content": [134 *[{"type": "image", "image": image} for image in images],135 {"type": "text", "text": text},136 ],137 }138 ]139 prompt = processor.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)140 inputs = processor(141 text=[prompt],142 images=images if images else None,143 return_tensors="pt",144 padding=True,145 ).to("cuda")146 streamer = TextIteratorStreamer(processor, skip_prompt=True, skip_special_tokens=True)147 generation_kwargs = dict(inputs, streamer=streamer, max_new_tokens=1024)148 thread = Thread(target=model.generate, kwargs=generation_kwargs)149 thread.start()150 buffer = ""151 yield progress_bar_html("Processing with Qwen2.5VL Model")152 for new_text in streamer:153 buffer += new_text154 time.sleep(0.01)155 yield buffer156 157examples = [158 [{"text": "Describe the Image?", "files": ["example_images/document.jpg"]}],159 [{"text": "@video-infer Explain the content of the Advertisement", "files": ["example_images/videoplayback.mp4"]}],160 [{"text": "@video-infer Explain the content of the video in detail", "files": ["example_images/breakfast.mp4"]}],161 [{"text": "@video-infer Explain the content of the video.", "files": ["example_images/sky.mp4"]}],162]163 164demo = gr.ChatInterface(165 fn=model_inference,166 description="# **Qwen2.5-VL-7B-Instruct `@video-infer for video understanding`**",167 examples=examples,168 fill_height=True,169 textbox=gr.MultimodalTextbox(label="Query Input", file_types=["image", "video"], file_count="multiple"),170 stop_btn="Stop Generation",171 multimodal=True,172 cache_examples=False,173)174 175demo.launch(debug=True)