import os import base64 import requests import gradio as gr from openai import OpenAI from duckduckgo_search import DDGS from PIL import Image from io import BytesIO NVIDIA_BASE_URL = "https://integrate.api.nvidia.com/v1" MODEL_NAME = "nvidia/minimaxai/minimax-m2.7" def encode_image_from_url(url): """Download and encode image from URL to base64.""" try: response = requests.get(url, timeout=10) response.raise_for_status() img = Image.open(BytesIO(response.content)) buffered = BytesIO() img.save(buffered, format=img.format or "PNG") return base64.b64encode(buffered.getvalue()).decode("utf-8") except Exception as e: return None def encode_image_from_file(file_obj): """Encode uploaded image file to base64.""" try: if hasattr(file_obj, 'name') and file_obj.name: img = Image.open(file_obj.name) else: img = Image.open(file_obj) buffered = BytesIO() img.save(buffered, format=img.format or "PNG") return base64.b64encode(buffered.getvalue()).decode("utf-8") except Exception as e: return None def get_minimax_relevance(question, image_data, client): """Get relevance score from MiniMax-M2.7 vision model.""" try: response = client.chat.completions.create( model=MODEL_NAME, messages=[ { "role": "user", "content": [ {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{image_data}"}}, {"type": "text", "text": f"Question: {question}\nAnalyze this image for relevance. Respond with only a number between 0.0 and 1.0 representing how relevant this image is to the question. 1.0 = highly relevant, 0.0 = not relevant. Response must be ONLY the number, no text."} ] } ], temperature=0.1, max_tokens=10 ) score_text = response.choices[0].message.content.strip() score = float(score_text) return min(max(score, 0.0), 1.0) except Exception as e: return 0.0 def get_duckduckgo_context(question, image_description=""): """Get search context from DuckDuckGo.""" try: query = f"{question} {image_description}".strip() with DDGS() as ddgs: results = list(ddgs.text(query, max_results=3)) return " ".join([r["body"] for r in results]) if results else "" except Exception as e: return "" def calculate_combined_score(minimax_score, search_context, question): """Combine MiniMax score with DuckDuckGo context for final score.""" if not search_context: return minimax_score return 0.7 * minimax_score + 0.3 * (1.0 if any(word in search_context.lower() for word in question.lower().split()) else 0.5) def rank_images(question, images, image_urls, search_context, api_key): """Rank images by relevance to question.""" if not api_key: return [], "Please provide NVIDIA API key in secrets (NVIDIA_API_KEY)" if not images and not image_urls: return [], "Please upload images or provide image URLs" if not question.strip(): return [], "Please enter a question" client = OpenAI(api_key=api_key, base_url=NVIDIA_BASE_URL) image_data_list = [] if images: for img_obj in images: encoded = encode_image_from_file(img_obj) if encoded: image_data_list.append(("upload", encoded)) if image_urls: for url in image_urls.strip().split("\n"): url = url.strip() if url: encoded = encode_image_from_url(url) if encoded: image_data_list.append(("url", encoded)) if not image_data_list: return [], "No valid images could be loaded" ranked_images = [] for idx, (source, image_data) in enumerate(image_data_list): minimax_score = get_minimax_relevance(question, image_data, client) search_result = "" if search_context: search_result = get_duckduckgo_context(question, f"image {idx+1}") final_score = calculate_combined_score(minimax_score, search_result, question) ranked_images.append((final_score, source, image_data)) ranked_images.sort(key=lambda x: x[0], reverse=True) result_gallery = [] for score, source, image_data in ranked_images: if source == "upload": result_gallery.append(f"data:image/png;base64,{image_data}") else: img = Image.open(BytesIO(base64.b64decode(image_data))) img_path = f"/tmp/ranked_image_{len(result_gallery)}.png" img.save(img_path) result_gallery.append(img_path) return result_gallery, None css = """ #title { text-align: center; font-size: 2em; font-weight: bold; margin-bottom: 1em; } #question-input { margin-bottom: 1em; } #image-section { margin-bottom: 1em; } #button-row { margin-bottom: 1em; } #error-box { color: red; margin-bottom: 1em; } """ with gr.Blocks(css=css) as demo: gr.Markdown("## IMAGE RANKER", elem_id="title") with gr.Row(): with gr.Column(scale=1): question = gr.Textbox(label="Question", placeholder="What are you looking for?", elem_id="question-input") api_key = gr.Textbox(label="NVIDIA API Key (or set in secrets)", type="password", visible=True) with gr.Column(elem_id="image-section"): images = gr.File(file_count="multiple", file_types=["image"], label="Upload Images (up to 5)") gr.Markdown("**OR**") image_urls = gr.Textbox(label="Image URLs (one per line)", placeholder="https://example.com/image1.png") with gr.Row(elem_id="button-row"): search_btn = gr.Button("Search Context (DuckDuckGo)", variant="secondary") rank_btn = gr.Button("Rank Images", variant="primary") error_output = gr.Textbox(label="Error", visible=False, elem_id="error-box") gallery = gr.Gallery(label="Ranked Results", columns=3, object_fit="contain") search_context_state = gr.State("") def search_context_handler(question): if not question.strip(): return "Please enter a question first", "" try: with DDGS() as ddgs: results = list(ddgs.text(question, max_results=5)) context = " | ".join([f"{r['title']}: {r['body'][:100]}" for r in results]) if results else "" return "", context except Exception as e: return f"Search error: {str(e)}", "" search_btn.click( fn=search_context_handler, inputs=[question], outputs=[error_output, search_context_state] ) rank_btn.click( fn=rank_images, inputs=[question, images, image_urls, search_context_state, api_key], outputs=[gallery, error_output] ) demo.launch(debug=False, show_error=True)