Spaces:
Sleeping
Sleeping
| import gradio as gr | |
| import torch | |
| from PIL import Image | |
| from transformers import AutoModel, AutoTokenizer | |
| import numpy as np | |
| import tempfile | |
| import os | |
| from decord import VideoReader, cpu | |
| from scipy.spatial import cKDTree | |
| import math | |
| import warnings | |
| warnings.filterwarnings("ignore") | |
| # Global variables for model and tokenizer | |
| model = None | |
| tokenizer = None | |
| def load_model(): | |
| """Load the MiniCPM-V-4.5 model and tokenizer""" | |
| global model, tokenizer | |
| if model is None: | |
| print("Loading MiniCPM-V-4.5 model...") | |
| model = AutoModel.from_pretrained( | |
| 'openbmb/MiniCPM-V-4_5', | |
| trust_remote_code=True, | |
| attn_implementation='sdpa', | |
| torch_dtype=torch.bfloat16, | |
| device_map="auto" | |
| ) | |
| model = model.eval() | |
| tokenizer = AutoTokenizer.from_pretrained( | |
| 'openbmb/MiniCPM-V-4_5', | |
| trust_remote_code=True | |
| ) | |
| print("Model loaded successfully!") | |
| return model, tokenizer | |
| def map_to_nearest_scale(values, scale): | |
| """Map values to nearest scale for temporal IDs""" | |
| tree = cKDTree(np.asarray(scale)[:, None]) | |
| _, indices = tree.query(np.asarray(values)[:, None]) | |
| return np.asarray(scale)[indices] | |
| def group_array(arr, size): | |
| """Group array into chunks of specified size""" | |
| return [arr[i:i+size] for i in range(0, len(arr), size)] | |
| def uniform_sample(l, n): | |
| """Uniformly sample n items from list l""" | |
| gap = len(l) / n | |
| idxs = [int(i * gap + gap / 2) for i in range(n)] | |
| return [l[i] for i in idxs] | |
| def encode_video(video_path, choose_fps=3, max_frames=180, max_packing=3, time_scale=0.1): | |
| """Encode video frames with temporal IDs for the model""" | |
| vr = VideoReader(video_path, ctx=cpu(0)) | |
| fps = vr.get_avg_fps() | |
| video_duration = len(vr) / fps | |
| if choose_fps * int(video_duration) <= max_frames: | |
| packing_nums = 1 | |
| choose_frames = round(min(choose_fps, round(fps)) * min(max_frames, video_duration)) | |
| else: | |
| packing_nums = math.ceil(video_duration * choose_fps / max_frames) | |
| if packing_nums <= max_packing: | |
| choose_frames = round(video_duration * choose_fps) | |
| else: | |
| choose_frames = round(max_frames * max_packing) | |
| packing_nums = max_packing | |
| frame_idx = [i for i in range(0, len(vr))] | |
| frame_idx = np.array(uniform_sample(frame_idx, choose_frames)) | |
| print(f'Video duration: {video_duration:.2f}s, frames: {len(frame_idx)}, packing: {packing_nums}') | |
| frames = vr.get_batch(frame_idx).asnumpy() | |
| frame_idx_ts = frame_idx / fps | |
| scale = np.arange(0, video_duration, time_scale) | |
| frame_ts_id = map_to_nearest_scale(frame_idx_ts, scale) / time_scale | |
| frame_ts_id = frame_ts_id.astype(np.int32) | |
| frames = [Image.fromarray(v.astype('uint8')).convert('RGB') for v in frames] | |
| frame_ts_id_group = group_array(frame_ts_id, packing_nums) | |
| return frames, frame_ts_id_group | |
| def process_input( | |
| file_input, | |
| user_prompt, | |
| system_prompt, | |
| fps, | |
| context_size, | |
| temperature, | |
| enable_thinking | |
| ): | |
| """Process user input and generate response""" | |
| try: | |
| # Load model if not already loaded | |
| model, tokenizer = load_model() | |
| if file_input is None: | |
| return "Please upload an image or video file." | |
| # Determine if input is image or video | |
| file_path = file_input.name if hasattr(file_input, 'name') else file_input | |
| file_ext = os.path.splitext(file_path)[1].lower() | |
| is_video = file_ext in ['.mp4', '.avi', '.mov', '.mkv', '.webm', '.m4v'] | |
| # Prepare messages | |
| msgs = [] | |
| # Add system prompt if provided | |
| if system_prompt and system_prompt.strip(): | |
| msgs.append({'role': 'system', 'content': system_prompt.strip()}) | |
| if is_video: | |
| # Process video | |
| frames, frame_ts_id_group = encode_video(file_path, choose_fps=fps) | |
| msgs.append({'role': 'user', 'content': frames + [user_prompt]}) | |
| # Generate response for video | |
| answer = model.chat( | |
| msgs=msgs, | |
| tokenizer=tokenizer, | |
| use_image_id=False, | |
| max_slice_nums=1, | |
| temporal_ids=frame_ts_id_group, | |
| enable_thinking=enable_thinking, | |
| max_new_tokens=context_size, | |
| temperature=temperature | |
| ) | |
| else: | |
| # Process image | |
| image = Image.open(file_path).convert('RGB') | |
| msgs.append({'role': 'user', 'content': [image, user_prompt]}) | |
| # Generate response for image | |
| answer = model.chat( | |
| msgs=msgs, | |
| tokenizer=tokenizer, | |
| enable_thinking=enable_thinking, | |
| max_new_tokens=context_size, | |
| temperature=temperature | |
| ) | |
| return answer | |
| except Exception as e: | |
| return f"Error processing input: {str(e)}" | |
| def create_interface(): | |
| """Create and configure Gradio interface""" | |
| with gr.Blocks(title="MiniCPM-V-4.5 Multimodal Chat", theme=gr.themes.Soft()) as iface: | |
| gr.Markdown(""" | |
| # 🚀 MiniCPM-V-4.5 Multimodal Chat | |
| A powerful 8B parameter multimodal model that can understand images and videos with GPT-4V level performance. | |
| **Features:** | |
| - 📸 Single/Multi-image understanding | |
| - 🎥 High refresh rate video understanding (up to 10 FPS) | |
| - 📄 Strong OCR and document parsing | |
| - 🧠 Controllable fast/deep thinking mode | |
| - 🌍 Multilingual support (30+ languages) | |
| """) | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| # File input | |
| file_input = gr.File( | |
| label="Upload Image or Video", | |
| file_types=["image", "video"], | |
| type="filepath" | |
| ) | |
| # Video FPS setting | |
| fps_slider = gr.Slider( | |
| minimum=1, | |
| maximum=30, | |
| value=5, | |
| step=1, | |
| label="Video FPS", | |
| info="Frames per second for video processing (only applies to videos)" | |
| ) | |
| # Context size | |
| context_size = gr.Slider( | |
| minimum=512, | |
| maximum=4096, | |
| value=2048, | |
| step=256, | |
| label="Max Output Tokens", | |
| info="Maximum number of tokens to generate" | |
| ) | |
| # Temperature | |
| temperature = gr.Slider( | |
| minimum=0.1, | |
| maximum=2.0, | |
| value=0.7, | |
| step=0.1, | |
| label="Temperature", | |
| info="Controls randomness in generation" | |
| ) | |
| # Thinking mode | |
| enable_thinking = gr.Checkbox( | |
| label="Enable Deep Thinking", | |
| value=False, | |
| info="Enable deep thinking mode for complex problem solving" | |
| ) | |
| with gr.Column(scale=2): | |
| # System prompt | |
| system_prompt = gr.Textbox( | |
| label="System Prompt (Optional)", | |
| placeholder="Enter system instructions here...", | |
| lines=3, | |
| info="Set the behavior and context for the model" | |
| ) | |
| # User prompt | |
| user_prompt = gr.Textbox( | |
| label="Your Question", | |
| placeholder="Describe what you see in the image/video, or ask a specific question...", | |
| lines=4 | |
| ) | |
| # Submit button | |
| submit_btn = gr.Button("🚀 Generate Response", variant="primary", size="lg") | |
| # Output | |
| output = gr.Textbox( | |
| label="Model Response", | |
| lines=15, | |
| max_lines=25, | |
| show_copy_button=True | |
| ) | |
| # Examples | |
| gr.Markdown("## 💡 Example Prompts") | |
| gr.Examples( | |
| examples=[ | |
| ["What objects do you see in this image?"], | |
| ["Describe the scene in detail."], | |
| ["What is the main action happening in this video?"], | |
| ["Read and transcribe any text visible in the image."], | |
| ["What emotions or mood does this image convey?"], | |
| ["Analyze the composition and visual elements."], | |
| ["What might happen next in this sequence?"] | |
| ], | |
| inputs=[user_prompt], | |
| label="Click any example to use it" | |
| ) | |
| # Event handlers | |
| submit_btn.click( | |
| fn=process_input, | |
| inputs=[ | |
| file_input, | |
| user_prompt, | |
| system_prompt, | |
| fps_slider, | |
| context_size, | |
| temperature, | |
| enable_thinking | |
| ], | |
| outputs=output, | |
| show_progress=True | |
| ) | |
| # Also allow Enter key submission | |
| user_prompt.submit( | |
| fn=process_input, | |
| inputs=[ | |
| file_input, | |
| user_prompt, | |
| system_prompt, | |
| fps_slider, | |
| context_size, | |
| temperature, | |
| enable_thinking | |
| ], | |
| outputs=output, | |
| show_progress=True | |
| ) | |
| # Information section | |
| with gr.Accordion("📋 Model Information", open=False): | |
| gr.Markdown(""" | |
| ### MiniCPM-V-4.5 Specifications | |
| - **Parameters**: 8B (Qwen3-8B + SigLIP2-400M) | |
| - **Video Compression**: 96x compression rate (6 frames → 64 tokens) | |
| - **Max Resolution**: Up to 1.8M pixels (1344x1344) | |
| - **Languages**: 30+ languages supported | |
| - **Performance**: Surpasses GPT-4o-latest on multiple benchmarks | |
| ### Usage Tips | |
| 1. **For Images**: Upload any image format and ask questions about content, objects, text, or analysis | |
| 2. **For Videos**: Adjust FPS based on video content (higher FPS for action, lower for static scenes) | |
| 3. **System Prompt**: Use to set specific roles like "You are an expert art critic" or "Analyze this from a medical perspective" | |
| 4. **Deep Thinking**: Enable for complex reasoning tasks, analysis, or problem-solving | |
| 5. **Temperature**: Lower (0.1-0.3) for factual responses, higher (0.7-1.0) for creative outputs | |
| ### Supported Formats | |
| - **Images**: JPG, PNG, JPEG, BMP, GIF, WEBP | |
| - **Videos**: MP4, AVI, MOV, MKV, WEBM, M4V | |
| """) | |
| return iface | |
| if __name__ == "__main__": | |
| # Create and launch interface | |
| demo = create_interface() | |
| demo.queue(max_size=20) | |
| demo.launch( | |
| share=True, | |
| server_name="0.0.0.0", | |
| server_port=7860, | |
| show_error=True | |
| ) |