Chapter 8 — Alternatives to Jev & Multi-Engine Benchmarks

Previously: Chapter 7 closed the loop: decide → gate → execute → verify.

You will build: openai_decisions_client.py, a DecisionClient for the OpenAI Decisions API, and multi_engine_eval.py, a harness that runs the same decision contract on the same labeled set across Jev, Kev, OpenAI Decisions and an LLM adapter. It reports accuracy, latency, cost, schema errors and calibration.

Capstone increment: benchmark.py plus a documented engine choice for the triage service, and a cascade that escalates only uncertain cases to a stronger model.

8.0 The problem

You have built a whole engine on Jev. A reasonable colleague will now ask: “Why Jev? Is something cheaper, faster or safer for us?” Answering with a vendor’s benchmark page isn’t good enough, because vendors publish their best case on their own data.

The honest answer is to measure on your own cases. Everything in Chapters 2 to 7 sits behind DecisionClient. That is what makes this chapter cheap: swapping the engine is a constructor change, and the labeled set from Chapter 6 is the referee.

💡 Mental Model: a driving test, not a brochure You wouldn’t pick a taxi driver from their brochure. You’d put each driver behind the wheel on the same route and time them. The labeled set is the route; DecisionClient is the car every driver sits in.

This chapter only benchmarks engines you can run today: hosted Jev, self-hosted Kev, the OpenAI Decisions API, and a general LLM behind the adapter. Everything else is listed in §8.1.1 as a link to evaluate on your own.

8.1 The candidates

ZDR (Zero Data Retention) means the provider doesn’t store your request or response after processing it.

The rule from the Codila playbook still decides most cases: use a decision engine when the answer space can be bounded and the decision repeats; use an LLM when the output must be created; use code when the answer can be calculated exactly.

8.1.1 Other options

These are worth a look, but they’re not part of this chapter’s harness. Either they can’t be run end to end today, or they’re another route to an engine already covered:

• THX-01 - THX-01 is an open-source, multilingual, non-autoregressive decision model developed by HAL-X AI. number, excerpt and citations are native THX-01 capabilities. They are not part of the TypeSafe question types (choice, noul, score), where they have to be emulated in the service layer with repeated choice calls. THX-01 also answers such emulation calls (value-range buckets and document chunks) through its native lookup, so existing service layers work unchanged.
• Cloudflare Clef — Cloudflare’s decision model on Workers AI (text and images; clef and clef-flash). See also this Jev-compatibility and pricing write-up.
• OpenRouter — serves both Jev and gpt-6-luna-decisions behind one key.
• Vercel AI Gateway and the Vercel AI SDK — the Decisions API as openai/gpt-6-luna-decisions.
• promptfoo openai:decisions provider — run eval suites against the Decisions API without writing a client.
• Laya — decision model on Vercel AI Gateway, with a self-hosted Jev-compatible deployment. Check its current docs for primitives and pricing.

8.2 Jev, Kev and OpenAI Decisions side by side

All three answer the same three primitive shapes, under different names:

⚠️ Watch Out: confidences don’t transfer. Each engine computes confidence its own way. A 0.80 from Jev, Kev and Decisions means three different things. Re-tune the Chapter 5 bands per engine on your labeled set.

OpenAI Decisions support image input that both Jev and Kev are not. You can check example at https://developers.openai.com/api/docs/guides/decisions#check-an-image-for-visible-damage

8.3 Published comparisons: read them carefully

How to read any Jev benchmark: name the baseline model and effort level, report quality beside latency and price, separate model time from observation/tools/waits, state the number of tasks, repeats and failures, and treat a demo as a lead to reproduce, not a constant.

8.4 The honest conclusion: cascades

A slightly weaker check that’s cheap enough to run on every item often beats a stronger check you can only afford to sample. The production pattern: run the decision engine continuously, escalate uncertain or high-risk cases to the stronger model, and measure accuracy and total cost of the cascade.

💡 Mental Model: triage nurse, then specialist A triage nurse sees every patient quickly (fast and cheap) and sends only the unclear or dangerous cases to the specialist (slow and expensive). Most patients never need the specialist, so the average visit is both faster and cheaper.

📄 Full source: capstone-project/src/triage/cascade.py

Snippet 1: the two engines and the rules for escalating. CascadeDecider is itself a DecisionClient, so the rest of the engine can’t tell it isn’t a plain client.

# capstone-project/src/triage/cascade.py
from __future__ import annotations

from dataclasses import dataclass, replace
from typing import Any, Mapping

from triage.jev_client import DecisionClient, DecisionError, DecisionResponse

@dataclass(slots=True)
class CascadeDecider:
    fast: DecisionClient          # Jev (or Kev on-prem)
    strong: DecisionClient        # AdapterDecisionClient(frontier model)
    escalate_below: float = 0.70  # dispatch_target confidence
    fraud_grey_zone: tuple[float, float] = (0.30, 0.85)
    backend: str = "cascade"      # satisfies the DecisionClient protocol

Snippet 2: the single decision call on the fast engine. Read the three numbers we need to decide whether to escalate:

    def decide(self, state: Mapping[str, Any], questions: Mapping[str, Any]) -> DecisionResponse:
        first = self.fast.decide(state, questions)
        conf = first.choice("dispatch_target").confidence
        fraud = first.noul("is_fraud_suspected").p_yes
        lo, hi = self.fraud_grey_zone
        if conf >= self.escalate_below and not (lo <= fraud < hi):
            return first

• conf below 0.70 means the destination is uncertain.
• A fraud probability between 0.30 and 0.85 is the grey zone: not clearly fraud, not clearly safe.
• If neither applies, the fast answer is returned and the strong model is never called.

Snippet 3: the guardrail on escalation. If the strong model is down, keep the calibrated first answer:

        try:
            second = self.strong.decide(state, questions)
        except DecisionError:
            return first                      # keep the calibrated answer; policy will gate it

Snippet 4: cap the strong model’s confidence. LLM probabilities aren’t calibrated, so the policy engine must never see an LLM’s 0.99 as if it were Jev’s 0.99.

        # LLM probabilities are not calibrated: cap confidence so policy lands in the medium band.
        dt = second.choice("dispatch_target")
        capped = replace(dt, confidence=min(dt.confidence, 0.84))
        answers = {**second.answers, "dispatch_target": capped}
        return replace(second, answers=answers, latency_ms=first.latency_ms + second.latency_ms,
                       backend=f"cascade({first.backend}->{second.backend})")

⚠️ Watch Out Without the 0.84 cap, a confident LLM answer would land in the “high” band and skip the confirmation step that Jev’s own medium-band answers get. The cap keeps the escalated answer in the medium band by construction.

Because every engine implements DecisionClient, the cascade is engine-agnostic. CascadeDecider(fast=KevDecisionClient(), strong=OpenAIDecisionsClient()) keeps routine traffic on your own hardware and sends only the grey zone to a hosted model. The confidence cap still applies: the second engine’s confidence isn’t calibrated against the first engine’s bands.

8.5 Mini-project 7: the multi-engine benchmark harness

8.5.1 Kev: no new code, just a server

Kev serves the System One API, so KevDecisionClient (in triage.jev_client) is Jev’s client pointed at another base_url. All you need is a running server. Keep it in its own folder and venv (Python 3.12/3.13, torch), outside the book’s workspace:

git clone https://github.com/jaredpalmer/kev.git && cd kev
uv sync --extra serve
uv run --extra serve python -m kev.serve --run jaredpalmer/kev-0.8b --port 8009   # laptop; use kev-4b for accuracy work

# back in the book workspace
export KEV_BASE_URL=http://127.0.0.1:8009        # default; KEV_API_KEY only if the server requires one
uv run python mini-projects/07-multi-engine-benchmark/multi_engine_eval.py --engines kev,offline

Chapter 10 covers model sizes, hardware, auth and fine-tuning. For this chapter, the point is that the same contract runs on a model you own, and the harness tells you what that costs in accuracy.

8.5.2 OpenAI Decisions: one adapter class

The Decisions API answers the same three shapes, but its wire format differs from Jev’s: questions and answers are arrays (answers come back in question order with a name), choice options are {value, description} objects, score levels are {label, description} objects, and probabilities are arrays rather than dicts. The client below keeps all of that in two pure functions, to_decision_questions() and from_decision_answers(), so the rest of the engine never sees the vendor schema.

💡 Mental Model: a translator at the border Jev speaks “dict”, OpenAI speaks “array”. Rather than teaching the whole engine two languages, we post one translator at the border. Everything on our side keeps using NoulResult, ChoiceResult and ScoreResult.

Three details matter in production:

• A single eligible option (for example, refund_pipeline pruned because nothing posted) isn’t sent. The API needs 2–255 choices, and the answer is certain anyway.
• A refusal answer raises a non-retryable DecisionError, so Chapter 6’s fallback path takes over instead of guessing.
• 429 and 5xx errors become retryable DecisionErrors, so the circuit breaker treats them like Jev outages.

📄 Full source: mini-projects/07-multi-engine-benchmark/openai_decisions_client.py

We build it in five parts: header → request mapping → response mapping → the client → errors and a smoke test.

Part 1: header and constants

# mini-projects/07-multi-engine-benchmark/openai_decisions_client.py
"""DecisionClient for the OpenAI Decisions API (POST /v1/decisions, model gpt-6-luna; public beta).

Maps the book's dict contract (noul / choice / score) onto the Decisions questions array
(predicate / choice / score) and maps the answers back to NoulResult / ChoiceResult / ScoreResult,
so the rest of the engine never sees the vendor schema.

  uv run python mini-projects/07-multi-engine-benchmark/openai_decisions_client.py --dry-run
  uv sync --group ch7 && export OPENAI_API_KEY=sk-...
  uv run python mini-projects/07-multi-engine-benchmark/openai_decisions_client.py
Used as an engine by multi_engine_eval.py (--engines openai-decisions).
"""

from __future__ import annotations

import argparse
import json
import os
import sys
import time
from collections.abc import Mapping, Sequence
from typing import Any

from triage.jev_client import (Answer, ChoiceResult, DecisionError, DecisionResponse, JSONState,
                               NoulResult, Questions, ScoreResult)

DEFAULT_MODEL = "gpt-6-luna"
RETRYABLE_STATUS = frozenset({408, 409, 429, 500, 502, 503, 504})

RETRYABLE_STATUS is the list of HTTP codes that mean “try again” (timeouts, rate limits, server errors).

Part 2: request mapping (our contract → Decisions questions)

The input shape and the setup. Returns two things: the questions to send, and answers we settled locally.

# ---------- Request mapping: book contract -> Decisions questions array ----------
def to_decision_questions(questions: Questions) -> tuple[list[dict[str, Any]], dict[str, Answer]]:
    """Return (questions to send, answers settled locally).

    A Choice with a single eligible option (e.g. refund_pipeline pruned away) is not a question:
    the API needs 2-255 choices, and asking would only add latency, so we answer it here.
    """
    out: list[dict[str, Any]] = []
    local: dict[str, Answer] = {}

Noul → predicate. The true/false criteria are folded into the instruction text.

    for qid, q in questions.items():
        kind, text = q["type"], str(q.get("instructions", ""))
        if kind == "noul":
            tf = q.get("criteria") or {}
            if tf:
                text += f"\nTrue when: {tf.get('true', '')}\nFalse when: {tf.get('false', '')}"
            out.append({"type": "predicate", "name": qid, "instructions": text})

Choice → choice. A one-option Choice is answered locally with probability 1.0 (the guardrail from the bullet above), and a list over 255 is rejected up front.

        elif kind == "choice":
            crit: Mapping[str, Any] = q["criteria"]
            if len(crit) == 1:
                only = next(iter(crit))
                local[qid] = ChoiceResult(only, {only: 1.0}, 1.0)
                continue
            if len(crit) > 255:
                raise DecisionError(f"{qid}: {len(crit)} choices exceeds the API limit of 255", retryable=False)
            out.append({"type": "choice", "name": qid, "instructions": text,
                        "choices": [{"value": k, "description": str(v or k)} for k, v in crit.items()]})

Score → score. Levels are numbered by position, and the 2 to 10 limit is enforced.

        elif kind == "score":
            levels = [_level_text(x) for x in q["criteria"]]
            if not 2 <= len(levels) <= 10:
                raise DecisionError(f"{qid}: score needs 2-10 levels, got {len(levels)}", retryable=False)
            out.append({"type": "score", "name": qid, "instructions": text,
                        "levels": [{"label": str(i), "description": d} for i, d in enumerate(levels)]})
        else:
            raise DecisionError(f"{qid}: unknown question type {kind!r}", retryable=False)
    return out, local

Two tiny helpers. Score levels from Chapter 3 can be rubric rows with examples; the helper flattens them into text:

def _level_text(level: Any) -> str:
    """Score levels are strings or {"what": ..., "examples": [...]} rubric rows (Chapter 2)."""
    if isinstance(level, Mapping):
        ex = "; ".join(map(str, level.get("examples", [])))
        return f"{level.get('what', '')}" + (f" (e.g. {ex})" if ex else "")
    return str(level)

def to_input(state: JSONState) -> str:
    return state if isinstance(state, str) else json.dumps(state, ensure_ascii=False, default=str)

Part 3: response mapping (Decisions answers → our result types)

Answers come back in question order. Match by name first, and fall back to position:

# ---------- Response mapping: Decisions answers array -> book result types ----------
def from_decision_answers(answers: Sequence[Mapping[str, Any]], sent: Sequence[Mapping[str, Any]],
                          questions: Questions) -> dict[str, Answer]:
    """Answers come back in question order; match by name and fall back to position."""
    by_name = {a.get("name"): a for a in answers if a.get("name")}
    out: dict[str, Answer] = {}
    for i, q in enumerate(sent):
        qid = q["name"]
        a = by_name.get(qid) or (answers[i] if i < len(answers) else None)
        if a is None:
            raise DecisionError(f"answer missing for question {qid!r}", retryable=False)

Then three guardrails before trusting any answer: a refusal, a mismatched type, and (below) a label outside the contract.

        kind = a.get("type")
        if kind == "refusal":                 # the host declined this question: policy must fall back
            raise DecisionError(f"{qid}: refused by the Decisions API", retryable=False)
        if kind != q["type"]:
            raise DecisionError(f"{qid}: expected {q['type']}, got {kind!r}", retryable=False)
        if kind == "predicate":
            out[qid] = NoulResult(float(a["probability"]))

        elif kind == "choice":
            probs = {str(p["value"]): float(p["probability"]) for p in a["probabilities"]}
            choice = str(a["choice"])
            if choice not in questions[qid]["criteria"]:
                raise DecisionError(f"{qid}: label outside contract: {choice!r}", retryable=False)
            out[qid] = ChoiceResult(choice, probs, float(a["confidence"]))
        else:  # score: `score` is the expected zero-based level index, like Jev's
            probs_i = {int(p["value"]): float(p["probability"]) for p in a["probabilities"]}
            out[qid] = ScoreResult(float(a["score"]), probs_i, float(a["confidence"]),
                                   len(questions[qid]["criteria"]))
    return out

Part 4: the client

The class docstring is a reminder: this engine’s confidence is not Jev’s. Then the constructor creates the SDK client, converting missing-SDK or missing-key problems into readable errors:

# ---------- The client ----------
class OpenAIDecisionsClient:
    """Its confidence is not Jev's: tune thresholds on this backend separately (Chapter 4)."""

    backend = "openai-decisions"

    def __init__(self, model: str | None = None, timeout_s: float = 3.0, max_retries: int = 2,
                 sdk_client: Any | None = None) -> None:

        self._model = model or os.getenv("OPENAI_DECISIONS_MODEL", DEFAULT_MODEL)
        if sdk_client is None:
            try:
                from openai import OpenAI          # openai>=3.26 (uv sync --group ch7)
            except ImportError as e:
                raise RuntimeError("openai SDK missing: run `uv sync --group ch7`") from e
            # Reads OPENAI_API_KEY; OPENAI_BASE_URL can point at a gateway that serves the same API.
            try:
                sdk_client = OpenAI(timeout=timeout_s, max_retries=max_retries)
            except Exception as e:                 # e.g. OpenAIError: no OPENAI_API_KEY
                raise RuntimeError(f"cannot create OpenAI client: {e}") from e
        self._sdk = sdk_client

decide() has four jobs: map the request, short-circuit if everything was settled locally, call the API, and map errors.

    def decide(self, state: JSONState, questions: Questions) -> DecisionResponse:
        sent, local = to_decision_questions(questions)
        t0 = time.perf_counter()
        if not sent:                               # everything was settled locally
            return DecisionResponse(local, self._model, 0, 0.0, self.backend)
        try:
            resp = self._sdk.decisions.create(model=self._model, input=to_input(state), questions=sent)
        except Exception as e:                     # map SDK errors without importing them at module level
            raise _as_decision_error(e) from e

Then it times the call and assembles a DecisionResponse, keeping the contract’s question order:

        latency_ms = (time.perf_counter() - t0) * 1000
        body = resp.model_dump() if hasattr(resp, "model_dump") else dict(resp)
        answers = {**local, **from_decision_answers(body["answers"], sent, questions)}
        usage = body.get("usage") or {}
        return DecisionResponse(
            answers={qid: answers[qid] for qid in questions},     # keep the contract's order
            model=str(body.get("model", self._model)),
            input_tokens=usage.get("input_tokens"),
            latency_ms=latency_ms,
            backend=self.backend,
            raw=body,
        )

Part 5: error mapping and a smoke test

HTTP statuses decide whether the breaker should count an error:

def _as_decision_error(e: Exception) -> DecisionError:
    status = getattr(e, "status_code", None)
    name = type(e).__name__
    if status is not None:
        return DecisionError(f"openai-decisions HTTP {status}: {e}", retryable=status in RETRYABLE_STATUS)
    if name in ("APIConnectionError", "APITimeoutError"):
        return DecisionError(f"openai-decisions transport: {e}", retryable=True)
    return DecisionError(f"openai-decisions client error: {name}: {e}", retryable=False)

The main() function lets you try one request. First the setup and the dry-run path, which prints the request body without any network:

def main(argv: list[str] | None = None) -> int:
    from triage.state_builder import build_questions, build_state, parse_report

    ap = argparse.ArgumentParser(description="Send one triage request to the OpenAI Decisions API")
    ap.add_argument("--text", default="Got OTPs I never asked for and 2 payments I didn't make just now")
    ap.add_argument("--model", default=None, help=f"default: $OPENAI_DECISIONS_MODEL or {DEFAULT_MODEL}")
    ap.add_argument("--dry-run", action="store_true", help="print the request body; send nothing")
    args = ap.parse_args(argv)

    report = parse_report({"report_id": "smoke-1", "body": args.text})
    state, questions = build_state(report), build_questions(report, speculative=False)

    if args.dry_run:
        sent, local = to_decision_questions(questions)
        model = args.model or os.getenv("OPENAI_DECISIONS_MODEL", DEFAULT_MODEL)
        print(json.dumps({"model": model, "input": to_input(state), "questions": sent}, indent=2))
        if local:
            print(f"\nanswered locally (single option): {sorted(local)}", file=sys.stderr)
        return 0

Then the live path, which refuses to run without a key:

    if not os.getenv("OPENAI_API_KEY"):
        print("set OPENAI_API_KEY first (or use --dry-run)", file=sys.stderr)
        return 2
    try:
        r = OpenAIDecisionsClient(model=args.model).decide(state, questions)
    except (RuntimeError, DecisionError) as e:
        print(f"decision failed: {e}", file=sys.stderr)
        return 1

    for qid, a in r.answers.items():
        print(f"{qid:20} {a}")
    print(f"\n{r.model}: {r.latency_ms:.0f} ms, {r.input_tokens} input tokens")
    return 0

if __name__ == "__main__":
    sys.exit(main())

Run it on its own. --dry-run prints the exact request body without any network or credentials. A live call needs OPENAI_API_KEY and the openai SDK from the ch7 group:

uv run python mini-projects/07-multi-engine-benchmark/openai_decisions_client.py --dry-run
uv sync --group ch7
export OPENAI_API_KEY=sk-...
uv run python mini-projects/07-multi-engine-benchmark/openai_decisions_client.py
uv run python mini-projects/07-multi-engine-benchmark/openai_decisions_client.py --text "I was charged twice for order A-104"

{
  "model": "gpt-6-luna",
  "input": "{\"report\": {\"channel\": \"app_chat\", \"body\": \"Got OTPs I never asked for ...\"}, ...}",
  "questions": [
    {"type": "predicate", "name": "is_fraud_suspected", "instructions": "Does `report.body` describe ...\nTrue when: ...\nFalse when: ..."},
    {"type": "score", "name": "urgency_rating", "instructions": "How urgent is ...",
     "levels": [{"label": "0", "description": "Informational; no money or access at risk (e.g. ...)"}, ...]},
    {"type": "choice", "name": "dispatch_target", "instructions": "Which team should own the next step for `report`?",
     "choices": [{"value": "tier1_support", "description": "How-to questions, app errors, ..."}, ...]}
  ]
}

sheikudumanali@sheikmac $ uv run python mini-projects/07-multi-engine-benchmark/openai_decisions_client.py
is_fraud_suspected   NoulResult(p_yes=1.0)
urgency_rating       ScoreResult(score=3.86, probabilities={0: 0.0, 1: 0.0, 2: 0.01, 3: 0.12, 4: 0.87}, confidence=0.88, levels=5)
dispatch_target      ChoiceResult(choice='secops_agent', probabilities={'tier1_support': 0.0, 'secops_agent': 0.99, 'manual_escalation': 0.01}, confidence=0.99)

gpt-6-luna: 1582 ms, 740 input tokens

To use a gateway that serves the same API (OpenRouter, Vercel AI Gateway), set OPENAI_BASE_URL and OPENAI_DECISIONS_MODEL to the gateway’s URL and model id. The SDK reads both from the environment.

Test the mapping offline. The tests feed the documented response shape through a fake SDK: types map back correctly, a refusal becomes a fallback, a 429 is retryable, and a label outside the contract is rejected.

📄 Full source: mini-projects/07-multi-engine-benchmark/test_openai_decisions_client.py

uv run pytest mini-projects/07-multi-engine-benchmark

======================== test session starts =============================
platform darwin -- Python 3.12.12, pytest-9.1.1, pluggy-1.6.0
rootdir: /Users/sheikudumanali/mastering-jev-internal
configfile: pyproject.toml
plugins: anyio-4.15.1
collected 5 items                                                                                                                                                                                                                                                    

mini-projects/07-multi-engine-benchmark/test_openai_decisions_client.py .....[100%]

====================== 5 passed in 0.02s ===================================

8.5.3 The harness

It measures six things per engine: acceptable-route accuracy, schema error rate, p50/p95 latency, cost per 1k decisions, ECE (expected calibration error) and Brier score for the dispatch_target Choice.

📄 Full source: mini-projects/07-multi-engine-benchmark/multi_engine_eval.py

Snippet 1: usage. The docstring tells you what the labeled set is and how to run each engine.

# mini-projects/07-multi-engine-benchmark/multi_engine_eval.py
"""Run the SAME triage contract on the SAME labeled cases across several engines.

The labeled cases are the Chapter 5 regression set: capstone-project/tests/labeled_cases.jsonl
(one JSON object per line: payload, expected, acceptable, kind, human_required). Used by default.

  uv run python mini-projects/07-multi-engine-benchmark/multi_engine_eval.py --engines offline
  uv run python mini-projects/07-multi-engine-benchmark/multi_engine_eval.py --engines kev,offline
  uv run python mini-projects/07-multi-engine-benchmark/multi_engine_eval.py --engines jev,openai-decisions,kev
  uv run python mini-projects/07-multi-engine-benchmark/multi_engine_eval.py --cases my_labels.jsonl --engines jev,adapter

Writes benchmark_results.json next to this script (input for benchmark_comparison_plot.py).
"""

Snippet 2: imports and paths.

from __future__ import annotations

import json
import statistics
import sys
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any, Mapping, Sequence

from triage.jev_client import Answer, ChoiceResult, DecisionClient, DecisionError, FakeDecisionClient, NoulResult, ScoreResult
from triage.state_builder import build_questions, build_state, parse_report

LABELS = ("tier1_support", "secops_agent", "refund_pipeline", "manual_escalation")
HERE = Path(__file__).resolve().parent
DEFAULT_CASES = HERE.parents[1] / "capstone-project" / "tests" / "labeled_cases.jsonl"
DEFAULT_OUT = HERE / "benchmark_results.json"

Snippet 3: the report shape. One row per engine, six metrics plus identity:

@dataclass(frozen=True, slots=True)
class EngineReport:
    engine: str
    n: int
    accuracy: float
    schema_error_rate: float
    p50_ms: float
    p95_ms: float
    usd_per_1k: float
    ece: float
    brier: float

Snippet 4: calibration metrics. ece groups answers into confidence bins and compares each bin’s stated confidence with its actual accuracy. brier is the squared error of the whole probability distribution.

def ece(conf: Sequence[float], correct: Sequence[bool], bins: int = 10) -> float:
    total, err = len(conf), 0.0
    for b in range(bins):
        lo, hi = b / bins, (b + 1) / bins
        idx = [i for i, c in enumerate(conf) if lo <= c < hi or (b == bins - 1 and c == 1.0)]
        if idx:
            acc = sum(correct[i] for i in idx) / len(idx)
            avg = sum(conf[i] for i in idx) / len(idx)
            err += len(idx) / total * abs(acc - avg)
    return err

def brier(probs: Sequence[Mapping[str, float]], gold: Sequence[str]) -> float:
    return statistics.fmean(sum((p.get(l, 0.0) - (l == g)) ** 2 for l in LABELS) for p, g in zip(probs, gold, strict=True))

💡 Mental Model: the weather forecaster A forecaster who says “70% rain” is calibrated if it rains on about 70% of those days. ECE is the average gap between what the forecaster said and what happened. Brier punishes both being wrong and being unsure when you should have been sure.

Snippet 5: run one engine over every case. For each labeled case, build the state and the same questions, make the single decision call, and record latency, confidence and whether the answer was acceptable. A bad or missing answer counts as a schema error, not a crash.

def run_engine(name: str, client: DecisionClient, cases: Sequence[Mapping[str, Any]],
               usd_per_m_input: float, fallback_tokens: int = 450) -> EngineReport:
    lat, conf, correct, probs, gold = [], [], [], [], []
    schema_errors, tokens = 0, []
    for c in cases:
        report = parse_report(c["payload"])
        state, qs = build_state(report), build_questions(report, speculative=False)
        try:
            r = client.decide(state, qs)
            dt = r.choice("dispatch_target")
        except (DecisionError, KeyError, TypeError):
            schema_errors += 1
            continue

        lat.append(r.latency_ms)
        tokens.append(r.input_tokens or fallback_tokens)
        conf.append(dt.confidence)
        correct.append(dt.choice in c["acceptable"])
        probs.append(dict(dt.probabilities))
        gold.append(c["expected"])

Snippet 6: roll up the metrics.

    ok = len(lat)
    q = statistics.quantiles(lat, n=20) if ok >= 2 else [float("nan")] * 19
    return EngineReport(
        engine=name, n=len(cases),
        accuracy=sum(correct) / ok if ok else 0.0,
        schema_error_rate=schema_errors / len(cases),
        p50_ms=statistics.median(lat) if lat else float("nan"), p95_ms=q[18],
        usd_per_1k=statistics.fmean(tokens) * usd_per_m_input / 1_000_000 * 1000 if tokens else 0.0,
        ece=ece(conf, correct) if ok else float("nan"),
        brier=brier(probs, gold) if ok else float("nan"),
    )

Snippet 7: an offline stand-in so the harness runs without keys. It measures the harness, not a model.

def offline_model(state: Any, questions: Mapping[str, Any]) -> Mapping[str, Answer]:
    """OFFLINE STAND-IN: keyword routing so the harness runs without keys. Measures the harness, not a model."""
    body = state["report"]["body"].lower()
    if re.search(r"system:|ignore (previous|all) instructions|i am the admin", body):
        target, conf = "manual_escalation", 0.6
    elif re.search(r"never (asked|made)|didn'?t make|not me|unauthori[sz]ed|otp", body):
        target, conf = "secops_agent", 0.85
    elif re.search(r"charged twice|duplicate|refund|charge .* wrong|looks wrong", body):
        target, conf = "refund_pipeline", 0.7
    elif re.search(r"passed away|lawyer|regulator|ombudsman", body):
        target, conf = "manual_escalation", 0.8
    else:
        target, conf = "tier1_support", 0.75

    offered = list(questions["dispatch_target"]["criteria"])
    if target not in offered:                          # e.g. refund_pipeline pruned: nothing posted
        target, conf = "manual_escalation", 0.5
    rest = (1 - conf) / max(1, len(offered) - 1)
    probs = {l: (conf if l == target else rest) for l in offered}
    k = len(offered)
    return {
        "is_fraud_suspected": NoulResult(0.9 if target == "secops_agent" else 0.05),
        "urgency_rating": ScoreResult(3.5 if target == "secops_agent" else 1.5, {}, 0.6, 5),
        "dispatch_target": ChoiceResult(target, probs, (conf - 1 / k) / (1 - 1 / k)),
    }

Snippet 8: choosing engines. Clients are created only when selected, so you only need credentials for the engines you pick:

ENGINES = ("offline", "jev", "kev", "openai-decisions", "adapter")


def build_engines(names: Sequence[str]) -> list[tuple[str, DecisionClient, float]]:
    """name -> (label, client, USD per million input tokens). Clients are created only when selected.

    Prices are list prices at the time of writing (Oct 2026): re-check them before you publish a number.
    """

    from triage.jev_client import AdapterDecisionClient, JevDecisionClient, KevDecisionClient
    engines: list[tuple[str, DecisionClient, float]] = []
    for name in names:
        if name == "offline":
            engines.append(("offline-standin", FakeDecisionClient(offline_model), 0.0))
        elif name == "jev":
            engines.append(("jev-1.13.0", JevDecisionClient(model="jev-1.13.0"), 0.042))
        elif name == "kev":
            engines.append(("kev (local)", KevDecisionClient(), 0.0))            # you pay for the box, not tokens
        elif name == "openai-decisions":
            from openai_decisions_client import OpenAIDecisionsClient           # uv sync --group ch7; OPENAI_API_KEY
            engines.append(("gpt-6-luna (decisions)", OpenAIDecisionsClient(), 0.10))  # output tokens are free

        elif name == "adapter":
            engines.append(("gpt-4o-mini (adapter)", AdapterDecisionClient("openai", "gpt-4o-mini"), 0.15))
        else:
            raise SystemExit(f"unknown engine {name!r}; choose from {', '.join(ENGINES)}")
    return engines

Snippet 9: the command line, run, and report.

def main(argv: Sequence[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description="Benchmark decision engines on one labeled set")
    ap.add_argument("--cases", type=Path, default=DEFAULT_CASES,
                    help=f"labeled JSONL (default: {DEFAULT_CASES.relative_to(HERE.parents[1])})")
    ap.add_argument("--engines", default="offline",
                    help=f"comma list from: {', '.join(ENGINES)} (default: offline)")
    ap.add_argument("--out", type=Path, default=DEFAULT_OUT)
    args = ap.parse_args(argv)

    cases = [json.loads(l) for l in args.cases.read_text(encoding="utf-8").splitlines() if l.strip()]
    if not cases:
        raise SystemExit(f"{args.cases} has no cases")
    try:
        engines = build_engines([e.strip() for e in args.engines.split(",") if e.strip()])
    except (ImportError, RuntimeError) as e:                 # missing SDK group or credentials
        raise SystemExit(f"cannot create engine: {e}") from e
    reports = [run_engine(n, c, cases, price) for n, c, price in engines]
    args.out.write_text(json.dumps([asdict(r) for r in reports], indent=2), encoding="utf-8")

    print(f"{len(cases)} labeled cases from {args.cases.name}\n")
    print(f"{'engine':24} {'acc':>6} {'schema':>7} {'p50':>7} {'p95':>7} {'$/1k':>9} {'ECE':>6} {'Brier':>6}")
    for r in reports:
        print(f"{r.engine:24} {r.accuracy:6.1%} {r.schema_error_rate:7.1%} {r.p50_ms:6.0f}ms {r.p95_ms:6.0f}ms "
              f"{r.usd_per_1k:9.5f} {r.ece:6.3f} {r.brier:6.3f}")
    print(f"\nwrote {args.out}")
    return 0

if __name__ == "__main__":
    sys.exit(main())

What it reads. The input is the Chapter 6 labeled set, capstone-project/tests/labeled_cases.jsonl, used by default. It’s JSON Lines (one JSON object per line), not a single JSON file. Each line is a raw report plus its expert label:

{"id": "L-014", "kind": "ordinary",
 "payload": {"report_id": "L-014", "body": "Got OTPs I never asked for and 2 payments I didn't make just now", "signals": {"new_device": true}},
 "expected": "secops_agent", "acceptable": ["secops_agent"], "human_required": false}

The harness turns each payload into a state with build_state(), asks every engine the same build_questions() contract, and scores dispatch_target against acceptable. To use your own labels, write the same shape and pass --cases my_labels.jsonl. The 5 sample rows only prove the plumbing; you want hundreds before comparing engines.

Run it. Choose engines with --engines (comma-separated). Each one is created only when selected, so you only need credentials for the engines you pick:

uv run python mini-projects/07-multi-engine-benchmark/multi_engine_eval.py --engines offline
uv run python mini-projects/07-multi-engine-benchmark/multi_engine_eval.py --engines jev,kev,openai-decisions
uv run python mini-projects/07-multi-engine-benchmark/multi_engine_eval.py --engines jev,openai-decisions,adapter --cases my_labels.jsonl

sheikudumanali@sheikmac $ uv run python mini-projects/07-multi-engine-benchmark/multi_engine_eval.py --engines jev,openai-decisions,offline,kev
5 labeled cases from labeled_cases.jsonl

engine                      acc  schema     p50     p95      $/1k    ECE  Brier
jev-1.13.0               100.0%    0.0%    367ms    648ms   0.03432  0.160  0.541
gpt-6-luna (decisions)   100.0%    0.0%    155ms    711ms   0.07364  0.248  0.415
offline-standin          100.0%    0.0%      0ms      0ms   0.00000  0.450  0.360
kev (local)                0.0%  100.0%    nan ms    nan ms   0.00000    nan    nan

wrote /Users/sheikudumanali/mastering-jev-internal/mini-projects/07-multi-engine-benchmark/benchmark_results.json

ECE (Expected Calibration Error) measures how far an engine’s stated confidence is from its actual accuracy, averaged across confidence bins: 0 means perfectly calibrated, and an engine that says 0.9 but is right only 60% of the time scores high.

The Brier score is the mean squared error between an engine’s predicted probabilities and the true outcome (1 for the correct label, 0 for the rest). Lower is better and 0 is perfect, and unlike ECE it rewards being both accurate and well-calibrated. In our harness it’s summed over the 4 dispatch labels, so it ranges from 0 to 2.

⚠️ Watch Out: graded on its own homework The offline stand-in scores 100 % because its keywords were written against these same five rows. That’s the trap from Chapter 13, and it’s why the stand-in only checks the harness. Its high ECE (0.45) also shows that a confident engine can be badly calibrated even when every answer is right.

When an engine can’t be created (SDK group not installed, key missing), the harness exits with one line saying why instead of a stack trace. A refusal or a malformed answer during the run counts as a schema error for that engine, which is the number you want to see next to accuracy.

The adapter cost column counts input only. Add the LLM’s output-token price for a fair comparison (Chapter 2’s llm_monthly_cost). Decisions and Jev bill no output tokens, so their column is already complete.

8.5.4 Plot it

📄 Full source: mini-projects/07-multi-engine-benchmark/benchmark_comparison_plot.py

Imports and the entry point. It reads the JSON the harness wrote:

# mini-projects/07-multi-engine-benchmark/benchmark_comparison_plot.py
"""Plot benchmark_results.json (written by multi_engine_eval.py). Needs: uv sync --group ch7

  uv run python mini-projects/07-multi-engine-benchmark/benchmark_comparison_plot.py
"""
from __future__ import annotations

import json
from pathlib import Path

import matplotlib.pyplot as plt

def main(path: str | Path = Path(__file__).with_name("benchmark_results.json")) -> None:
    rows = json.loads(Path(path).read_text(encoding="utf-8"))
    fig, ax = plt.subplots(figsize=(7, 4.5))
    for r in rows:
        ax.scatter(max(r["p95_ms"], 1.0), r["accuracy"] * 100, s=60 + 4000 * r["usd_per_1k"], alpha=0.7)
        ax.annotate(f"{r['engine']}\nECE {r['ece']:.2f}", (max(r["p95_ms"], 1.0), r["accuracy"] * 100),
                    textcoords="offset points", xytext=(8, 4), fontsize=8)

Axes and output. Log-scale latency, accuracy on the y-axis, bubble size for cost:

    ax.set_xscale("log")
    ax.set_xlabel("p95 latency (ms, log scale)")
    ax.set_ylabel("Acceptable-route accuracy (%)")
    ax.set_title("Triage contract: engines on the same labeled set (bubble = cost/1k)")
    ax.grid(True, which="both", alpha=0.3)
    fig.tight_layout()
    out = Path(path).with_name("benchmark_comparison.png")
    fig.savefig(out, dpi=160)
    print(f"wrote {out}")

if __name__ == "__main__":
    main()

uv sync --group ch7                                   # matplotlib
uv run python mini-projects/07-multi-engine-benchmark/benchmark_comparison_plot.py

It reads benchmark_results.json from the mini-project folder and writes benchmark_comparison.png next to it.

8.6 Classical baseline: always include one

If you already have a few thousand labeled reports, a supervised classifier is the cheapest thing that could work. It takes ten lines and belongs in every bake-off:

📄 Full source: mini-projects/07-multi-engine-benchmark/sklearn_baseline.py

# mini-projects/07-multi-engine-benchmark/sklearn_baseline.py
from __future__ import annotations

from sklearn.calibration import CalibratedClassifierCV
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import make_pipeline


def train(texts: list[str], labels: list[str]):
    base = make_pipeline(TfidfVectorizer(ngram_range=(1, 2), min_df=2), LogisticRegression(max_iter=2000))
    return CalibratedClassifierCV(base, method="isotonic", cv=5).fit(texts, labels)

Decision engines usually win when options change often (new queue, new product), labels are scarce, or inputs are long and varied. The classifier wins when options are frozen, labels are plentiful, and you need sub-millisecond latency with no network. You can also combine them: TypeSafe’s feature-discovery cookbook turned 38 Jev questions into 67 numeric columns feeding a CatBoost regressor.

8.7 Capstone increment: the engine decision record

Run the harness on the Chapter 6 labeled set and write the result down next to the policy version:

# ADR-007: Decision engine for triage-contract/1.0.0

- Labeled set: labeled_cases.jsonl @ a1b2c3 (n = 400; 280 ordinary, 60 ambiguous, 30 no-match, 30 adversarial)
- Engines run: jev-1.13.0, kev-4b (on-prem, L4 GPU), gpt-6-luna (Decisions, beta), gpt-4o-mini (adapter)
- Region: ap-south-1 -> api.typesafe.ai / api.openai.com
- Results: see benchmark_results.json (attached), benchmark_comparison.png
- Decision: Jev primary; OpenAI Decisions as second source behind the circuit breaker
  (bands re-tuned for it separately); cascade to frontier LLM via adapter for
  dispatch confidence < 0.70 or fraud p in [0.30, 0.85); classical baseline kept in shadow as drift canary.
- Re-run triggers: model upgrade, Decisions GA, contract/policy version bump, monthly.

The numbers in an ADR must come from your run. Don’t copy vendor figures into it.

8.8 Checklist

• Every engine sits behind the same DecisionClient interface
• Same contract, same labeled set, same region for every engine
• Report quality and calibration and p95 and total cost, never one number
• Vendor wire formats mapped in one place (to_decision_questions / from_decision_answers), with offline tests
• Refusals and out-of-contract labels become DecisionError, never a guess
• Thresholds re-tuned per engine; LLM confidences capped or ignored
• Beta APIs pinned to a model id and re-benchmarked at GA
• Engine choice recorded as an ADR tied to contract + policy versions

What’s next

Every part exists: client, contract, state builder, policy, resilience, safety gate, executors, verifier, cascade, benchmark. Chapter 9 wires them into one service, engine.py, runs it end to end on real-shaped payloads, adds the operational glue (budgets, metrics, runbook), and lays out the rollout from shadow mode to production.