import os.path as osp import gradio as gr HEADER = """ # Penguin-VL Gradio Interface Developed by [Penguin-VL](https://github.com/tencent-ailab/Penguin-VL) team at Tencent AI Lab. Note: speed on ZeroGPU does not reflect real model speed and may be influenced by the shared environment. For stable and fast Gradio Space deployment and running, please visit [the local UI instructions](https://github.com/tencent-ailab/Penguin-VL?tab=readme-ov-file#-gradio-demo-local-ui). For usage examples and expected results, please refer to [here](https://github.com/tencent-ailab/Penguin-VL/blob/master/inference/notebooks/01_penguinvl_inference_recipes.public.ipynb). Please login with your Hugging Face account first. """ IMAGE_FORMATS = ("png", "jpg", "jpeg") VIDEO_FORMATS = ("mp4", "mov") # (filename, prompt) pairs sourced from the official inference notebook. EXAMPLE_PAIRS = [ ("leetcode.png", "please think this problem step by step and give the python code solution"), ("newspaper.png", "please output the text in the image"), ("horse_poet.png", "Write a short poem inspired by this image. Capture the relationship between the man and the horse, as well as the traditional, historical atmosphere of the painting."), ("2b_table_result.png", "please output the table content in markdown format."), ("chart_understanding.png", "Look at the 'Nonmetropolitan' line. In what approximate year does it reach its absolute lowest point on the chart, and what is the approximate percent change at that time?"), ("video-example.mp4", "please describe the video in details"), ("polar_bear.mp4", "Describe what happens in this video."), ] class PenguinVLQwen3GradioInterface: def __init__(self, model_client, example_dir=None, default_system_prompt="You are a helpful assistant developed by Tencent AI Lab PenguinVL team.", **server_kwargs): self.model_client = model_client self.server_kwargs = server_kwargs self.default_system_prompt = (default_system_prompt or "").strip() examples = [] if example_dir: for filename, prompt in EXAMPLE_PAIRS: path = osp.join(example_dir, filename) if osp.isfile(path): examples.append([{"text": prompt, "files": [path]}]) self.interface = gr.ChatInterface( fn=self._predict, multimodal=True, description=HEADER, additional_inputs=[ gr.Textbox(label="System Prompt", value=self.default_system_prompt, lines=4), gr.Checkbox(label="Do Sample", value=True), gr.Slider(label="Temperature", minimum=0.0, maximum=1.0, value=0.1), gr.Slider(label="Top P", minimum=0.0, maximum=1.0, value=0.9), gr.Slider(label="Max New Tokens", minimum=0, maximum=4096, value=1536, step=1), gr.Slider(label="FPS (video)", minimum=0.0, maximum=10.0, value=1), gr.Slider(label="Max Frames (video)", minimum=0, maximum=256, value=180, step=1), ], additional_inputs_accordion=gr.Accordion(label="Settings", open=False), examples=examples or None, ) def _classify_file(self, path): ext = osp.splitext(path)[1].lower().lstrip(".") if ext in VIDEO_FORMATS: return "video" if ext in IMAGE_FORMATS: return "image" return None def _normalize_content(self, content, fps, max_frames): """Convert a single history content entry to model conversation format.""" if isinstance(content, str): return [{"type": "text", "text": content}] if isinstance(content, dict): path = content.get("path") or content.get("url", "") if path: media_type = self._classify_file(path) if media_type == "video": return [{"type": "video", "video": {"video_path": path, "fps": fps, "max_frames": max_frames}}] if media_type == "image": return [{"type": "image", "image": {"image_path": path}}] text = content.get("text") if isinstance(text, str): return [{"type": "text", "text": text}] if isinstance(content, list): result = [] for item in content: result.extend(self._normalize_content(item, fps, max_frames)) return result return [] def _build_conversation(self, message, history, system_prompt, fps, max_frames): conversation = [] active_system_prompt = (system_prompt or self.default_system_prompt).strip() if active_system_prompt: conversation.append({ "role": "system", "content": [{"type": "text", "text": active_system_prompt}], }) # History — merge consecutive user messages into one turn for entry in history: role = entry["role"] if role == "assistant": conversation.append({"role": "assistant", "content": entry["content"]}) elif role == "user": normalized = self._normalize_content(entry["content"], fps, max_frames) if not normalized: continue if conversation and conversation[-1]["role"] == "user": conversation[-1]["content"].extend(normalized) else: conversation.append({"role": "user", "content": normalized}) # Current message current_content = [] for f in message.get("files") or []: path = f if isinstance(f, str) else f.get("path", "") media_type = self._classify_file(path) if media_type == "video": current_content.append({"type": "video", "video": {"video_path": path, "fps": fps, "max_frames": max_frames}}) elif media_type == "image": current_content.append({"type": "image", "image": {"image_path": path}}) text = (message.get("text") or "").strip() if text: current_content.append({"type": "text", "text": text}) if current_content: if conversation and conversation[-1]["role"] == "user": conversation[-1]["content"].extend(current_content) else: conversation.append({"role": "user", "content": current_content}) return conversation def _predict(self, message, history, system_prompt, do_sample, temperature, top_p, max_new_tokens, fps, max_frames): conversation = self._build_conversation(message, history, system_prompt, fps, max_frames) if not conversation or conversation[-1]["role"] != "user": yield "" return generation_config = { "do_sample": do_sample, "temperature": temperature, "top_p": top_p, "max_new_tokens": max_new_tokens, } response = self.model_client.submit({"conversation": conversation, "generation_config": generation_config}) if isinstance(response, str): yield response return text = "" for token in response: text += token yield text def launch(self): self.interface.launch(ssr_mode=False, **self.server_kwargs)