Route support tickets
An incoming support ticket needs a processing queue. The input is the original notebook's fictional delivery-address message; the output is a ticket ID, intent queue, address-change probability, explicitness score (0–2), and review priority. The program only proposes a queue: it does not modify an order.
Ask all three questions about the same input. A choice needs selected probability ≥ 0.8 and a top-two gap ≥ 0.2 to route automatically. A predicted address change also needs its independent predicate probability ≥ 0.8; disagreement goes to human_review. Rank review work by address_change_probability × explicitness_score / 2. A score near 2 indicates a more explicit request, not a 100% likelihood of successful fulfillment.
These thresholds illustrate a workflow. They are not calibrated accuracy guarantees. Measure false routing costs on your own labeled tickets and tune thresholds before operational use. Ambiguous, novel and low-confidence messages remain reviewable.
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
{
"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
python ticket_routing.py --backend remote --model valen-2b --asset-dir . --output result.json
# An existing local HTTP deployment instead:
python ticket_routing.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
"""Turn an incoming ticket into a queue and a review priority. Apache-2.0."""
from common import choice_status, client, emit, parser, request, samples
def run(predict, sample):
answers = request(predict, sample["text"], sample["questions"])
intent = answers["intent"]
confidence, certain = choice_status(intent)
probability = answers["address_change"]["probability"]
explicitness = answers["explicitness"]["score"]
# These are illustrative workflow thresholds, not calibrated accuracy claims.
route = intent["choice"] if certain else "human_review"
if route == "change_address" and probability < 0.8:
route = "human_review"
return {"ticket_id": sample["id"], "queue": route, "confidence": confidence,
"address_change_probability": probability, "explicitness_score": explicitness,
"review_priority": round(probability * explicitness / 2, 6),
"action": "review_request_before_changing_order", "answers": list(answers.values())}
def main():
args = parser(__doc__).parse_args()
with client(args) as predict:
emit(run(predict, samples()["ticket"]), 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
{
"ticket_id": "ticket-001",
"queue": "change_address",
"confidence": 0.9997590278141875,
"address_change_probability": 0.9862781119951404,
"explicitness_score": 1.7783824425004437,
"review_priority": 0.87699,
"action": "review_request_before_changing_order",
"answers": [
{
"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"
},
{
"name": "address_change",
"type": "predicate",
"probability": 0.9862781119951404
},
{
"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
}
]
}
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