API basics
This migrates the original API notebook without duplicating the application tutorials. Send a delivery-address message, identify the animal group in an image, and label the sauce color in one sampled video frame. The output contains named answers and the exact sampled frame position.
The HTTP client uses Decisions. The API accepts an image data URL; the tutorial extracts a frame before sending the video example. This is not direct video inference on the hosted service. Set VALEN_API_KEY only if your chosen deployment requires it.
Install and prepare inputs
Use Python 3.11+. Download the small source bundle, extract it, then install its dependencies:
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
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.
Bash
curl -fL https://raw.githubusercontent.com/Valen-Team/Valen/main/assets/cookbook/sheep.jpg -o sheep.jpg
curl -fL https://raw.githubusercontent.com/Valen-Team/Valen/main/assets/cookbook/momos.mp4 -o momos.mp4
samples.json
{
"id": "ticket-001",
"text": "I moved. Can you ship my order to 12 Elm Street instead of the old address?",
"questions": [
{
"name": "intent",
"type": "choice",
"instructions": "Which intent does the user's message express?",
"choices": [
{
"value": "billing_question",
"description": "Asks about a charge, invoice or payment"
},
{
"value": "cancel_order",
"description": "Wants to cancel an order"
},
{
"value": "change_address",
"description": "Wants to change the delivery address"
},
{
"value": "report_damage",
"description": "Received an item that is broken or damaged"
},
{
"value": "track_order",
"description": "Wants to know where an order is or when it arrives"
}
]
},
{
"name": "address_change",
"type": "predicate",
"instructions": "The user requests a delivery address change."
},
{
"name": "explicitness",
"type": "score",
"instructions": "How explicit is the request to change the delivery address?",
"levels": [
{
"label": "none",
"description": "No request"
},
{
"label": "possible",
"description": "Asks whether a change is possible"
},
{
"label": "explicit",
"description": "Explicitly requests a change"
}
]
}
]
}
Run
Bash
python api_quickstart.py --backend remote --model valen-2b --asset-dir . --output result.json
# An existing local HTTP deployment instead:
python api_quickstart.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
api_quickstart.py
"""API basics: text, one image and a sampled video frame. Apache-2.0."""
from common import client, emit, image_bytes, image_input, middle_frame, parser, request, samples
def main():
args = parser(__doc__).parse_args()
data = samples()
with client(args) as predict:
text = request(predict, data["ticket"]["text"], data["ticket"]["questions"])
image = request(predict, image_input(image_bytes(args.asset_dir / data["product"]["image"])), [data["product"]["question"]])
payload, sampling = middle_frame(args.asset_dir / data["video"]["path"])
video_frame = request(predict, image_input(payload), [data["video"]["question"]])
emit({"text": text, "image": image, "video_frame": video_frame, "sampling": sampling}, 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
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
Actual output
{
"text": {
"intent": {
"name": "intent",
"type": "choice",
"probabilities": [
{
"value": "billing_question",
"probability": 0.0000014444915728188846
},
{
"value": "cancel_order",
"probability": 0.00013284262147016516
},
{
"value": "change_address",
"probability": 0.9997590278141875
},
{
"value": "report_damage",
"probability": 0.000004776970793180753
},
{
"value": "track_order",
"probability": 0.00010190810197642957
}
],
"confidence": 0.9996987847677343,
"choice": "change_address"
},
"address_change": {
"name": "address_change",
"type": "predicate",
"probability": 0.9862781119951404
},
"explicitness": {
"name": "explicitness",
"type": "score",
"probabilities": [
{
"label": "none",
"value": 0,
"probability": 0.015131870978155534
},
{
"label": "possible",
"value": 1,
"probability": 0.1913538155432454
},
{
"label": "explicit",
"value": 2,
"probability": 0.7935143134785991
}
],
"confidence": 0.6675736637506653,
"score": 1.7783824425004437
}
},
"image": {
"group": {
"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"
}
},
"video_frame": {
"sauce_color": {
"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"
}
},
"sampling": {
"frame_index": 98,
"total_frames": 196,
"timestamp_seconds": 3.2666666666666666
}
}
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