商品图片筛选

一个市场出售动物摄影挂画。两个商品记录使用同一张羊群照片：第一个声明为羊群，第二个声明为猫群。这个小样例不伪造第二张图片，同时覆盖分类匹配与不匹配两条业务分支。

复用旧 Notebook 的动物群体问题，每张图片只推理一次。程序将识别结果与商品声明分类对比，并按声明分类的匹配概率排序，输出 accept、category_mismatch 或 manual_review，不会删除商品。选中概率至少为 0.8 且前两名概率差至少为 0.2 才自动判断；结果不确定时两个商品均进入人工复核。

本例验证该素材的分类一致性，不验证商品质量、版权归属、图片安全性或真实市场准确率。这里将羊群照片当作挂画商品展示，并非完整商品数据集。扩展前应使用自有标注商品集和实际业务分类体系。

安装与输入

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/sheep.jpg -o sheep.jpg

samples.json

{
  "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"
      }
    ]
  }
}

运行

Bash

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

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

完整可执行代码

product_screening.py

"""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()

共享运行器与 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. 机器可读结果

实际业务输出

{
  "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"
  }
}

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

素材、许可与相关文档

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

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