Tag short videos
A clip index needs a searchable visual tag. Read the original short dumpling clip, extract its middle frame, ask the original sauce-color question, and emit a JSON annotation containing the clip ID, tag, confidence, frame index, timestamp and total decoded frames.
This recipe deliberately produces a single-frame visual attribute. It does not infer the clip's action or assign the tag to every moment. Direct video and the original action question are retained in the local-inference tutorial. The bounded decoder counts at most 1800 frames, limits frame dimensions to 4K, decodes in two passes, and holds only one selected image. It converts the selected frame to a JPEG with longest edge ≤ 512 pixels before sending one image to the backend.
Emit the tag only when its selected probability ≥ 0.8 and the top-two gap ≥ 0.2. Otherwise emit no tag and set manual_review. Confidence can prioritize the review queue, but it is not a calibrated correctness estimate. Sample multiple time segments and validate temporal coverage if your application needs clip-wide labels.
Install and prepare inputs
Use Python 3.11+. Download the small source bundle, extract it, then install its dependencies:
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
Fetch only these small assets when you choose to run (image ≈58 KB, clip ≈395 KB), or provide authorized files with the same names through --asset-dir.
curl -fL https://raw.githubusercontent.com/Valen-Team/Valen/main/assets/cookbook/momos.mp4 -o momos.mp4
{
"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"
}
]
}
}
Run
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
For direct ONNX execution, use --backend onnx --model /path/to/Valen-0.8B-ONNX and first install valen-sdk[onnx,video]==0.2.0. Connect a GGUF service through --backend remote; a GGUF file is not a Valen.local model.
Complete executable code
"""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()
The shared runner and samples.json are included in the download bundle, from the same maintained source as this page.
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="")
Real validation results
Real-model verification: 2026-10-26 · Valen-0.8B / ONNX FP32 · SDK 0.2.0 · ONNX Runtime 1.31.0 / CPU / SDK HTTP service.
Source model revision: a0a10112716f54457bad6f60b792ad334f9e89fd. Machine-readable results
{
"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"
}
}
Results belong to the recorded model and backend; they do not imply identical versions, scores or speed on the default public service. Policy unit tests use constructed answers. Real-model execution is recorded separately and is not replaced by those tests.
Sources, licensing and related documentation
Code and authored JSON text samples use Apache-2.0. The ticket is fictional. sheep.jpg and momos.mp4 reuse the original notebooks' evaluation excerpts and stable asset URLs. The old directory did not record their original dataset attribution or a separate media license; this page does not grant redistribution or commercial media rights. Use your own authorized media for production or redistribution. The code license does not relicense the media, and this migration creates no additional media copies.
Decisions API · SystemOne API · ONNX / SDK · Native deployment · GGUF deployment · Evaluation · Validation and performance