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 )