| # VLM Utilities | |
| # Simple functions for VLM data preprocessing | |
| from qwen_vl_utils import process_vision_info | |
| from typing import List, Dict, Any, Tuple | |
| import logging | |
| logger = logging.getLogger(__name__) | |
| def preprocess_vlm_messages(text_instruction: str, image_pil, processor): | |
| """ | |
| Complete VLM preprocessing - create messages, process vision, and get final inputs. | |
| Args: | |
| text_instruction: Robot task instruction | |
| image_pil: PIL Image object | |
| processor: VLM processor (AutoProcessor) | |
| Returns: | |
| VLM inputs ready for model forward | |
| """ | |
| # Create VLM messages format | |
| messages = [ | |
| { | |
| "role": "user", | |
| "content": [ | |
| {"type": "image", "image": image_pil}, | |
| {"type": "text", "text": text_instruction} | |
| ] | |
| } | |
| ] | |
| # Apply chat template | |
| text = processor.apply_chat_template( | |
| messages, tokenize=False, add_generation_prompt=True | |
| ) | |
| # Process vision info | |
| image_inputs, video_inputs = process_vision_info(messages) | |
| # Get final processor inputs | |
| inputs = processor( | |
| text=[text], | |
| images=image_inputs, | |
| videos=video_inputs, | |
| padding=True, | |
| return_tensors="pt", | |
| ) | |
| return inputs |