Chapter 7 — Alternatives to Jev & Multi-Engine Benchmarks

Previously: Chapter 6 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.

Everything in Chapters 1–6 sits behind DecisionClient. That’s what makes this chapter cheap: swapping the engine is a constructor change, and the labeled set from Chapter 5 is the referee. 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 7.1.1 as a link to evaluate on your own.

7.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.

7.2 Published comparisons: read them carefully

7.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:

• 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.
• Von on Hugging Face (wfzyx/von) — small open-weights decision model. Its serving format isn’t documented well enough to run reliably.
• Laya — decision model on Vercel AI Gateway, with a self-hosted Jev-compatible deployment. Check its current docs for primitives and pricing.

7.2 Jev, Kev and OpenAI Decisions side by side

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

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 4 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

7.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.

7.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.

# 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

    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
        try:
            second = self.strong.decide(state, questions)
        except DecisionError:
            return first                      # keep the calibrated answer; policy will gate it
        # 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})")

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.

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

7.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 9 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.

7.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.

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 5’s fallback path takes over instead of guessing.
• 429 and 5xx errors become retryable DecisionErrors, so the circuit breaker treats them like Jev outages.

# 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})


# ---------- 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] = {}
    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})
        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()]})
        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


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)


# ---------- 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)
        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


# ---------- 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

    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
        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,
        )


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)


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
    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.

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 ===================================

7.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.

# 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).
"""
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"

@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


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))


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"])
    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"),
    )


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)),
    }


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


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 5 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.

The offline stand-in scores 100 % because its keywords were written against these same five rows. That’s the “graded on its own homework” trap from Chapter 12, 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 1’s llm_monthly_cost). Decisions and Jev bill no output tokens, so their column is already complete.

7.5.4 Plot it

# 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)
    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.

7.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:

# 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.

7.7 Capstone increment: the engine decision record

Run the harness on the Chapter 5 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.

7.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 8 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.