Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

Chapter 2 — Foundations: System One AI & Decision Economics

You will build: a swappable decision client (Jev, adapter, or offline fake) plus a latency and cost harness.

Capstone increment: capstone-project/src/triage/jev_client.py


2.0 The problem

Every application makes small judgments: Is this message angry? Which team should get this ticket? Is this payment suspicious?

Today you have two ways to automate them, and both have a catch:

  • Hand-written rules (if "refund" in text) are instant, but they only catch what you predicted. Real customers never read your keyword list.
  • A chat LLM understands meaning, but it takes seconds, charges you for every word it writes back, and answers in prose like “This appears likely to be fraud, although it could also be…”. You then need more code just to read the answer, and the answer can change on the next run.

What you actually want is a component that understands meaning and behaves like a function: same input, typed output, a few milliseconds, a number you can trust.

That component is a System One decision model, and this chapter introduces it in four small steps:

  1. what it is and how it differs from an LLM (§2.1),
  2. who owns what in your system (§2.2),
  3. what it costs (§2.3),
  4. how to call it and wrap it safely (§2.4 to §2.6).

2.1 The one-sentence model: a smart if

Ordinary code branches on values it can compute:

if order.total > 100:
    apply_free_shipping(order)

That works until the condition is a judgment: is this customer describing fraud? How urgent is it? Which team should get it? Before Jev you had three options:

OptionFastCheapUnderstands meaningNo training dataCalibrated probability
if / regex / rules✅✅❌✅n/a
Trained classifier✅✅✅ (narrow)❌✅ (per task)
LLM + structured output❌ (2–8 s)❌✅✅❌
Jev (System One)✅ (~70–500 ms)✅ ($0.042 / M input)✅✅✅ (trained for it)

Jev takes a state (text or JSON) and a set of typed questions, and returns one typed answer per question with probabilities. It doesn’t write text, call tools, or run a loop. It returns numbers, and your code branches on them.

💡 Mental Model: a smart if Think of Jev as an if statement that can read. You hand it evidence and a list of questions, and it hands back numbers like 0.97. Your code, not the model, still decides what to do with 0.97.

Three terms appear in the shaded notes below. You do not need to master them now; they come back in Chapter 3 and Chapter 5.

OpenAI Decisions - an Jev counterpart from OpenAI describes requests as three parts model (jev-1.13.0 or gpt-6-luna), state and questions. OpenAI is called state as input.

Calibration means its probabilities match reality: when it says 0.8 it’s right about 80% of the time, because it was trained for that (RLCD).

RLCD (Reinforcement Learning for Calibrated Decisions) is TypeSafe’s post-training method. It rewards Jev for probabilities that match how often it’s actually right, not just for picking the right answer. TypeSafe has published only the name, so the exact recipe (likely log-loss or Brier-style scoring) is inferred.

Jev Use Cases

2.2 Where Jev sits in a system

The rule for the whole book is Code computes. LLMs create. Jev decides. Fresh state proves. Every design decision in later chapters comes back to it.

A quick way to decide who owns a piece of work:

If the answer…OwnerExample from the capstone
can be computed exactlyCode“Is the amount > ₹50,000?”, “3 failed logins in 10 min?”
is one of a closed set, or a position on a rubricJev“Is fraud suspected?”, “Which queue?”
has to be writtenLLMThe apology email to the customer
is irreversible or regulatedHuman (with Jev pre-sorting)Freezing an account, reversing a payout

💡 Mental Model: the four-way split Ask one question about any piece of work: “Could a calculator do this?” If yes, it is code. If it is a judgment from a short menu, it is Jev. If it needs sentences written, it is an LLM. If a wrong answer cannot be undone, a human signs off.

The customer message is state: the evidence for this decision. A question states the judgment we want. Keeping those separate makes the request easier to review.

Step 1: Define the evidence

state = {
    "report_id": "R-104",
    "message": "Someone used my card. That payment was not mine.",
}

report_id connects later logs to this report. message contains the customer’s words. Neither field authorises an action.

Define one question

A Noul estimates the probability that a yes/no condition holds. We will explore it fully in Chapter 2; here we need only one direct condition.

questions = {
    "fraud": {
        "type": "noul",
        "instructions": "Does `message` describe an unauthorised payment?",
    },
}

fraud is our lookup key. type selects the answer shape. instructions supplies the complete condition; the identifier alone is not an adequate question.

Step 2: Make the decision call

from triage.jev_client import client_from_env

client = client_from_env()  # The book wrapper selects the configured backend.
result = client.decide(state, questions)

The first argument is evidence. The second is the question contract. The wrapper normalizes the backend response so application code can use the same interface.

Step 3: Inspect the typed answer

fraud = result.noul("fraud")
if fraud.p_yes >= 0.85:  # Teaching threshold; validate it on labeled reports.
    destination = "secops_agent"
else:
    destination = "manual_escalation"

This assigns a destination. It does not freeze an account, move money, or establish that fraud occurred. A low fraud estimate also does not prove a report belongs in ordinary support; we have not asked that question yet.

Understand what a probability promises

If a model is calibrated on a workload, cases assigned probability 0.8 should satisfy the condition about 80% of the time. That is a statement about groups of predictions. It cannot guarantee one answer, and calibration can change on unfamiliar data.

The source draft discusses TypeSafe’s RLCD training approach. Training for calibrated decisions is a design goal; we still need labeled evaluation in the deployment domain. The book’s policy never treats a model’s training method as a substitute for that evaluation.

⚠️ Watch Out “The answer is typed,” “the estimate is confident,” and “the answer is correct” are three different claims.

2.3 Economics you can put in a design doc

Published figures (TypeSafe, Sep 2026): $0.042 per million input tokens, output free, typically 70–500 ms end to end (most around 100 ms, measured from US West). Rate limits during early access: 250k tokens/s, 1,200 requests/min.

Don’t reason about cost per token. Reason about cost per solved task. Here is a small calculator you can keep in the repo. We build it in three pieces.

📄 Full source: mini-projects/01-starter-client/economics.py

Piece 1: describe a workload. A workload is just “how many calls, how big, and what an LLM would charge for the same job”.

# economics.py
from __future__ import annotations

from dataclasses import dataclass

JEV_USD_PER_M_INPUT: float = 0.042
@dataclass(frozen=True, slots=True)
class Workload:
    name: str
    calls_per_month: int
    input_tokens_per_call: int
    llm_usd_per_m_input: float
    llm_usd_per_m_output: float
    llm_output_tokens_per_call: int

Piece 2: price it on both sides. Jev bills input tokens only; an LLM bills input and output.

def jev_monthly_cost(w: Workload) -> float:
    return w.calls_per_month * w.input_tokens_per_call * JEV_USD_PER_M_INPUT / 1_000_000


def llm_monthly_cost(w: Workload) -> float:
    inp = w.calls_per_month * w.input_tokens_per_call * w.llm_usd_per_m_input
    out = w.calls_per_month * w.llm_output_tokens_per_call * w.llm_usd_per_m_output
    return (inp + out) / 1_000_000

Piece 3: the blended saving. Most apps use an LLM for many things, so only the decision part of the bill shrinks.

def blended_savings(total_ai_bill: float, decision_share: float, new_cost_ratio: float) -> float:
    """total savings = decision share x savings on those calls.

    decision_share: fraction of the bill spent on classify/route/score/verify (0..1)
    new_cost_ratio: new path cost / old path cost for those calls (e.g. 0.05)
    """
    if not 0 <= decision_share <= 1 or not 0 <= new_cost_ratio <= 1:
        raise ValueError("shares and ratios must be within [0, 1]")
    return total_ai_bill * decision_share * (1 - new_cost_ratio)

Run it. The __main__ block compares the capstone’s workload (3 million reports a month, about 450 tokens each):

if __name__ == "__main__":
    triage = Workload(
        name="fintech-triage",
        calls_per_month=3_000_000,
        input_tokens_per_call=450,        # payload + 3 questions
        llm_usd_per_m_input=0.15,         # small LLM with JSON output
        llm_usd_per_m_output=0.60,
        llm_output_tokens_per_call=60,
    )
    print(f"Jev : ${jev_monthly_cost(triage):>10,.2f}/month")
    print(f"LLM : ${llm_monthly_cost(triage):>10,.2f}/month")
    # The Deep Dive example: $10k bill, 60% decisions, new path costs 5% of old.
    print(f"Blended saving: ${blended_savings(10_000, 0.60, 0.05):,.0f}")  # $5,700
Jev : $     56.70/month
LLM : $    310.50/month
Blended saving: $5,700

Two points to take from this:

  1. Savings are capped by the decision share. If only 10% of your bill is classification, making it free saves at most 10%.
  2. The Jevons paradox. Jev is named after William Stanley Jevons: cheaper steam engines drove coal consumption up. When a judgment costs a hundred-thousandth of a cent, you stop sampling and start checking every message, candidate, tool call and handoff. The real payoff is decision density, meaning checks you couldn’t afford to run before.

2.4 Getting access: two paths

First call with curl

Before any Python, here is the smallest possible decision. We send one sentence and ask one yes/no question (a Noul, which Chapter 3 explains properly).

export TYPESAFE_API_KEY=your_key_here
curl -s https://api.typesafe.ai/v1/systemone \
  -H "Authorization: Bearer $TYPESAFE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "jev-latest",
    "state": {"message": "Someone used my card in another city. I did not make this payment."},
    "questions": {
      "is_fraud_suspected": {
        "type": "noul",
        "instructions": "Does `message` describe a transaction the customer did not authorise?"
      }
    }
  }'

The reply carries the answer and the bill:

{
  "model": "jev-1.13.0",
  "answers": { "is_fraud_suspected": { "type": "noul", "noul": 0.97 } },
  "usage": { "input_tokens": 302, "output_tokens": 26 }
}

(Numbers are illustrative. The shape is exact.) Read it like this: noul: 0.97 means “97% likely the customer is describing an unauthorised transaction”, and usage tells you how many input tokens you paid for.

⚠️ Watch Out Always log the model field. jev-latest moves, and your thresholds are calibrated against one specific version.

Error codes you will actually see

StatusSDK exceptionMeaningAction
401TypeSafeAuthenticationErrorKey missing/malformed/wrong shellFix config, don’t retry
422TypeSafeUnprocessableEntityErrorBad question/criteria shapeFix contract, don’t retry
429TypeSafeRateLimitErrorRate limitedRetry with backoff (SDK does)
529 / 5xxTypeSafeInternalServerErrorOverloadedRetry with backoff, then fall back
—TypeSafeAPITimeoutError / TypeSafeAPIConnectionErrorNetworkRetry, then fall back

The pattern is simple: mistakes on our side (401, 422) are never retried; trouble on their side (429, 5xx, timeouts) is retried, then we fall back. Keep that split in mind; the client below encodes it.

2.5 Capstone increment: a swappable decision client

We’ll protect the rest of the codebase from vendor types. The engine talks to one small interface, DecisionClient, and gets back our own normalized result types. That buys us three things:

  • switch Jev ↔ adapter ↔ offline fake by environment variable,
  • unit tests with no network,
  • one place to record latency, tokens and model version for every decision.

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

We build the file in the order a request flows: input shape → answer types → the interface → one real backend → the other backends → the factory.

Step 1: the input shape

Everything starts with what we send. Three aliases name the pieces, and a logger records every decision.

# capstone-project/src/triage/jev_client.py
"""Swappable decision client. The rest of the engine never imports a vendor SDK."""
from __future__ import annotations

import logging
import os
import time
from collections.abc import Callable, Mapping
from dataclasses import dataclass, field
from typing import Any, Protocol, TypeAlias

log = logging.getLogger("triage.jev_client")
JSONState: TypeAlias = str | Mapping[str, Any] | list[Any]
QuestionSpec: TypeAlias = Mapping[str, Any]          # {"type": "noul"|"choice"|"score", ...}
Questions: TypeAlias = Mapping[str, QuestionSpec]
  • JSONState: the evidence. Plain text, a dict, or a list.
  • QuestionSpec: one question, e.g. {"type": "noul", "instructions": "..."}.
  • Questions: a dict of question_id → QuestionSpec. The id comes back on the answer.

Why plain dicts? The SDK accepts question dictionaries as well as Choice/Score/Noul objects. Dicts are easy to version, diff, hash and store in a trace. Chapter 3 builds on that.

Step 2: the answer types (one per primitive)

Jev answers come in three shapes. We wrap each in a small frozen dataclass so nothing downstream can mutate a probability.

Noul (a yes/no probability) is the simplest:

# ---------- Normalised result types (vendor-neutral) ----------
@dataclass(frozen=True, slots=True)
class NoulResult:
    p_yes: float

Choice (pick one option) carries the winner plus every option’s probability. margin() is a quick “was it close?” diagnostic:

@dataclass(frozen=True, slots=True)
class ChoiceResult:
    choice: str
    probabilities: Mapping[str, float]
    confidence: float

    def margin(self) -> float:
        """Gap between the top two options: a useful diagnostic next to `confidence`."""
        top = sorted(self.probabilities.values(), reverse=True)
        return top[0] - (top[1] if len(top) > 1 else 0.0)

Score (a position on a rubric) adds normalized() so rubrics of different lengths can be compared on a 0 to 1 scale:

@dataclass(frozen=True, slots=True)
class ScoreResult:
    score: float
    probabilities: Mapping[int, float]
    confidence: float
    levels: int

    def normalized(self) -> float:
        """Map 0..(levels-1) onto 0..1 so scores with different rubric lengths can be combined."""
        return self.score / (self.levels - 1) if self.levels > 1 else 0.0

Step 3: one response wrapper

A single DecisionResponse holds all answers plus the metadata we want in every trace: model version, tokens, latency and backend.

Answer: TypeAlias = NoulResult | ChoiceResult | ScoreResult


@dataclass(frozen=True, slots=True)
class DecisionResponse:
    answers: Mapping[str, Answer]
    model: str
    input_tokens: int | None
    latency_ms: float
    backend: str
    raw: Any = field(default=None, repr=False, compare=False)

Typed accessors turn a wrong-shape lookup into an immediate error instead of a silent bug:

    def noul(self, qid: str) -> NoulResult:
        a = self.answers[qid]
        if not isinstance(a, NoulResult):
            raise TypeError(f"{qid} is {type(a).__name__}, expected NoulResult")
        return a

    def choice(self, qid: str) -> ChoiceResult:
        a = self.answers[qid]
        if not isinstance(a, ChoiceResult):
            raise TypeError(f"{qid} is {type(a).__name__}, expected ChoiceResult")
        return a
    def score(self, qid: str) -> ScoreResult:
        a = self.answers[qid]
        if not isinstance(a, ScoreResult):
            raise TypeError(f"{qid} is {type(a).__name__}, expected ScoreResult")
        return a

⚠️ Watch Out resp.noul("urgency_rating") raises TypeError, because that question is a Score. This is on purpose. A quiet type mix-up in a routing decision is far worse than a loud crash in a test.

Step 4: the interface and the failure signal

The engine only needs two things: a way to ask, and a way to hear that asking failed.

class DecisionError(RuntimeError):
    """Raised when no typed decision could be obtained. Callers must fall back."""

    def __init__(self, message: str, *, retryable: bool) -> None:
        super().__init__(message)
        self.retryable = retryable
class DecisionClient(Protocol):
    backend: str

    def decide(self, state: JSONState, questions: Questions) -> DecisionResponse: ...

retryable is the field that matters. Auth and schema errors set it to False (retrying can’t fix them); outages set it to True. Chapter 6 hangs the circuit breaker on this flag.

Step 5: turning SDK output into our types

Every backend returns slightly different objects. One helper converts them all into the types above. Read it in three pieces:

# ---------- Conversion from any SDK-shaped response ----------
def _normalise(raw_answers: Mapping[str, Any], questions: Questions) -> dict[str, Answer]:
    out: dict[str, Answer] = {}
    for qid, spec in questions.items():
        if qid not in raw_answers:
            raise DecisionError(f"answer missing for question {qid!r}", retryable=False)
        a = raw_answers[qid]
        kind = spec["type"]
        if kind == "noul":
            out[qid] = NoulResult(p_yes=float(a.noul))
        elif kind == "choice":
            out[qid] = ChoiceResult(
                choice=str(a.choice),
                probabilities={str(k): float(v) for k, v in a.probabilities.items()},
                confidence=float(a.confidence),
            )
        elif kind == "score":
            out[qid] = ScoreResult(
                score=float(a.score),
                probabilities={int(k): float(v) for k, v in a.probabilities.items()},
                confidence=float(a.confidence),
                levels=len(spec["criteria"]),
            )
        else:  # pragma: no cover - guarded by contract validation
            raise DecisionError(f"unknown question type {kind!r}", retryable=False)
    return out

Note how a missing answer raises DecisionError(retryable=False): a half-answered contract is a bug, not an outage.

Step 6: Backend 1, TypeSafe Jev

The real backend. The constructor wires timeouts and retries:

# ---------- Backend 1: TypeSafe Jev ----------
class JevDecisionClient:
    backend = "jev"

    def __init__(self, model: str | None = None, timeout_s: float = 2.0, max_retries: int = 2,
                 *, base_url: str | None = None, api_key: str | None = None) -> None:
        from typesafe_sdk import RetryPolicy, TypeSafeClient  # lazy: optional dependency

        self._model = model or os.getenv("JEV_MODEL", "jev-latest")
        self._client = TypeSafeClient(
            model=self._model,
            api_key=api_key,            # None -> TYPESAFE_API_KEY
            base_url=base_url,          # None -> TYPESAFE_BASE_URL or the hosted API (Ch9 points it at Kev)
            timeout=timeout_s,
            retry=RetryPolicy(max_retries=max_retries, timeout=timeout_s * (max_retries + 1)),
        )

decide() has three jobs: call the SDK, translate errors into DecisionError, and log the result. First, the imports and timer:

    def decide(self, state: JSONState, questions: Questions) -> DecisionResponse:
        from typesafe_sdk import (
            TypeSafeAPIConnectionError,
            TypeSafeAPIError,
            TypeSafeAuthenticationError,
            TypeSafeError,
            TypeSafeUnprocessableEntityError,
        )

Then the call and the error translation. This is the guardrail:

        t0 = time.perf_counter()
        try:
            r = self._client.system_one(state=state, questions=dict(questions))
        except (TypeSafeAuthenticationError, TypeSafeUnprocessableEntityError) as e:
            raise DecisionError(f"non-retryable Jev error: {e}", retryable=False) from e
        except (TypeSafeAPIConnectionError, TypeSafeAPIError) as e:
            # SDK already retried 408/429/5xx with backoff; surface to the circuit breaker.
            raise DecisionError(f"Jev unavailable after retries: {e}", retryable=True) from e
        except TypeSafeError as e:
            raise DecisionError(f"Jev client error: {e}", retryable=False) from e

Finally, wrap the result and log it:

        latency_ms = (time.perf_counter() - t0) * 1000
        resp = DecisionResponse(
            answers=_normalise(r.answers, questions),
            model=r.model,
            input_tokens=r.usage.input_tokens,
            latency_ms=latency_ms,
            backend=self.backend,
            raw=r,
        )
        log.info("decision backend=%s model=%s tokens=%s latency_ms=%.1f",
                 self.backend, resp.model, resp.input_tokens, latency_ms)
        return resp
    def close(self) -> None:
        self._client.close()

⚠️ Watch Out The try block catches vendor exceptions and raises our DecisionError. Nothing outside this file ever sees a TypeSafe...Error. That is how rule 4 of the checklist (“vendor SDK imported in exactly one module”) stays true.

Step 7: Backend 2, the System One Adapter

Don’t have a Jev key yet? The adapter exposes the same interface but uses a general LLM underneath, so you can build the whole engine first.

# ---------- Backend 2: System One Adapter (build before access) ----------
class AdapterDecisionClient:
    """Same interface backed by a general LLM. NOT calibrated like Jev: never tune thresholds on it."""

    backend = "adapter"

    def __init__(self, provider: str = "openai", model: str = "gpt-4o-mini") -> None:
        from system_one_adapter import SystemOneAdapterClient

        self._provider, self._model = provider, model
        self._client = SystemOneAdapterClient(
            structured_outputs=True,
            llm_answer_mode="probabilities",
            normalize_probabilities=True,
        )
    def decide(self, state: JSONState, questions: Questions) -> DecisionResponse:
        t0 = time.perf_counter()
        try:
            r = self._client.system_one(
                state=state, questions=dict(questions), provider=self._provider, model=self._model
            )
        except Exception as e:  # adapter wraps several provider SDKs
            raise DecisionError(f"adapter failure: {e}", retryable=True) from e
        return DecisionResponse(
            answers=_normalise(r.answers, questions),
            model=f"{self._provider}:{self._model}",
            input_tokens=None,
            latency_ms=(time.perf_counter() - t0) * 1000,
            backend=self.backend,
            raw=r,
        )

⚠️ Watch Out The adapter is not calibrated like Jev. A “0.8” from a general LLM is not right 80% of the time. Use it to build plumbing, never to tune thresholds.

Step 8: Backend 3, an offline fake

For unit tests and demos you want zero network. The fake takes a rule function and can simulate latency and outages (fail_times), which is exactly what the circuit-breaker tests in Chapter 6 need.

# ---------- Backend 3: offline fake for unit tests and demos ----------
FakeRule: TypeAlias = Callable[[JSONState, Questions], Mapping[str, Answer]]


class FakeDecisionClient:
    backend = "fake"

    def __init__(self, rule: FakeRule, latency_ms: float = 0.0, fail_times: int = 0) -> None:
        self._rule, self._latency_ms, self._fail_left = rule, latency_ms, fail_times
    def decide(self, state: JSONState, questions: Questions) -> DecisionResponse:
        if self._fail_left > 0:
            self._fail_left -= 1
            raise DecisionError("fake outage", retryable=True)
        if self._latency_ms:
            time.sleep(self._latency_ms / 1000)
        return DecisionResponse(
            answers=dict(self._rule(state, questions)),
            model="fake-0",
            input_tokens=None,
            latency_ms=self._latency_ms,
            backend=self.backend,
        )

Step 9: choosing a backend from the environment

One factory reads DECISION_BACKEND and hands back the right client:

def client_from_env() -> DecisionClient:
    backend = os.getenv("DECISION_BACKEND", "jev")
    if backend == "jev":
        return JevDecisionClient()
    if backend == "adapter":
        return AdapterDecisionClient(
            provider=os.getenv("ADAPTER_PROVIDER", "openai"),
            model=os.getenv("ADAPTER_MODEL", "gpt-4o-mini"),
        )
    raise ValueError(f"Unsupported DECISION_BACKEND={backend!r} (fake is for tests only)")

The fake backend is deliberately missing: it is for tests only, so production config can never pick it by accident.

2.6 Mini-project 1: latency & cost harness

The harness runs the same contract N times, reports p50/p95/p99, and compares one fan-out call with N sequential single-question calls. Run it on day one and keep running it: numbers measured from Tirunelveli (a culturally rich city in Tamil Nadu, South India) won’t match numbers measured from San Francisco.

Note: p50/p95/p99 are latency percentiles: p50 is the typical (median) request time, while p95 and p99 are the times that 95% and 99% of requests finish within, which shows how slow the worst 1 in 20 and 1 in 100 calls get

💡 Mental Model: fan-out vs sequential Fan-out is asking a waiter for starter, main and dessert in one trip. Sequential is calling the waiter back three times. Same food, three times the walking.

📄 Full source: mini-projects/01-starter-client/bench.py

Part 1: a result record and a percentile helper.

# mini-projects/01-starter-client/bench.py
from __future__ import annotations

import statistics
import time
from collections.abc import Sequence
from dataclasses import dataclass

from triage.jev_client import DecisionClient, DecisionError, Questions, JSONState, client_from_env

JEV_USD_PER_M_INPUT = 0.042
@dataclass(frozen=True, slots=True)
class BenchResult:
    label: str
    n: int
    p50_ms: float
    p95_ms: float
    p99_ms: float
    errors: int
    usd_per_1k: float
def _pct(xs: Sequence[float], p: float) -> float:
    if not xs:
        return float("nan")
    s = sorted(xs)
    k = min(len(s) - 1, max(0, round(p / 100 * (len(s) - 1))))
    return s[k]

Part 2: time one strategy N times. Failed calls are counted, not timed, so one outage can’t flatter your p50.

def bench(label: str, fn, n: int) -> BenchResult:
    lat: list[float] = []
    tokens: list[int] = []
    errors = 0
    for _ in range(n):
        t0 = time.perf_counter()
        try:
            used = fn()
        except DecisionError:
            errors += 1
            continue
        lat.append((time.perf_counter() - t0) * 1000)
        tokens.append(used)
    mean_tokens = statistics.fmean(tokens) if tokens else 0.0
    return BenchResult(label, n, _pct(lat, 50), _pct(lat, 95), _pct(lat, 99), errors,
                       mean_tokens * JEV_USD_PER_M_INPUT / 1_000_000 * 1000)

Part 3: the two strategies under test. One sends all questions together; the other loops.

def fan_out(client: DecisionClient, state: JSONState, qs: Questions) -> int:
    r = client.decide(state, qs)
    return r.input_tokens or 0


def sequential(client: DecisionClient, state: JSONState, qs: Questions) -> int:
    total = 0
    for qid, spec in qs.items():
        total += client.decide(state, {qid: spec}).input_tokens or 0
    return total

Part 4: run both and print.

if __name__ == "__main__":
    from triage.contract import TRIAGE_QUESTIONS  # built in Chapter 2

    client = client_from_env()
    state = {"report": "Two card payments of ₹ 49,999 I don't recognise, both at 3am. Please block my card."}
    for r in (
        bench("fan-out (1 call, 3 questions)", lambda: fan_out(client, state, TRIAGE_QUESTIONS), 50),
        bench("sequential (3 calls)", lambda: sequential(client, state, TRIAGE_QUESTIONS), 50),
    ):
        print(f"{r.label:32} p50={r.p50_ms:6.0f}ms p95={r.p95_ms:6.0f}ms "
              f"p99={r.p99_ms:6.0f}ms err={r.errors} ${r.usd_per_1k:.5f}/1k")

Output

Strategyp50p95p99errCost ($/1k)
fan-out (1 call, 3 questions)406ms515ms561ms00.03205
sequential (3 calls)1058ms1233ms1352ms00.05666

Reference numbers to compare yours against (published, not ours):

ExperimentRequestsLatencyCost
13 questions, one call, 53,777-char GDPR doc10.27 s$0.000497
Same 13 questions, sequential132.71 s$0.006090
Advantage13× fewer10.0× faster12.2× cheaper

Most of that saving comes from sending a big state once. With a small state the cost gap shrinks, but latency still favours fan-out. Concurrent separate calls close the latency gap but never the cost gap.

2.7 Checklist

  • TYPESAFE_API_KEY in env / secret manager, never in state, logs or git
  • JEV_MODEL pinned to a versioned id (e.g. jev-1.13.0) before any threshold tuning
  • Every decision logs model, input_tokens, latency_ms, backend
  • Vendor SDK imported in exactly one module (jev_client.py)
  • Harness run from your region; p95 recorded in the design doc

What’s next

We have a client that can ask something. Chapter 3 is about what to ask: the three primitives, how to write questions Jev answers well, and the versioned decision contract for the triage engine (is_fraud_suspected, urgency_rating, dispatch_target) that the bench above already imports.