# coding=utf-8 # Copyright 2026 NAVER Cloud Corp. and the HuggingFace Inc. team. All rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """HyperCLOVAX-SEED multimodal processor""" import base64 import copy import io import ipaddress import json import mimetypes import os import re import socket import tempfile from typing import Any, Dict, List, Optional, Tuple, Union import numpy as np import requests import torch import PIL from PIL import Image from transformers import ( AutoTokenizer, AutoFeatureExtractor, AutoImageProcessor, AutoVideoProcessor, ) from transformers.dynamic_module_utils import get_class_from_dynamic_module from transformers.utils import cached_file from transformers.audio_utils import AudioInput from transformers.image_processing_utils import BatchFeature from transformers.image_utils import ImageInput from transformers.processing_utils import ( AudioKwargs, ProcessingKwargs, ProcessorMixin, SpecificProcessorType, TextKwargs, Unpack, VideosKwargs, ) from transformers.tokenization_utils_base import PreTokenizedInput, TextInput from transformers.utils import logging from transformers.video_utils import VideoInput logger = logging.get_logger(__name__) # Default timeout for HTTP requests (connect, read) in seconds _DEFAULT_REQUEST_TIMEOUT = (5, 60) def _validate_url_safe(url: str) -> None: """Validate that a URL does not point to a private/internal network address (SSRF protection).""" from urllib.parse import urlparse parsed = urlparse(url) if parsed.scheme not in ("http", "https"): raise ValueError(f"Unsupported URL scheme: {parsed.scheme!r}") hostname = parsed.hostname if not hostname: raise ValueError(f"No hostname in URL: {url}") try: resolved_ips = socket.getaddrinfo(hostname, None) except socket.gaierror: raise ValueError(f"Cannot resolve hostname: {hostname}") for _, _, _, _, sockaddr in resolved_ips: ip = ipaddress.ip_address(sockaddr[0]) if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved: raise ValueError( f"URL resolves to a private/internal address ({ip}), " f"which is blocked for security: {url}" ) def _safe_request_get(url: str, timeout=_DEFAULT_REQUEST_TIMEOUT, **kwargs) -> "requests.Response": """Wrapper around requests.get() with SSRF protection and mandatory timeout.""" _validate_url_safe(url) response = requests.get(url, timeout=timeout, **kwargs) response.raise_for_status() return response def _detect_audio_suffix(audio_bytes: bytes) -> str: """Return a file-extension suffix (e.g. '.wav') from magic bytes, defaulting to '.wav'.""" header = audio_bytes[:12] if header[:4] == b"RIFF" and header[8:12] == b"WAVE": return ".wav" if header[:4] == b"fLaC": return ".flac" if header[:4] == b"OggS": return ".ogg" if header[:3] == b"ID3" or header[:2] in (b"\xff\xfb", b"\xff\xf3", b"\xff\xf2"): return ".mp3" if header[4:8] == b"ftyp": return ".m4a" return ".wav" class HyperCLOVAXSeedAudioKwargs(AudioKwargs, total=False): sample_rate: int chunk_unit: int min_chunk_size: int class HyperCLOVAXSeedTextKwargs(TextKwargs, total=False): return_mm_token_type_ids: bool class HyperCLOVAXSeedVideosKwargs(VideosKwargs, total=False): max_num_frames: int class HyperCLOVAXSeedProcessorKwargs(ProcessingKwargs, total=False): audio_kwargs: HyperCLOVAXSeedAudioKwargs text_kwargs: HyperCLOVAXSeedTextKwargs videos_kwargs: HyperCLOVAXSeedVideosKwargs _defaults = { "audio_kwargs": { "sample_rate": 16_000, "chunk_unit": 80, "min_chunk_size": 1_600, }, "images_kwargs": {}, "text_kwargs": { "padding": False, "return_mm_token_type_ids": False, }, "videos_kwargs": { "max_num_frames": 120, }, } class HyperCLOVAXSeedProcessor(ProcessorMixin): r""" Processor for HyperCLOVAX-SEED multimodal model. Combines a tokenizer, image processor, video processor, and audio feature extractor into a single processor that handles text, image, video, and audio inputs. Supports both continuous and discrete representations for each modality. Args: audio_processor ([`HyperCLOVAXSeedAudioProcessor`], *optional*): Audio feature extractor for continuous and discrete audio processing. chat_template (`str`, *optional*): Jinja2 chat template string. Falls back to the tokenizer's chat template if not provided. image_processor ([`HyperCLOVAXSeedImageProcessor`], *optional*): Image processor for continuous and discrete image processing. video_processor ([`HyperCLOVAXSeedVideoProcessor`], *optional*): Video processor for continuous video processing. tokenizer ([`PreTrainedTokenizer`] or [`PreTrainedTokenizerFast`]): Tokenizer for text encoding and special token management. ```python >>> from transformers import AutoProcessor >>> processor = AutoProcessor.from_pretrained("naver-hyperclovax/HyperCLOVAX-SEED-Think-4B") ``` """ attributes = [ "audio_processor", "image_processor", "video_processor", "tokenizer", ] audio_processor_class = "AutoFeatureExtractor" image_processor_class = "AutoImageProcessor" tokenizer_class = ("PreTrainedTokenizer", "PreTrainedTokenizerFast") video_processor_class = "AutoVideoProcessor" def __init__( self, audio_processor: Optional[AutoFeatureExtractor] = None, chat_template: Optional[str] = None, image_processor: Optional[AutoImageProcessor] = None, video_processor: Optional[AutoVideoProcessor] = None, tokenizer: Optional[AutoTokenizer] = None, **kwargs, ): # Prefer explicit chat_template; fall back to tokenizer's if available if chat_template is None and hasattr(tokenizer, "chat_template"): chat_template = tokenizer.chat_template # Pass all processors including None ones. check_argument_for_proper_class # is overridden below to accept None for optional sub-processors. ProcessorMixin.__init__( self, audio_processor=audio_processor, image_processor=image_processor, video_processor=video_processor, tokenizer=tokenizer, chat_template=chat_template, ) # Trim class-level attributes to only those that are actually present, # so save_pretrained does not try to serialise absent processors. self.attributes = [a for a in self.__class__.attributes if getattr(self, a) is not None] self.modalities = list() if self.audio_processor is not None: self.modalities.append("audio") self.audio_token = self.audio_processor.audio_token self.audio_token_id = tokenizer.convert_tokens_to_ids(self.audio_processor.audio_token) self.audio_start_token_id = tokenizer.convert_tokens_to_ids(self.audio_processor.audio_start_token) self.audio_end_token_id = tokenizer.convert_tokens_to_ids(self.audio_processor.audio_end_token) self.discrete_audio_token_id = None self.discrete_audio_start_token_id = None self.discrete_audio_end_token_id = None if self.audio_processor.use_discrete_token: self.discrete_audio_token_id = tokenizer.convert_tokens_to_ids(self.audio_processor.discrete_audio_token) self.discrete_audio_start_token_id = tokenizer.convert_tokens_to_ids(self.audio_processor.discrete_audio_start_token) self.discrete_audio_end_token_id = tokenizer.convert_tokens_to_ids(self.audio_processor.discrete_audio_end_token) if self.image_processor is not None: self.modalities.append("image") self.image_token = self.image_processor.image_token self.image_token_id = tokenizer.convert_tokens_to_ids(self.image_processor.image_token) self.image_start_token_id = tokenizer.convert_tokens_to_ids(self.image_processor.image_start_token) self.image_end_token_id = tokenizer.convert_tokens_to_ids(self.image_processor.image_end_token) self.discrete_image_token_id = None self.discrete_image_start_token_id = None self.discrete_image_end_token_id = None if self.image_processor.use_discrete_token: self.discrete_image_token_id = tokenizer.convert_tokens_to_ids(self.image_processor.discrete_image_token) self.discrete_image_start_token_id = tokenizer.convert_tokens_to_ids(self.image_processor.discrete_image_start_token) self.discrete_image_end_token_id = tokenizer.convert_tokens_to_ids(self.image_processor.discrete_image_end_token) if self.video_processor is not None: self.modalities.append("video") self.video_token = self.video_processor.video_token self.video_token_id = tokenizer.convert_tokens_to_ids(self.video_processor.video_token) self.video_audio_token = self.video_processor.video_audio_token self.video_start_token_id = tokenizer.convert_tokens_to_ids(self.video_processor.video_start_token) self.video_end_token_id = tokenizer.convert_tokens_to_ids(self.video_processor.video_end_token) self.video_audio_token_id = tokenizer.convert_tokens_to_ids(self.video_processor.video_audio_token) def check_argument_for_proper_class(self, argument_name, argument): """Allow None for optional sub-processors (audio, image, video). ProcessorMixin.__init__ calls this for every kwarg it receives. The base implementation raises TypeError when the argument is not an instance of the expected processor class, so None would be rejected. We short-circuit for None here because audio_processor, image_processor, and video_processor are all optional in HyperCLOVAXSeedProcessor. """ if argument is None: return None return super().check_argument_for_proper_class(argument_name, argument) @classmethod def from_pretrained( cls: "type[SpecificProcessorType]", pretrained_model_name_or_path: Union[str, os.PathLike], **kwargs, ) -> "HyperCLOVAXSeedProcessor": audio_processor_kwargs = kwargs.pop("audio_processor_kwargs", dict()) image_processor_kwargs = kwargs.pop("image_processor_kwargs", dict()) video_processor_kwargs = kwargs.pop("video_processor_kwargs", dict()) if "tokenizer" not in kwargs: kwargs["tokenizer"] = AutoTokenizer.from_pretrained( pretrained_model_name_or_path, **kwargs, ) if not kwargs.get("audio_processor", None): try: # AutoFeatureExtractor does not support specifying a config file kwargs["audio_processor"] = cls._load_processor_from_config( pretrained_model_name_or_path, "audio_preprocessor_config.json", "AutoFeatureExtractor", **audio_processor_kwargs, **kwargs, ) except Exception as ex: logger.warning("Failed to load audio_processor: %s", ex) kwargs["audio_processor"] = None if not kwargs.get("image_processor", None): try: kwargs["image_processor"] = cls._load_processor_from_config( pretrained_model_name_or_path, "image_preprocessor_config.json", "AutoImageProcessor", **image_processor_kwargs, **kwargs, ) except Exception as ex: logger.warning("Failed to load image_processor: %s", ex) kwargs["image_processor"] = None if not kwargs.get("video_processor", None): try: kwargs["video_processor"] = cls._load_processor_from_config( pretrained_model_name_or_path, "video_preprocessor_config.json", "AutoVideoProcessor", **video_processor_kwargs, **kwargs, ) except Exception as ex: logger.warning("Failed to load video_processor: %s", ex) kwargs["video_processor"] = None return cls( audio_processor=kwargs.get("audio_processor"), image_processor=kwargs.get("image_processor"), video_processor=kwargs.get("video_processor"), tokenizer=kwargs.get("tokenizer"), chat_template=kwargs.get("chat_template"), ) @staticmethod def _load_processor_from_config( pretrained_model_name_or_path: Union[str, os.PathLike], config_filename: str, auto_class_key: str, **kwargs, ) -> Any: """Load a processor from a non-standard config filename. Standard Auto classes (AutoImageProcessor, AutoFeatureExtractor) only read from ``preprocessor_config.json``. This method reads from a custom config file (e.g., ``image_preprocessor_config.json``, ``audio_preprocessor_config.json``), resolves the class via ``auto_map``, and instantiates it with the config fields. Args: pretrained_model_name_or_path: Model path or HF Hub repo ID. config_filename: JSON config filename (e.g., "image_preprocessor_config.json"). auto_class_key: Key in auto_map (e.g., "AutoImageProcessor"). """ resolved_path = cached_file( pretrained_model_name_or_path, config_filename, **{k: v for k, v in kwargs.items() if k in ( "cache_dir", "force_download", "proxies", "token", "revision", "local_files_only", )}, ) with open(resolved_path, "r") as f: config_dict = json.load(f) auto_map = config_dict.pop("auto_map", {}) class_ref = auto_map.get(auto_class_key) if class_ref is None: raise ValueError( f"No '{auto_class_key}' found in auto_map of {config_filename}" ) processor_class = get_class_from_dynamic_module( class_ref, pretrained_model_name_or_path, **{k: v for k, v in kwargs.items() if k in ( "cache_dir", "force_download", "proxies", "token", "revision", "local_files_only", "code_revision", )}, ) # Remove meta fields that are not __init__ parameters config_dict.pop("image_processor_type", None) config_dict.pop("feature_extractor_type", None) config_dict.pop("processor_class", None) return processor_class(**config_dict) def save_pretrained( self, save_directory: Union[str, os.PathLike], *args, **kwargs, ) -> None: # self.attributes is already filtered in __init__, so no class-level # mutation is needed. Just make sure register_for_auto_class() sees the # correct (instance-level) attributes list. self.register_for_auto_class() super().save_pretrained(save_directory, *args, **kwargs) def load_multimodal_inputs( self, conversation: Union[List[Dict[str, Any]], List[List[Dict[str, Any]]]], use_audio_in_video: Optional[bool] = None, ) -> Dict[str, Any]: """Load audio, image, and video data referenced in conversations. Extracts media references from conversation messages and loads the actual data. Each message may reference media either via top-level keys (``audio_files``, ``image_files``, ``video_files``) or via structured content blocks with ``type`` set to ``"audio"``, ``"image"``, or ``"video"``. Supported input formats for each media type: - Local file path (e.g., ``"/path/to/file.wav"``) - HTTP/HTTPS URL (e.g., ``"https://example.com/image.jpg"``) - Base64 data URI (e.g., ``"data:audio/wav;base64,..."``) Args: conversation: A single conversation (list of message dicts) or a batch of conversations (list of lists). Each message dict should have a ``"role"`` and ``"content"`` key following the chat format, matching the input accepted by ``processor.tokenizer.apply_chat_template``. use_audio_in_video: If ``True``, extract audio tracks from video files and include them in the returned audio list. Returns: Plain dict with flat parallel lists:: { "audios": List[np.ndarray] | None, "sampling_rates": List[int] | None, "images": List[PIL.Image.Image] | None, "videos": List[List[PIL.Image.Image]] | None, "video_audios": List[np.ndarray | None] | None, "video_sampling_rates": List[int | None] | None, "video_fps_list": List[float] | None, } Pass directly to the processor via ``processor(**mm)``. """ if use_audio_in_video is None: use_audio_in_video = bool( self.video_processor is not None and getattr(self.video_processor, "use_audio_in_video", False) ) conversations = conversation if isinstance(conversation[0], list) else [conversation] audios: List[np.ndarray] = [] sampling_rates: List[int] = [] images: List[Image.Image] = [] videos: List[List[Image.Image]] = [] video_audios: List[Optional[np.ndarray]] = [] video_sampling_rates: List[Optional[int]] = [] video_fps_list: List[float] = [] target_sr = 16_000 if ( self.audio_processor is not None and hasattr(self.audio_processor, "sampling_rate") ): target_sr = self.audio_processor.sampling_rate for conv in conversations: for message in conv: # Handle media files at message level (when content is a string) if message.get("audio_files"): for audio_path in message["audio_files"]: info = self._load_audio(audio_path, sr=target_sr) audios.append(info["waveform"]) sampling_rates.append(info["sampling_rate"]) if message.get("image_files"): for image_path in message["image_files"]: info = self._load_image(image_path) images.append(info["image"]) if message.get("video_files"): for video_path in message["video_files"]: info = self._load_video( video_path, sr=target_sr, use_audio_in_video=use_audio_in_video, ) videos.append(info["frames"]) video_audios.append(info["audio"]) video_sampling_rates.append(info["sampling_rate"]) video_fps_list.append(info["fps"]) content = message.get("content", []) if not isinstance(content, list): continue for ele in content: type_ = ele.get("type") if type_ in ("audio", "audio_url"): raw = ele.get("audio", ele.get("audio_url")) # OpenAI-style: {"type": "audio_url", "audio_url": {"url": "..."}} if isinstance(raw, dict): raw = raw.get("url", raw) path = raw if path: if "mime_type" not in ele and isinstance(path, str): filename = ele.get("filename", path if not path.startswith("http") else "a.wav") mime_type = mimetypes.guess_type(filename)[0] if mime_type: ele["mime_type"] = mime_type info = self._load_audio( path, sr=target_sr, start=ele.get("audio_start", 0.0), end=ele.get("audio_end", None), ) audios.append(info["waveform"]) sampling_rates.append(info["sampling_rate"]) elif type_ in ("image", "image_url"): raw = ele.get("image", ele.get("image_url")) # OpenAI-style: {"type": "image_url", "image_url": {"url": "..."}} if isinstance(raw, dict): raw = raw.get("url", raw) path = raw if path: if "mime_type" not in ele and isinstance(path, str): filename = ele.get("filename", path if not path.startswith("http") else "a.jpg") mime_type = mimetypes.guess_type(filename)[0] if mime_type: ele["mime_type"] = mime_type info = self._load_image(path) images.append(info["image"]) elif type_ in ("video", "video_url"): raw = ele.get("video", ele.get("video_url")) # OpenAI-style: {"type": "video_url", "video_url": {"url": "..."}} if isinstance(raw, dict): raw = raw.get("url", raw) path = raw if path: if "mime_type" not in ele and isinstance(path, str): filename = ele.get("filename", path if not path.startswith("http") else "a.mp4") mime_type = mimetypes.guess_type(filename)[0] if mime_type: ele["mime_type"] = mime_type info = self._load_video( path, start=ele.get("video_start", 0.0), end=ele.get("video_end", None), max_num_frames=ele.get("max_num_frames", None), sr=target_sr, use_audio_in_video=use_audio_in_video, ) videos.append(info["frames"]) video_audios.append(info["audio"]) video_sampling_rates.append(info["sampling_rate"]) video_fps_list.append(info["fps"]) return { "audios": audios if audios else None, "sampling_rates": sampling_rates if sampling_rates else None, "images": images if images else None, "videos": videos if videos else None, "video_audios": video_audios if video_audios else None, "video_sampling_rates": video_sampling_rates if video_sampling_rates else None, "video_fps_list": video_fps_list if video_fps_list else None, } def apply_chat_template( self, conversation: Union[List[Dict[str, Any]], List[List[Dict[str, Any]]]], chat_template: Optional[str] = None, tokenize: bool = False, return_dict: bool = False, **kwargs, ): """ Apply the chat template to a conversation and optionally tokenize the result. This override extends the base class behaviour by also loading and processing multimodal inputs (audio, images, videos, video-audio tracks) that are referenced inside the conversation, then forwarding everything to :meth:`__call__` in a single step. Args: conversation: A single conversation or a batch of conversations. chat_template: Jinja2 template string or template name. Falls back to the processor's default chat template when not provided. tokenize: If ``True``, tokenize the rendered prompt and process all multimodal inputs, returning a :class:`~transformers.BatchFeature`. If ``False`` (default), return the rendered prompt string only. return_dict: If ``True`` and ``tokenize=True``, return the full :class:`~transformers.BatchFeature` dict. If ``False``, return only ``input_ids``. **kwargs: Split automatically into two groups: - *Template kwargs* — any key not recognised by :class:`HyperCLOVAXSeedProcessorKwargs` is forwarded to ``tokenizer.apply_chat_template`` as a Jinja variable or standard tokenizer argument (e.g. ``add_generation_prompt``, ``use_audio_in_video``, ``skip_reasoning``, ``tools``). - *Processor kwargs* — keys declared in :class:`HyperCLOVAXSeedProcessorKwargs` (or its modality sub-dicts such as ``text_kwargs``) are forwarded exclusively to :meth:`__call__` (e.g. ``return_tensors``, ``padding``). Returns: ``str`` when ``tokenize=False``; :class:`~transformers.BatchFeature` when ``tokenize=True`` and ``return_dict=True``; ``list[list[int]]`` otherwise. """ # Build the set of kwarg keys that __call__ recognises via _merge_kwargs. # These are the flat field names declared in HyperCLOVAXSeedProcessorKwargs # and its modality-specific sub-TypedDicts (text_kwargs, images_kwargs, …). _processor_keys: set = set() for _modality_annot in HyperCLOVAXSeedProcessorKwargs.__annotations__.values(): if hasattr(_modality_annot, "__annotations__"): _processor_keys.update(_modality_annot.__annotations__) # Also allow passing modality sub-dicts directly (e.g. text_kwargs={…}) _processor_keys.update(HyperCLOVAXSeedProcessorKwargs.__annotations__) call_kwargs = {k: v for k, v in kwargs.items() if k in _processor_keys} template_kwargs = {k: v for k, v in kwargs.items() if k not in _processor_keys} # Step 1: render chat template → plain text (no media loading). # Use tokenizer.apply_chat_template directly so that the full tokenizer # context (special tokens, helper variables, etc.) is available to the # Jinja template, matching the behaviour users expect. # template_kwargs flows through freely — no hard-coded list needed. if "use_audio_in_video" not in template_kwargs: template_kwargs["use_audio_in_video"] = bool( self.video_processor is not None and getattr(self.video_processor, "use_audio_in_video", False) ) prompt = self.tokenizer.apply_chat_template( conversation, chat_template=chat_template, tokenize=False, **template_kwargs, ) if not tokenize: return prompt # Step 2: load all multimodal inputs from the conversation mm = self.load_multimodal_inputs(conversation) # Step 3: call __call__ with text + all modalities. out = self( text=prompt, audios=mm.get("audios"), images=mm.get("images"), videos=mm.get("videos"), video_audios=mm.get("video_audios"), sampling_rates=mm.get("sampling_rates"), video_sampling_rates=mm.get("video_sampling_rates"), video_fps_list=mm.get("video_fps_list"), **call_kwargs, ) if return_dict: return out return out["input_ids"] def _load_audio( self, path: Union[str, bytes, io.BytesIO, np.ndarray], sr: int = 16000, start: float = 0.0, end: Optional[float] = None, ) -> Dict[str, Any]: """Load an audio clip from a file path, URL, base64 string, bytes, or numpy array. Supports the following input formats: - ``np.ndarray``: Used directly (multi-channel arrays are averaged to mono). - ``bytes`` / ``io.BytesIO``: Written to a temp file, then loaded via ``librosa.load``. - Local file path: Loaded via ``librosa.load``. - HTTP/HTTPS URL: Downloaded with SSRF protection, then loaded via ``librosa.load``. - Base64 data URI (``data:audio/...;base64,...``): Decoded then loaded. Args: path: Audio source — file path, URL, base64 data URI, bytes, or numpy array. sr: Target sampling rate in Hz. start: Start time in seconds for slicing. end: End time in seconds for slicing. ``None`` means until the end. Returns: Dict with keys ``"waveform"`` (1-D float32 numpy array) and ``"sampling_rate"`` (int). """ import librosa duration = (end - start) if end is not None else None if isinstance(path, np.ndarray): audio = path.mean(axis=1) if path.ndim > 1 else path start_idx = int(sr * start) end_idx = int(sr * end) if end is not None else None return {"waveform": audio[start_idx:end_idx], "sampling_rate": sr} if isinstance(path, io.BytesIO): path = path.getvalue() if isinstance(path, bytes): # Detect format from magic bytes and write to temp file for librosa suffix = _detect_audio_suffix(path) with tempfile.NamedTemporaryFile(mode="wb", suffix=suffix, delete=True) as fp: fp.write(path) fp.flush() y, _ = librosa.load(fp.name, sr=sr, offset=start, duration=duration, mono=True) return {"waveform": y, "sampling_rate": sr} # str path if path.startswith("data:audio"): _, base64_data = path.split("base64,", 1) raw = base64.b64decode(base64_data) return self._load_audio(raw, sr=sr, start=start, end=end) if path.startswith("http://") or path.startswith("https://"): response = _safe_request_get(path, timeout=_DEFAULT_REQUEST_TIMEOUT) return self._load_audio(response.content, sr=sr, start=start, end=end) if path.startswith("file://"): path = path[len("file://"):] # Local file path y, _ = librosa.load(path, sr=sr, offset=start, duration=duration, mono=True) return {"waveform": y, "sampling_rate": sr} def _load_image( self, path: Union[str, bytes, np.ndarray, "PIL.Image.Image"], ) -> Dict[str, Any]: """Load an image from a file path, URL, base64 string, bytes, ndarray, or PIL Image. Supports the following input formats: - ``PIL.Image.Image``: Used directly (converted to RGB if needed). - ``np.ndarray``: Converted via ``Image.fromarray``. - ``bytes``: Opened via ``Image.open(BytesIO(...))``. - Local file path: Opened via ``PIL.Image.open``. - HTTP/HTTPS URL: Downloaded with SSRF protection, then opened. - Base64 data URI (``data:image/...;base64,...``): Decoded then opened. Args: path: Image source. Returns: Dict with key ``"image"`` (PIL Image in RGB mode). """ if isinstance(path, Image.Image): image = path elif isinstance(path, np.ndarray): image = Image.fromarray(path) elif isinstance(path, bytes): image = Image.open(io.BytesIO(path)) elif isinstance(path, str): if path.startswith("data:image"): _, base64_data = path.split("base64,", 1) image = Image.open(io.BytesIO(base64.b64decode(base64_data))) elif path.startswith("http://") or path.startswith("https://"): response = _safe_request_get(path, timeout=_DEFAULT_REQUEST_TIMEOUT) image = Image.open(io.BytesIO(response.content)) elif path.startswith("file://"): image = Image.open(path[len("file://"):]) else: # local file path image = Image.open(path) else: raise TypeError(f"Unsupported image type: {type(path)}") if image.mode != "RGB": image = image.convert("RGB") return {"image": image} def _load_video( self, path: Union[str, bytes, io.BytesIO], start: float = 0.0, end: Optional[float] = None, max_num_frames: Optional[int] = None, fps: float = 2.0, sr: int = 16000, use_audio_in_video: bool = False, ) -> Dict[str, Any]: """Load video frames (and optionally audio) from a file, URL, bytes, or base64 string. Args: path: Video source — local file path, HTTP/HTTPS URL, raw bytes, BytesIO, or base64 data URI. start: Start time in seconds. end: End time in seconds. ``None`` means until the end. max_num_frames: Maximum number of frames to return. fps: Target frame rate for uniform sampling. sr: Target audio sampling rate in Hz (used only when ``use_audio_in_video=True``). use_audio_in_video: If ``True``, also extract audio from the video stream. Returns: Dict with keys: - ``"frames"``: ``List[PIL.Image.Image]``; - ``"audio"``: 1-D float32 ``np.ndarray`` or ``None``; - ``"sampling_rate"``: audio sampling rate (int) or ``None``; - ``"fps"``: actual frames-per-second at which frames were sampled (float). """ video_source = self._resolve_video_source(path) # decord (primary) try: import decord from decord import cpu as decord_cpu frames, actual_fps, tmp_path = self._decord_read_frames( video_source, start, end, max_num_frames, fps, decord_cpu ) audio = None if use_audio_in_video: audio = self._decord_read_audio( video_source if tmp_path is None else tmp_path, sr=sr, start=start, end=end, ) if tmp_path is not None: os.remove(tmp_path) return { "frames": frames, "audio": audio, "sampling_rate": sr if audio is not None else None, "fps": actual_fps, } except ImportError: pass # torchvision (fallback) import torchvision tmp_path = None if isinstance(video_source, io.BytesIO): with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as tmp: tmp.write(video_source.getvalue()) tmp_path = tmp.name video_source = tmp_path try: video_tensor, _, info = torchvision.io.read_video( video_source, start_pts=start, end_pts=end, pts_unit="sec" ) total_frames = video_tensor.shape[0] video_fps = info.get("video_fps", 24.0) nframes = max(2, round(total_frames / video_fps * fps)) if max_num_frames is not None: nframes = min(nframes, max_num_frames) nframes = max(2, nframes - (nframes % 2)) nframes = min(nframes, total_frames) idx = torch.linspace(0, total_frames - 1, nframes).round().long() sampled = video_tensor[idx].numpy() frames = [Image.fromarray(sampled[i]) for i in range(sampled.shape[0])] audio = None if use_audio_in_video: logger.warning( "Audio extraction is not supported in the torchvision fallback path; " "install decord for full audio-from-video support." ) clip_duration = total_frames / video_fps if video_fps > 0 else 1.0 actual_fps = len(frames) / clip_duration if clip_duration > 0 else fps return { "frames": frames, "audio": audio, "sampling_rate": sr if audio is not None else None, "fps": actual_fps, } finally: if tmp_path is not None: os.remove(tmp_path) # Private helpers for _load_video def _resolve_video_source( self, path: Union[str, bytes, io.BytesIO], ) -> Union[str, io.BytesIO]: """Resolve raw video input to a file path string or an BytesIO object.""" if isinstance(path, (bytes, bytearray)): return io.BytesIO(path) if isinstance(path, io.BytesIO): return path if isinstance(path, str): if path.startswith("data:video"): _, base64_data = path.split("base64,", 1) return io.BytesIO(base64.b64decode(base64_data)) if path.startswith("http://") or path.startswith("https://"): response = _safe_request_get(path, timeout=(5, 600)) return io.BytesIO(response.content) if path.startswith("file://"): path = path[len("file://"):] # local file path — return as-is return path raise TypeError(f"Unsupported video source type: {type(path)}") @staticmethod def _decord_read_frames( video_source: Union[str, io.BytesIO], start: float, end: Optional[float], max_num_frames: Optional[int], fps: float, decord_cpu, ) -> Tuple[List[Image.Image], float, Optional[str]]: """Read frames using decord. Returns (frames, actual_fps, tmp_path). ``actual_fps`` is the sampling rate (frames per second) at which frames were selected from the clip. ``tmp_path`` is set when a BytesIO had to be flushed to a temporary file (decord on some platforms cannot read BytesIO directly); the caller is responsible for deleting it. """ import decord source = video_source tmp_path = None # decord.VideoReader may not accept BytesIO on all platforms — try directly first try: vr = decord.VideoReader(source, ctx=decord_cpu(0)) except Exception: if isinstance(source, io.BytesIO): with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as tmp: tmp.write(source.getvalue()) tmp_path = tmp.name vr = decord.VideoReader(tmp_path, ctx=decord_cpu(0)) else: raise total_frames = len(vr) video_fps = vr.get_avg_fps() start_frame = int(start * video_fps) if start else 0 end_frame = int(end * video_fps) if end is not None else total_frames - 1 end_frame = min(end_frame, total_frames - 1) if start_frame >= end_frame: start_frame, end_frame = 0, total_frames - 1 available_frames = end_frame - start_frame + 1 nframes = max(2, round(available_frames / video_fps * fps)) if max_num_frames is not None: nframes = min(nframes, max_num_frames) nframes = max(2, nframes - (nframes % 2)) nframes = min(nframes, available_frames) idx = torch.linspace(start_frame, end_frame, nframes).round().long().tolist() frames_np = vr.get_batch(idx).asnumpy() del vr frames = [Image.fromarray(frames_np[i]) for i in range(frames_np.shape[0])] clip_duration = available_frames / video_fps if video_fps > 0 else 1.0 actual_fps = len(frames) / clip_duration if clip_duration > 0 else fps return frames, actual_fps, tmp_path @staticmethod def _decord_read_audio( source: Union[str, io.BytesIO], sr: int, start: float, end: Optional[float], ) -> Optional[np.ndarray]: """Extract audio from a video file using decord.AudioReader.""" try: from decord import AudioReader from decord import cpu as decord_cpu # AudioReader requires a file path string tmp_path = None if isinstance(source, io.BytesIO): with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as tmp: tmp.write(source.getvalue()) tmp_path = tmp.name source = tmp_path try: ar = AudioReader(source, ctx=decord_cpu(0), sample_rate=sr, mono=True) total_samples = ar.shape[1] start_sample = int(start * sr) end_sample = int(end * sr) if end is not None else total_samples end_sample = min(end_sample, total_samples) audio = ar[start_sample:end_sample].asnumpy().flatten().astype(np.float32) return audio finally: if tmp_path is not None: os.remove(tmp_path) except Exception as e: logger.warning("Failed to extract audio from video: %s", e) return None def __call__( self, text: Union[TextInput, PreTokenizedInput, List[TextInput], List[PreTokenizedInput]] = None, audios: Optional[AudioInput] = None, images: Optional[ImageInput] = None, videos: Optional[VideoInput] = None, video_audios: Optional[AudioInput] = None, sampling_rates: Optional[List[int]] = None, video_sampling_rates: Optional[List[Optional[int]]] = None, video_fps_list: Optional[List[float]] = None, **kwargs: Unpack[HyperCLOVAXSeedProcessorKwargs], ) -> BatchFeature: """ Main method to prepare text, audio, image, and video inputs for the model. This method forwards `text` and `kwargs` to the tokenizer if `text` is not `None`, processes images/videos through their respective processors, and handles audio feature extraction. Args: images (`PIL.Image.Image`, `np.ndarray`, `torch.Tensor`, `list[PIL.Image.Image]`, `list[np.ndarray]`, `list[torch.Tensor]`): The image or batch of images to be prepared. Each image can be a PIL image, NumPy array or PyTorch tensor. Both channels-first and channels-last formats are supported. text (`str`, `list[str]`, `list[list[str]]`): The sequence or batch of sequences to be encoded. Each sequence can be a string or a list of strings (pretokenized string). If the sequences are provided as list of strings (pretokenized), you must set `is_split_into_words=True` (to lift the ambiguity with a batch of sequences). videos (`np.ndarray`, `torch.Tensor`, `list[np.ndarray]`, `list[torch.Tensor]`): The image or batch of videos to be prepared. Each video can be a 4D NumPy array or PyTorch tensor, or a nested list of 3D frames. Both channels-first and channels-last formats are supported. return_tensors (`str` or [`~utils.TensorType`], *optional*): If set, will return tensors of a particular framework. Acceptable values are: - `'tf'`: Return TensorFlow `tf.constant` objects. - `'pt'`: Return PyTorch `torch.Tensor` objects. - `'np'`: Return NumPy `np.ndarray` objects. - `'jax'`: Return JAX `jnp.ndarray` objects. Returns: [`BatchFeature`]: A [`BatchFeature`] with the following fields: - **input_ids** -- List of token ids to be fed to a model. Returned when `text` is not `None`. - **attention_mask** -- List of indices specifying which tokens should be attended to by the model (when `return_attention_mask=True` or if *"attention_mask"* is in `self.model_input_names` and if `text` is not `None`). - **pixel_values** -- Pixel values to be fed to a model. Returned when `images` is not `None`. - **pixel_values_videos** -- Pixel values of videos to be fed to a model. Returned when `videos` is not `None`. - **image_grid_thw** -- List of image 3D grid in LLM. Returned when `images` is not `None`. - **video_grid_thw** -- List of video 3D grid in LLM. Returned when `videos` is not `None`. """ output_kwargs = self._merge_kwargs( HyperCLOVAXSeedProcessorKwargs, tokenizer_init_kwargs=self.tokenizer.init_kwargs, **kwargs, ) # [Text Processing] (Placeholder Replacement) if text is None: pass else: if isinstance(text, str): text = [ text, ] # below lines change text in-place text = copy.deepcopy(text) # [Audio Processing] audio_inputs = dict() discrete_audio_inputs = dict() if ( audios is not None and self.audio_processor is not None ): # Normalize to List[waveform] if isinstance(audios, np.ndarray): # (T,) single → [arr]; (B,T) batch → list of 1-D arrays audios = [audios] if audios.ndim == 1 else list(audios) elif isinstance(audios, torch.Tensor): audios = [audios] if audios.ndim == 1 else list(audios.unbind(0)) _no_concat_keys = {"num_audio_tokens", "num_discrete_audio_tokens"} for _audio in audios: # Normalize each element to a list accepted by audio_processor if isinstance(_audio, np.ndarray): _audio = [_audio] elif isinstance(_audio, torch.Tensor): _audio = [_audio] _audio_features = self.audio_processor( audios=_audio, **output_kwargs.get("audio_kwargs", {}), ) for _k, _v in _audio_features.items(): if _k in [ "discrete_audio_values", "num_discrete_audio_tokens", ]: if _k not in discrete_audio_inputs: discrete_audio_inputs[_k] = list() discrete_audio_inputs[_k].append(_v) else: if _k not in audio_inputs: audio_inputs[_k] = list() audio_inputs[_k].append(_v) audio_inputs = { _k: torch.cat(_v, dim=0) if isinstance(_v[0], torch.Tensor) and _k not in _no_concat_keys else _v for _k, _v in audio_inputs.items() } if discrete_audio_inputs: discrete_audio_inputs = { _k: torch.cat(_v, dim=0) if isinstance(_v[0], torch.Tensor) and _k not in _no_concat_keys else _v for _k, _v in discrete_audio_inputs.items() } # [Image Processing] image_inputs = dict() discrete_image_inputs = dict() if ( images is not None and self.image_processor is not None ): # Normalize to List[image] if isinstance(images, PIL.Image.Image): images = [images] elif isinstance(images, np.ndarray): # (H,W,C) single → [arr]; (B,H,W,C) batch → list of arrays images = [images] if images.ndim == 3 else list(images) elif isinstance(images, torch.Tensor): images = [images] if images.ndim == 3 else list(images.unbind(0)) _no_concat_keys = {"num_image_tokens", "num_discrete_image_tokens"} for _image in images: # Normalize each element to a list accepted by image_processor if isinstance(_image, PIL.Image.Image): _image = [_image] elif isinstance(_image, np.ndarray): _image = [_image] elif isinstance(_image, torch.Tensor): _image = [_image] _image_features = self.image_processor( images=_image, **output_kwargs.get("images_kwargs", {}), ) for _k, _v in _image_features.items(): if _k in [ "discrete_pixel_values", "discrete_image_ratios", "num_discrete_image_tokens", ]: if _k not in discrete_image_inputs: discrete_image_inputs[_k] = list() discrete_image_inputs[_k].append(_v) else: if _k not in image_inputs: image_inputs[_k] = list() image_inputs[_k].append(_v) image_inputs = { _k: torch.cat(_v, dim=0) if isinstance(_v[0], torch.Tensor) and _k not in _no_concat_keys else _v for _k, _v in image_inputs.items() } if discrete_image_inputs: discrete_image_inputs = { _k: torch.cat(_v, dim=0) if isinstance(_v[0], torch.Tensor) and _k not in _no_concat_keys else _v for _k, _v in discrete_image_inputs.items() } # [Video Processing] video_inputs = dict() if ( videos is not None and self.video_processor is not None ): # Normalize to List[video] if isinstance(videos, np.ndarray): # (T,H,W,C) single → [(T,H,W,C)]; (B,T,H,W,C) batch → list of (T,H,W,C) videos = [videos] if videos.ndim == 4 else list(videos) elif isinstance(videos, torch.Tensor): videos = [videos] if videos.ndim == 4 else list(videos.unbind(0)) elif isinstance(videos, (list, tuple)) and len(videos) > 0: # List[PIL.Image] = single video given as a flat frame list if isinstance(videos[0], Image.Image): videos = [list(videos)] _no_concat_keys = { "num_video_tokens", "num_discrete_video_tokens", "num_video_audio_tokens", "num_discrete_video_audio_tokens" } for _video in videos: # Normalize each video to List[np.ndarray (T,H,W,C)] expected by video_processor if isinstance(_video, (list, tuple)) and len(_video) > 0 and isinstance(_video[0], Image.Image): # List[PIL.Image] frames → (T,H,W,C) numpy array _video = [np.stack([np.array(f) for f in _video], axis=0)] elif isinstance(_video, np.ndarray) and _video.ndim == 4: _video = [_video] elif isinstance(_video, torch.Tensor) and _video.ndim == 4: _video = [_video] _video_features = self.video_processor( videos=_video, **output_kwargs.get("videos_kwargs", {}), ) for _k, _v in _video_features.items(): if _k not in video_inputs: video_inputs[_k] = list() video_inputs[_k].append(_v) if ( self.video_processor.use_audio_in_video and isinstance(video_audios, (list, tuple)) and len(video_audios) == len(videos) and self.audio_processor is not None ): for _video_audio in video_audios: if _video_audio is None: continue if ( not isinstance(_video_audio, (list, tuple)) and isinstance(_video_audio, np.ndarray) ): _video_audio = [_video_audio] _video_audio_features = self.audio_processor( audios=_video_audio, prefix="video_", **output_kwargs.get("audio_kwargs", {}), ) for _k, _v in _video_audio_features.items(): if _k not in video_inputs: video_inputs[_k] = list() video_inputs[_k].append(_v) video_inputs = { _k: torch.cat(_v, dim=0) if isinstance(_v[0], torch.Tensor) and _k not in _no_concat_keys else _v for _k, _v in video_inputs.items() } # [Duration Replacement] - Replace <|audio_duration|> placeholders with actual values if ( text is not None and audios is not None ): sr = 16000 if ( self.audio_processor is not None and isinstance(getattr(self.audio_processor, "sampling_rate", None), int) ): sr = self.audio_processor.sampling_rate # audios can be [batch][audio_idx] or [audio_idx] format flat_audios = audios if len(audios) > 0 and isinstance(audios[0], list): flat_audios = [a for batch in audios for a in batch] audio_dur_idx = 0 for _batch_idx, _text in enumerate(text): while "<|audio_duration|>" in _text and audio_dur_idx < len(flat_audios): audio_data = flat_audios[audio_dur_idx] duration_sec = len(audio_data) / sr # Add quotes around duration to maintain valid JSON format _text = _text.replace("<|audio_duration|>", f'"{duration_sec:.2f}s"', 1) audio_dur_idx += 1 text[_batch_idx] = _text # [Duration Replacement] - Replace <|video_duration|> placeholders with actual values if ( text is not None and videos is not None ): # videos is in [batch] format, each batch is a list of PIL images fps = output_kwargs.get("videos_kwargs", {}).get("fps", 2.0) flat_videos = videos if ( len(videos) > 0 and isinstance(videos[0], list) and len(videos[0]) > 0 and isinstance(videos[0][0], list) ): flat_videos = [v for batch in videos for v in batch] video_dur_idx = 0 for _batch_idx, _text in enumerate(text): while "<|video_duration|>" in _text and video_dur_idx < len(flat_videos): video_frames = flat_videos[video_dur_idx] num_frames = len(video_frames) if isinstance(video_frames, list) else video_frames.shape[0] duration_sec = round(num_frames / fps, 2) _text = _text.replace("<|video_duration|>", f"{duration_sec}s", 1) video_dur_idx += 1 text[_batch_idx] = _text # [Expansion] - Audio (discrete) if ( text is not None and discrete_audio_inputs and self.audio_processor is not None and self.audio_processor.use_discrete_token ): for _batch_idx, (_text_before, _num_discrete_audio_tokens) in enumerate( zip(text, discrete_audio_inputs["num_discrete_audio_tokens"]) ): discrete_audio_block_pattern = ( re.escape(self.audio_processor.discrete_audio_start_token) + r".*?" + re.escape(self.audio_processor.discrete_audio_token) + r".*?" + re.escape(self.audio_processor.discrete_audio_end_token) ) _find_iters = list(re.finditer(discrete_audio_block_pattern, _text_before)) if len(_find_iters) > 0: _text_after = "" _prev_end_idx = 0 for _sample_idx, _discrete_audio_match in enumerate(_find_iters): _inplace_str = self.get_audio_token_replacement( num_audio_tokens=None, num_discrete_audio_tokens=_num_discrete_audio_tokens[_sample_idx], include_boundary_tokens=True, tokenize=False, ) _text_after += _text_before[_prev_end_idx : _discrete_audio_match.start()] _text_after += _inplace_str _prev_end_idx = _discrete_audio_match.end() _text_after += _text_before[_prev_end_idx:] text[_batch_idx] = _text_after # [Expansion] - Audio (continuous) if ( text is not None and audio_inputs and self.audio_processor is not None ): for _batch_idx, (_text_before, _num_audio_tokens) in enumerate( zip(text, audio_inputs["num_audio_tokens"]) ): cont_audio_block_pattern = ( re.escape(self.audio_processor.audio_start_token) + r".*?" + re.escape(self.audio_processor.audio_token) + r".*?" + re.escape(self.audio_processor.audio_end_token) ) _find_iters = list(re.finditer(cont_audio_block_pattern, _text_before)) if len(_find_iters) > 0: _text_after = "" _prev_end_idx = 0 for _sample_idx, _continuous_audio_match in enumerate(_find_iters): _inplace_str = self.get_audio_token_replacement( num_audio_tokens=_num_audio_tokens[_sample_idx], num_discrete_audio_tokens=None, include_boundary_tokens=True, tokenize=False, ) _text_after += _text_before[_prev_end_idx : _continuous_audio_match.start()] _text_after += _inplace_str _prev_end_idx = _continuous_audio_match.end() _text_after += _text_before[_prev_end_idx:] text[_batch_idx] = _text_after # [Expansion] - Image (discrete) if ( text is not None and discrete_image_inputs and self.image_processor is not None and self.image_processor.use_discrete_token ): _item_idx = 0 for _batch_idx, (_text_before, _num_discrete_image_tokens) in enumerate( zip(text, discrete_image_inputs["num_discrete_image_tokens"]) ): discrete_image_block_pattern = ( re.escape(self.image_processor.discrete_image_start_token) + r".*?" + re.escape(self.image_processor.discrete_image_token) + r".*?" + re.escape(self.image_processor.discrete_image_end_token) ) _find_iters = list(re.finditer(discrete_image_block_pattern, _text_before)) if len(_find_iters) > 0: _text_after = "" _prev_end_idx = 0 for _sample_idx, _discrete_image_match in enumerate(_find_iters): _inplace_str = self.get_image_token_replacement( num_image_tokens=None, num_discrete_image_tokens=_num_discrete_image_tokens[_sample_idx], discrete_image_ratio=discrete_image_inputs["discrete_image_ratios"][_item_idx], include_boundary_tokens=True, tokenize=False, ) _text_after += _text_before[_prev_end_idx : _discrete_image_match.start()] _text_after += _inplace_str _prev_end_idx = _discrete_image_match.end() _item_idx += 1 _text_after += _text_before[_prev_end_idx:] text[_batch_idx] = _text_after # [Expansion] - Image (continuous) if ( text is not None and image_inputs and self.image_processor is not None ): for _batch_idx, (_text_before, _num_image_tokens) in enumerate( zip(text, image_inputs["num_image_tokens"]) ): cont_image_block_pattern = ( re.escape(self.image_processor.image_start_token) + r".*?" + re.escape(self.image_processor.image_token) + r".*?" + re.escape(self.image_processor.image_end_token) ) _find_iters = list(re.finditer(cont_image_block_pattern, _text_before)) if len(_find_iters) > 0: _text_after = "" _prev_end_idx = 0 for _sample_idx, _continuous_image_match in enumerate(_find_iters): _inplace_str = self.get_image_token_replacement( num_image_tokens=_num_image_tokens[_sample_idx], num_discrete_image_tokens=None, discrete_image_ratio=None, include_boundary_tokens=True, tokenize=False, ) _text_after += _text_before[_prev_end_idx : _continuous_image_match.start()] _text_after += _inplace_str _prev_end_idx = _continuous_image_match.end() _text_after += _text_before[_prev_end_idx:] text[_batch_idx] = _text_after # [Expansion] - Video if ( text is not None and video_inputs and self.video_processor is not None ): _use_video_audio = getattr(self.video_processor, "use_audio_in_video", False) _num_video_audio_tokens_all = ( video_inputs.get("num_video_audio_tokens") if _use_video_audio else None ) if _use_video_audio: # Template: <|video_start|><|VIDEO_PAD|><|VIDEO_AUDIO_PAD|><|video_end|> video_block_pattern = ( re.escape(self.video_processor.video_start_token) + r".*?" + re.escape(self.video_processor.video_token) + r".*?" + re.escape(self.video_audio_token) + r".*?" + re.escape(self.video_processor.video_end_token) ) else: # Template: <|video_start|><|VIDEO_PAD|><|video_end|> video_block_pattern = ( re.escape(self.video_processor.video_start_token) + r".*?" + re.escape(self.video_processor.video_token) + r".*?" + re.escape(self.video_processor.video_end_token) ) for _batch_idx, (_text_before, _num_video_tokens) in enumerate(zip( text, video_inputs["num_video_tokens"], )): _num_va_per_batch = ( _num_video_audio_tokens_all[_batch_idx] if _num_video_audio_tokens_all is not None else None ) _find_iters = list(re.finditer(video_block_pattern, _text_before)) if len(_find_iters) > 0: _text_after = "" _prev_end_idx = 0 for _sample_idx, _continuous_video_match in enumerate(_find_iters): _num_va = ( _num_va_per_batch[_sample_idx] if _num_va_per_batch is not None else None ) _inplace_str = self.get_video_token_replacement( num_video_tokens=_num_video_tokens[_sample_idx], num_video_audio_tokens=_num_va, include_boundary_tokens=True, tokenize=False, ) _text_after += _text_before[_prev_end_idx:_continuous_video_match.start()] _text_after += _inplace_str _prev_end_idx = _continuous_video_match.end() _text_after += _text_before[_prev_end_idx:] text[_batch_idx] = _text_after return_tensors = output_kwargs["text_kwargs"].pop("return_tensors", None) return_mm_token_type_ids = output_kwargs["text_kwargs"].pop("return_mm_token_type_ids", False) text_inputs = dict() if text is not None: text_inputs = self.tokenizer( text, **output_kwargs["text_kwargs"], return_tensors=None, ) self._check_special_mm_tokens( text, text_inputs, modalities=self.modalities, ) if ( return_mm_token_type_ids and hasattr(self, "image_token_id") ): array_ids = np.array(text_inputs["input_ids"]) mm_token_type_ids = np.zeros_like(text_inputs["input_ids"]) mm_token_type_ids[array_ids == self.image_token_id] = 1 data = { **audio_inputs, **image_inputs, **text_inputs, **video_inputs, } if ( discrete_audio_inputs and self.audio_processor is not None and self.audio_processor.use_discrete_token ): data.update(discrete_audio_inputs) if ( discrete_image_inputs and self.image_processor is not None and self.image_processor.use_discrete_token ): data.update(discrete_image_inputs) model_inputs = BatchFeature(data=data, tensor_type=return_tensors) return model_inputs def get_audio_placeholder( self, tokenize: bool = False, include_boundary_tokens: bool = True, ) -> Union[str, List[int]]: """Build the audio placeholder string (or token ids) for a single audio input. Includes discrete audio tokens (if enabled) followed by continuous audio tokens, each wrapped in their respective start/end boundary tokens. Args: tokenize: If ``True``, return token ids instead of the raw string. Returns: Placeholder string, or list of token ids when ``tokenize`` is ``True``. """ audio_placeholder = "" if self.audio_processor is None: if tokenize: return list() else: return audio_placeholder if self.audio_processor.use_discrete_token: _discrete_audio_placeholder = f'{self.audio_processor.discrete_audio_token}' if include_boundary_tokens: _discrete_audio_placeholder += f'{self.audio_processor.discrete_audio_start_token}{_discrete_audio_placeholder}{self.audio_processor.discrete_audio_end_token}' audio_placeholder += f'{_discrete_audio_placeholder}\n' _continuous_audio_placeholder = f'{self.audio_processor.audio_token}' if include_boundary_tokens: _continuous_audio_placeholder = f'{self.audio_processor.audio_start_token}{_continuous_audio_placeholder}{self.audio_processor.audio_end_token}' audio_placeholder += f'{_continuous_audio_placeholder}' if tokenize: audio_placeholder = self.tokenizer.encode(audio_placeholder) return audio_placeholder def get_audio_token_replacement( self, num_audio_tokens: Optional[Union[int, List[int], Tuple[int, ...], torch.Tensor]] = None, num_discrete_audio_tokens: Optional[Union[int, List[int], Tuple[int, ...], torch.Tensor]] = None, include_boundary_tokens: bool = False, tokenize: bool = False, audio_token: str = None, discrete_audio_token: str = None, return_tuple: Optional[bool] = None, ) -> Union[str, List[int], Tuple[str, str], Tuple[List[int], List[int]]]: """Build a replacement string (or token ids) for audio placeholder tokens. Expands the placeholder into the correct number of continuous and/or discrete audio tokens, optionally wrapped with boundary tokens. Args: num_audio_tokens: Number of continuous audio tokens (or a 1-element list/tensor). ``None`` to skip continuous replacement. num_discrete_audio_tokens: Number of discrete audio tokens (or a 1-element list/tensor). ``None`` to skip discrete replacement. include_boundary_tokens: Whether to wrap with start/end tokens. tokenize: If ``True``, return token ids instead of the raw string. return_tuple: If ``True``, return ``(continuous, discrete)`` tuple. Returns: Replacement string, token id list, or a tuple of two depending on ``tokenize`` and ``return_tuple``. """ if not audio_token: audio_token = self.audio_processor.audio_token if not discrete_audio_token: discrete_audio_token = self.audio_processor.discrete_audio_token continuous_replacement, discrete_replacement = "", "" if self.audio_processor is None: if return_tuple: return (continuous_replacement, discrete_replacement) else: return "" if num_audio_tokens is not None: if ( isinstance(num_audio_tokens, (list, tuple)) or (isinstance(num_audio_tokens, torch.Tensor) and num_audio_tokens.dim() >= 1) ): num_audio_tokens = num_audio_tokens[0] continuous_replacement = audio_token * num_audio_tokens if include_boundary_tokens: continuous_replacement = f"{self.audio_processor.audio_start_token}{continuous_replacement}{self.audio_processor.audio_end_token}" if ( num_discrete_audio_tokens is not None and self.audio_processor.use_discrete_token ): if ( isinstance(num_discrete_audio_tokens, (list, tuple)) or (isinstance(num_discrete_audio_tokens, torch.Tensor) and num_discrete_audio_tokens.dim() >= 1) ): num_discrete_audio_tokens = num_discrete_audio_tokens[0] discrete_replacement = discrete_audio_token * num_discrete_audio_tokens if include_boundary_tokens: discrete_replacement = f"{self.audio_processor.discrete_audio_start_token}{discrete_replacement}{self.audio_processor.discrete_audio_end_token}" discrete_replacement = f'{discrete_replacement}\n' if return_tuple: if tokenize: continuous_replacement = self.tokenizer.encode(continuous_replacement) discrete_replacement = self.tokenizer.encode(discrete_replacement) return (continuous_replacement, discrete_replacement) else: replacement = f'{discrete_replacement}{continuous_replacement}' if tokenize: replacement = self.tokenizer.encode(replacement) return replacement def get_image_placeholder( self, tokenize: bool = False, include_boundary_tokens: bool = True, ) -> Union[str, List[int]]: """Build the image placeholder string (or token ids) for a single image input. Includes discrete image tokens (if enabled) followed by continuous image tokens, each wrapped in their respective start/end boundary tokens. Args: tokenize: If ``True``, return token ids instead of the raw string. Returns: Placeholder string, or list of token ids when ``tokenize`` is ``True``. """ image_placeholder = "" if self.image_processor is None: if tokenize: return list() else: return image_placeholder if self.image_processor.use_discrete_token: _discrete_audio_placeholder = f'{self.image_processor.discrete_image_token}' if include_boundary_tokens: _discrete_audio_placeholder = f'{self.image_processor.discrete_image_start_token}{_discrete_audio_placeholder}{self.image_processor.discrete_image_end_token}' image_placeholder += f'{_discrete_audio_placeholder}\n' _continuous_image_placeholder = f'{self.image_processor.image_token}' if include_boundary_tokens: _continuous_image_placeholder = f'{self.image_processor.image_start_token}{_continuous_image_placeholder}{self.image_processor.image_end_token}' image_placeholder += f'{_continuous_image_placeholder}' if tokenize: image_placeholder = self.tokenizer.encode(image_placeholder) return image_placeholder def get_image_token_replacement( self, num_image_tokens: Optional[Union[int, List[int], Tuple[int, ...], torch.Tensor]] = None, num_discrete_image_tokens: Optional[Union[int, List[int], Tuple[int, ...], torch.Tensor]] = None, discrete_image_ratio: Optional[Union[List[int], Tuple[int, ...], torch.Tensor]] = None, include_boundary_tokens: bool = False, tokenize: bool = False, return_tuple: Optional[bool] = None, ) -> Union[str, List[int], Tuple[str, str], Tuple[List[int], List[int]]]: """Build a replacement string (or token ids) for image placeholder tokens. Expands the placeholder into the correct number of continuous and/or discrete image tokens, optionally prefixed with a ratio token and wrapped with boundary tokens. Args: num_image_tokens: Number of continuous image tokens (or a 1-element list/tensor). ``None`` to skip continuous replacement. num_discrete_image_tokens: Number of discrete image tokens (or a 1-element list/tensor). ``None`` to skip discrete replacement. discrete_image_ratio: Aspect ratio ``[h, w]`` for the discrete image ratio token. ``None`` to omit the ratio prefix. include_boundary_tokens: Whether to wrap with start/end tokens. tokenize: If ``True``, return token ids instead of the raw string. return_tuple: If ``True``, return ``(continuous, discrete)`` tuple. Returns: Replacement string, token id list, or a tuple of two depending on ``tokenize`` and ``return_tuple``. """ continuous_replacement, discrete_replacement = "", "" if self.image_processor is None: if return_tuple: return (continuous_replacement, discrete_replacement) else: return "" if num_image_tokens is not None: if ( isinstance(num_image_tokens, (list, tuple)) or (isinstance(num_image_tokens, torch.Tensor) and num_image_tokens.dim() >= 1) ): num_image_tokens = num_image_tokens[0] continuous_replacement = self.image_processor.image_token * num_image_tokens if include_boundary_tokens: continuous_replacement = f"{self.image_processor.image_start_token}{continuous_replacement}{self.image_processor.image_end_token}" if ( num_discrete_image_tokens is not None and self.image_processor.use_discrete_token ): if ( isinstance(discrete_image_ratio, (list, tuple)) or (isinstance(discrete_image_ratio, torch.Tensor) and discrete_image_ratio.dim() >= 2) ) and len(discrete_image_ratio) == 1: # [[16, 9]], or torch.Tensor([[16, 9]]) discrete_image_ratio = discrete_image_ratio[0] row_str = self.image_processor.discrete_image_token * self.image_processor.discrete_token_size discrete_replacement = row_str * self.image_processor.discrete_token_size if discrete_image_ratio is not None: if isinstance(discrete_image_ratio, (list, tuple)): ratio_key = f"{int(discrete_image_ratio[0])}:{int(discrete_image_ratio[1])}" elif isinstance(discrete_image_ratio, torch.Tensor): ratio_key = f"{discrete_image_ratio[0].item()}:{discrete_image_ratio[1].item()}" discrete_image_ratio_token = self.image_processor.discrete_image_ratio_tokens[ratio_key] discrete_replacement = f"{discrete_image_ratio_token}{discrete_replacement}" if include_boundary_tokens: discrete_replacement = f"{self.image_processor.discrete_image_start_token}{discrete_replacement}{self.image_processor.discrete_image_end_token}" discrete_replacement = f'{discrete_replacement}\n' if return_tuple: if tokenize: continuous_replacement = self.tokenizer.encode(continuous_replacement) discrete_replacement = self.tokenizer.encode(discrete_replacement) return (continuous_replacement, discrete_replacement) else: replacement = f'{discrete_replacement}{continuous_replacement}' if tokenize: replacement = self.tokenizer.encode(replacement) return replacement def get_video_placeholder( self, tokenize: bool = False, include_boundary_tokens: bool = True, ) -> Union[str, List[int]]: """Build the video placeholder string (or token ids) for a single video input. The placeholder consists of continuous video tokens wrapped in start/end boundary tokens. Args: tokenize: If ``True``, return token ids instead of the raw string. Returns: Placeholder string, or list of token ids when ``tokenize`` is ``True``. """ video_placeholder = "" if self.video_processor is None: if tokenize: return list() else: return video_placeholder _continuous_video_placeholder = f'{self.video_processor.video_token}' if include_boundary_tokens: _continuous_video_placeholder = f'{self.video_processor.video_start_token}{_continuous_video_placeholder}{self.video_processor.video_end_token}' video_placeholder += f'{_continuous_video_placeholder}' if tokenize: video_placeholder = self.tokenizer.encode(video_placeholder) return video_placeholder def get_video_audio_placeholder( self, tokenize: bool = False, include_boundary_tokens: bool = False, ) -> Union[str, List[int]]: """Build the video placeholder string (or token ids) for a single video input. The placeholder consists of continuous video tokens wrapped in start/end boundary tokens. Args: tokenize: If ``True``, return token ids instead of the raw string. Returns: Placeholder string, or list of token ids when ``tokenize`` is ``True``. """ video_audio_placeholder = "" if ( self.video_processor is None or self.audio_processor is None ): if tokenize: return list() else: return video_audio_placeholder # do not have start or end token since video_audio_placeholder is embedded in video_placeholder _continuous_video_audio_placeholder = f'{self.video_processor.video_audio_token}' if include_boundary_tokens: _continuous_video_audio_placeholder = f'{self.video_processor.video_audio_start_token}{_continuous_video_audio_placeholder}{self.video_processor.video_audio_end_token}' video_audio_placeholder += f'{_continuous_video_audio_placeholder}' if tokenize: video_audio_placeholder = self.tokenizer.encode(video_audio_placeholder) return video_audio_placeholder def get_video_token_replacement( self, num_video_tokens: Optional[Union[int, List[int], Tuple[int, ...], torch.Tensor]] = None, num_video_audio_tokens: Optional[Union[int, List[int], Tuple[int, ...], torch.Tensor]] = None, include_boundary_tokens: bool = False, tokenize: bool = False, return_tuple: Optional[bool] = None, ) -> Union[str, List[int], Tuple[str, str], Tuple[List[int], List[int]]]: """Build a replacement string (or token ids) for video placeholder tokens. Expands the placeholder into the correct number of continuous video tokens, optionally followed by video-audio tokens, all wrapped with boundary tokens. Args: num_video_tokens: Number of continuous video tokens (or a 1-element list/tensor). ``None`` to skip replacement. num_video_audio_tokens: Number of video-audio tokens to append after the video tokens. ``None`` or ``0`` to omit. include_boundary_tokens: Whether to wrap with start/end tokens. tokenize: If ``True``, return token ids instead of the raw string. return_tuple: If ``True``, return ``(continuous, discrete)`` tuple. Returns: Replacement string, token id list, or a tuple of two depending on ``tokenize`` and ``return_tuple``. """ continuous_replacement, discrete_replacement = "", "" if self.video_processor is None: if return_tuple: return (continuous_replacement, discrete_replacement) else: return "" if num_video_tokens is not None: if ( isinstance(num_video_tokens, (list, tuple)) or (isinstance(num_video_tokens, torch.Tensor) and num_video_tokens.dim() >= 1) ): num_video_tokens = num_video_tokens[0] continuous_replacement = self.video_processor.video_token * int(num_video_tokens) if num_video_audio_tokens is not None: if ( isinstance(num_video_audio_tokens, (list, tuple)) or (isinstance(num_video_audio_tokens, torch.Tensor) and num_video_audio_tokens.dim() >= 1) ): num_video_audio_tokens = num_video_audio_tokens[0] _n_va = int(num_video_audio_tokens) if _n_va > 0: continuous_replacement += self.video_audio_token * _n_va if include_boundary_tokens: continuous_replacement = ( f"{self.video_processor.video_start_token}" f"{continuous_replacement}" f"{self.video_processor.video_end_token}" ) if return_tuple: if tokenize: continuous_replacement = self.tokenizer.encode(continuous_replacement) discrete_replacement = self.tokenizer.encode(discrete_replacement) return (continuous_replacement, discrete_replacement) else: replacement = f'{discrete_replacement}{continuous_replacement}' if tokenize: replacement = self.tokenizer.encode(replacement) return replacement