python/jina_omni_utils.py

from __future__ import annotations

import json
import sys
from pathlib import Path
from typing import Iterable

import numpy as np
import torch
import torch.nn.functional as F
from PIL import Image
from safetensors import safe_open
from transformers import AutoProcessor, AutoTokenizer, WhisperFeatureExtractor

IMAGE_SUFFIXES = {".jpg", ".jpeg", ".png", ".bmp", ".webp", ".gif", ".tif", ".tiff"}
VIDEO_SUFFIXES = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".flv", ".wmv"}
AUDIO_SUFFIXES = {".wav", ".mp3", ".flac", ".ogg", ".m4a", ".opus", ".aac"}
TASK_NAMES = ("retrieval", "text-matching", "clustering", "classification")
PROMPT_PREFIX = {
    "query": "Query: ",
    "document": "Document: ",
}


def repo_root() -> Path:
    return Path(__file__).resolve().parent.parent


def resolve_device(device: str | None = None) -> torch.device:
    if device:
        return torch.device(device)
    if torch.cuda.is_available():
        return torch.device("cuda")
    return torch.device("cpu")


def torch_dtype_from_name(name: str | None) -> torch.dtype:
    if not name:
        return torch.bfloat16
    lowered = name.lower()
    mapping = {
        "bf16": torch.bfloat16,
        "bfloat16": torch.bfloat16,
        "fp16": torch.float16,
        "float16": torch.float16,
        "fp32": torch.float32,
        "float32": torch.float32,
    }
    if lowered not in mapping:
        raise ValueError(f"Unsupported dtype: {name}")
    return mapping[lowered]


def _ensure_model_repo_on_sys_path(model_dir: Path) -> None:
    model_dir = model_dir.resolve()
    model_dir_str = str(model_dir)
    if model_dir_str not in sys.path:
        sys.path.insert(0, model_dir_str)


def load_model_classes(model_dir: str | Path):
    model_dir = Path(model_dir)
    _ensure_model_repo_on_sys_path(model_dir)
    from modeling_jina_embeddings_v5_omni import (
        JinaEmbeddingsV5OmniBase,
        JinaEmbeddingsV5OmniConfig,
    )

    return JinaEmbeddingsV5OmniConfig, JinaEmbeddingsV5OmniBase


def merge_task_lora_inplace(model: torch.nn.Module, model_dir: str | Path, task: str) -> None:
    model_dir = Path(model_dir)
    if task not in TASK_NAMES:
        raise ValueError(f"Unsupported task: {task}")

    adapter_dir = model_dir / "adapters" / task
    with open(adapter_dir / "adapter_config.json", "r", encoding="utf-8") as handle:
        adapter_cfg = json.load(handle)

    scale = adapter_cfg["lora_alpha"] / adapter_cfg["r"]
    with safe_open(str(adapter_dir / "adapter_model.safetensors"), framework="pt") as handle:
        adapter = {key: handle.get_tensor(key) for key in handle.keys()}

    with torch.no_grad():
        for a_key in list(adapter.keys()):
            if "lora_A" not in a_key:
                continue
            b_key = a_key.replace("lora_A", "lora_B")
            if b_key not in adapter:
                continue
            target_name = a_key.replace("base_model.model.", "").replace(".lora_A.weight", ".weight")
            parts = target_name.split(".")
            module = model
            for part in parts[:-2]:
                module = module[int(part)] if part.isdigit() else getattr(module, part)
            proj = getattr(module, parts[-2])
            delta = (adapter[b_key].float() @ adapter[a_key].float()) * scale
            proj.weight.data.add_(delta.to(device=proj.weight.device, dtype=proj.weight.dtype))


def load_base_model(
    model_dir: str | Path,
    modality: str,
    task: str,
    dtype_name: str = "bfloat16",
    device: str | None = None,
    merge_task_lora: bool = True,
    attn_implementation: str | None = None,
):
    model_dir = Path(model_dir)
    config_cls, base_cls = load_model_classes(model_dir)
    config = config_cls.from_pretrained(str(model_dir))
    config.modality = modality
    if attn_implementation is not None:
        config._attn_implementation = attn_implementation
        for attr_name in ("vision_config", "text_config", "audio_config"):
            sub_cfg = getattr(config, attr_name, None)
            if sub_cfg is not None:
                setattr(sub_cfg, "_attn_implementation", attn_implementation)
    dtype = torch_dtype_from_name(dtype_name)
    model = base_cls.from_pretrained(str(model_dir), config=config, torch_dtype=dtype)
    if merge_task_lora:
        merge_task_lora_inplace(model, model_dir, task)
    model.set_task(task)
    resolved_device = resolve_device(device)
    model.to(device=resolved_device)
    if dtype == torch.float32:
        model.float()
    else:
        model.to(dtype=dtype)
    model.eval()
    return model


def load_processor(model_dir: str | Path, pixel_budget: int | None = None):
    kwargs = {"trust_remote_code": True}
    if pixel_budget is not None:
        kwargs["min_pixels"] = pixel_budget
        kwargs["max_pixels"] = pixel_budget
    return AutoProcessor.from_pretrained(str(model_dir), **kwargs)


def load_tokenizer(model_dir: str | Path):
    return AutoTokenizer.from_pretrained(str(model_dir), trust_remote_code=True)


def prompt_prefix(prompt_name: str) -> str:
    if prompt_name not in PROMPT_PREFIX:
        raise ValueError(f"Unsupported prompt name: {prompt_name}")
    return PROMPT_PREFIX[prompt_name]


def get_image_token_id(config) -> int:
    token_id = getattr(config, "image_token_id", None)
    if token_id is None:
        token_id = getattr(config, "image_token_index", None)
    if token_id is None:
        raise AttributeError("Config does not define image_token_id or image_token_index")
    return int(token_id)


def get_video_token_id(config) -> int:
    token_id = getattr(config, "video_token_id", None)
    if token_id is not None:
        return int(token_id)
    return get_image_token_id(config)


def build_image_prompt(processor, prompt_name: str = "query") -> str:
    image_token = getattr(processor, "image_token", "<|image_pad|>")
    text = f"{prompt_prefix(prompt_name)}<|vision_start|>{image_token}<|vision_end|>"
    return processor.apply_chat_template(
        [{"role": "user", "content": text}],
        tokenize=False,
        add_generation_prompt=False,
    )


def build_video_prompt(processor, prompt_name: str = "query") -> str:
    video_token = getattr(processor, "video_token", None) or getattr(processor, "image_token", "<|image_pad|>")
    text = f"{prompt_prefix(prompt_name)}<|vision_start|>{video_token}<|vision_end|>"
    return processor.apply_chat_template(
        [{"role": "user", "content": text}],
        tokenize=False,
        add_generation_prompt=False,
    )


def resize_image_object(image: Image.Image, width: int, height: int) -> Image.Image:
    resampling = getattr(Image, "Resampling", Image).BICUBIC
    return image.convert("RGB").resize((width, height), resampling)


def open_and_resize_image(image_path: str | Path, width: int, height: int) -> Image.Image:
    image = Image.open(image_path)
    return resize_image_object(image, width=width, height=height)


def is_video_path(path: str | Path) -> bool:
    return Path(path).suffix.lower() in VIDEO_SUFFIXES


def is_audio_path(path: str | Path) -> bool:
    return Path(path).suffix.lower() in AUDIO_SUFFIXES


def prepare_fixed_image_inputs(
    model_dir: str | Path,
    image_path: str | Path,
    width: int,
    height: int,
    prompt_name: str = "query",
    processor=None,
):
    pixel_budget = width * height
    processor = processor or load_processor(model_dir, pixel_budget=pixel_budget)
    prompt = build_image_prompt(processor, prompt_name=prompt_name)
    image = open_and_resize_image(image_path, width=width, height=height)
    inputs = processor(images=image, text=prompt, return_tensors="pt", truncation=False)
    return inputs, {"pixel_budget": pixel_budget, "prompt": prompt}


def prepare_text_inputs(
    model_dir: str | Path,
    text: str,
    prompt_name: str = "query",
    max_length: int = 32768,
):
    tokenizer = load_tokenizer(model_dir)
    encoded = tokenizer(
        [f"{prompt_prefix(prompt_name)}{text}"],
        return_tensors="pt",
        padding=True,
        truncation=True,
        max_length=max_length,
    )
    return encoded


def _extract_audio_from_video(video_path: str | Path, target_sr: int = 16000) -> np.ndarray:
    import av
    from av.audio.resampler import AudioResampler

    container = av.open(str(video_path))
    resampler = AudioResampler(format="flt", layout="mono", rate=target_sr)
    samples: list[np.ndarray] = []
    try:
        for frame in container.decode(audio=0):
            for resampled in resampler.resample(frame):
                samples.append(resampled.to_ndarray().flatten())
        for resampled in resampler.resample(None):
            samples.append(resampled.to_ndarray().flatten())
    finally:
        container.close()
    if not samples:
        return np.zeros((0,), dtype=np.float32)
    return np.concatenate(samples).astype(np.float32)


def load_audio_array(audio_path: str | Path, target_sr: int = 16000) -> np.ndarray:
    audio_path = Path(audio_path)
    if is_video_path(audio_path):
        audio = _extract_audio_from_video(audio_path, target_sr=target_sr)
    else:
        import librosa

        audio, _ = librosa.load(str(audio_path), sr=target_sr, mono=True)
        audio = audio.astype(np.float32)
    if audio.size == 0:
        raise ValueError(f"No audio samples decoded from: {audio_path}")
    peak = float(np.max(np.abs(audio)))
    if peak > 1.0:
        audio = audio / peak
    return audio.astype(np.float32)


def extract_uniform_video_frames(
    video_path: str | Path,
    *,
    num_frames: int,
    width: int,
    height: int,
) -> tuple[list[Image.Image], list[int], int]:
    if num_frames <= 0:
        raise ValueError(f"num_frames must be positive, got {num_frames}")

    video_path = Path(video_path)
    if video_path.is_dir():
        frame_paths = sorted(
            path for path in video_path.iterdir()
            if path.is_file() and path.suffix.lower() in IMAGE_SUFFIXES
        )
        frames = [Image.open(path).convert("RGB") for path in frame_paths]
    else:
        import av

        container = av.open(str(video_path))
        try:
            frames = [frame.to_image().convert("RGB") for frame in container.decode(video=0)]
        finally:
            container.close()

    if not frames:
        raise ValueError(f"No video frames decoded from: {video_path}")

    if len(frames) <= num_frames:
        indices = list(range(len(frames)))
    else:
        indices = np.linspace(0, len(frames) - 1, num_frames, dtype=int).tolist()
    selected = [resize_image_object(frames[index], width=width, height=height) for index in indices]
    return selected, indices, len(frames)


def expand_multimodal_placeholder_ids(
    tokenizer,
    prompt: str,
    *,
    placeholder_token_id: int,
    placeholder_count: int,
) -> list[int]:
    encoded = tokenizer([prompt], return_tensors="pt", padding=False, truncation=False)
    base_ids = encoded["input_ids"][0].tolist()
    current_count = sum(1 for token_id in base_ids if token_id == placeholder_token_id)
    if current_count == placeholder_count:
        return base_ids
    if current_count != 1:
        raise ValueError(
            f"Unexpected placeholder token count in prompt: got {current_count}, expected 1 or {placeholder_count}"
        )

    expanded: list[int] = []
    replaced = False
    for token_id in base_ids:
        if token_id == placeholder_token_id and not replaced:
            expanded.extend([placeholder_token_id] * placeholder_count)
            replaced = True
            continue
        expanded.append(token_id)
    return expanded


def build_audio_prompt(processor, tokenizer, config, prompt_name: str = "query", n_audio_tokens: int = 750) -> str:
    start = tokenizer.convert_ids_to_tokens(config.audio_start_token_id)
    token = tokenizer.convert_ids_to_tokens(config.audio_token_id)
    end = tokenizer.convert_ids_to_tokens(config.audio_end_token_id)
    text = f"{prompt_prefix(prompt_name)}{start}{token * n_audio_tokens}{end}"
    return processor.apply_chat_template(
        [{"role": "user", "content": text}],
        tokenize=False,
        add_generation_prompt=False,
    )


def prepare_fixed_audio_inputs(
    model_dir: str | Path,
    audio_path: str | Path,
    prompt_name: str = "query",
    max_frames: int = 3000,
    processor=None,
    tokenizer=None,
    config=None,
):
    model_dir = Path(model_dir)
    processor = processor or load_processor(model_dir)
    tokenizer = tokenizer or load_tokenizer(model_dir)
    if config is None:
        config_cls, _ = load_model_classes(model_dir)
        config = config_cls.from_pretrained(str(model_dir))

    audio = load_audio_array(audio_path)
    feat_ext = WhisperFeatureExtractor(feature_size=128)
    max_samples = int(max_frames) * int(feat_ext.hop_length)
    audio_inputs = feat_ext(
        audio,
        sampling_rate=16000,
        return_tensors="pt",
        padding="max_length",
        max_length=max_samples,
        return_attention_mask=True,
    )
    input_features = audio_inputs["input_features"]
    feature_attention_mask = audio_inputs["attention_mask"]
    if input_features.shape[-1] != max_frames:
        raise ValueError(f"Expected {max_frames} mel frames, got {input_features.shape[-1]}")

    n_audio_tokens = input_features.shape[-1] // 4
    prompt = build_audio_prompt(processor, tokenizer, config, prompt_name=prompt_name, n_audio_tokens=n_audio_tokens)
    text_inputs = processor(text=[prompt], return_tensors="pt", padding=False, truncation=False)
    text_inputs["input_features"] = input_features
    text_inputs["feature_attention_mask"] = feature_attention_mask
    return text_inputs, {
        "prompt": prompt,
        "n_audio_tokens": n_audio_tokens,
        "feature_frames": int(input_features.shape[-1]),
        "real_frames": int(feature_attention_mask.sum().item()),
    }


def prepare_fixed_video_inputs(
    model_dir: str | Path,
    video_path: str | Path,
    *,
    num_frames: int,
    width: int,
    height: int,
    prompt_name: str = "query",
    max_prefill_tokens: int | None = None,
    processor=None,
    tokenizer=None,
    config=None,
):
    model_dir = Path(model_dir)
    processor = processor or load_processor(model_dir, pixel_budget=width * height)
    tokenizer = tokenizer or load_tokenizer(model_dir)
    if config is None:
        config_cls, _ = load_model_classes(model_dir)
        config = config_cls.from_pretrained(str(model_dir))

    frames, sampled_indices, total_frames = extract_uniform_video_frames(
        video_path,
        num_frames=num_frames,
        width=width,
        height=height,
    )
    image_prompt = build_image_prompt(processor, prompt_name=prompt_name)
    video_prompt = build_video_prompt(processor, prompt_name=prompt_name)

    per_frame_pixel_values: list[torch.Tensor] = []
    frame_grid = None
    for frame in frames:
        frame_inputs = processor(images=frame, text=image_prompt, return_tensors="pt", truncation=False)
        pixel_values = frame_inputs["pixel_values"].cpu()
        image_grid_thw = frame_inputs["image_grid_thw"].cpu()
        if frame_grid is None:
            frame_grid = image_grid_thw[0]
        elif not torch.equal(frame_grid, image_grid_thw[0]):
            raise ValueError("All selected video frames must produce the same image_grid_thw for static-shape export")
        per_frame_pixel_values.append(pixel_values)

    if frame_grid is None or not per_frame_pixel_values:
        raise ValueError(f"No usable video frames after preprocessing: {video_path}")

    spatial_merge_size = int(getattr(getattr(config, "vision_config", None), "spatial_merge_size", 2) or 2)
    tokens_per_frame = int(frame_grid.prod().item()) // (spatial_merge_size ** 2)
    video_token_id = get_video_token_id(config)
    base_ids = tokenizer([video_prompt], return_tensors="pt", padding=False, truncation=False)["input_ids"][0].tolist()
    base_placeholder_count = sum(1 for token_id in base_ids if token_id == video_token_id)
    prompt_overhead = len(base_ids) - base_placeholder_count

    requested_num_frames = len(per_frame_pixel_values)
    max_frames_by_budget = None
    if max_prefill_tokens is not None:
        max_frames_by_budget = max(1, (int(max_prefill_tokens) - prompt_overhead) // tokens_per_frame)
        if requested_num_frames > max_frames_by_budget:
            per_frame_pixel_values = per_frame_pixel_values[:max_frames_by_budget]
            sampled_indices = sampled_indices[:max_frames_by_budget]

    used_num_frames = len(per_frame_pixel_values)
    total_video_tokens = used_num_frames * tokens_per_frame
    input_ids = torch.tensor(
        [
            expand_multimodal_placeholder_ids(
                tokenizer,
                video_prompt,
                placeholder_token_id=video_token_id,
                placeholder_count=total_video_tokens,
            )
        ],
        dtype=torch.long,
    )
    attention_mask = torch.ones_like(input_ids)
    pixel_values_frames = torch.stack([value for value in per_frame_pixel_values], dim=0)
    pixel_values_videos = torch.cat([value for value in per_frame_pixel_values], dim=0)
    video_grid_thw = torch.tensor(
        [[used_num_frames, int(frame_grid[1].item()), int(frame_grid[2].item())]],
        dtype=frame_grid.dtype,
    )

    return {
        "input_ids": input_ids,
        "attention_mask": attention_mask,
        "image_grid_thw": frame_grid.unsqueeze(0),
        "pixel_values_frames": pixel_values_frames,
        "pixel_values_videos": pixel_values_videos,
        "video_grid_thw": video_grid_thw,
    }, {
        "prompt": video_prompt,
        "requested_num_frames": int(num_frames),
        "used_num_frames": used_num_frames,
        "sampled_frame_indices": sampled_indices,
        "decoded_total_frames": int(total_frames),
        "tokens_per_frame": tokens_per_frame,
        "total_video_tokens": total_video_tokens,
        "prompt_overhead": prompt_overhead,
        "max_prefill_tokens": int(max_prefill_tokens) if max_prefill_tokens is not None else None,
        "max_frames_by_budget": int(max_frames_by_budget) if max_frames_by_budget is not None else None,
    }


def move_tensor_inputs(inputs: dict, device: torch.device) -> dict:
    return {key: value.to(device) if torch.is_tensor(value) else value for key, value in inputs.items()}


def last_token_embedding(last_hidden_state: torch.Tensor, attention_mask: torch.Tensor, truncate_dim: int | None = None):
    seq_lens = attention_mask.sum(dim=1) - 1
    pooled = last_hidden_state[torch.arange(last_hidden_state.shape[0], device=last_hidden_state.device), seq_lens]
    if truncate_dim is not None:
        pooled = pooled[:, :truncate_dim]
    return F.normalize(pooled, dim=-1)


def build_bidi_attention_mask(
    inputs_embeds: torch.Tensor,
    attention_mask: torch.Tensor | None,
) -> torch.Tensor | None:
    if attention_mask is None or attention_mask.dim() != 2:
        return attention_mask
    dtype = inputs_embeds.dtype
    seq_len = inputs_embeds.shape[1]
    bidi = attention_mask[:, None, None, :].to(dtype=dtype)
    bidi = (1.0 - bidi) * torch.finfo(dtype).min
    return bidi.expand(-1, -1, seq_len, -1)


def run_embedding_with_replaced_multimodal_tokens(
    model: torch.nn.Module,
    inputs: dict,
    replacement_tokens: np.ndarray | torch.Tensor,
    modality: str,
    truncate_dim: int | None = None,
) -> torch.Tensor:
    if modality not in {"vision", "audio", "video"}:
        raise ValueError(f"Unsupported modality: {modality}")

    device = next(model.parameters()).device
    torch_inputs = move_tensor_inputs(inputs, device)
    input_ids = torch_inputs["input_ids"]
    attention_mask = torch_inputs.get("attention_mask")
    inputs_embeds = model.get_input_embeddings()(input_ids)

    if isinstance(replacement_tokens, np.ndarray):
        replacement_tokens = torch.from_numpy(replacement_tokens)
    replacement_tokens = replacement_tokens.to(device=device, dtype=inputs_embeds.dtype)
    flat_tokens = replacement_tokens.reshape(-1, replacement_tokens.shape[-1])

    if modality == "vision":
        token_id = get_image_token_id(model.config)
    elif modality == "audio":
        token_id = model.config.audio_token_id
    else:
        token_id = get_video_token_id(model.config)
    token_count = int((input_ids == token_id).sum().item())
    if token_count != flat_tokens.shape[0]:
        raise ValueError(
            f"Replacement token count mismatch for {modality}: "
            f"mask expects {token_count}, got {flat_tokens.shape[0]}"
        )

    mask = (input_ids == token_id).unsqueeze(-1).expand_as(inputs_embeds)
    inputs_embeds = inputs_embeds.masked_scatter(mask, flat_tokens)
    language_model_attention_mask = build_bidi_attention_mask(inputs_embeds, attention_mask)

    with torch.no_grad():
        out = model.language_model(
            inputs_embeds=inputs_embeds,
            attention_mask=language_model_attention_mask,
        )
    return last_token_embedding(out[0], attention_mask, truncate_dim=truncate_dim)


def run_text_embedding(
    model_dir: str | Path,
    text: str,
    task: str = "retrieval",
    prompt_name: str = "query",
    dtype_name: str = "bfloat16",
    device: str | None = None,
    truncate_dim: int | None = None,
):
    model = load_base_model(model_dir, modality="text", task=task, dtype_name=dtype_name, device=device)
    inputs = prepare_text_inputs(model_dir, text=text, prompt_name=prompt_name)
    inputs = move_tensor_inputs(inputs, next(model.parameters()).device)
    with torch.no_grad():
        outputs = model(**inputs)
    return last_token_embedding(outputs.last_hidden_state, inputs["attention_mask"], truncate_dim=truncate_dim)


def run_image_embedding(
    model_dir: str | Path,
    image_path: str | Path,
    width: int,
    height: int,
    task: str = "retrieval",
    prompt_name: str = "query",
    dtype_name: str = "bfloat16",
    device: str | None = None,
    truncate_dim: int | None = None,
):
    model = load_base_model(model_dir, modality="vision", task=task, dtype_name=dtype_name, device=device)
    inputs, _ = prepare_fixed_image_inputs(model_dir, image_path=image_path, width=width, height=height, prompt_name=prompt_name)
    inputs = move_tensor_inputs(inputs, next(model.parameters()).device)
    with torch.no_grad():
        outputs = model(**inputs)
    return last_token_embedding(outputs.last_hidden_state, inputs["attention_mask"], truncate_dim=truncate_dim)


def run_audio_embedding(
    model_dir: str | Path,
    audio_path: str | Path,
    task: str = "retrieval",
    prompt_name: str = "query",
    dtype_name: str = "bfloat16",
    device: str | None = None,
    truncate_dim: int | None = None,
    max_frames: int = 3000,
):
    model_dir = Path(model_dir)
    model = load_base_model(model_dir, modality="audio", task=task, dtype_name=dtype_name, device=device)
    processor = load_processor(model_dir)
    tokenizer = load_tokenizer(model_dir)
    inputs, _ = prepare_fixed_audio_inputs(
        model_dir,
        audio_path=audio_path,
        prompt_name=prompt_name,
        max_frames=max_frames,
        processor=processor,
        tokenizer=tokenizer,
        config=model.config,
    )
    inputs = move_tensor_inputs(inputs, next(model.parameters()).device)
    del inputs["feature_attention_mask"]
    with torch.no_grad():
        outputs = model(**inputs)
    return last_token_embedding(outputs.last_hidden_state, inputs["attention_mask"], truncate_dim=truncate_dim)


def get_video_tokens_from_processed(model, pixel_values_videos: torch.Tensor, video_grid_thw: torch.Tensor) -> torch.Tensor:
    with torch.no_grad():
        features = model.get_image_features(pixel_values_videos, video_grid_thw)
    if len(features) != 1:
        raise ValueError(f"Expected one video sample, got {len(features)}")
    return features[0].unsqueeze(0)


def run_video_embedding(
    model_dir: str | Path,
    video_path: str | Path,
    *,
    num_frames: int,
    width: int,
    height: int,
    task: str = "retrieval",
    prompt_name: str = "query",
    dtype_name: str = "bfloat16",
    device: str | None = None,
    truncate_dim: int | None = None,
    max_prefill_tokens: int | None = None,
):
    model_dir = Path(model_dir)
    model = load_base_model(model_dir, modality="vision", task=task, dtype_name=dtype_name, device=device)
    processor = load_processor(model_dir, pixel_budget=width * height)
    tokenizer = load_tokenizer(model_dir)
    inputs, _ = prepare_fixed_video_inputs(
        model_dir,
        video_path=video_path,
        num_frames=num_frames,
        width=width,
        height=height,
        prompt_name=prompt_name,
        max_prefill_tokens=max_prefill_tokens,
        processor=processor,
        tokenizer=tokenizer,
        config=model.config,
    )
    inputs = move_tensor_inputs(inputs, next(model.parameters()).device)
    replacement_tokens = get_video_tokens_from_processed(
        model,
        inputs["pixel_values_videos"],
        inputs["video_grid_thw"],
    )
    lm_inputs = {
        "input_ids": inputs["input_ids"],
        "attention_mask": inputs["attention_mask"],
    }
    return run_embedding_with_replaced_multimodal_tokens(
        model,
        lm_inputs,
        replacement_tokens,
        modality="video",
        truncate_dim=truncate_dim,
    )


def get_vision_tokens_from_processed(model, pixel_values: torch.Tensor, image_grid_thw: torch.Tensor) -> torch.Tensor:
    with torch.no_grad():
        features = model.get_image_features(pixel_values, image_grid_thw)
    if len(features) != 1:
        raise ValueError(f"Expected one image, got {len(features)}")
    return features[0].unsqueeze(0)


def to_numpy(tensor: torch.Tensor) -> np.ndarray:
    return tensor.detach().cpu().float().numpy()


def list_image_files(dataset_dir: str | Path) -> list[Path]:
    dataset_dir = Path(dataset_dir)
    return sorted(
        path for path in dataset_dir.rglob("*")
        if path.is_file() and path.suffix.lower() in IMAGE_SUFFIXES
    )


def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
    lhs = a.reshape(-1).astype(np.float64)
    rhs = b.reshape(-1).astype(np.float64)
    denom = (np.linalg.norm(lhs) * np.linalg.norm(rhs)) + 1e-12
    return float(np.dot(lhs, rhs) / denom)


def format_vector_preview(array: np.ndarray, count: int = 8) -> list[float]:
    flat = array.reshape(-1)
    return [float(v) for v in flat[:count]]


def ensure_parent(path: str | Path) -> Path:
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    return path