VLM-based scene matching using AVoCaDO (Qwen2.5-Omni).

Replaces VBench's Tag2Text-based scene metric with a modern VLM caption. The algorithm follows VBench: 1. Caption the video 2. Check if all scene keywords appear in the caption 3. Score = 1.0 if all match, 0.0 otherwise

Unlike VBench (which captions each frame separately with Tag2Text), AVoCaDO captions the entire video in one pass with rich natural language, making the keyword check more robust.

`fastvideo.eval.metrics.vbench.scene.metric.BaseMetric`

BaseMetric()

Abstract base class for all eval metrics.

Two execution shapes:

Per-sample (is_set_metric=False, default) — implement :meth:compute. The Evaluator calls it once per input sample and returns one :class:MetricResult per sample.
Set-vs-set (is_set_metric=True) — implement :meth:accumulate (called once per sample to buffer features) and :meth:finalize (called once after all samples to compute the corpus-level result). Use :meth:reset to clear buffers and :meth:merge_from to fold multi-GPU per-worker state together.

Optionally override :meth:setup to eagerly load models. Metrics that chunk along the time dim for memory hardcode their own chunk size in __init__ (see optical_flow for the canonical example). Eval always processes one video per :meth:Evaluator.evaluate call; compute / accumulate receive a single sample, not a batch.

`fastvideo.eval.metrics.vbench.scene.metric.BaseMetric.to`

to(device: str | torch.device) -> BaseMetric

Move metric (and its internal models) to device.

`fastvideo.eval.metrics.vbench.scene.metric.BaseMetric.setup`

setup() -> None

Eagerly load models. Called once by :class:EvalWorker.

Default is a no-op; metrics with no eager state (pixel math, closed-form ops) inherit this. Override only if your metric needs to load weights.

`fastvideo.eval.metrics.vbench.scene.metric.BaseMetric.compute`

compute(sample: dict) -> MetricResult

Per-sample metrics: compute the score for one sample.

sample["video"] is (T, C, H, W) float in [0, 1]. sample["reference"] (if used) has the same shape. Return self._skip(sample, reason) for missing inputs.

`fastvideo.eval.metrics.vbench.scene.metric.BaseMetric.reset`

reset() -> None

Clear accumulator state at the start of each evaluate() call.

`fastvideo.eval.metrics.vbench.scene.metric.BaseMetric.accumulate`

accumulate(sample: dict) -> None

Buffer per-sample features for a corpus-level metric.

`fastvideo.eval.metrics.vbench.scene.metric.BaseMetric.finalize`

finalize() -> MetricResult

Compute the corpus-level result from buffered state.

`fastvideo.eval.metrics.vbench.scene.metric.BaseMetric.merge_from`

merge_from(other: BaseMetric) -> None

Multi-GPU: fold another worker's accumulator state into this one.

`fastvideo.eval.metrics.vbench.scene.metric.register`

register(name: str)

Decorator to register a metric class.

Usage::

@register("ssim")
class SSIMMetric(BaseMetric):
    ...

`fastvideo.eval.metrics.vbench.scene.metric.MetricResult`

MetricResult(name: str, score: float | None, details: dict[str, Any] = dict())

Standard result container returned by all metrics.

score is None when the metric was skipped (e.g. missing required input). Check details["skipped"] for the reason.

`fastvideo.eval.metrics.vbench.scene.metric.SceneMetric`

SceneMetric(model_path: str = 'AVoCaDO-Captioner/AVoCaDO')

Source

@register("vbench.scene")
class SceneMetric(BaseMetric):

    name = "vbench.scene"
    requires_reference = False
    higher_is_better = True
    needs_gpu = True
    dependencies = ["transformers", "qwen_omni_utils"]
    backbone = "avocado"

    def __init__(self, model_path: str = "AVoCaDO-Captioner/AVoCaDO") -> None:
        super().__init__()
        self._model: Any = None
        self._processor: Any = None
        self._model_path = model_path

    def to(self, device):
        super().to(device)
        if self._model is not None:
            self._model = self._model.to(self.device)
        return self

    def setup(self) -> None:
        if self._model is not None:
            return
        from transformers import Qwen2_5OmniForConditionalGeneration, Qwen2_5OmniProcessor

        # AVoCaDO uses Qwen2.5-Omni — large multimodal model
        envs.setdefault_external("VIDEO_MAX_PIXELS", str(20070400))  # 512*28*28*50

        self._model = Qwen2_5OmniForConditionalGeneration.from_pretrained(
            self._model_path,
            torch_dtype=torch.bfloat16,
            device_map=str(self.device) if self.device.type != "cpu" else None,
        )
        self._model.disable_talker()
        self._model.eval()
        self._processor = Qwen2_5OmniProcessor.from_pretrained(self._model_path)

    def _save_temp_video(self, video: torch.Tensor) -> str:
        """Save (T, C, H, W) float [0,1] tensor as a temp mp4 file."""
        # torchvision expects (T, H, W, C) uint8
        frames = (video * 255).clamp(0, 255).to(torch.uint8).permute(0, 2, 3, 1).cpu()
        # Caller owns the resulting file (it's read by Qwen2.5-Omni and
        # cleaned up at end of compute()), so we just need a unique path.
        fd, path = tempfile.mkstemp(suffix=".mp4")
        os.close(fd)
        torchvision.io.write_video(path, frames, fps=8, video_codec="libx264", options={"crf": "18"})
        return path

    def _generate_caption(self, video_path: str) -> str:
        from qwen_omni_utils import process_mm_info

        conversation = [
            {
                "role":
                "system",
                "content": [{
                    "type":
                    "text",
                    "text": ("You are Qwen, a virtual human developed by the Qwen Team, "
                             "Alibaba Group, capable of perceiving auditory and visual inputs.")
                }],
            },
            {
                "role":
                "user",
                "content": [
                    {
                        "type": "video",
                        "video": video_path,
                        "max_pixels": 401408
                    },
                    {
                        "type": "text",
                        "text": _SCENE_PROMPT
                    },
                ],
            },
        ]

        text = self._processor.apply_chat_template(conversation, add_generation_prompt=True, tokenize=False)
        # AVoCaDO is video+audio; for scene matching we don't need audio,
        # but the model expects it so let it process
        audios, images, videos = process_mm_info(conversation, use_audio_in_video=False)
        inputs = self._processor(
            text=text,
            audio=audios,
            images=images,
            videos=videos,
            return_tensors="pt",
            padding=True,
            use_audio_in_video=False,
        )
        inputs = inputs.to(self._model.device).to(self._model.dtype)

        with torch.no_grad():
            text_ids = self._model.generate(
                **inputs,
                use_audio_in_video=False,
                return_audio=False,
                do_sample=False,
                thinker_max_new_tokens=512,
            )

        decoded = self._processor.batch_decode(text_ids, skip_special_tokens=True,
                                               clean_up_tokenization_spaces=False)[0]
        return decoded.split("\nassistant\n")[-1].lower()

    @torch.no_grad()
    def compute(self, sample: dict) -> MetricResult:
        video = sample["video"]  # (T, C, H, W)
        aux = sample.get("auxiliary_info") or {}
        if "scene" not in aux:
            return self._skip(sample, "missing 'scene' in auxiliary_info")

        scene_keywords = aux["scene"]
        keywords = [k.strip().lower() for k in scene_keywords.split() if k.strip()]

        tmp_path = self._save_temp_video(video)
        try:
            caption = self._generate_caption(tmp_path)
        finally:
            os.unlink(tmp_path)

        matched = [kw for kw in keywords if kw in caption]
        score = 1.0 if len(matched) == len(keywords) else 0.0
        return MetricResult(
            name=self.name,
            score=score,
            details={
                "caption": caption[:500],
                "keywords": keywords,
                "matched": matched,
            },
        )

`fastvideo.eval.metrics.vbench.scene.metric.SceneMetric.to`

to(device)

Source

def to(self, device):
    super().to(device)
    if self._model is not None:
        self._model = self._model.to(self.device)
    return self

`fastvideo.eval.metrics.vbench.scene.metric.SceneMetric.setup`

setup() -> None

Source

def setup(self) -> None:
    if self._model is not None:
        return
    from transformers import Qwen2_5OmniForConditionalGeneration, Qwen2_5OmniProcessor

    # AVoCaDO uses Qwen2.5-Omni — large multimodal model
    envs.setdefault_external("VIDEO_MAX_PIXELS", str(20070400))  # 512*28*28*50

    self._model = Qwen2_5OmniForConditionalGeneration.from_pretrained(
        self._model_path,
        torch_dtype=torch.bfloat16,
        device_map=str(self.device) if self.device.type != "cpu" else None,
    )
    self._model.disable_talker()
    self._model.eval()
    self._processor = Qwen2_5OmniProcessor.from_pretrained(self._model_path)

`fastvideo.eval.metrics.vbench.scene.metric.SceneMetric.compute`

compute(sample: dict) -> MetricResult

Source

@torch.no_grad()
def compute(self, sample: dict) -> MetricResult:
    video = sample["video"]  # (T, C, H, W)
    aux = sample.get("auxiliary_info") or {}
    if "scene" not in aux:
        return self._skip(sample, "missing 'scene' in auxiliary_info")

    scene_keywords = aux["scene"]
    keywords = [k.strip().lower() for k in scene_keywords.split() if k.strip()]

    tmp_path = self._save_temp_video(video)
    try:
        caption = self._generate_caption(tmp_path)
    finally:
        os.unlink(tmp_path)

    matched = [kw for kw in keywords if kw in caption]
    score = 1.0 if len(matched) == len(keywords) else 0.0
    return MetricResult(
        name=self.name,
        score=score,
        details={
            "caption": caption[:500],
            "keywords": keywords,
            "matched": matched,
        },
    )