Skip to content
FastVideo
Esc
↑↓navigate↵open⌘Jpreview
On this page

metric

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,
        },
    )</code></pre></details>


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(&quot;VIDEO_MAX_PIXELS&quot;, 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 != &quot;cpu&quot; else None,
)
self._model.disable_talker()
self._model.eval()
self._processor = Qwen2_5OmniProcessor.from_pretrained(self._model_path)</code></pre></details>


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&#91;&quot;scene&quot;&#93;
keywords = &#91;k.strip().lower() for k in scene_keywords.split() if k.strip()&#93;

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

matched = &#91;kw for kw in keywords if kw in caption&#93;
score = 1.0 if len(matched) == len(keywords) else 0.0
return MetricResult(
    name=self.name,
    score=score,
    details=&#123;
        &quot;caption&quot;: caption&#91;:500&#93;,
        &quot;keywords&quot;: keywords,
        &quot;matched&quot;: matched,
    &#125;,
)</code></pre></details>

Was this page helpful?