短视频标注

视频索引需要可检索的视觉标签。读取旧教程的饺子短视频，抽取中间帧，复用酱汁颜色问题，生成包含片段 ID、颜色标签、置信度、帧索引、时间戳与总解码帧数的 JSON 标注。

本例只输出单帧视觉属性，不推断整段视频的动作，也不把标签赋给所有时刻。直接视频及原动作问题保留在本地推理教程中。解码器最多计数 1800 帧、限制单帧为 4K、分两遍解码，只保留目标图像；提交前将该帧转换为最长边不超过 512 像素的 JPEG，后端仅收到一张图片。

选中概率至少为 0.8 且前两名概率差至少为 0.2 时才输出标签，否则输出空标签并进入 manual_review。置信度可用于安排审核优先级，但不是经过校准的正确率。需要覆盖整段视频时，应抽取多个时间片段并验证时间覆盖率。

安装与输入

Python 3.11+。先下载小型源文件包并解压，然后安装所需依赖：

Bash

curl -fLO https://valen-team.github.io/cookbook/valen-cookbook.zip
python -m zipfile -e valen-cookbook.zip .
cd valen-cookbook
python -m pip install -r requirements.txt

明确执行后才获取以下小素材（图片约 58 KB，视频约 395 KB）；也可将有权使用的同名文件放在 --asset-dir 指定目录。

Bash

curl -fL https://raw.githubusercontent.com/Valen-Team/Valen/main/assets/cookbook/momos.mp4 -o momos.mp4

samples.json

{
  "id": "clip-001",
  "path": "momos.mp4",
  "question": {
    "name": "sauce_color",
    "type": "choice",
    "instructions": "What is the color of the sauce in the foil container on the left?",
    "choices": [
      {
        "value": "brown",
        "description": "Brown"
      },
      {
        "value": "white",
        "description": "White"
      },
      {
        "value": "red",
        "description": "Red"
      },
      {
        "value": "orange",
        "description": "Orange"
      }
    ]
  },
  "action_question": {
    "name": "action",
    "type": "choice",
    "instructions": "What action does the person perform with one of the dumplings?",
    "choices": [
      {
        "value": "A",
        "description": "Breaks it open with their hands"
      },
      {
        "value": "B",
        "description": "Places it back in the container"
      },
      {
        "value": "C",
        "description": "Dips it in sauce"
      },
      {
        "value": "D",
        "description": "Tosses it in the air"
      }
    ]
  }
}

运行

Bash

python video_tagging.py --backend remote --model valen-2b --asset-dir . --output result.json
# An existing local HTTP deployment instead:
python video_tagging.py --backend remote --base-url http://127.0.0.1:8000/v1 --model valen --asset-dir . --output result.json

本地 ONNX 也可直接运行：把参数改为 --backend onnx --model /path/to/Valen-0.8B-ONNX，并先安装 valen-sdk[onnx,video]==0.2.0。GGUF 服务通过 --backend remote 接入；不能把 GGUF 文件传给 Valen.local。

完整可执行代码

video_tagging.py

"""Create a reviewable visual tag from one sampled clip frame. Apache-2.0."""
from common import choice_status, client, emit, image_input, middle_frame, parser, request, samples


def run(predict, sample, asset_dir):
    payload, sampling = middle_frame(asset_dir / sample["path"])
    answer = request(predict, image_input(payload), [sample["question"]])["sauce_color"]
    confidence, certain = choice_status(answer)
    return {"clip_id": sample["id"], "tags": {"sauce_color": answer["choice"]} if certain else {},
            "status": "tagged" if certain else "manual_review", "confidence": confidence,
            "sampling": sampling, "scope": "single_frame_visual_attribute_only", "answer": answer}


def main():
    args = parser(__doc__).parse_args()
    with client(args) as predict:
        emit(run(predict, samples()["video"], args.asset_dir), args)


if __name__ == "__main__":
    main()

共享运行器与 samples.json 已包含在下载包中，与网页展示内容使用同一个维护源。

common.py

common.py

"""Shared cookbook runner. Apache-2.0. No implicit model or media downloads."""
import argparse
import base64
from contextlib import contextmanager
import json
import math
import os
from pathlib import Path

HERE = Path(__file__).resolve().parent


def samples():
    return json.loads((HERE / "samples.json").read_text())


def parser(description):
    p = argparse.ArgumentParser(description=description)
    p.add_argument("--backend", choices=["remote", "onnx", "native"], default="remote")
    p.add_argument("--base-url", default=os.environ.get("VALEN_BASE_URL", "https://llm-proxy.openxxlab.com/v1"))
    p.add_argument("--model", default=os.environ.get("VALEN_MODEL"))
    p.add_argument("--threads", type=int, default=4)
    p.add_argument("--asset-dir", type=Path, default=Path("."))
    p.add_argument("--output", type=Path)
    return p


@contextmanager
def client(args):
    if args.threads < 1:
        raise ValueError("--threads must be positive")
    if args.backend == "native":
        if not args.model or not Path(args.model).is_dir():
            raise ValueError("native requires --model pointing to an existing local Valen model directory")
        import torch
        from transformers import AutoModel, AutoProcessor
        torch.set_num_threads(args.threads)
        model = AutoModel.from_pretrained(args.model, trust_remote_code=True, local_files_only=True,
                                          dtype=torch.float32, attn_implementation="eager").float().cpu().eval()
        model.backbone_autocast_dtype = None
        processor = AutoProcessor.from_pretrained(args.model, trust_remote_code=True, local_files_only=True)

        def predict(body):
            with torch.inference_mode():
                return model.predict(body, processor=processor, execution="shared_state",
                                     temperature=body.get("temperature", 1.0))
        yield predict
    else:
        from valen_sdk import Valen
        if args.backend == "onnx":
            if not args.model or not Path(args.model).is_dir():
                raise ValueError("onnx requires --model pointing to an existing local ONNX bundle")
            instance = Valen.local(args.model, variant="fp32", local_files_only=True, num_threads=args.threads)
        else:
            instance = Valen.remote(args.base_url, model=args.model or "valen-2b",
                                    api_key=os.environ.get("VALEN_API_KEY"), max_retries=0)
        with instance:
            yield instance.predict


def image_input(payload):
    """Inline media works across HTTP services; never send a client-local path."""
    return [{"role": "user", "content": [
        {"type": "input_text", "text": "Answer using only the visible image."},
        {"type": "input_image", "image_url": "data:image/jpeg;base64," + base64.b64encode(payload).decode("ascii")},
    ]}]


def image_bytes(path):
    import io
    from PIL import Image, ImageOps
    with Image.open(path) as image:
        image = ImageOps.exif_transpose(image).convert("RGB")
        image.thumbnail((512, 512))
        buffer = io.BytesIO()
        image.save(buffer, format="JPEG", quality=95)
        return buffer.getvalue()


def middle_frame(path):
    """Decode a bounded clip twice: count first, then keep only its middle frame."""
    import av
    import io
    from PIL import ImageOps
    with av.open(str(path)) as video:
        stream = video.streams.video[0]
        if stream.width * stream.height > 4096 * 2160:
            raise ValueError("Clip exceeds the cookbook's 4K frame limit")
        count = 0
        for _ in video.decode(video=0):
            count += 1
            if count > 1800:
                raise ValueError("Clip exceeds the cookbook's 1800-frame limit")
    if not count:
        raise ValueError("Clip contains no decodable frames")
    target = count // 2
    with av.open(str(path)) as video:
        for index, frame in enumerate(video.decode(video=0)):
            if index == target:
                image = ImageOps.exif_transpose(frame.to_image()).convert("RGB")
                image.thumbnail((512, 512))
                buffer = io.BytesIO()
                image.save(buffer, format="JPEG", quality=95)
                return buffer.getvalue(), {"frame_index": target, "total_frames": count,
                                            "timestamp_seconds": float(frame.time) if frame.time is not None else None}
    raise ValueError("Clip changed while decoding")


def request(predict, source, questions):
    response = predict({"input": source, "questions": questions, "temperature": 1.0})
    answers = response["answers"]
    if [a["name"] for a in answers] != [q["name"] for q in questions]:
        raise ValueError("Service returned missing or reordered answers")
    return {a["name"]: a for a in answers}


def choice_status(answer, *, threshold=0.8, margin=0.2):
    probabilities = [float(x["probability"]) for x in answer["probabilities"]]
    if (not probabilities or any(not math.isfinite(x) or not 0 <= x <= 1 for x in probabilities)
            or abs(sum(probabilities) - 1) > 1e-4):
        raise ValueError("Invalid choice probability distribution")
    selected = next(x for x in answer["probabilities"]
                    if type(x["value"]) is type(answer["choice"]) and x["value"] == answer["choice"])
    ordered = sorted(probabilities, reverse=True)
    confidence = float(selected["probability"])
    gap = ordered[0] - (ordered[1] if len(ordered) > 1 else 0)
    return confidence, confidence >= threshold and gap >= margin


def emit(value, args):
    text = json.dumps(value, indent=2, ensure_ascii=False, allow_nan=False) + "\n"
    if args.output:
        args.output.write_text(text)
    print(text, end="")

真实验证结果

真实模型验证: 2026-10-26 · Valen-0.8B / ONNX FP32 · SDK 0.2.0 · ONNX Runtime 1.31.0 / CPU / SDK HTTP service.

源模型 revision: a0a10112716f54457bad6f60b792ad334f9e89fd. 机器可读结果

实际业务输出

{
  "clip_id": "clip-001",
  "tags": {
    "sauce_color": "white"
  },
  "status": "tagged",
  "confidence": 0.9773176758616222,
  "sampling": {
    "frame_index": 98,
    "total_frames": 196,
    "timestamp_seconds": 3.2666666666666666
  },
  "scope": "single_frame_visual_attribute_only",
  "answer": {
    "name": "sauce_color",
    "type": "choice",
    "probabilities": [
      {
        "value": "brown",
        "probability": 0.012937617960974548
      },
      {
        "value": "white",
        "probability": 0.9773176758616222
      },
      {
        "value": "red",
        "probability": 0.006636453849260038
      },
      {
        "value": "orange",
        "probability": 0.0031082523281432115
      }
    ],
    "confidence": 0.9697569011488296,
    "choice": "white"
  }
}

这里记录的是指定版本与后端的实测，不意味着默认公共服务的模型版本、分数或速度完全相同。业务阈值单元测试使用构造响应，真实模型执行单独记录，二者不互相替代。

素材、许可与相关文档

代码与作者编写的 JSON 文本样例采用 Apache-2.0。工单为虚构示例。sheep.jpg 与 momos.mp4 沿用旧 Notebook 的评测片段和稳定素材 URL；旧目录未提供原始数据集归属及单独的媒体许可记录，因此这里不授予媒体再分发或商业使用许可。生产或再分发时请替换为有权使用的素材，勿把代码许可视为媒体许可。官网不额外复制这些媒体文件。

Decisions API · SystemOne API · ONNX / SDK · Native deployment · GGUF deployment · Evaluation · Validation and performance