VideoScore2 — VLM-based video quality scoring.

Uses a Qwen2.5-VL model fine-tuned to score generated videos on three dimensions: visual quality, text-to-video alignment, and physical consistency. Scores are extracted from token logits as upstream's ll_based_soft_score_normed weighting (1-5 scale).

Reference: TIGER-AI-Lab/VideoScore2 (vs2_inference.py).

`fastvideo.eval.metrics.videoscore2.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.videoscore2.metric.BaseMetric.to`

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

Move metric (and its internal models) to device.

`fastvideo.eval.metrics.videoscore2.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.videoscore2.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.videoscore2.metric.BaseMetric.reset`

reset() -> None

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

`fastvideo.eval.metrics.videoscore2.metric.BaseMetric.accumulate`

accumulate(sample: dict) -> None

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

`fastvideo.eval.metrics.videoscore2.metric.BaseMetric.finalize`

finalize() -> MetricResult

Compute the corpus-level result from buffered state.

`fastvideo.eval.metrics.videoscore2.metric.BaseMetric.merge_from`

merge_from(other: BaseMetric) -> None

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

`fastvideo.eval.metrics.videoscore2.metric.register`

register(name: str)

Decorator to register a metric class.

Usage::

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

`fastvideo.eval.metrics.videoscore2.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.videoscore2.metric.VideoScore2Metric`

VideoScore2Metric(model_name: str = 'TIGER-Lab/VideoScore2', infer_fps: float = 2.0, max_tokens: int = 1024, temperature: float = 0.7, do_sample: bool = False)

VideoScore2: VLM-based video quality scoring (3 dimensions).

Requires sample["text_prompt"] for text-to-video alignment. Supports batched generation for GPU efficiency.

Source

@register("videoscore2")
class VideoScore2Metric(BaseMetric):
    """VideoScore2: VLM-based video quality scoring (3 dimensions).

    Requires ``sample["text_prompt"]`` for text-to-video alignment.
    Supports batched generation for GPU efficiency.
    """

    name = "videoscore2"
    requires_reference = False
    higher_is_better = True
    needs_gpu = True
    dependencies = ["transformers", "qwen_vl_utils"]

    def __init__(
        self,
        model_name: str = "TIGER-Lab/VideoScore2",
        infer_fps: float = 2.0,
        max_tokens: int = 1024,
        temperature: float = 0.7,
        # Greedy by default — eval metrics should be deterministic on a
        # given input.  Upstream's vs2_inference.py defaults to
        # do_sample=True with temperature=0.7, but stochastic decoding
        # makes the score irreproducible run-to-run AND can land the
        # chain-of-thought outside the regex anchor (yielding score=0).
        # Pass do_sample=True explicitly if you want BoN/ensembling.
        do_sample: bool = False,
    ) -> None:
        super().__init__()
        self._model_name = model_name
        self.infer_fps = infer_fps
        self.max_tokens = max_tokens
        self.temperature = temperature
        self.do_sample = do_sample
        self._model: Any = None
        self._processor: Any = None
        self._tokenizer: Any = None

    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 AutoProcessor, AutoTokenizer

        # transformers ≥4.45 prefers AutoModelForImageTextToText for
        # vision-language models; AutoModelForVision2Seq is the legacy
        # alias and may go away in a future release.
        try:
            from transformers import AutoModelForImageTextToText as _AutoVisionModel
        except ImportError:
            from transformers import AutoModelForVision2Seq as _AutoVisionModel

        self._model = _AutoVisionModel.from_pretrained(
            self._model_name,
            trust_remote_code=True,
            dtype=torch.bfloat16,
        ).to(self.device)
        self._model.eval()
        self._processor = AutoProcessor.from_pretrained(
            self._model_name,
            trust_remote_code=True,
        )
        self._tokenizer = getattr(self._processor, "tokenizer", None)
        if self._tokenizer is None:
            self._tokenizer = AutoTokenizer.from_pretrained(
                self._model_name,
                trust_remote_code=True,
                use_fast=False,
            )

    def _tensor_to_pil_list(self, video: torch.Tensor) -> list[Image.Image]:
        """Convert (T, C, H, W) float [0,1] tensor to list of PIL images."""
        frames = (video.permute(0, 2, 3, 1).cpu().numpy() * 255).astype(np.uint8)
        return [Image.fromarray(frames[t]) for t in range(frames.shape[0])]

    def _subsample_frames(self,
                          pil_frames: list[Image.Image],
                          source_fps: float | None,
                          max_frames: int = 64,
                          max_resolution: int = 960) -> list[Image.Image]:
        """Subsample to ``infer_fps`` worth of frames, capped at ``max_frames``.

        Mirrors upstream's path-based qwen_vl_utils flow, which samples at
        ``infer_fps``.  Without ``source_fps`` we can't convert frame count
        to duration, so we fall back to ``max_frames`` evenly spaced — the
        pre-fps-derivation behavior.  Pass ``sample["fps"]`` from the
        caller to get the upstream-aligned sampling rate.
        """
        n = len(pil_frames)
        if source_fps is not None and source_fps > 0:
            duration = n / source_fps
            target = max(1, min(max_frames, int(round(duration * self.infer_fps))))
        else:
            target = min(n, max_frames)
        if target < n:
            indices = np.linspace(0, n - 1, target, dtype=int)
            pil_frames = [pil_frames[i] for i in indices]

        w, h = pil_frames[0].size
        if max(w, h) > max_resolution:
            scale = max_resolution / max(w, h)
            new_w, new_h = int(w * scale), int(h * scale)
            pil_frames = [f.resize((new_w, new_h), Image.LANCZOS) for f in pil_frames]

        return pil_frames

    @torch.no_grad()
    def compute(self, sample: dict) -> MetricResult:
        if self._model is None:
            self.setup()

        from qwen_vl_utils import process_vision_info

        video = sample["video"]  # (T, C, H, W)
        text = sample.get("text_prompt", "")
        if isinstance(text, list):
            text = text[0] if text else ""

        pil_frames = self._tensor_to_pil_list(video)
        source_fps = sample.get("fps")
        pil_frames = self._subsample_frames(
            pil_frames,
            source_fps=float(source_fps) if source_fps is not None else None,
        )
        user_prompt = VS2_QUERY_TEMPLATE.substitute(t2v_prompt=text)

        messages = [{
            "role":
            "user",
            "content": [
                {
                    "type": "video",
                    "video": pil_frames,
                    "fps": self.infer_fps
                },
                {
                    "type": "text",
                    "text": user_prompt
                },
            ]
        }]
        chat_text = self._processor.apply_chat_template(
            messages,
            tokenize=False,
            add_generation_prompt=True,
        )
        _, vid_inputs = process_vision_info(messages)

        inputs = self._processor(
            text=[chat_text],
            videos=vid_inputs if vid_inputs else None,
            fps=self.infer_fps,
            padding=True,
            return_tensors="pt",
        ).to(self.device)

        input_len = inputs["input_ids"].shape[1]
        gen_kwargs: dict[str, Any] = dict(
            max_new_tokens=self.max_tokens,
            output_scores=True,
            return_dict_in_generate=True,
            do_sample=self.do_sample,
        )
        if self.do_sample:
            gen_kwargs["temperature"] = self.temperature
        gen_out = self._model.generate(**inputs, **gen_kwargs)

        gen_ids = gen_out.sequences[0, input_len:].tolist()
        pad_id = self._tokenizer.pad_token_id
        if pad_id is not None:
            gen_ids = [t for t in gen_ids if t != pad_id]
        output_text = self._tokenizer.decode(gen_ids, skip_special_tokens=True)

        parsed = _parse_output(
            output_text,
            gen_out.scores,
            self._tokenizer,
            gen_ids,
            seq_idx=0,
        )
        soft_vals = [
            v for v in (parsed["visual_quality"], parsed["text_alignment"], parsed["physical_consistency"])
            if v is not None
        ]
        combined = sum(soft_vals) / len(soft_vals) if soft_vals else 0.0
        return MetricResult(name=self.name, score=combined, details=parsed)

`fastvideo.eval.metrics.videoscore2.metric.VideoScore2Metric.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.videoscore2.metric.VideoScore2Metric.setup`

setup() -> None

Source

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

    # transformers ≥4.45 prefers AutoModelForImageTextToText for
    # vision-language models; AutoModelForVision2Seq is the legacy
    # alias and may go away in a future release.
    try:
        from transformers import AutoModelForImageTextToText as _AutoVisionModel
    except ImportError:
        from transformers import AutoModelForVision2Seq as _AutoVisionModel

    self._model = _AutoVisionModel.from_pretrained(
        self._model_name,
        trust_remote_code=True,
        dtype=torch.bfloat16,
    ).to(self.device)
    self._model.eval()
    self._processor = AutoProcessor.from_pretrained(
        self._model_name,
        trust_remote_code=True,
    )
    self._tokenizer = getattr(self._processor, "tokenizer", None)
    if self._tokenizer is None:
        self._tokenizer = AutoTokenizer.from_pretrained(
            self._model_name,
            trust_remote_code=True,
            use_fast=False,
        )

`fastvideo.eval.metrics.videoscore2.metric.VideoScore2Metric.compute`

compute(sample: dict) -> MetricResult

Source

@torch.no_grad()
def compute(self, sample: dict) -> MetricResult:
    if self._model is None:
        self.setup()

    from qwen_vl_utils import process_vision_info

    video = sample["video"]  # (T, C, H, W)
    text = sample.get("text_prompt", "")
    if isinstance(text, list):
        text = text[0] if text else ""

    pil_frames = self._tensor_to_pil_list(video)
    source_fps = sample.get("fps")
    pil_frames = self._subsample_frames(
        pil_frames,
        source_fps=float(source_fps) if source_fps is not None else None,
    )
    user_prompt = VS2_QUERY_TEMPLATE.substitute(t2v_prompt=text)

    messages = [{
        "role":
        "user",
        "content": [
            {
                "type": "video",
                "video": pil_frames,
                "fps": self.infer_fps
            },
            {
                "type": "text",
                "text": user_prompt
            },
        ]
    }]
    chat_text = self._processor.apply_chat_template(
        messages,
        tokenize=False,
        add_generation_prompt=True,
    )
    _, vid_inputs = process_vision_info(messages)

    inputs = self._processor(
        text=[chat_text],
        videos=vid_inputs if vid_inputs else None,
        fps=self.infer_fps,
        padding=True,
        return_tensors="pt",
    ).to(self.device)

    input_len = inputs["input_ids"].shape[1]
    gen_kwargs: dict[str, Any] = dict(
        max_new_tokens=self.max_tokens,
        output_scores=True,
        return_dict_in_generate=True,
        do_sample=self.do_sample,
    )
    if self.do_sample:
        gen_kwargs["temperature"] = self.temperature
    gen_out = self._model.generate(**inputs, **gen_kwargs)

    gen_ids = gen_out.sequences[0, input_len:].tolist()
    pad_id = self._tokenizer.pad_token_id
    if pad_id is not None:
        gen_ids = [t for t in gen_ids if t != pad_id]
    output_text = self._tokenizer.decode(gen_ids, skip_special_tokens=True)

    parsed = _parse_output(
        output_text,
        gen_out.scores,
        self._tokenizer,
        gen_ids,
        seq_idx=0,
    )
    soft_vals = [
        v for v in (parsed["visual_quality"], parsed["text_alignment"], parsed["physical_consistency"])
        if v is not None
    ]
    combined = sum(soft_vals) / len(soft_vals) if soft_vals else 0.0
    return MetricResult(name=self.name, score=combined, details=parsed)