候选项都不对: 未知类别处理
你的自动工单系统能处理退款、查物流和开发票。客户要求“发货前修改收货地址”时，应该转给人工：查物流的队列并不负责修改地址。
但在一次真实的 Valen 测试中，这条请求被分给了物流查询，选中概率达到 99.3%。即使应用要求概率至少为 99%，它也会通过。
给模型一个“都不适用”的选项
我们保持消息和三个业务类别不变，增加一个 none 候选，再运行一次：
Please change the delivery address before you ship my order.
| 同一条请求 | Valen 的选择 | 应用处理 |
|---|---|---|
| 原候选＋阈值 | shipping99.3% | 自动接收，分错队列 |
| 增加 none | none51.2% | 交给人工 |
这次，模型选择了 none，应用把请求交给人工。Valen 仍然返回结构化选择和候选概率；你的代码只需要为这个选择安排一条兜底路径。
原因很直接：Choice 总会在你提供的候选中分配概率。当正确类别缺席时，“最可能的选项”仍可能是错误的。none 给了模型一个可表达的出口。把它作为普通 Choice 候选，就能接入现有流程。
少一些错分，也会多一些人工
我们用 Valen-0.8B 比较了原来的阈值策略和增加 none 后的处理方式。下面是 200 条未参与选门槛的测试请求的结果：
| 你需要关心的结果 | 原候选＋阈值 | 增加 none |
|---|---|---|
| 未知请求被错误接收 | 8 / 70 11.4% | 1 / 70 1.4% |
| 正常请求转人工 | 0 / 100 0.0% | 3 / 100 3.0% |
| 自动处理中的错误 | 8 / 108 7.4% | 2 / 99 2.0% |
| 自动处理量 | 108 / 200 54.0% | 99 / 200 49.5% |
这组测试里，不属于这些类别、却被自动接收的请求从 8 条降到 1 条，代价是 3 条原本可以处理的正常请求被送给人工。是否值得，取决于你更在意错误派单，还是人工处理量。
我们也测试了 none 加概率／差值门槛。验证集选出的额外门槛为零，因此它与仅使用 none 的结果相同；本次测试没有显示叠加门槛的额外收益。
none 也会选错。给开头那条改地址请求加上 “Please let me know the next step.”，模型又把它分给了物流查询。一条询问包裹为何滞留海关的正常请求，则被判为 none。下面保留了这些失败记录。
测试集由 40 个虚构场景的表述变体组成，完整分组和评测方法见文末。
接到自己的工单系统
先把 none 的含义写清楚。在这个示例中，它表示“退款、物流查询、发票都不适用，或者消息不足以判断”。业务中的“其他”队列可以保留自己的类别和处理流程。
下载示例源码包（约 22 KB），使用 Python 3.11+ 和已有的 Valen 服务即可运行，不需要下载模型：
Bash
python -m zipfile -e valen-cookbook.zip .
cd valen-cookbook
python -m pip install valen-sdk==0.2.0
python unknown_categories.py demo --backend remote \
--base-url http://127.0.0.1:8000/v1 --model valen \
--text "Please change the delivery address before you ship my order."
替换服务地址和模型名；鉴权服务通过环境变量 VALEN_API_KEY 提供凭证。上面的命令演示仅使用 none 的策略，输出选中类别、候选概率以及最终处理路径。
放进现有调用时，核心逻辑是：
Python
from unknown_categories import question, route
response = predict({
"input": message,
"questions": [question(with_none=True)],
"temperature": 1.0,
})
result = route(response["answers"][0], "B_none")
# none → 人工复核；普通类别 → 对应业务队列
完整脚本还会检查概率格式，把请求失败交给人工，并支持加载验证集选定的概率／差值规则。门槛使用候选概率，不使用衡量分布集中程度的 confidence 字段。若要拟合概率校准，继续阅读校准工单分流。
上线前，用自己的工单同时检查两件事：未知请求是否仍被错误接收，正常请求是否被过多送给人工。 把“类别不支持”和“信息不足”分别标注，之后才能决定是补充业务类别，还是让人工追问。
查看证据和完整实现
以下内容按需展开。代码、样本和原始预测都可以下载；复算报告不会再次调用模型。
完整策略对比、指标定义和评测方法
对比三种策略
评测使用单意图的英文虚构消息。
| 策略 | 候选集合 | 自动处理条件 |
|---|---|---|
| A：阈值 | 三个业务类别 | 选中概率与前两项差值均达到验证集冻结门槛 |
| B：none | 三个类别＋none | 选中普通类别，不额外检查概率 |
| C：none＋阈值 | 三个类别＋none | 选中普通类别，并满足两个冻结门槛 |
A、C 分别在验证集选定门槛，再用于测试集。
记录范围: Valen-0.8B · native FP32 · CPU: INTEL(R) XEON(R) PLATINUM 8558 · eager · shared_state · T=1 · 600 次文本请求；仅测试划分结果如下
测试集：100 条候选内、70 条候选外、30 条信息不足，来自 40 个来源场景。区间为来源组 bootstrap 的经验 95% CI；零事件或全成功时，经验区间可能退化为单点。表格可横向滚动。
| 指标 | A：阈值 | B：none | C：none＋阈值 |
|---|---|---|---|
| 验证集选定规则 | p ≥ 0.99; gap ≥ 0 | 仅按 none 判断 | p ≥ 0; gap ≥ 0 |
| 候选外漏检率 ↓ | 11.4%95% CI 0.0%–30.0% | 1.4%95% CI 0.0%–4.7% | 1.4%95% CI 0.0%–4.7% |
| 候选内误转人工率 ↓ | 0.0%95% CI 0.0%–0.0% | 3.0%95% CI 0.0%–7.8% | 3.0%95% CI 0.0%–7.8% |
| 自动处理覆盖率 | 54.0%95% CI 37.5%–69.0% | 49.5%95% CI 34.0%–64.0% | 49.5%95% CI 34.0%–64.0% |
| 自动处理错误率 ↓ | 7.4%95% CI 0.0%–19.4% | 2.0%95% CI 0.0%–5.8% | 2.0%95% CI 0.0%–5.8% |
| 候选外检测率（不计请求失败） | 88.6%95% CI 70.0%–100.0% | 98.6%95% CI 95.3%–100.0% | 98.6%95% CI 95.3%–100.0% |
| 信息不足转人工率 | 100.0%95% CI 100.0%–100.0% | 96.7%95% CI 90.0%–100.0% | 96.7%95% CI 90.0%–100.0% |
| 自动处理数量 / 错误数量 | 108 / 8 | 99 / 2 | 99 / 2 |
| 请求错误 | 0 | 0 | 0 |
门槛选择的完整规则
A、C 分别扫描同一组预先声明的门槛：概率 0, .5, .6, .7, .8, .85, .9, .95, .98, .99, 1，差值 0, .1, .2, .3, .4, .5。在验证集已接受样本观测错误率不超过 5%、且至少接受五个来源场景组的条件下，选择覆盖率最高的规则；并列时依次选择错误更少、概率门槛更低、差值门槛更低的规则。没有可行规则就全部转人工。
覆盖率分母是全部测试请求。候选外漏检率＝候选外请求中被自动分入普通类别的数量／全部候选外请求。候选内误转人工率＝正常请求中转人工的数量／全部正常请求。自动处理错误率同时计入普通类别选错、错误接受候选外请求和信息不足请求；没有自动接受样本时无定义，显示“—”。
候选外检测率只统计语义／策略判断产生的转人工，请求失败不算成功检测。失败仍保留在对应分母及人工量中。信息不足转人工率包含运行失败兜底，因此须同时查看错误数量。
成功与失败案例，包括同一输入的前后对比
| 案例类型 | 原始输入 / 真值 | 策略与实际结果 |
|---|---|---|
| 候选外请求转人工 | I forgot my password. Help me regain access to my account. outside; 业务类别真值: 不适用 | C：none＋阈值none → human_reviewnone_selected; p=99.5% |
| 漏检：候选外被自动接受 | Please change the delivery address before you ship my order. Please let me know the next step. outside; 业务类别真值: 不适用 | C：none＋阈值shipping → automaticaccepted; p=59.1% |
| 误报：正常请求转人工 | Can you check why my package is waiting at customs? Please let me know the next step. in_scope; 业务类别真值: shipping | C：none＋阈值none → human_reviewnone_selected; p=50.2% |
| 信息不足仍被自动接受 | I have a question about my purchase. insufficient; 业务类别真值: 不适用 | C：none＋阈值invoice → automaticaccepted; p=53.8% |
| 正常请求正确处理 | Please email me a copy of the invoice for my purchase. in_scope; 业务类别真值: invoice | C：none＋阈值invoice → automaticaccepted; p=100.0% |
| 高概率漏检：阈值策略 A | Please change the delivery address before you ship my order. outside; 业务类别真值: 不适用 | A：阈值shipping → automaticaccepted; p=99.3% |
| 同一输入加入 none：策略 C | Please change the delivery address before you ship my order. outside; 业务类别真值: 不适用 | C：none＋阈值none → human_reviewnone_selected; p=51.2% |
完整报告保留每条测试决策，包括只改变表述就改变结果的情况。案例按固定规则从实测记录中选出。选择时依次优先 C、B、A；同一策略中优先无包装的原始消息，再按行 ID 排序。另外补充 A 中首个候选外漏检（优先原始消息，再按行 ID）及同一输入在 C 下的结果。
下载原始预测、离线复算，或评测自己的数据
数据包括 100 条验证请求（20 个来源场景）和 200 条测试请求（另外 40 个场景）。一半请求属于候选内，另一半属于候选外或信息不足。每个场景有五种表述变体，同一场景保留在同一个划分中。区间通过固定随机种子的来源组 bootstrap 计算。
标注数据 · 原始预测 gzip · 完整报告 gzip · 运行环境和模型文件哈希 · 实测执行脚本
下载并解压预测文件后，用 Python 标准库离线复算：
Bash
python -c "import gzip,pathlib; pathlib.Path('unknown-predictions.jsonl').write_bytes(gzip.open('predictions.jsonl.gz','rb').read())"
python unknown_categories.py report --cache unknown-predictions.jsonl --output replay.json
评测自己的数据时，复制并编辑 unknown_categories_data.json。保持唯一行 ID、稳定 group_id、validation／test 划分，以及 in_scope、outside、insufficient 标签。候选内的 gold 填普通类别，其余填 JSON null。真值不进入模型请求。替换类别还需修改脚本中的 CHOICES、NONE 及描述；协议哈希会改变，旧缓存会被拒绝。
每条请求采集两份预测，离线比较三种策略。B、C 复用相同预测。记录中的实测使用已有原版 0.8B 模型、CPU（八线程）和 FP32；以下路径需替换为自己的已有模型目录：
Bash
python unknown_categories.py collect --backend native --device cpu --threads 8 \
--model /path/to/Valen-0.8B --revision YOUR_PINNED_ARTIFACT_ID \
--dataset my-data.json --cache my-predictions.jsonl
python unknown_categories.py report --dataset my-data.json \
--cache my-predictions.jsonl --output my-report.json
python unknown_categories.py demo --backend native --device cpu --threads 8 \
--model /path/to/Valen-0.8B --revision YOUR_PINNED_ARTIFACT_ID \
--policy my-report.json --text "Please send my invoice."
原版运行使用支持 Decisions 的模型目录。采集支持从同一缓存补齐缺失请求，已记录的失败不会自动重试。首条请求失败会停止，避免浪费调用。远端运行改用 --backend remote --base-url ... --model ... --revision ...。--revision 用于记录部署版本。加载策略时会检查模型路径／名称、后端、声明的版本、原版运行设备、线程数及协议，使用与采集时一致的配置。
本例固定 temperature=1。更换候选、描述、模型或后端后，可用自己的工单重新选择门槛；概率校准方法见校准工单分流。
unknown_categories.py：采集、路由、规则选择和离线报告
unknown_categories.py
"""Choice with an explicit none option, optional review gates and offline replay. Apache-2.0."""
import hashlib
import json
import math
from pathlib import Path
import random
from collections import Counter
from contextlib import contextmanager
from common import client, parser
HERE = Path(__file__).resolve().parent
CHOICES = [
{"value": "refund", "description": "Requests money back, a payment reversal, or reimbursement."},
{"value": "shipping", "description": "Asks about shipment progress, tracking, delivery time, or parcel location."},
{"value": "invoice", "description": "Requests or corrects an invoice or receipt, or asks about its contents."},
]
NONE = {"value": "none", "description": "None of the three categories applies, or the message lacks enough information to choose one."}
INSTRUCTIONS = "Classify the customer's current request using only the supplied message and the candidate descriptions. Do not invent missing context."
STRATEGIES = ("A_threshold", "B_none", "C_none_threshold")
@contextmanager
def prediction_client(args):
if args.backend != "native":
with client(args) as predict:
yield predict
return
if not args.model or not Path(args.model).is_dir():
raise ValueError("native requires an existing model directory; no implicit download")
import torch
from transformers import AutoModel, AutoProcessor
if args.threads < 1:
raise ValueError("--threads must be positive")
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().to(args.device).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=1.0)
yield predict
def runtime_identity(args):
return {"backend": args.backend, "model": args.model, "revision": args.revision,
"base_url": args.base_url if args.backend == "remote" else None,
"device": args.device if args.backend == "native" else None,
"threads": args.threads, "temperature": 1.0}
def digest(value):
return hashlib.sha256(json.dumps(value, sort_keys=True, ensure_ascii=False, allow_nan=False).encode()).hexdigest()
def question(with_none):
return {"name": "intent", "type": "choice", "instructions": INSTRUCTIONS,
"choices": CHOICES + ([NONE] if with_none else [])}
def validated_answer(answer, with_none):
expected = [c["value"] for c in question(with_none)["choices"]]
values = answer["probabilities"]
if (len(values) != len(expected) or {v["value"] for v in values} != set(expected)
or any(type(v["probability"]) not in (int, float) or not math.isfinite(v["probability"])
or not 0 <= v["probability"] <= 1 for v in values)
or abs(sum(v["probability"] for v in values) - 1) > 1e-4):
raise ValueError("Invalid candidate probabilities")
probabilities = {v["value"]: v["probability"] for v in values}
selected = answer["choice"]
if selected not in probabilities or probabilities[selected] < max(probabilities.values()) - 1e-7:
raise ValueError("Selected value is not a maximum-probability candidate")
return probabilities
def route(answer, strategy, policy=None):
if strategy not in STRATEGIES:
raise ValueError("Unknown strategy")
if answer is None:
return {"selected": None, "route": "human_review", "reason": "request_error"}
with_none = strategy != "A_threshold"
p = validated_answer(answer, with_none)
selected = answer["choice"]
ordered = sorted(p.values(), reverse=True)
result = {"selected": selected, "probability": p[selected], "margin": ordered[0] - ordered[1],
"probabilities": answer["probabilities"], "route": "human_review"}
if with_none and selected == "none":
result["reason"] = "none_selected"
elif strategy == "B_none":
result.update(route="automatic", reason="accepted")
elif policy is None:
raise ValueError("A frozen policy is required")
elif policy.get("reject_all"):
result["reason"] = "no_validation_rule"
elif p[selected] < policy["threshold"]:
result["reason"] = "low_probability"
elif result["margin"] < policy["margin"]:
result["reason"] = "small_margin"
else:
result.update(route="automatic", reason="accepted")
return result
def load_dataset(path):
data = json.loads(Path(path).read_text())
ids, groups, texts = set(), {}, {}
for row in data["rows"]:
if (not row["id"] or row["id"] in ids or not row["group_id"] or not row["text"].strip()
or row["split"] not in ("validation", "test")
or row["kind"] not in ("in_scope", "outside", "insufficient")
or (row["kind"] == "in_scope" and row["gold"] not in {c["value"] for c in CHOICES})
or (row["kind"] != "in_scope" and row["gold"] is not None)):
raise ValueError("Invalid dataset row")
ids.add(row["id"])
identity = (row["split"], row["kind"], row["gold"])
if groups.setdefault(row["group_id"], identity) != identity:
raise ValueError("Source group crosses splits or labels")
if texts.setdefault(row["text"], row["group_id"]) != row["group_id"]:
raise ValueError("Duplicate text across source groups")
if {r["split"] for r in data["rows"]} != {"validation", "test"}:
raise ValueError("Both validation and test are required")
return data
def protocol():
return {"temperature": 1.0, "without_none": question(False), "with_none": question(True)}
def collect(args, data):
identity = runtime_identity(args)
header = {"schema_version": 1, "dataset_sha256": digest(data), "protocol_sha256": digest(protocol()),
"identity": identity, "source_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest()}
args.cache.parent.mkdir(parents=True, exist_ok=True)
existing = []
if args.cache.exists():
lines = args.cache.read_text().splitlines()
if not lines or json.loads(lines[0]) != header:
raise ValueError("Cache identity changed; use a new cache path")
existing = [json.loads(line) for line in lines[1:]]
keys = [(r["id"], r["variant"]) for r in existing]
expected = {(r["id"], v) for r in data["rows"] for v in ("without_none", "with_none")}
if len(keys) != len(set(keys)) or not set(keys) <= expected:
raise ValueError("Duplicate or unexpected cache rows")
seen = set(keys)
# Failures are retained and never silently retried; a resume only fills missing requests.
with args.cache.open("a") as f:
if not args.cache.stat().st_size:
f.write(json.dumps(header) + "\n")
f.flush()
if seen == expected:
return
with prediction_client(args) as predict:
for row in data["rows"]:
for variant in ("without_none", "with_none"):
if (row["id"], variant) in seen:
continue
result = {"id": row["id"], "variant": variant}
try:
response = predict({"input": row["text"], "questions": [question(variant == "with_none")], "temperature": 1.0})
answers = response["answers"]
if len(answers) != 1 or answers[0]["name"] != "intent":
raise ValueError("Missing or unexpected answer")
answer = answers[0]
validated_answer(answer, variant == "with_none")
result.update(status="ok", answer=answer, resolved_model=response.get("model"))
except Exception as exc:
result.update(status="error", error_type=type(exc).__name__)
f.write(json.dumps(result, ensure_ascii=False, allow_nan=False) + "\n")
f.flush()
if result["status"] == "error" and not seen:
raise RuntimeError("First request failed; stopped to avoid unnecessary traffic. Inspect cache.")
seen.add((row["id"], variant))
def decisions(rows, cache, strategy, policy):
variant = "without_none" if strategy == "A_threshold" else "with_none"
out = []
for row in rows:
record = cache[(row["id"], variant)]
result = route(record["answer"] if record["status"] == "ok" else None, strategy, policy)
out.append({**row, **result})
return out
def counts(rows):
accepted = [r for r in rows if r["route"] == "automatic"]
return {"requests": len(rows), "source_groups": len({r["group_id"] for r in rows}),
"accepted": len(accepted),
"accepted_errors": sum(r["kind"] != "in_scope" or r["selected"] != r["gold"] for r in accepted),
"outside": sum(r["kind"] == "outside" for r in rows),
"outside_accepted": sum(r["kind"] == "outside" for r in accepted),
"in_scope": sum(r["kind"] == "in_scope" for r in rows),
"in_scope_review": sum(r["kind"] == "in_scope" and r["route"] == "human_review" for r in rows),
"insufficient": sum(r["kind"] == "insufficient" for r in rows),
"insufficient_review": sum(r["kind"] == "insufficient" and r["route"] == "human_review" for r in rows),
"request_errors": sum(r["reason"] == "request_error" for r in rows),
"outside_detected": sum(r["kind"] == "outside" and r["route"] == "human_review" and r["reason"] != "request_error" for r in rows)}
RATIOS = {"coverage": ("accepted", "requests"), "accepted_error_rate": ("accepted_errors", "accepted"),
"outside_miss_rate": ("outside_accepted", "outside"),
"in_scope_review_rate": ("in_scope_review", "in_scope"),
"insufficient_review_rate": ("insufficient_review", "insufficient"),
"outside_detection_rate": ("outside_detected", "outside")}
def metrics(rows, bootstrap=1000):
c = counts(rows)
result = {"counts": c, "reasons": dict(Counter(r["reason"] for r in rows))}
groups = {}
for r in rows:
groups.setdefault(r["group_id"], []).append(r)
grouped = list(groups.values())
rng = random.Random(20261027)
samples = {key: [] for key in RATIOS}
for _ in range(bootstrap):
sampled = counts([r for group in rng.choices(grouped, k=len(grouped)) for r in group])
for key, (num, den) in RATIOS.items():
if sampled[den]:
samples[key].append(sampled[num] / sampled[den])
for key, (num, den) in RATIOS.items():
values = sorted(samples[key])
result[key] = {"value": c[num] / c[den] if c[den] else None,
"ci95": [values[int((len(values)-1)*q)] for q in (.025, .975)] if values else None}
return result
def select_policy(rows, cache, strategy, max_error=.05):
best = None
for threshold in (0, .5, .6, .7, .8, .85, .9, .95, .98, .99, 1.0):
for margin in (0, .1, .2, .3, .4, .5):
policy = {"threshold": threshold, "margin": margin, "reject_all": False}
routed = decisions(rows, cache, strategy, policy)
c = counts(routed)
accepted_groups = len({r["group_id"] for r in routed if r["route"] == "automatic"})
if c["accepted"] and accepted_groups >= 5 and c["accepted_errors"] / c["accepted"] <= max_error:
rank = (c["accepted"], -c["accepted_errors"], -threshold, -margin)
if best is None or rank > best[0]:
best = (rank, policy)
return {**(best[1] if best else {"reject_all": True, "threshold": None, "margin": None}),
"validation_observed_error_target": max_error, "min_accepted_source_groups": 5,
"selection": "maximum validation coverage, then fewer errors, then lower threshold and margin; no test tuning"}
def replay(args, data):
lines = [json.loads(line) for line in args.cache.read_text().splitlines()]
header = lines[0]
if (header["dataset_sha256"] != digest(data) or header["protocol_sha256"] != digest(protocol())
or header["source_sha256"] != hashlib.sha256(Path(__file__).read_bytes()).hexdigest()):
raise ValueError("Dataset, protocol or collection source changed")
keys = [(r["id"], r["variant"]) for r in lines[1:]]
expected = {(r["id"], v) for r in data["rows"] for v in ("without_none", "with_none")}
if len(keys) != len(set(keys)) or set(keys) != expected:
raise ValueError("Cache must contain exactly two predictions per row")
cache = dict(zip(keys, lines[1:]))
for (_, variant), record in cache.items():
if record["status"] == "ok":
validated_answer(record["answer"], variant == "with_none")
elif record["status"] != "error":
raise ValueError("Unknown prediction status")
validation = [r for r in data["rows"] if r["split"] == "validation"]
test = [r for r in data["rows"] if r["split"] == "test"]
policies = {s: select_policy(validation, cache, s) if s != "B_none" else None for s in STRATEGIES}
report = {"schema_version": 1, "validation_kind": "real_model_execution", "identity": header["identity"],
"dataset_sha256": digest(data), "protocol_sha256": digest(protocol()),
"cache_sha256": hashlib.sha256(args.cache.read_bytes()).hexdigest(),
"source_sha256": header["source_sha256"], "resolved_models": sorted({r["resolved_model"] for r in lines[1:] if r.get("resolved_model")}),
"confidence_interval": "95% percentile bootstrap, 1000 resamples of source groups, seed 20261027; descriptive, not an operational guarantee",
"scope": data["description"], "policies": policies, "results": {}, "cases": []}
for strategy in STRATEGIES:
test_rows = decisions(test, cache, strategy, policies[strategy])
report["results"][strategy] = {"validation": metrics(decisions(validation, cache, strategy, policies[strategy])),
"test": metrics(test_rows)}
# Publish all test decisions; tutorials choose examples without hiding failures.
report["cases"].extend({"strategy": strategy, **r} for r in test_rows)
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2, allow_nan=False) + "\n")
def main(argv=None):
p = parser(__doc__)
p.add_argument("command", choices=("collect", "report", "demo"))
p.add_argument("--dataset", type=Path, default=HERE / "unknown_categories_data.json")
p.add_argument("--cache", type=Path, default=Path("unknown-predictions.jsonl"))
p.add_argument("--device", default="cpu", help="Existing native model execution device, e.g. cpu or cuda")
p.add_argument("--revision", help="Pinned revision or explicit deployment identifier, recorded with predictions")
p.add_argument("--policy", type=Path, help="Report from this model/deployment and protocol")
p.add_argument("--text", default="Please delete my account and erase my personal data.")
args = p.parse_args(argv)
if args.command == "collect":
if not args.revision or not args.model:
p.error("collect requires --model and --revision; do not mix deployments in one cache")
collect(args, load_dataset(args.dataset))
elif args.command == "report":
if args.output is None:
p.error("report requires --output")
replay(args, load_dataset(args.dataset))
else:
frozen = None
if args.policy:
report = json.loads(args.policy.read_text())
identity = report["identity"]
actual = runtime_identity(args)
if report["protocol_sha256"] != digest(protocol()) or identity != actual:
raise ValueError("Policy model/deployment or protocol mismatch; collect and fit your own policy")
frozen = report["policies"]["C_none_threshold"]
with prediction_client(args) as predict:
try:
response = predict({"input": args.text, "questions": [question(True)], "temperature": 1.0})
answer = response["answers"][0]
validated_answer(answer, True)
except Exception:
answer = None
result = route(answer, "C_none_threshold" if frozen is not None else "B_none", frozen)
print(json.dumps(result, ensure_ascii=False, indent=2, allow_nan=False))
if __name__ == "__main__":
main()
下载包包含 common.py 和标注 JSON。代码及虚构文本使用 Apache-2.0。