metric
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:MetricResultper 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:resetto clear buffers and :meth:merge_fromto 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) -> BaseMetricMove metric (and its internal models) to device.
fastvideo.eval.metrics.videoscore2.metric.BaseMetric.setup
setup() -> NoneEagerly 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) -> MetricResultPer-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() -> NoneClear accumulator state at the start of each evaluate() call.
fastvideo.eval.metrics.videoscore2.metric.BaseMetric.accumulate
accumulate(sample: dict) -> NoneBuffer per-sample features for a corpus-level metric.
fastvideo.eval.metrics.videoscore2.metric.BaseMetric.finalize
finalize() -> MetricResultCompute the corpus-level result from buffered state.
fastvideo.eval.metrics.videoscore2.metric.BaseMetric.merge_from
merge_from(other: BaseMetric) -> NoneMulti-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)</code></pre></details>
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,
)</code></pre></details>
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)</code></pre></details>