Personalize this lesson
Adapt explanations and teaching visuals to your background and preferred voice.
Consider this ticket:
1The production API key was disabled during an incident freeze. The portal rejected my restore request. Please make an exception.Given a new access ticket, should it enter human_review_now, or may it enter the guarded agent workflow?
That exception isn't the same as a routine key-rotation question. The intended route is immediate human review; routine questions can use a guarded workflow without consuming the same review capacity. The classifier does routing, not authorization. It doesn't decide access eligibility, send a reply, or change permissions. It can misclassify a ticket, so the agent's independent approval controls must remain in place.
Why use a dedicated, fine-tuned encoder rather than prompting a frontier generative LLM? In production intake systems, four engineering realities decide the choice:
- Latency budget and webhook SLAs: Webhook ingress requires an immediate routing lane in 5 to 15 milliseconds. A full autoregressive LLM API roundtrip costs 400 to 1,500 milliseconds, introducing queue backlog and token-generation latency jitter at high ticket volume.
- Deterministic logit geometry: A fine-tuned encoder classification head produces continuous, unnormalized logits through an exact affine projection . That exposes raw score distributions for threshold tuning, temperature calibration, and out-of-distribution audits. Prompting an LLM yields sampled tokens that can break JSON formatting, hallucinate unsupported keys, or surrender to prompt injections embedded in customer ticket bodies.
- Operational cost and throughput: Serving an encoder like DeBERTa-v3 or DistilBERT on CPU or small GPU instances costs roughly $0.00001 per ticket, compared to $0.001 to $0.005 for commercial LLM calls. At hundreds of thousands of daily tickets, that's the difference between single-digit server bills and thousands of dollars in token fees.
- Asymmetric risk and critical triage: Support triage has radically asymmetric costs. Routing a routine key-rotation question to a human reviewer (a false positive) wastes two minutes of an engineer's time ($1.50). Routing an emergency incident-freeze bypass, suspected account takeover, or unauthorized privilege escalation into the automated agent (a false negative) risks automated token reactivation during an active security breach. That asymmetry means we can't rely on symmetric averages like accuracy or macro-F1. High-risk security slices demand a hard release gate with 0% false-negative tolerance.
The evaluation dashboard in the previous capstone made one rule hard to dodge: a release is blocked by the failures that matter, even when the average looks fine. Bring that exact-receipt habit to a learned classifier. Document QA already answers policy questions with permitted citations. A later production-agent capstone will draft an access response and request approval. This intake classifier is the boundary in front of that agent.

By the end, one bundle will carry:
- a versioned label guide and split rule
- a fitted baseline that a contextual model must actually improve on
- a small PyTorch encoder training loop that exposes the mechanics
- a practical Hugging Face fine-tuning recipe for a pretrained encoder
- held-out threshold and slice gates
- a serving bundle the production-agent capstone can consume
BERT showed how a pretrained bidirectional encoder can adapt to a downstream classification task with a small output layer.[1] Hugging Face exposes that sequence-classification pattern through models and training utilities.[2] Modern encoders like DeBERTa-v3 advance this by disentangling content and relative position representations, while distilled options like DistilBERT provide lightweight alternatives running at double the speed. The discipline here isn't choosing a fashionable checkpoint. It's proving that a trained classifier satisfies a routing contract.
The executable examples have three evidence levels: a fitted miniature baseline, a six-row neural training demonstration, and hand-authored score receipts for testing routing logic. The separate DistilBERT project script needs your reviewed dataset and a checkpoint download. None of the printed gate scores comes from that pretrained model, and a passing demonstration doesn't authorize deployment or shadow traffic.
Write the routing contract before training
Before training, give the model one narrow decision to make:
| Label | Route | Meaning |
|---|---|---|
1 | human_review_now | An access-recovery risk, suspected takeover, or policy exception requires a person before the agent workflow. |
0 | guarded_agent | A routine request may enter the agent, which still follows retrieval evidence and approval rules. |
Read the second row carefully. Label 0 doesn't mean "approve automatically." It means "safe to enter an automation path with its own controls." The classifier acts as an intake firewall. When a routine request enters guarded_agent, the downstream agent still has to retrieve permitted documentation and request human approval before executing any state change. But when an incident-freeze exception or account takeover enters the agent, the system risks unauthorized automated actions. That's why the routing boundary treats missed escalations as critical errors.

The model will produce a score, but the score isn't a route yet. A pinned threshold turns it into an intake lane, and either lane preserves the agent's downstream evidence and approval boundaries.
Carry the evidence forward
| Capstone artifact | Contract it contributes | What this classifier needs from it |
|---|---|---|
| Document QA product | An answer must cite permitted policy evidence. | Routine questions still need evidence checks after routing. |
| Evaluation dashboard | Hard failures block a release even when averages improve. | Missed urgent slices must block a classifier release. |
| Fine-tuned classifier | Ticket becomes a risk route under a pinned threshold. | Export the intake contract for the production agent. |
Write labeled fixtures before training. They make the route contract concrete and give support operations something to review before a model can hide an ambiguous case.
1label_version = "escalation-policy-v1"
2fixtures = [
3 {
4 "id": "access_override_exception",
5 "text": "The production API key is disabled, but the incident freeze still blocks restore.",
6 "gold_route": "human_review_now",
7 "reason": "access_override_exception",
8 },
9 {
10 "id": "incident_freeze_exception",
11 "text": "The production API key was disabled during an incident freeze. Please allow an exception.",
12 "gold_route": "human_review_now",
13 "reason": "incident_freeze_exception",
14 },
15 {
16 "id": "account_takeover",
17 "text": "Someone changed my recovery email and I can't sign in.",
18 "gold_route": "human_review_now",
19 "reason": "account_takeover",
20 },
21 {
22 "id": "routine_key_rotation",
23 "text": "Where is the key rotation guide?",
24 "gold_route": "guarded_agent",
25 "reason": "routine_key_rotation",
26 },
27 {
28 "id": "access_policy_question",
29 "text": "What is the access policy for disabled keys?",
30 "gold_route": "guarded_agent",
31 "reason": "access_policy_question",
32 },
33]
34
35allowed_routes = {"human_review_now", "guarded_agent"}
36assert all(row["gold_route"] in allowed_routes for row in fixtures)
37assert {row["reason"] for row in fixtures if row["gold_route"] == "human_review_now"} == {
38 "access_override_exception",
39 "incident_freeze_exception",
40 "account_takeover",
41}
42
43print("label guide:", label_version)
44for row in fixtures:
45 print(f"{row['id']}: {row['gold_route']} ({row['reason']})")1label guide: escalation-policy-v1
2access_override_exception: human_review_now (access_override_exception)
3incident_freeze_exception: human_review_now (incident_freeze_exception)
4account_takeover: human_review_now (account_takeover)
5routine_key_rotation: guarded_agent (routine_key_rotation)
6access_policy_question: guarded_agent (access_policy_question)A label guide should name ambiguous cases too. Frustrated tone alone isn't a reason for immediate human routing, while a calm request for an incident-freeze exception may be. Support operations should adjudicate disagreements and store that policy version beside every dataset generated from it.
Now the labels have meaning. The next question is whether future tickets can reach the evaluation split without bringing their account history with them.
Build a dataset you can defend
These rows form a hypothetical teaching dataset, not measured production traffic. Their job is to make failure modes visible:
| Field | Teaching dataset choice |
|---|---|
| Source | Synthetic access tickets modeled after access, incident-freeze, and account-recovery requests |
| Label version | escalation-policy-v1 |
| Exclusions | Spam, internal QA, tickets without requester text |
| High-risk slices | access_override_exception, incident_freeze_exception, account_takeover |
| Split policy | Earlier accounts train; later, unseen accounts validate and test |
| Demonstration gate | No missed positive in required gate slices, with at most one routine ticket sent to review |
For a real data card, replace each teaching assumption with provenance, consent and retention policy, labeling instructions, class balance, disagreement rate, and known coverage gaps.
Keep related tickets together
Multiple messages from one workspace account can share service names or incident wording. Putting one thread in training and another in validation can inflate an estimate meant to describe new accounts. For this target, use non-overlapping accounts and later time windows. If an account spans windows, exclude its earlier rows from training when reserving it for evaluation. Group splits and time splits solve different problems; neither alone enforces both.[3]
1records = [
2 {"ticket": "t01", "account": "acme", "month": 1, "split": "train"},
3 {"ticket": "t02", "account": "acme", "month": 2, "split": "train"},
4 {"ticket": "t03", "account": "north", "month": 2, "split": "train"},
5 {"ticket": "t04", "account": "north", "month": 2, "split": "train"},
6 {"ticket": "v01", "account": "summit", "month": 3, "split": "validation"},
7 {"ticket": "v02", "account": "summit", "month": 3, "split": "validation"},
8 {"ticket": "x01", "account": "harbor", "month": 4, "split": "test"},
9 {"ticket": "x02", "account": "harbor", "month": 4, "split": "test"},
10]
11
12accounts = {
13 split: {row["account"] for row in records if row["split"] == split}
14 for split in ("train", "validation", "test")
15}
16assert accounts["train"].isdisjoint(accounts["validation"])
17assert accounts["train"].isdisjoint(accounts["test"])
18assert accounts["validation"].isdisjoint(accounts["test"])
19assert len({row["ticket"] for row in records}) == len(records)
20months = {
21 split: [row["month"] for row in records if row["split"] == split]
22 for split in ("train", "validation", "test")
23}
24assert max(months["train"]) < min(months["validation"])
25assert max(months["validation"]) < min(months["test"])
26
27for split in ("train", "validation", "test"):
28 count = sum(row["split"] == split for row in records)
29 names = ",".join(sorted(accounts[split]))
30 print(f"{split}: {count} tickets, accounts={names}")1train: 4 tickets, accounts=acme,north
2validation: 2 tickets, accounts=summit
3test: 2 tickets, accounts=harborUse development data to pick features and checkpoints, then a disjoint selection partition to choose the threshold. Keep the final gate/test partition untouched until those choices are frozen. The eight-row manifest illustrates group and time checks; it isn't the actual split behind the later synthetic score tables. In a real pipeline, ticket IDs, normalized text hashes, account groups, and timestamps must join to the same frozen dataset.
The split output is simple on purpose: each account appears in one partition, and later months stand in for future traffic. That gives every later metric a clearer question to answer.
Two messages from account acme describe the same access incident. Why is putting one in training and one in validation invalid even when the text differs?
Answer
The rows share incident and account context, so the model can learn wording or facts from the training thread that reappear in validation. Keep the whole account or incident group on one side of the split, and reserve the final test groups until model and threshold choices are frozen.
Compare a cheap baseline first
Before paying for a contextual transformer, give a cheap baseline a chance. Start with a deterministic rule or a bag-of-words logistic-regression model. If it satisfies the routing gate on representative held-out data, a neural model may add maintenance without adding safety.
The rule below looks for contiguous phrases such as key disabled. Before reading its output, predict what happens to "Production API key disabled..." and to "The production API key was disabled during the incident freeze." The second sentence contains the same idea, but was sits between key and disabled.
1validation = [
2 ("explicit_access_override_exception", "Production API key disabled and the incident freeze blocks restore.", 1),
3 ("indirect_incident_freeze_exception", "The production API key was disabled during the incident freeze.", 1),
4 ("routine_key_rotation", "Where is the key rotation guide?", 0),
5 ("routine_policy", "Where can I read the access policy?", 0),
6]
7terms = ("key disabled", "can't sign in", "admin override")
8
9predictions = []
10for fixture_id, text, label in validation:
11 predicted = int(any(term in text.lower() for term in terms))
12 predictions.append((fixture_id, label, predicted))
13
14positives = sum(label for _, label, _ in predictions)
15true_positives = sum(label == 1 and predicted == 1 for _, label, predicted in predictions)
16missed = [fixture_id for fixture_id, label, predicted in predictions if label == 1 and predicted == 0]
17
18print(f"positive recall: {true_positives}/{positives}")
19print("missed positive:", missed[0])
20print("next experiment: fit a text-feature baseline")1positive recall: 1/2
2missed positive: indirect_incident_freeze_exception
3next experiment: fit a text-feature baselineThe missed phrase exposes a weakness in this rule, not a proof that a transformer is necessary. Fit TF-IDF plus logistic regression next. The vectorizer learns its vocabulary and inverse document frequencies from training rows only; a pipeline keeps that fit boundary explicit.[4][5]
This small comparison uses the same four held-out teaching tickets as the rule. It prints positive recall and false positives instead of rewarding an all-human classifier for perfect recall. The training rows are deliberately similar to the validation wording, so these results test the implementation, not realistic generalization.
1from sklearn.feature_extraction.text import TfidfVectorizer
2from sklearn.linear_model import LogisticRegression
3from sklearn.pipeline import make_pipeline
4
5baseline_train = [
6 ("prod key disabled incident freeze", 1),
7 ("prod key disabled during incident freeze", 1),
8 ("recovery email changed cannot sign in", 1),
9 ("where is key rotation guide", 0),
10 ("show key rotation guide", 0),
11 ("show access policy", 0),
12]
13baseline_validation = [
14 ("Production API key disabled and the incident freeze blocks restore.", 1),
15 ("The production API key was disabled during the incident freeze.", 1),
16 ("Where is the key rotation guide?", 0),
17 ("Where can I read the access policy?", 0),
18]
19baseline = make_pipeline(
20 TfidfVectorizer(ngram_range=(1, 2)),
21 LogisticRegression(C=1.0, max_iter=500, random_state=7),
22)
23baseline.fit([text for text, _ in baseline_train], [gold for _, gold in baseline_train])
24predicted = baseline.predict([text for text, _ in baseline_validation])
25tp = sum(gold == 1 and pred == 1 for (_, gold), pred in zip(baseline_validation, predicted))
26fn = sum(gold == 1 and pred == 0 for (_, gold), pred in zip(baseline_validation, predicted))
27fp = sum(gold == 0 and pred == 1 for (_, gold), pred in zip(baseline_validation, predicted))
28print(f"TF-IDF baseline: tp={tp} fn={fn} fp={fp}")
29print("validation predictions:", predicted.tolist())1TF-IDF baseline: tp=2 fn=0 fp=0
2validation predictions: [1, 1, 0, 0]Keep this baseline when it meets the actual acceptance criteria. Fine-tuning is a candidate experiment, not the required winner. Why do linear baselines eventually fall short on complex enterprise traffic? While TF-IDF handles straightforward vocabulary overlap, it fails on complex syntactic negation (such as "I am not requesting a key restore, just asking where the guide is"), clause embedding, and cross-sentence semantic dependencies. That's precisely where contextual encoders like DeBERTa-v3 and DistilBERT justify their compute footprint by capturing bidirectional semantic dependencies across the entire sequence.
Keep the comparison empirical. A fine-tuned encoder isn't cheaper, faster, or more accurate than prompting or this baseline until those candidates run on the same task and hardware.
Turn a score into a route
The model's raw output isn't a route yet. An encoder classifier emits one unnormalized logit for each class. Softmax turns the pair into class scores that sum to one. A separate decision threshold turns the positive-class score into the intake route.
The small implementation detail matters: subtract the largest logit before exponentiating so softmax stays numerically stable. With logits [-1.0, 2.0], predict which class should receive the larger score before running the function.
1from math import exp, isfinite
2
3examples = [
4 ("incident_freeze", [-1.0, 2.0]),
5 ("key_rotation", [1.0, -1.0]),
6]
7
8def softmax(logits: list[float]) -> list[float]:
9 if not logits or not all(isfinite(value) for value in logits):
10 raise ValueError("Expected nonempty finite logits")
11 offset = max(logits)
12 values = [exp(value - offset) for value in logits]
13 total = sum(values)
14 return [value / total for value in values]
15
16for fixture_id, logits in examples:
17 routine, human = softmax(logits)
18 print(f"{fixture_id}: human_review_score={human:.2f}, routine_score={routine:.2f}")1incident_freeze: human_review_score=0.95, routine_score=0.05
2key_rotation: human_review_score=0.12, routine_score=0.88Call 0.95 a score until held-out reliability evidence supports calling it a probability. A bounded output between zero and one isn't proof of calibration. The classification head computes an affine projection . Softmax normalizes those logits so they sum to one, but overparameterized neural networks routinely output overconfident scores on out-of-distribution inputs. The number is a continuous routing score for threshold comparison, not an objective Bayesian probability, until you measure calibration on held-out test data.[1]
Make the training mechanics visible
Before downloading a checkpoint, trace one trainable path end to end. The sandbox builds a tiny word-embedding encoder, averages token vectors, attaches a classification head, and updates every parameter with cross-entropy loss.
This is a classifier trained from scratch, not BERT fine-tuning. Its job is to make each moving part visible before random embeddings give way to pretrained representations.
1import torch
2from torch import nn
3from io import BytesIO
4
5torch.manual_seed(7)
6torch.set_num_threads(1)
7
8training_rows = [
9 ("prod key disabled incident freeze", 1),
10 ("prod key disabled during incident freeze", 1),
11 ("recovery email changed cannot sign in", 1),
12 ("where is key rotation guide", 0),
13 ("show key rotation guide", 0),
14 ("show access policy", 0),
15]
16vocabulary = {"<pad>": 0, "<unk>": 1}
17for text, _ in training_rows:
18 for token in text.split():
19 vocabulary.setdefault(token, len(vocabulary))
20
21def encode(texts: list[str]) -> torch.Tensor:
22 rows = [[vocabulary.get(token, 1) for token in text.lower().split()] for text in texts]
23 width = max(1, max(len(row) for row in rows))
24 return torch.tensor([row + [0] * (width - len(row)) for row in rows])
25
26inputs = encode([text for text, _ in training_rows])
27labels = torch.tensor([label for _, label in training_rows])
28
29class TinyEncoderClassifier(nn.Module):
30 def __init__(self, vocab_size: int) -> None:
31 super().__init__()
32 self.embedding = nn.Embedding(vocab_size, 8, padding_idx=0)
33 self.head = nn.Linear(8, 2)
34
35 def forward(self, token_ids: torch.Tensor) -> torch.Tensor:
36 mask = (token_ids != 0).unsqueeze(-1)
37 embedded = self.embedding(token_ids) * mask
38 pooled = embedded.sum(dim=1) / mask.sum(dim=1).clamp_min(1)
39 return self.head(pooled)
40
41model = TinyEncoderClassifier(len(vocabulary))
42criterion = nn.CrossEntropyLoss()
43optimizer = torch.optim.Adam(model.parameters(), lr=0.05)
44initial_loss = criterion(model(inputs), labels).item()
45original_head = model.head.weight.detach().clone()
46
47for _ in range(80):
48 loss = criterion(model(inputs), labels)
49 optimizer.zero_grad()
50 loss.backward()
51 optimizer.step()
52
53model.eval()
54with torch.inference_mode():
55 final_logits = model(inputs)
56 final_loss = criterion(final_logits, labels).item()
57print("logit shape:", tuple(final_logits.shape))
58print("loss decreased:", final_loss < initial_loss)
59print("classification head updated:", not torch.equal(original_head, model.head.weight))
60
61# Evaluate after fitting. This set was not used for gradient updates.
62held_out = [
63 ("prod key was disabled during incident freeze", 1),
64 ("recovery email changed cannot sign in today", 1),
65 ("where is the key rotation guide", 0),
66 ("show the access policy", 0),
67]
68with torch.inference_mode():
69 predictions = model(encode([text for text, _ in held_out])).argmax(dim=1).tolist()
70tp = sum(gold == 1 and pred == 1 for (_, gold), pred in zip(held_out, predictions))
71fn = sum(gold == 1 and pred == 0 for (_, gold), pred in zip(held_out, predictions))
72fp = sum(gold == 0 and pred == 1 for (_, gold), pred in zip(held_out, predictions))
73print(f"tiny held-out check: tp={tp} fn={fn} fp={fp}")
74
75# Verify a state-dict round trip without writing a deployment artifact.
76buffer = BytesIO()
77torch.save(model.state_dict(), buffer)
78buffer.seek(0)
79restored = TinyEncoderClassifier(len(vocabulary))
80restored.load_state_dict(torch.load(buffer, weights_only=True))
81restored.eval()
82with torch.inference_mode():
83 print("restored logits match:", torch.equal(final_logits, restored(inputs)))1logit shape: (6, 2)
2loss decreased: True
3classification head updated: True
4tiny held-out check: tp=2 fn=0 fp=0
5restored logits match: TrueThe sandbox can memorize six training examples. The four new strings check inference after fitting, but their near-duplicate wording doesn't establish generalization to new accounts. Follow the wiring: text becomes tokens, pooled representations become logits, and gradients update weights. The restore check uses the same vocabulary and architecture; saving weights without those would be incomplete.
One architecture gap is deliberate. This sandbox averages token vectors, so it can't distinguish word order: swapping tokens leaves the pooled representation unchanged. BERT-style sequence classification uses contextual token representations and a leading [CLS] summary.[1] Cross-entropy expects unnormalized logits, so don't apply softmax before passing the output to CrossEntropyLoss.[6]
Fine-tune a pretrained encoder
Now replace random embeddings with a pretrained encoder such as DistilBERT or DeBERTa-v3 and fine-tune it for two labels. Bidirectional encoders read context on both sides of each token simultaneously, avoiding the unidirectional attention blind spot of causal decoder models. DeBERTa-v3 improves classification accuracy by disentangling token content from relative position vectors, while DistilBERT distills the 12 transformer layers of BERT-base into 6 layers, running at double the inference speed. Both adapt to sequence classification by attaching a linear projection head to the pooled [CLS] summary vector.[1] Hugging Face exposes that sequence-classification workflow through AutoModelForSequenceClassification and Trainer.[2]
The script uses three epochs, a learning rate of , and batch size 16 as starting settings, not tuned values for this dataset. While DistilBERT's default configuration supports 512 positions, the serving policy here enforces a 256-token ceiling including special tokens.[7]
Our truncation rule is an explicit security control: never silently truncate inputs from the right. When an engineer pastes a long diagnostic log followed by a closing sentence like "The key failed during the incident freeze, please bypass", right-truncation chops off the critical risk signal at token 256. That would leave the classifier scoring only the routine preamble and mistakenly routing the ticket to automation. Instead, oversized tickets fail closed directly to human triage.
This project script isn't a marked, executed fine-tuning result. It needs your reviewed JSONL splits, the pinned checkpoint download, and torch, transformers, datasets, accelerate, and numpy. Each row needs a unique fixture_id, nonempty text, and integer label (0 for guarded_agent, 1 for human_review_now). Apply the account/time manifest before export. The script loads only training and checkpoint-development data; threshold-selection and final-gate rows stay outside training.
1import hashlib
2import json
3from pathlib import Path
4import numpy as np
5from datasets import load_dataset
6from transformers import (
7 AutoModelForSequenceClassification,
8 AutoTokenizer,
9 DataCollatorWithPadding,
10 Trainer,
11 TrainingArguments,
12 set_seed,
13)
14
15checkpoint = "distilbert/distilbert-base-uncased"
16revision = "12040accade4e8a0f71eabdb258fecc2e7e948be"
17run_root = Path("artifacts/encoder_v1")
18if run_root.exists():
19 raise FileExistsError("Use a new versioned directory for each training run")
20set_seed(7) # before initializing the new classification head
21dataset = load_dataset(
22 "json",
23 data_files={
24 "train": "data/train.jsonl",
25 "validation": "data/validation.jsonl",
26 },
27)
28tokenizer = AutoTokenizer.from_pretrained(checkpoint, revision=revision)
29seen_ids = set()
30seen_text = set()
31for partition in dataset.values():
32 if set(partition["label"]) != {0, 1}:
33 raise ValueError("Each development partition must contain both classes")
34 for row in partition:
35 if type(row["label"]) is not int or not isinstance(row["text"], str) or not row["text"].strip():
36 raise ValueError("Invalid label or blank text")
37 if row["fixture_id"] in seen_ids:
38 raise ValueError("Repeated fixture ID across development data")
39 seen_ids.add(row["fixture_id"])
40 normalized = " ".join(row["text"].lower().split())
41 if normalized in seen_text:
42 raise ValueError("Repeated normalized text in development data")
43 seen_text.add(normalized)
44
45def tokenize(batch):
46 encoded = tokenizer([text.strip() for text in batch["text"]], truncation=False)
47 if any(len(ids) > 256 for ids in encoded["input_ids"]):
48 raise ValueError("Oversize ticket requires the manual-input policy")
49 return encoded
50
51encoded = dataset.map(tokenize, batched=True)
52model = AutoModelForSequenceClassification.from_pretrained(
53 checkpoint,
54 revision=revision,
55 num_labels=2,
56 id2label={0: "guarded_agent", 1: "human_review_now"},
57 label2id={"guarded_agent": 0, "human_review_now": 1},
58)
59def compute_metrics(eval_pred):
60 logits, labels = eval_pred
61 predicted = np.argmax(logits, axis=-1)
62 tp = np.sum((predicted == 1) & (labels == 1))
63 fn = np.sum((predicted == 0) & (labels == 1))
64 fp = np.sum((predicted == 1) & (labels == 0))
65 return {"recall_at_argmax": float(tp / (tp + fn)), "false_positives_at_argmax": int(fp)}
66
67arguments = TrainingArguments(
68 output_dir="artifacts/encoder_v1",
69 learning_rate=2e-5,
70 per_device_train_batch_size=16,
71 per_device_eval_batch_size=32,
72 num_train_epochs=3,
73 eval_strategy="epoch",
74 save_strategy="epoch",
75 load_best_model_at_end=True,
76 metric_for_best_model="eval_loss",
77 greater_is_better=False,
78 seed=7,
79 data_seed=7,
80 report_to=[],
81 push_to_hub=False,
82)
83trainer = Trainer(
84 model=model,
85 args=arguments,
86 train_dataset=encoded["train"],
87 eval_dataset=encoded["validation"],
88 processing_class=tokenizer,
89 data_collator=DataCollatorWithPadding(tokenizer=tokenizer),
90 compute_metrics=compute_metrics,
91)
92trainer.train()
93print(trainer.evaluate(encoded["validation"]))
94export = Path("artifacts/encoder_v1/model")
95trainer.save_model(str(export))
96tokenizer.save_pretrained(str(export))
97files = {
98 path.name: hashlib.sha256(path.read_bytes()).hexdigest()
99 for path in sorted(export.iterdir()) if path.is_file()
100}
101manifest = {
102 "base_checkpoint": checkpoint, "base_revision": revision,
103 "selected_checkpoint": trainer.state.best_model_checkpoint,
104 "label2id": model.config.label2id, "max_length": 256,
105 "seed": 7, "sha256": files,
106}
107Path("artifacts/encoder_v1/export-manifest.json").write_text(json.dumps(manifest, indent=2))Minimum development loss chooses a checkpoint here; it doesn't approve a routing policy. load_best_model_at_end restores the checkpoint selected by that metric.[8] The reported argmax recall uses the implicit two-class 0.5 boundary, not the later 0.35 routing threshold (at an exact logit tie, NumPy's argmax chooses class 0). Selecting only maximum recall could favor sending everything to people. Compare candidates using required-slice misses and review capacity after threshold selection, and record exact dependency versions with the export.
The base revision identifies pretrained weights, while the export hashes identify your fine-tuned model and tokenizer. A human-readable name such as encoder_v1 is only a label. Reload the exported directory, verify its hashes against an independently retained manifest, and use model.eval() with torch.inference_mode() when scoring. Take softmax(logits, dim=-1)[:, 1]: a generic pipeline's confidence in its winning label is not always the human-review score.
After completing the project training run, this separate scorer reloads its local export and writes one score per evaluator-owned fixture. gate-fixtures.json is a list of objects containing fixture_id and text; the trusted evaluator retains labels separately. Hash checks detect mismatched files, not a malicious replacement of both the files and their manifest. Store the approved manifest outside the scorer's write boundary.
1import hashlib
2import json
3from pathlib import Path
4import torch
5from transformers import AutoModelForSequenceClassification, AutoTokenizer
6
7export = Path("artifacts/encoder_v1/model")
8manifest = json.loads(Path("artifacts/encoder_v1/export-manifest.json").read_text())
9for name, expected in manifest["sha256"].items():
10 if hashlib.sha256((export / name).read_bytes()).hexdigest() != expected:
11 raise ValueError(f"Export mismatch: {name}")
12tokenizer = AutoTokenizer.from_pretrained(export, local_files_only=True)
13classifier = AutoModelForSequenceClassification.from_pretrained(export, local_files_only=True)
14if classifier.config.label2id != {"guarded_agent": 0, "human_review_now": 1}:
15 raise ValueError("Positive-class mapping changed")
16classifier.eval()
17fixtures = json.loads(Path("gate-fixtures.json").read_text())
18scores = []
19for fixture in fixtures:
20 text = fixture["text"].strip()
21 if not text:
22 raise ValueError("Blank gate input")
23 encoded = tokenizer(text, return_tensors="pt", truncation=False)
24 if encoded["input_ids"].shape[1] > manifest["max_length"]:
25 raise ValueError("Gate input exceeds serving policy; route manually")
26 with torch.inference_mode():
27 score = torch.softmax(classifier(**encoded).logits, dim=-1)[0, 1].item()
28 scores.append({"fixture_id": fixture["fixture_id"], "score": score})
29Path("gate-scores.json").write_text(json.dumps(scores, indent=2, allow_nan=False))Run this only after freezing the model and threshold. Retain the scored fixture-file hash, export-manifest hash, tokenizer policy, and package versions with the output. The next sections exercise the evaluator with synthetic values instead of pretending this scoring run has happened.
Select a threshold on a selection set, not the freeze gate
Don't accept a library default threshold of 0.50 without auditing its asymmetric error tradeoff. A default 0.50 cutoff assumes symmetric loss, treating a missed security escalation as mathematically equivalent to sending a routine question to a person. In enterprise access triage, the cost matrix is heavily asymmetric: false negatives on critical risk categories (account takeover, incident-freeze bypass, privilege elevation) risk severe compromise, while false positives merely add two minutes of human review.
Under asymmetric costs, the Bayes-optimal threshold shifts downward:
When , is much lower than 0.50. Choose that operational cutoff on a dedicated selection set, then evaluate a separate frozen gate receipt for release. Tuning until required false negatives disappear on the same rows you later treat as release evidence contaminates the gate.
The next scores are hand-authored numbers for the threshold exercise, not exports from either trained example. Suppose a future, independently evaluated encoder_v1 produced them. Which cutoff retains required positives without flooding the human queue?

| Fixture | Role | Slice | Gold | Score |
|---|---|---|---|---|
access_override_exception | selection | access_override_exception | 1 | 0.89 |
incident_freeze_exception | selection | incident_freeze_exception | 1 | 0.42 |
account_takeover | selection | account_takeover | 1 | 0.64 |
angry_key_rotation | selection | routine_key_rotation | 0 | 0.71 |
routine_key_rotation | selection | routine_key_rotation | 0 | 0.23 |
access_policy_question | selection | access_policy_question | 0 | 0.18 |
1# Selection set: used only to propose a threshold. Not the release gate.
2selection_rows = [
3 ("access_override_exception", 1, 0.89),
4 ("incident_freeze_exception", 1, 0.42),
5 ("account_takeover", 1, 0.64),
6 ("angry_key_rotation", 0, 0.71),
7 ("routine_key_rotation", 0, 0.23),
8 ("access_policy_question", 0, 0.18),
9]
10
11def summarize(threshold: float) -> tuple[int, int, int, float]:
12 tp = fp = fn = 0
13 for _, gold, score in selection_rows:
14 predicted = int(score >= threshold)
15 tp += predicted == 1 and gold == 1
16 fp += predicted == 1 and gold == 0
17 fn += predicted == 0 and gold == 1
18 recall = tp / (tp + fn)
19 return tp, fp, fn, recall
20
21for threshold in (0.35, 0.50, 0.75):
22 tp, fp, fn, recall = summarize(threshold)
23 print(f"threshold={threshold:.2f} tp={tp} fp={fp} fn={fn} recall={recall:.2f}")1threshold=0.35 tp=3 fp=1 fn=0 recall=1.00
2threshold=0.50 tp=2 fp=1 fn=1 recall=0.67
3threshold=0.75 tp=1 fp=0 fn=2 recall=0.33The sweep answers one narrow question. On this selection set, threshold 0.35 is the only candidate shown that avoids a missed escalation. It sends one angry but routine key-rotation ticket to human review, so the tradeoff is acceptable only if queue capacity allows it.
After locking a threshold, score a frozen gate receipt that wasn't used to choose it. Nested cross-validation is one way to separate model selection from evaluation when data is limited; it isn't a replacement for this application's group, time, and leakage checks.[3]
Don't report 1.00 recall from three positive selection fixtures as proof of production reliability. It's evidence for this selection slice, not for unseen traffic.
Threshold 0.35 has zero false negatives on the selection set, while 0.50 misses one required escalation. What must happen before 0.35 can control shadow traffic?
Answer
Freeze 0.35 as part of a versioned serving bundle, then score a disjoint gate receipt that wasn't used to choose it. The gate must still satisfy required-slice false-negative and queue-capacity limits; selection-set recall alone can't authorize release.
Freeze the evidence gate
The dashboard capstone compared runs only when dataset, grader, corpus, and fixture set matched. Carry that discipline here. A classifier receipt pins its dataset, label guide, split manifest, model, and exact gate fixture IDs before scores are aggregated. The evaluator owns the fixture manifest, including each fixture's gold route and required slice. A scoring worker may submit only a fixture ID and the model's score. It can't submit a replacement label or slice. Gate fixtures must not be the set used to choose the threshold.
In the simulation, intake_bundle_v1 uses threshold 0.50 and misses freeze_window_ticket at score 0.41. intake_bundle_v2 uses 0.35 and passes the arithmetic gate. The two comparisons illustrate a policy change, not permission to retune after seeing the final gate. Real gate feedback becomes development data once used for tuning, and the next candidate needs new independent evidence. A threshold change versions the serving bundle, not the model weights.
Here eligible_for_shadow is a demonstration return value. These synthetic scores can't establish eligibility for real shadow traffic. That decision also needs approved data handling, measured model outputs, sufficient slice coverage, and operational review.
The evaluator checks declared model identity, bounded scores, and exact fixture coverage. It owns gold routes, slice membership, and bundle thresholds; submitted replacements are rejected. These checks don't prove which model ran or which text it saw. A real harness must score evaluator-owned inputs and retain authenticated output provenance. Disjoint fixture IDs also don't prove disjoint text, accounts, or incidents; check the split manifest and duplicates before scoring.
1from collections import Counter
2from dataclasses import dataclass
3from types import MappingProxyType
4
5EXPECTED_IDENTITY = {
6 "dataset_version": "access-intake-gate-v1",
7 "label_version": "escalation-policy-v1",
8 "split_manifest_version": "access-intake-split-2026-05",
9 "model_version": "encoder_v1",
10 "threshold_selection_set": "access-intake-selection-v1",
11}
12# The frozen gate receipt uses fixture IDs that never appeared in the threshold
13# selection set above. It is scored only after the threshold is locked.
14SELECTION_FIXTURES = {
15 "access_override_exception",
16 "incident_freeze_exception",
17 "account_takeover",
18 "angry_key_rotation",
19 "routine_key_rotation",
20 "access_policy_question",
21}
22TRUSTED_BUNDLES = MappingProxyType({
23 "intake_bundle_v1": MappingProxyType({"threshold": 0.50}),
24 "intake_bundle_v2": MappingProxyType({"threshold": 0.35}),
25})
26
27@dataclass(frozen=True)
28class FixtureSpec:
29 text: str
30 slice: str
31 gold: int
32
33# This in-process fixture mapping is read-only. That is not authentication:
34# the production evaluator and scoring worker need separate trust boundaries.
35AUTHORITATIVE_FIXTURES = MappingProxyType({
36 "after_hours_access_override": FixtureSpec(
37 "The production API key is disabled outside business hours. Please make an exception.",
38 "access_override_exception",
39 1,
40 ),
41 "freeze_window_ticket": FixtureSpec(
42 "The production API key was disabled during an incident freeze. Please restore it.",
43 "incident_freeze_exception",
44 1,
45 ),
46 "takeover_recovery_request": FixtureSpec(
47 "Someone changed the recovery email and I can't sign in.",
48 "account_takeover",
49 1,
50 ),
51 "routine_rotation_followup": FixtureSpec(
52 "The key rotation guide failed. Can I see it again?",
53 "routine_key_rotation",
54 0,
55 ),
56 "routine_access_question": FixtureSpec(
57 "Which document explains who may rotate a service key?",
58 "access_policy_question",
59 0,
60 ),
61 "routine_status_request": FixtureSpec(
62 "Can you tell me whether the key rotation completed?",
63 "routine_key_rotation",
64 0,
65 ),
66})
67EXPECTED_FIXTURES = frozenset(AUTHORITATIVE_FIXTURES)
68REQUIRED_SLICES = frozenset(
69 spec.slice for spec in AUTHORITATIVE_FIXTURES.values() if spec.gold == 1
70)
71MAX_FALSE_POSITIVES = 1 # queue capacity for this teaching receipt
72# Production scores must come from the pinned model runner. The values below
73# are synthetic fixtures for the evaluator, not output from a trained model.
74SUBMITTED_SCORE_FIELDS = frozenset({"fixture_id", "score"})
75gate_scores = [
76 {"fixture_id": "after_hours_access_override", "score": 0.87},
77 {"fixture_id": "freeze_window_ticket", "score": 0.41},
78 {"fixture_id": "takeover_recovery_request", "score": 0.62},
79 {"fixture_id": "routine_rotation_followup", "score": 0.69},
80 {"fixture_id": "routine_access_question", "score": 0.21},
81 {"fixture_id": "routine_status_request", "score": 0.14},
82]
83assert REQUIRED_SLICES == {
84 "access_override_exception",
85 "incident_freeze_exception",
86 "account_takeover",
87}
88
89def receipt(bundle_version: str, rows: list[dict[str, object]], **overrides: object) -> dict[str, object]:
90 return {
91 **EXPECTED_IDENTITY,
92 "bundle_version": bundle_version,
93 "threshold": TRUSTED_BUNDLES[bundle_version]["threshold"],
94 "rows": rows,
95 **overrides,
96 }
97
98def release_decision(run: dict[str, object]) -> tuple[str, str]:
99 if not isinstance(run, dict):
100 return "hold", "invalid:receipt"
101 for field, expected in EXPECTED_IDENTITY.items():
102 actual = run.get(field)
103 if actual != expected:
104 return "hold", f"{field}:{actual}"
105 bundle_version = run.get("bundle_version")
106 if not isinstance(bundle_version, str):
107 return "hold", f"bundle_version:{bundle_version}"
108 bundle = TRUSTED_BUNDLES.get(bundle_version)
109 if bundle is None:
110 return "hold", f"bundle_version:{bundle_version}"
111 # A submitted threshold is metadata, not policy. Use the evaluator's
112 # pinned bundle manifest for every comparison and hold a report that disagrees.
113 if run.get("threshold") != bundle["threshold"]:
114 return "hold", f"threshold:{run.get('threshold')}"
115 rows = run.get("rows")
116 if not isinstance(rows, list) or any(not isinstance(row, dict) for row in rows):
117 return "hold", "invalid:rows"
118
119 scores_by_fixture = {}
120 fixture_ids = []
121 for row in rows:
122 if set(row) != SUBMITTED_SCORE_FIELDS:
123 fields = sorted(str(field) for field in set(row) ^ SUBMITTED_SCORE_FIELDS)
124 return "hold", f"invalid:submitted_fields:{','.join(fields)}"
125 fixture_id = row["fixture_id"]
126 if not isinstance(fixture_id, str):
127 return "hold", "invalid:fixture_id"
128 if fixture_id in SELECTION_FIXTURES:
129 return "hold", f"selection_reuse:{fixture_id}"
130 if fixture_id not in AUTHORITATIVE_FIXTURES:
131 return "hold", f"unexpected:{fixture_id}"
132 raw_score = row["score"]
133 # JSON numbers only: booleans and numeric-looking strings are not scores.
134 if type(raw_score) not in (int, float) or not 0.0 <= raw_score <= 1.0:
135 return "hold", f"invalid:score:{fixture_id}"
136 score = float(raw_score)
137 fixture_ids.append(fixture_id)
138 scores_by_fixture[fixture_id] = score
139
140 counts = Counter(fixture_ids)
141 missing = sorted(EXPECTED_FIXTURES - set(fixture_ids))
142 unexpected = sorted(set(fixture_ids) - EXPECTED_FIXTURES)
143 duplicated = sorted(fixture_id for fixture_id, count in counts.items() if count != 1)
144 if missing:
145 return "hold", f"missing:{','.join(missing)}"
146 if unexpected:
147 return "hold", f"unexpected:{','.join(unexpected)}"
148 if duplicated:
149 return "hold", f"duplicate:{','.join(duplicated)}"
150 threshold = bundle["threshold"]
151 misses = [
152 fixture_id
153 for fixture_id in fixture_ids
154 if AUTHORITATIVE_FIXTURES[fixture_id].slice in REQUIRED_SLICES
155 and AUTHORITATIVE_FIXTURES[fixture_id].gold == 1
156 and scores_by_fixture[fixture_id] < threshold
157 ]
158 if misses:
159 return "hold", f"missed:{','.join(misses)}"
160 false_positives = sum(
161 1
162 for fixture_id, spec in AUTHORITATIVE_FIXTURES.items()
163 if spec.gold == 0 and scores_by_fixture[fixture_id] >= threshold
164 )
165 if false_positives > MAX_FALSE_POSITIVES:
166 return "hold", f"false_positives:{false_positives}>max:{MAX_FALSE_POSITIVES}"
167 return "eligible_for_shadow", "exact_receipt_pass"
168
169runs = {
170 "intake_bundle_v1": receipt("intake_bundle_v1", gate_scores),
171 "intake_bundle_v2": receipt("intake_bundle_v2", gate_scores),
172 "intake_bundle_incomplete": receipt(
173 "intake_bundle_v2",
174 [row for row in gate_scores if row["fixture_id"] != "takeover_recovery_request"],
175 ),
176 "intake_bundle_padded": receipt(
177 "intake_bundle_v2",
178 [*gate_scores, {"fixture_id": "easy_status_extra", "score": 0.02}],
179 ),
180 "intake_bundle_duplicated": receipt("intake_bundle_v2", [*gate_scores, gate_scores[-1]]),
181 "intake_bundle_drifted": receipt("intake_bundle_v2", gate_scores, dataset_version="access-intake-gate-v2"),
182 "intake_bundle_wrong_threshold": receipt("intake_bundle_v2", gate_scores, threshold=0.20),
183 "intake_bundle_tampered_labels": receipt(
184 "intake_bundle_v2",
185 [
186 {**row, "slice": "routine_key_rotation", "gold": 0}
187 if row["fixture_id"] == "freeze_window_ticket"
188 else row
189 for row in gate_scores
190 ],
191 ),
192 "intake_bundle_invalid_score": receipt(
193 "intake_bundle_v2",
194 [*gate_scores[:-1], {"fixture_id": "routine_status_request", "score": float("nan")}],
195 ),
196 "intake_bundle_string_score": receipt(
197 "intake_bundle_v2",
198 [*gate_scores[:-1], {"fixture_id": "routine_status_request", "score": "0.14"}],
199 ),
200 "intake_bundle_fp_flood": receipt(
201 "intake_bundle_v2",
202 [
203 {
204 **row,
205 "score": (
206 0.90
207 if AUTHORITATIVE_FIXTURES[row["fixture_id"]].gold == 0
208 else row["score"]
209 ),
210 }
211 for row in gate_scores
212 ],
213 ),
214}
215for bundle_version, run in runs.items():
216 decision, reason = release_decision(run)
217 print(f"{bundle_version}: decision={decision} reason={reason}")1intake_bundle_v1: decision=hold reason=missed:freeze_window_ticket
2intake_bundle_v2: decision=eligible_for_shadow reason=exact_receipt_pass
3intake_bundle_incomplete: decision=hold reason=missing:takeover_recovery_request
4intake_bundle_padded: decision=hold reason=unexpected:easy_status_extra
5intake_bundle_duplicated: decision=hold reason=duplicate:routine_status_request
6intake_bundle_drifted: decision=hold reason=dataset_version:access-intake-gate-v2
7intake_bundle_wrong_threshold: decision=hold reason=threshold:0.2
8intake_bundle_tampered_labels: decision=hold reason=invalid:submitted_fields:gold,slice
9intake_bundle_invalid_score: decision=hold reason=invalid:score:routine_status_request
10intake_bundle_string_score: decision=hold reason=invalid:score:routine_status_request
11intake_bundle_fp_flood: decision=hold reason=false_positives:3>max:1The rejected label mutation never reaches the gate calculation: gold and slice aren't part of the scoring-worker schema, so the evaluator keeps its own values. A malformed score also holds before aggregation, and a report that changes the threshold can't replace the evaluator's pinned policy. Lowering the threshold isn't automatically the right fix. If false positives overwhelm reviewers, the executable gate holds on max_false_positives rather than only on required-slice FNs. This receipt therefore keeps threshold selection separate from the freeze gate, refuses to hide a required-slice miss inside an average, and rejects easy rows that pad the comparison.
Audit scores without overclaiming calibration
A threshold can be useful even when the score isn't a well-calibrated probability. Calibration asks whether tickets scored near 0.70, across enough held-out examples, need human review about 70 percent of the time. Modern neural classifiers can be miscalibrated; temperature scaling is one common post-training adjustment, fit on held-out data, that rescales logits without changing the argmax class.[9]
That last detail matters for this product. Guo et al. showed temperature scaling leaves predicted labels unchanged because it doesn't change which logit is largest. This router doesn't use argmax. It uses score >= 0.35 on the positive class, so rescaling logits can move a ticket across the cutoff even when the predicted class stays the same. Fit temperature on held-out data, then re-run the gate receipt before that change enters the serving bundle.
Six fixtures are enough to demonstrate a calculation, not to justify operator-facing percentages:
1scored_rows = [
2 (0.89, 1),
3 (0.64, 1),
4 (0.42, 1),
5 (0.71, 0),
6 (0.23, 0),
7 (0.18, 0),
8]
9buckets = {
10 "low [0.0,0.4)": [(score, gold) for score, gold in scored_rows if score < 0.4],
11 "middle [0.4,0.7)": [(score, gold) for score, gold in scored_rows if 0.4 <= score < 0.7],
12 "high [0.7,1.0]": [(score, gold) for score, gold in scored_rows if score >= 0.7],
13}
14
15for name, bucket in buckets.items():
16 if not bucket:
17 print(f"{name}: no observations")
18 continue
19 average_score = sum(score for score, _ in bucket) / len(bucket)
20 observed_rate = sum(gold for _, gold in bucket) / len(bucket)
21 print(f"{name}: n={len(bucket)}, score={average_score:.2f}, observed={observed_rate:.2f}")
22print("decision: diagnostic only; collect more held-out labels")1low [0.0,0.4): n=2, score=0.21, observed=0.00
2middle [0.4,0.7): n=2, score=0.53, observed=1.00
3high [0.7,1.0]: n=2, score=0.80, observed=0.50
4decision: diagnostic only; collect more held-out labelsThe high bucket exposes why calibrated language matters: an average score of 0.80 with one positive among two tickets isn't evidence that 0.80 means an 80 percent escalation rate. These are the same synthetic selection scores, not a fresh calibration evaluation. If you fit a temperature, choose the threshold under that transformed score scale and evaluate both together on untouched data.
Ship the serving contract
A model file alone can't reproduce the route. Deploying a serialized weights file (model.safetensors) without its surrounding runtime environment creates silent production drift. If a serving container boots with a newer tokenizer release that splits domain tokens differently, or if an engineer leaves an implicit 0.50 cutoff in place, routing decisions break even though the weights hash matches.
The production release artifact is an immutable serving bundle:
| Component | Versioned value |
|---|---|
| Label guide | escalation-policy-v1 |
| Gate receipt | access-intake-gate-v1, six exact fixtures (held out from threshold selection) |
| Dataset split | access-intake-split-2026-05 |
| Tokenizer and model | encoder_v1 alias plus verified export hashes and base revision |
| Serving release | intake_bundle_v2 |
| Input policy | access-intake-input-v1: trim outer whitespace, preserve punctuation, reject over 256 tokenizer tokens including special tokens |
| Threshold | 0.35, chosen on access-intake-selection-v1, then frozen for the gate |
| Failure fallback | human_review_now |
| Downstream consumer | access_agent_v2 |
![Fail-closed intake graph for intake_bundle_v2. A ticket is checked for a finite score in [0, 1]. Blank text, a down model, or NaN go to human_review_now. A valid score at or above 0.35 also goes to a person. Only a valid score below 0.35 may enter the guarded agent. The admission table then accepts r-104 from bundle v2 with route guarded_agent, blocks r-105 as human review, and blocks r-106 because its bundle provenance is still v1.](/cdn/content-image/foundations/capstone-fine-tuned-classifier/illustrations/_generated/serving_contract_bundle_dark.png?v=ff57d0d8d184)
The endpoint must fail closed. In Python, evaluating float("nan") >= 0.35 returns False. If your code naively checks if score >= threshold: return "human_review_now" followed by a default fallthrough, any NaN output from a floating-point overflow or model exception would slip directly into guarded_agent! That is a catastrophic fail-open vulnerability.
A defensive intake contract enforces strict verification order:
- Validate input text: empty, whitespace-only, or malformed inputs immediately take
fallback_route. - Verify token length: inputs exceeding 256 tokens bypass the model and take
fallback_route. - Check model availability: timeouts or worker crashes take
fallback_route. - Validate the score: non-numeric,
None, or non-finite values (NaN,Inf) takefallback_route. - Only finite numeric scores strictly within are evaluated against threshold
0.35.
The trusted tokenizer counts the full input before model inference; a client-supplied token count isn't acceptable. The function below tests the routing layer with supplied scores and token counts, not the tokenizer or model itself. Blank, malformed, oversized, and unavailable inputs stay out of automation.
1bundle = {
2 "bundle_version": "intake_bundle_v2",
3 "label_version": "escalation-policy-v1",
4 "gate_receipt_version": "access-intake-gate-v1",
5 "split_manifest_version": "access-intake-split-2026-05",
6 "tokenizer_version": "encoder_v1",
7 "model_version": "encoder_v1",
8 "input_policy_version": "access-intake-input-v1",
9 "max_length": 256,
10 "threshold": 0.35,
11 "fallback_route": "human_review_now",
12}
13
14def route_ticket(text: object, score: object, token_count: object) -> tuple[str, str]:
15 if not isinstance(text, str):
16 return bundle["fallback_route"], "invalid_text"
17 if not text.strip():
18 return bundle["fallback_route"], "empty_text"
19 if type(token_count) is not int or token_count <= 0:
20 return bundle["fallback_route"], "invalid_token_count"
21 if token_count > bundle["max_length"]:
22 return bundle["fallback_route"], "input_too_long"
23 if score is None:
24 return bundle["fallback_route"], "model_unavailable"
25 if type(score) not in (int, float) or not 0.0 <= score <= 1.0:
26 return bundle["fallback_route"], "invalid_score"
27 if score >= bundle["threshold"]:
28 return "human_review_now", "threshold"
29 return "guarded_agent", "below_threshold"
30
31checks = [
32 ("Production API key disabled during the incident freeze", 0.72, 12, "human_review_now", "threshold"),
33 ("Where is the access policy?", 0.18, 8, "guarded_agent", "below_threshold"),
34 ("", 0.04, 2, "human_review_now", "empty_text"),
35 ("Where is the key rotation guide?", None, 9, "human_review_now", "model_unavailable"),
36 ("Malformed model score", float("nan"), 5, "human_review_now", "invalid_score"),
37 ("Boolean is not a score", False, 8, "human_review_now", "invalid_score"),
38 ("Oversize ticket fixture", 0.01, 257, "human_review_now", "input_too_long"),
39]
40print("bundle:", bundle["bundle_version"], bundle["gate_receipt_version"])
41for text, score, token_count, expected_route, expected_reason in checks:
42 route, reason = route_ticket(text, score, token_count)
43 assert (route, reason) == (expected_route, expected_reason)
44 print(f"{route}: {reason}")1bundle: intake_bundle_v2 access-intake-gate-v1
2human_review_now: threshold
3guarded_agent: below_threshold
4human_review_now: empty_text
5human_review_now: model_unavailable
6human_review_now: invalid_score
7human_review_now: invalid_score
8human_review_now: input_too_longAdmit only pinned routes
A valid serving bundle still needs an explicit downstream handoff contract. In production deployments, multiple model versions run concurrently across canary environments, rolling updates, and blue-green clusters. If an intake payload generated by an experimental canary bundle or an outdated service instance (intake_bundle_v1) reaches the downstream production agent, admitting it risks executing automated actions under an unapproved routing policy.
The downstream agent therefore enforces three admission criteria:
- Pinned bundle identity: The ticket's
bundle_versionmust match the agent's pinned active configuration (intake_bundle_v2). - Route verification: The assigned route must strictly equal
guarded_agent. - Producer provenance: The intake artifact must originate from the authenticated classifier service.
Matching a caller-supplied version string alone isn't cryptographic authentication; a production pipeline signs provenance tokens. Even after admission, the agent capstone enforces independent security boundaries: document QA retrieval grounding and human approval gates before executing state changes.
1handoff = {
2 "producer": "support_ticket_escalation_classifier",
3 "bundle_version": "intake_bundle_v2",
4 "model_version": "encoder_v1",
5 "threshold": 0.35,
6 "allowed_agent_route": "guarded_agent",
7 "blocked_or_manual_route": "human_review_now",
8 "policy_evidence_service": "document_qa_v2",
9 "evaluation_report": "classifier_dashboard_run_intake_bundle_v2",
10}
11
12agent_inputs = [
13 {"ticket_id": "r-104", "bundle_version": "intake_bundle_v2", "route": "guarded_agent"},
14 {"ticket_id": "r-105", "bundle_version": "intake_bundle_v2", "route": "human_review_now"},
15 {"ticket_id": "r-106", "bundle_version": "intake_bundle_v1", "route": "guarded_agent"},
16]
17
18def admitted_by_pinned_bundle(row: dict[str, str]) -> bool:
19 return (
20 isinstance(row, dict)
21 and isinstance(row.get("ticket_id"), str)
22 and bool(row["ticket_id"].strip())
23 and row.get("bundle_version") == handoff["bundle_version"]
24 and row.get("route") == handoff["allowed_agent_route"]
25 )
26
27accepted = [
28 row["ticket_id"]
29 for row in agent_inputs
30 if admitted_by_pinned_bundle(row)
31]
32blocked = [
33 row["ticket_id"]
34 for row in agent_inputs
35 if not admitted_by_pinned_bundle(row)
36]
37
38print("agent accepts:", ",".join(accepted))
39print("agent blocked:", ",".join(blocked))
40print("agent authority: draft_with_evidence_and_request_approval")1agent accepts: r-104
2agent blocked: r-105,r-106
3agent authority: draft_with_evidence_and_request_approvalThe handoff now has four traceable boundaries:
- Document QA supplies evidence-bound policy answers.
- Exact-receipt evaluation rows and gates expose failures before release.
- The classifier proposes a route, with fail-closed handling for invalid inputs.
- The agent uses only admitted routine intake and still needs approval for consequential actions.
Keep the release evidence together
A notebook screenshot can't answer the release questions. Keep these files or equivalent artifacts together:
| Artifact | Question it answers |
|---|---|
label_guide.md | What does human_review_now mean, including edge cases? |
data_card.md | Where did rows come from and what traffic is missing? |
| Frozen split manifest | Did account or time leakage contaminate evaluation? |
| Evaluator-owned fixture manifest | Which exact ticket text, gold route, and required slice define the gate? |
| Baseline report | Did the encoder solve a measured baseline failure? |
| Training configuration | Which checkpoint, seed, hyperparameters, and label mapping produced the model? |
| Versioned scoring output | Did the pinned model return one finite score for every exact fixture ID? |
| Evaluator report | Which required-slice failures passed or blocked release under the manifest's gold labels? |
| Serving bundle | Are tokenizer, threshold, fallback, and model restored together? |
| Shadow-monitor plan | Which reviewed production sample can trigger rollback? |
Before changing live routing, run shadow traffic. Record scores and proposed routes, retain human decisions, and compare missed-escalation rate by required slice. Roll back to manual review when a required slice misses its agreed floor; don't wait for average accuracy to fall.
Practice: break the intake contract
Treat each mutation as a controlled failure. Predict which evidence should reject it, run the relevant cells, then revert that mutation before starting the next.
- Inspect
intake_bundle_incomplete. Why does it hold even though every remaining score clears threshold0.35? - Inspect
intake_bundle_padded,intake_bundle_duplicated, andintake_bundle_drifted. Why can't easy extras, duplicate rows, or a newer dataset silently change the release decision? - Re-version the serving policy at threshold
0.50whilefreeze_window_ticketstill scores0.41. Which required fixture blocks shadow eligibility? - Inspect
intake_bundle_fp_flood. Why does it hold even though required-slice positives still clear a low threshold? - Change
fallback_routetoguarded_agent. Whichserving-fallback.pyassertions fail, and why is that unsafe? - Remove the bundle-version comparison from
admitted_by_pinned_bundle. Which stale route artifact now reaches the agent? - Inspect
intake_bundle_tampered_labelsandintake_bundle_invalid_score. Which submitted fields does the evaluator trust, and why must the other fields hold before any metric is computed?
What should each mutation prove?
Answer
- Missing fixture evidence must fail closed; absence isn't a passing result. 2. The receipt rejects unexpected easy rows, duplicate rows, and dataset drift before aggregation. 3.
freeze_window_ticketblocks the re-versioned threshold, because its held-out score 0.41 falls below 0.50. 4.max_false_positivesholds a threshold that floods the review queue even when required FNs are zero. 5. Blank input, unavailable model, and malformed score must stay out of automation. 6.r-106must stay blocked because a route string without pinned-bundle provenance is stale evidence. 7. Onlyfixture_idand a finite score belong in submitted rows. The evaluator owns fixture text, gold, and slice values, and it must reject added labels or non-finite scores before aggregation.
What the agent receives next
The demonstration has tested threshold arithmetic, exact fixture coverage, fail-closed routing, and version checks. It hasn't deployed or validated a pretrained classifier. To finish the project, replace synthetic scores with authenticated outputs from the hashed export, compare the baseline on the same untouched gate, and obtain operational approval. The later agent capstone must retain these boundaries while adding retrieval, approval gates, traces, and evaluation.