工单分流
客服系统收到工单，需要生成处理队列。输入复用旧 Notebook 中虚构的收货地址变更消息；输出包括工单 ID、意图队列、地址变更概率、明确程度评分（0–2）及审核优先级。程序只提出分流建议，不修改订单。
对同一输入同时提问三道题。选择题的选中概率至少为 0.8、前两名概率差至少为 0.2 才自动分流。若选择“修改地址”，独立判断题的地址变更概率还须至少为 0.8；两个答案冲突时进入 human_review。审核优先级按 地址变更概率 × 明确程度评分 / 2 排序。score 接近 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
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"
}
]
}
]
}
运行
Bash
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
本地 ONNX 也可直接运行：把参数改为 --backend onnx --model /path/to/Valen-0.8B-ONNX，并先安装 valen-sdk[onnx,video]==0.2.0。GGUF 服务通过 --backend remote 接入；不能把 GGUF 文件传给 Valen.local。
完整可执行代码
ticket_routing.py
"""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()
共享运行器与 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. 机器可读结果
实际业务输出
{
"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
}
]
}
这里记录的是指定版本与后端的实测，不意味着默认公共服务的模型版本、分数或速度完全相同。业务阈值单元测试使用构造响应，真实模型执行单独记录，二者不互相替代。
素材、许可与相关文档
代码与作者编写的 JSON 文本样例采用 Apache-2.0。工单为虚构示例。sheep.jpg 与 momos.mp4 沿用旧 Notebook 的评测片段和稳定素材 URL；旧目录未提供原始数据集归属及单独的媒体许可记录，因此这里不授予媒体再分发或商业使用许可。生产或再分发时请替换为有权使用的素材，勿把代码许可视为媒体许可。官网不额外复制这些媒体文件。
Decisions API · SystemOne API · ONNX / SDK · Native deployment · GGUF deployment · Evaluation · Validation and performance