Screen product images
A marketplace sells photographic animal prints. It receives two listing records pointing to the same sheep photo: one claims a flock of sheep, the other a clowder of cats. This intentionally small test lets you verify both the matching and mismatching business branches without inventing a second image.
The original notebook's animal-group question is reused. The program asks it once per image, compares the selected group with each listing's declared category, and ranks listings by the probability of their declared group. Outputs are accept, category_mismatch, or manual_review; no listing is deleted. The choice needs selected probability ≥ 0.8 and a top-two gap ≥ 0.2. A weak result sends both listings to review.
This validates category consistency for this sample, not product quality, copyright ownership, image safety, or real marketplace accuracy. A photo of sheep is being used as a print product; it is not a general product catalog dataset. Use a labeled product set and your own business taxonomy before expanding the task.
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/sheep.jpg -o sheep.jpg
{
"image": "sheep.jpg",
"listings": [
{
"id": "print-001",
"title": "Flock of sheep photographic print",
"expected_group": "flock"
},
{
"id": "print-002",
"title": "Clowder of cats photographic print",
"expected_group": "clowder"
}
],
"question": {
"name": "group",
"type": "choice",
"instructions": "What is a group of these animals called?",
"choices": [
{
"value": "pride",
"description": "pride"
},
{
"value": "flock",
"description": "flock"
},
{
"value": "clowder",
"description": "clowder"
},
{
"value": "gang",
"description": "gang"
}
]
}
}
Run
python product_screening.py --backend remote --model valen-2b --asset-dir . --output result.json
# An existing local HTTP deployment instead:
python product_screening.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
"""Screen photographic print listings for image/category mismatch. Apache-2.0."""
from common import choice_status, client, emit, image_bytes, image_input, parser, request, samples
def run(predict, sample, asset_dir):
answer = request(predict, image_input(image_bytes(asset_dir / sample["image"])), [sample["question"]])["group"]
confidence, certain = choice_status(answer)
listings = []
for listing in sample["listings"]:
probability = next(x["probability"] for x in answer["probabilities"] if x["value"] == listing["expected_group"])
status = "manual_review" if not certain else "accept" if answer["choice"] == listing["expected_group"] else "category_mismatch"
listings.append({**listing, "status": status, "category_match_probability": probability})
return {"listings": sorted(listings, key=lambda x: x["category_match_probability"], reverse=True),
"predicted_group": answer["choice"], "confidence": confidence, "answer": answer}
def main():
args = parser(__doc__).parse_args()
with client(args) as predict:
emit(run(predict, samples()["product"], 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
{
"listings": [
{
"id": "print-001",
"title": "Flock of sheep photographic print",
"expected_group": "flock",
"status": "accept",
"category_match_probability": 0.9960099221422577
},
{
"id": "print-002",
"title": "Clowder of cats photographic print",
"expected_group": "clowder",
"status": "category_mismatch",
"category_match_probability": 0.002903469483494964
}
],
"predicted_group": "flock",
"confidence": 0.9960099221422577,
"answer": {
"name": "group",
"type": "choice",
"probabilities": [
{
"value": "pride",
"probability": 0.00028894281584105957
},
{
"value": "flock",
"probability": 0.9960099221422577
},
{
"value": "clowder",
"probability": 0.002903469483494964
},
{
"value": "gang",
"probability": 0.0007976655584062285
}
],
"confidence": 0.9946798961896769,
"choice": "flock"
}
}
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