Spaces:
Configuration error
Configuration error
| 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) | |