System One models vs LLMs: fast, typed decisions for automation

WhyMost automation is System 1 work

Most of the decisions in automation are fast judgements, and we've been paying for slow reasoning to make them. This part covers the idea, the decisions it applies to, what it's worth and what it changes for the business.

System 1 vs System 2, in summary

The names come from Daniel Kahneman's Thinking, Fast and Slow. System 1 is the fast, automatic judgement you make without deliberating: this email looks like spam, that driver is about to pull out. System 2 is slow and effortful. It's the reasoning you do when you work through a problem step by step.

That maps onto the models quite neatly:

The point isn't that one replaces the other. Most of what automation does all day is System 1 work: route this, gate that, flag the other. We've been paying System 2 prices and waiting System 2 latencies for it because an LLM was the only fuzzy decision-maker to hand.

The decisions enterprises already make

The out-of-the-box examples are deliberately simple. The OpenJev README triages a support message ("is it urgent, which team, how upset are they"). TypeSafe shows churn classification, a bot playing Doom and a Wikipedia-racing agent. They show the three primitives well, but they aren't what keeps a platform team up at night. I pick three of them for the What and How parts.

Put an automation layer on top of CI and CD and the decisions look like this. Every one of them is currently either a brittle rule, an LLM call we wait several seconds for, or a person:

The shape is the same as the demo: unstructured state, a handful of typed questions. What changes is the cost of being wrong and the volume. A misrouted support ticket is annoying. A wrongly approved deploy is an incident. That changes how you use the answer more than which model you use:

• Every question gets a threshold, and the threshold reflects the stakes. Auto-merge a docs-only change at P ≥ 0.9, but only auto-approve a deploy at P ≥ 0.99, and only when a noul says the canary is healthy.
• Low confidence sends the item to a human. It doesn't get a guess. The confidence is the thing that lets you automate the easy 80% safely and escalate the rest.
• The decisions are small enough to run on every event. At 100 ms and fractions of a penny each, you can gate every PR and every build, not a sampled few.
• Irreversible actions still need a person or a System 2 check: production data migrations, anything with a regulatory sign-off.

What it's worth

Here's a worked example that's easy to adjust. Take an automation layer making 10,000 decisions a day across the pipeline stages above. Each decision sends about 2,000 tokens of state: a diff summary, a log excerpt, some metadata. An LLM answering the same questions as JSON writes about 300 tokens, and that's before any reasoning trace, which can easily be ten times that.

Change the inputs and the chart updates. It uses a log scale, because the gap is several orders of magnitude.

Claude list prices per million tokens (input / output): Fable 5.1 $10 / $50, Opus 5 $5 / $25, Sonnet 5 $2 / $10, Haiku 4.5 $1 / $5, from Anthropic's pricing, excluding caching and batch discounts. Jev is $0.042 per million input tokens with free output, per TypeSafe's launch post. The self-hosted row is an assumption: a rented 48–96 GB GPU at the hourly rate you enter, running all month, with one extra GPU per ~50 decisions/second at 5× peak-to-average. Enter your own rate. Codiv's free tier and a Mac you already own are effectively $0 at this volume, so they're left off the chart.

What the numbers say at the default volume:

• Hosted Jev is roughly 400× cheaper than a frontier LLM and 40× cheaper than the smallest Claude. Most of the saving is output tokens, because the answer is a single token per question and it isn't charged for. At about $25 a month, it's a rounding error, provided you're happy to send the data to TypeSafe.
• Self-hosting is a fixed cost, not a per-token one, and at low volume it isn't the cheap option. A rented GPU running all month costs more than Haiku at 10,000 decisions a day. Try 100,000 decisions a day and it beats every LLM on the list. Against hosted Jev it only breaks even at around half a million decisions a day, close to what one card can serve.
• So the case for OpenJev isn't price, it's control. The data never leaves your network, nobody can deprecate the model under you, and a Mac or a GPU you already own costs you only the electricity.
• Latency is a cost too. A pipeline that waits 5–30 s for an LLM on every gate spends that in runner minutes and developer time. At ~100 ms the gate stops being something anyone notices.

The business case: process automation

Business process automation has always hit the same wall. The process diagram is full of decision diamonds, and each diamond needs either a hard rule (which breaks on the first messy input) or a person (who becomes the bottleneck). LLMs made the fuzzy diamonds possible, but at a latency and price that pushed them into batch jobs and pilots.

System One models change the economics of the diamond itself:

• More straight-through processing. Invoices, claims, onboarding checks, access requests and change approvals can be decided inline, with only the low-confidence cases sent to a person. The confidence score is the exception-handling policy.
• Rules engines with judgement. A typed answer drops straight into a BPMN gateway or a DMN decision table. "Is this supplier email a change of bank details?" becomes a noul you can branch on, not a regex nobody dares touch.
• An audit trail that's easier to reason about. Every decision is stored with the state, the question and the full distribution. "Why did it approve this?" has a numeric answer, and you can replay it after changing a threshold.
• Coverage, not sampling. When a check costs next to nothing you can run it on 100% of transactions. That matters most for compliance and fraud controls.
• Data can stay on premises. With OpenJev on your own GPU, customer and financial data never goes to a third party, and that's often the real blocker for BPA in regulated industries.

The catch is governance. A process that decides in 100 ms can make a lot of bad decisions very quickly. You need evaluation sets per decision, calibration checks, drift monitoring and a named owner for each threshold. That's the same discipline we'd apply to any model in production, but people will be tempted to skip it because each call feels so cheap.

WhatJev, the alternatives, and three use cases

What Jev is and how it differs from an LLM, the open alternatives, where a System One read sits in an agent graph, and the three use cases I'd start with.

Jev vs standard LLMs

TypeSafe launched Jev as "the first System One model". You send it a state (any unstructured text, and with OpenJev images too) plus a set of typed questions. It returns each answer with a probability and a confidence, and it generates no prose at all. TypeSafe says it trains Jev with Reinforcement Learning for Calibrated Decisions (RLCD), so that the probabilities mean what they say.

Jev vs the open variants

Jev is closed weights and API only. Within days of the launch, people started rebuilding the idea in the open. The rebuilds fall into four groups:

1. Wire-compatible servers on a big open model. OpenJev, which I've forked here, is one. It runs Google/NVIDIA's DiffusionGemma, a diffusion language model that denoises a whole canvas of tokens at once. OpenJev leaves one masked slot per question and reads the probability distribution for every slot in a single pass. It speaks Jev's API, so TypeSafe's own SDK works against it unchanged.
2. Logit readers on a frozen small model. These take an ordinary autoregressive model (Qwen 4B, for example) and read the next-token probabilities of the label tokens. They're cheap and they run on a laptop, but the probabilities are relative, not calibrated.
3. Small models fine-tuned for decisions. These are trained or given a decision head so they answer in Jev's shape. Some are small enough to run on a CPU.
4. Structured-output libraries. Outlines, Instructor and DSPy put a typed interface on any LLM. You still get System 2 cost and latency, just with better manners.

The rows other than Jev and OpenJev are summarised from the community list at systemonemodels.org. I haven't run them myself.

My take: nobody outside TypeSafe knows how to do RLCD, because it hasn't been described in enough detail to copy. So every open variant gives you the shape of Jev (typed answers, parallel questions, a probability on everything). None of them gives you Jev's calibration guarantee for free. OpenJev is the most complete of the lot. It has the same wire API, a strong 26B model underneath, images, and an OpenAI-compatible chat endpoint from the same weights. Its confidence comes from a real distribution rather than a number the model makes up. Even so, if you're going to act on thresholds you should check its calibration on your own data. More on that under caveats.

Where it fits in agent graphs

Most serious agent systems I see now are graphs: LangGraph, Google's ADK or a home-grown state machine. The nodes do the work and the edges decide what happens next. Today both jobs usually go to an LLM, which means every edge in the graph costs seconds.

My view is simple: System 2 models belong in the nodes, and System 1 models belong on the edges. These are the places a System One read fits:

• Routers and conditional edges: "which specialist agent handles this?" as a choice over the agents.
• Supervisors: "is the task done, blocked or still in progress?" on every turn of a loop, cheaply enough to ask every time.
• Guards: before a tool call, "does this action match the user's intent? Is it destructive?" Jailbreak and prompt-injection checks on inputs.
• Verifiers: after an agent produces something, "does this fix address the failing test? Does the summary contain claims that aren't in the source?"
• Escalation: low confidence anywhere sends the item to a human node, not to a retry.

Two things follow from this. First, the graph gets faster and cheaper in proportion to how many edges it has, which is usually more than the number of nodes. Second, the control flow becomes inspectable. Each branch was taken because a probability crossed a threshold you set, not because an LLM happened to write the word "approve".

OpenJev also serves an OpenAI-compatible /v1/chat/completions from the same weights, so a small graph can run both its System 1 edges and its lighter System 2 nodes on one local model. I'd still use a frontier model for the hard node, like the review-and-fix agent above.

The three use cases below are graphs of the same shape.

Three use cases to start with

The CI/CD table in the Why part is the broad picture. These are the three problems I'd pick first in most organisations, because each one is currently either expensive, leaky or done on a sample. For each: why today's approach hurts, what changes with a System One read, and the graph it becomes. The code for each is in the How part.

Evals: a judge that knows when it's unsure

Why: how it's done today. Most teams grade LLM output with another LLM. The judge reads the case, writes a paragraph of reasoning and gives a score out of five. It works, but it takes seconds per case and costs real money, so it runs nightly on a sample. The criteria bleed into each other: an answer that reads well gets marked as faithful too. The score is an integer the model made up, and every so often the JSON doesn't parse.

What changes: with a System One judge the rubric becomes a set of typed questions, answered in parallel and in isolation against the same case.

• No halo effect. Each criterion is its own read, so "well written" can't lift "faithful".
• A probability, not a grade. P ≥ 0.8 passes, P ≤ 0.2 fails, and anything in between goes to a person. The unsure queue is where human review time pays off most.
• All traffic, not a sample. At ~100 ms and a fraction of a penny per case you can grade every production answer, not just last night's 200.
• The same judge offline and inline. The rubric that scores the eval set also guards production answers, so what you measure is what you enforce.
• Replayable. Scores are stored as distributions, so a changed threshold can be re-applied to history without re-grading.

How to build it →

Personal data in logs

Why: how it's done today. A DLP rule set and a pile of regexes. They're good at the structured cases: email addresses, card numbers that pass a Luhn check, cloud keys. They miss the ones that turn into incidents: a customer's name and address pasted into a support note, a health condition in a claims comment, a search query that says more than it should. Scanning every line with an LLM is unaffordable, and it means shipping raw logs to a third party.

What changes: with a System One read every line gets four questions that match the categories your privacy policy already uses.

• Regex first, then the model. The cheap deterministic scan still runs and still cuts exact spans. The read covers the free text the patterns can't see into.
• Policy, not a verdict. A regex hit or P ≥ 0.8 means redact before shipping. P from 0.3 to 0.8 means quarantine for a person. Below that, forward. The thresholds belong to the DPO, not the model.
• It stays on your network. This is the example where OpenJev on your own GPU isn't optional. Sending logs to a hosted API to find out whether they contain personal data is itself the disclosure.

How to build it →

Model routing for log events

Why: how it's done today. Either everything goes to one model, usually a big one because nobody wants to be blamed for the miss, or a static rule maps severity to model, or an LLM decides which LLM to call. That last one costs seconds and money per event before any work gets done.

What changes: with a System One router one read decides whether an event needs a model at all and how much model it needs. A second read checks the answer and moves the event up a tier if it falls short.

This is where the money is. The mix of events is my assumption, but the prices are list prices. At 100,000 events a day, with 60% needing no model, 30% Haiku, 7% Sonnet, 2.5% Opus and 0.5% Fable, routing costs about $210 a day against $2,850 for sending everything to Opus 5. That's roughly 13× cheaper, or around $79k a month. The System One reads themselves come to about $1.50 a day at Jev's price. The less obvious win is that the security incident goes to the strongest model rather than whatever the default happens to be.

How to build it →

HowRun it, then build the use cases

Running the model, a one-off setup, then each use case step by step. Every code block has a copy button.

Run it locally, and how big it is

This is where the open variants earn their keep. Jev is a hosted API, but OpenJev fits on a single workstation GPU or a decent Mac.

The model is DiffusionGemma 26B-A4B, a diffusion version of Gemma 4 26B released by NVIDIA under Apache-2.0. It's a mixture of experts with 25.2B parameters in total, of which 3.8B are active per token. It needs the memory of a 25B model but does the compute of a ~4B one, and that's why it's quick. It has a 262K-token context and takes text, images and video as input.

* These are my own estimates from the parameter count (25.2B × 1 or 2 bytes, plus overhead), not published figures. The other numbers come from the OpenJev README.

On a Mac it really is two commands:

pip install -e '.[mlx]'
OPENJEV_BACKEND=mlx python -m openjev     # http://127.0.0.1:8080

On an NVIDIA box, docker compose up -d pulls a prebuilt image with the patched vLLM. Either way the first start downloads the weights into ~/.cache/huggingface, and after that nothing leaves the machine. For regulated data that matters as much as the price does.

Minimum spec for my demo

My own demo is a small, dependency-free Python project. It has four scripts, one for each primitive plus a triage pipeline that combines all three, and a script that starts OpenJev on a Mac with the 4-bit MLX weights. This is what you need to run it against the real model:

Why 32 GB and not 16 or 24? The 4-bit weights alone are 16.6 GB, and they have to sit in the part of unified memory the GPU is allowed to use. By default macOS gives the GPU roughly two-thirds to three-quarters of RAM. That's about 16 GB on a 24 GB machine, which is just short, and a 16 GB Mac has no chance at all. On 24 GB you can raise the limit with sudo sysctl iogpu.wired_limit_mb=…, but you'll be running with nothing to spare. Treat 32 GB as the real floor.

There are two other ways to run it:

• NVIDIA instead of a Mac. A 24 GB GPU (RTX 3090/4090 class or better) with Docker and the NVIDIA container toolkit runs the same server. Point the demo at it with OPENJEV_BASE_URL.
• No model at all. Every example takes --mock, which starts an offline mock of the API with the real request and response shapes. It answers by keyword overlap, so it's good for learning the API, not for judging quality. It runs on any machine with Python 3.10 or later.

Throughput is the other half. On one RTX PRO 6000 using 38% of the card, OpenJev did 57 decisions per second at 64 concurrent requests, with p95 latency around 1.1 s. The Mac backend handles one read at a time, about 4 req/s under load. That's fine for a developer loop or a small team, but not for serving production.

If you don't want to run anything, Codiv hosts OpenJev with a free allowance of 100M System One tokens, and the API is the same.

Set up once

One client serves all three examples. TypeSafe's SDK talks to OpenJev unchanged, so moving between Jev, hosted OpenJev and OpenJev on your own GPU is one environment variable, not a code change.

git clone https://github.com/SteFletcher/openjev && cd openjev
pip install -e '.[examples]'          # typesafe-sdk, langgraph, anthropic

# pick a System One backend
export TYPESAFE_BASE_URL=http://127.0.0.1:8080       # OpenJev on this machine
# export TYPESAFE_BASE_URL=https://api.codiv.ai      # hosted OpenJev
# export TYPESAFE_BASE_URL=https://api.typesafe.ai   # Jev
export TYPESAFE_API_KEY=...          # any value for a local server without auth
export ANTHROPIC_API_KEY=...         # for the Claude (System 2) nodes

# TYPESAFE_BASE_URL  https://api.typesafe.ai  -> Jev
#                    https://api.codiv.ai     -> hosted OpenJev
#                    http://127.0.0.1:8080    -> OpenJev on your own hardware (the default)
# TYPESAFE_API_KEY   the SDK requires one; any value works for a local server without auth
def client() -> TypeSafeClient:
    return TypeSafeClient(
        base_url=os.environ.get("TYPESAFE_BASE_URL", "http://127.0.0.1:8080"),
        api_key=os.environ.get("TYPESAFE_API_KEY", "local"),
        model=os.environ.get("TYPESAFE_MODEL", "jev-latest"),  # OpenJev accepts Jev's name as an alias
    )

Build the eval judge

The code is in examples/worked/evals/. What it does and why is in the What part.

1Write the rubric as questions. Each criterion is a noul or a score in Jev's wire format, so it works with any client.

RUBRIC = {
    "faithful": {
        "type": "noul",
        "instructions": "Every factual claim in the answer is supported by the context.",
        "criteria": {"true": "all claims appear in or follow directly from the context",
                     "false": "the answer adds, changes or contradicts facts in the context"},
    },
    "answers_question": {
        "type": "noul",
        "instructions": "The answer directly addresses the user's question.",
    },
    "personal_data": {
        "type": "noul",
        "instructions": "The answer discloses personal data about an identifiable individual, "
                        "such as a name together with an email address or phone number.",
    },
    "quality": {
        "type": "score",
        "instructions": "Overall quality as a reply from an internal policy assistant.",
        "criteria": ["unusable", "poor", "acceptable", "good", "excellent"],
    },
}

2Turn probabilities into a verdict. The thresholds live in your code, where they can be reviewed, not in the model. The reasons go back to the generator if it gets a second try.

PASS_AT, FAIL_AT = 0.8, 0.2   # P >= 0.8 is a yes, P <= 0.2 a no, anything between goes to a person


@dataclass(frozen=True)
class Verdict:
    outcome: str                  # "pass", "fail" or "human"
    faithful: float
    answers_question: float
    personal_data: float
    quality: float                # expected level, 0-4
    reasons: tuple[str, ...]      # fed back to the generator on a retry

    @property
    def p_good(self):
        """The weakest link, as one number for calibration checks."""
        return min(self.faithful, self.answers_question, 1 - self.personal_data)


def judge(s1, question, context, answer):
    r = s1.system_one({"question": question, "context": context, "answer": answer}, RUBRIC)
    n = {k: a.noul for k, a in r.nouls.items()}
    checks = {  # every check is phrased so that 1.0 is good
        "not supported by the context": n["faithful"],
        "does not answer the question": n["answers_question"],
        "discloses personal data": 1 - n["personal_data"],
    }
    if all(p >= PASS_AT for p in checks.values()):
        outcome = "pass"
    elif any(FAIL_AT < p < PASS_AT for p in checks.values()):
        outcome = "human"
    else:
        outcome = "fail"
    reasons = tuple(f"{why} (P={p:.2f})" for why, p in checks.items() if p < PASS_AT)
    return Verdict(outcome, n["faithful"], n["answers_question"], n["personal_data"],
                   r.scores["quality"].score, reasons)

3Grade the eval set in parallel, and gate on it. Agreement with the human labels and the Brier score tell you whether the judge can be trusted before you let it block anything.

def run(s1, cases, workers=8):
    with ThreadPoolExecutor(workers) as pool:   # reads are independent, so grade them in parallel
        verdicts = list(pool.map(lambda c: judge(s1, c["question"], c["context"], c["answer"]), cases))
    rows = [{"id": c["id"], "label": c.get("label"), **asdict(v)} for c, v in zip(cases, verdicts)]

    decided = [r for r in rows if r["outcome"] != "human" and r["label"]]
    labelled = [(v.p_good, r["label"] == "pass") for v, r in zip(verdicts, rows) if r["label"]]
    return {
        "cases": len(rows),
        "pass_rate": sum(r["outcome"] == "pass" for r in rows) / len(rows),
        "to_human": [r["id"] for r in rows if r["outcome"] == "human"],
        # how often the judge agrees with a person, on the cases it was willing to decide
        "agreement": sum(r["outcome"] == r["label"] for r in decided) / len(decided) if decided else None,
        # Brier score of the judge's P(good) against the labels: 0 is perfect, 0.25 is a coin toss
        "brier": sum((p - y) ** 2 for p, y in labelled) / len(labelled) if labelled else None,
        "rows": rows,
    }


def gate(report, min_pass, min_agreement):
    failures = []
    if report["pass_rate"] < min_pass:
        failures.append(f"pass rate {report['pass_rate']:.2f} < {min_pass}")
    if report["agreement"] is not None and report["agreement"] < min_agreement:
        failures.append(f"judge agreement {report['agreement']:.2f} < {min_agreement}")
    return failures

4Put the gate in CI. First check that the judge still agrees with the human labels. Then answer the question set with the changed prompt and grade the answers. A change that drops the pass rate fails the pull request.

name: eval-gate
on:
  pull_request:
    paths: ["examples/worked/**"]   # in your repo: your prompts, assistant code and eval sets

jobs:
  eval:
    runs-on: ubuntu-latest
    env:
      TYPESAFE_BASE_URL: ${{ vars.TYPESAFE_BASE_URL }}
      TYPESAFE_API_KEY: ${{ secrets.TYPESAFE_API_KEY }}
      ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - run: pip install -e '.[examples]'
      - name: Check the judge still agrees with the human labels
        run: python examples/worked/evals/run_eval.py examples/worked/evals/dataset.jsonl --min-agreement 0.85
      - name: Answer the eval questions with the changed prompt
        run: python examples/worked/evals/generate_answers.py examples/worked/evals/questions.jsonl > answers.jsonl
      - name: Judge the answers, and fail the build below the bar
        run: python examples/worked/evals/run_eval.py answers.jsonl --out eval-report.json --min-pass 0.9
      - uses: actions/upload-artifact@v4
        if: always()
        with: { name: eval-report, path: eval-report.json }

5Use the same judge inline. The guard in the diagram above, as a LangGraph StateGraph. The System One read is the conditional edge.

class GuardState(TypedDict, total=False):
    question: str
    context: str
    answer: str
    attempts: int
    verdict: object
    outcome: str


def build_guard(s1, claude):
    def prompt(st, feedback=""):
        return f"Policy:\n{st['context']}\n\nQuestion: {st['question']}{feedback}"

    def generate(st):
        return {"answer": claude("sonnet", SYSTEM, prompt(st)).text, "attempts": 1}

    def judge_node(st):
        return {"verdict": judge(s1, st["question"], st["context"], st["answer"])}

    def regenerate(st):
        why = "; ".join(st["verdict"].reasons)
        feedback = f"\n\nA reviewer rejected this draft ({why}):\n{st['answer']}\nWrite a corrected answer."
        return {"answer": claude("sonnet", SYSTEM, prompt(st, feedback)).text, "attempts": st["attempts"] + 1}

    def after_judge(st):
        outcome = st["verdict"].outcome
        if outcome == "pass":
            return "deliver"
        if outcome == "fail" and st["attempts"] < MAX_ATTEMPTS:
            return "regenerate"
        return "human"

    g = StateGraph(GuardState)
    g.add_node("generate", generate)
    g.add_node("judge", judge_node)
    g.add_node("regenerate", regenerate)
    g.add_node("deliver", lambda st: {"outcome": "delivered"})
    g.add_node("human", lambda st: {"outcome": "queued for a person"})
    g.add_edge(START, "generate")
    g.add_edge("generate", "judge")
    g.add_conditional_edges("judge", after_judge, ["deliver", "regenerate", "human"])
    g.add_edge("regenerate", "judge")
    g.add_edge("deliver", END)
    g.add_edge("human", END)
    return g.compile()

6Run it.

python examples/worked/evals/run_eval.py examples/worked/evals/dataset.jsonl \
    --out report.json --min-pass 0.5 --min-agreement 0.85

python examples/worked/evals/generate_answers.py examples/worked/evals/questions.jsonl > answers.jsonl
python examples/worked/evals/run_eval.py answers.jsonl --min-pass 0.9

python examples/worked/evals/guard.py "Can I expense a taxi home after working late?" \
    --context "Taxis are reimbursable when travel after 21:00 is required for business reasons and approved in advance by a line manager."

Where to be careful. A judge is a model too. Check it against human labels (agreement, and whether P = 0.8 really means right 80% of the time) before you put it on the critical path. Where possible, don't let a model judge its own family's output. And for things with a checkable answer, like code or arithmetic, running the tests beats any judge.

Build the PII filter

The code is in examples/worked/pii/. What it does and why is in the What part.

1Ask the questions your privacy policy already asks, and keep the regexes for the structured cases, where they're exact and give you a span to cut.

QUESTIONS = {
    "person": {"type": "noul", "instructions":
               "The log line identifies a person: a name, home address, phone number or email address."},
    "special_category": {"type": "noul", "instructions":
               "The log line mentions a person's medical condition, symptoms or treatment, or their religion, "
               "ethnicity, sexuality, trade-union membership or biometrics (special-category data, UK GDPR Article 9)."},
    "secret": {"type": "noul", "instructions":
               "The log line contains a credential or secret: a password, API key, access token or private key."},
    "financial": {"type": "noul", "instructions":
               "The log line contains a payment card number or a bank account number."},
}

# the structured cases stay with regexes: they are exact, free, and give a span to cut
PATTERNS = {
    "EMAIL": re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+"),
    "IPV4": re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b"),
    "CARD": re.compile(r"\b(?:\d[ -]?){13,19}\b"),
    "AWS_SECRET": re.compile(r"(?<=AWS_SECRET_ACCESS_KEY=)\S+"),
}

2Decide per line. A regex hit or a confident read means redact, an unsure read means quarantine, anything else ships.

class PiiDetector:
    def __init__(self, s1, redact_at=0.8, review_at=0.3):
        self.s1, self.redact_at, self.review_at = s1, redact_at, review_at

    def check(self, line):
        hits = [(name, m.span()) for name, rx in PATTERNS.items() for m in rx.finditer(line)
                if name != "CARD" or luhn(m.group())]
        r = self.s1.system_one(line, QUESTIONS)          # the whole line, free text included
        p = {k: a.noul for k, a in r.nouls.items()}
        top = max(p.values())

        if hits or top >= self.redact_at:
            return Decision("redact", self._redact(line, hits, p), p, tuple(hits))
        if top >= self.review_at:
            return Decision("quarantine", line, p)           # unsure: a person decides
        return Decision("forward", line, p)

    def _redact(self, line, hits, p):
        for name, (a, b) in sorted(hits, key=lambda h: -h[1][0]):
            line = line[:a] + f"[{name}]" + line[b:]
        found = [k for k, v in p.items() if v >= self.redact_at]
        if found and not hits:
            # the model says *that* the line holds personal data, not *where*: drop the free text
            tag = f'"[REDACTED:{",".join(found)}]"'
            line = QUOTED.sub(tag, line) if QUOTED.search(line) else f"{line.split(' ', 1)[0]} [REDACTED:{','.join(found)}]"
        return line

3Filter the stream. Reads run concurrently, lines leave in the order they arrived, and memory stays flat on an endless stream.

def filter_stream(detector, lines, out, quarantine, workers=16):
    """Check lines concurrently, emit them in order. Returns a count per route."""
    counts = Counter()
    pending = deque()

    def emit(future):
        d = future.result()
        counts[d.route] += 1
        (quarantine if d.route == "quarantine" else out).write(d.line + "\n")

    with ThreadPoolExecutor(workers) as pool:
        for line in lines:
            line = line.rstrip("\n")
            if not line:
                continue
            pending.append(pool.submit(detector.check, line))
            while len(pending) >= workers * 2:   # bounded: memory stays flat on an endless stream
                emit(pending.popleft())
        while pending:
            emit(pending.popleft())
    return counts

4Put it in front of whatever ships logs off the box.

python examples/worked/pii/log_filter.py --quarantine quarantine.log \
    < examples/worked/pii/sample.log > clean.log

# in a pipeline, with OpenJev on your own network
kubectl logs -f deploy/support-api \
  | python examples/worked/pii/log_filter.py --quarantine /var/log/pii-review.log \
  | vector --config examples/worked/pii/vector.toml

The wording of the questions matters. My first version asked about "special-category data about a person: health, religion, …". On the real model it flagged INFO health /ready ok as health data (P = 0.81), because the service is called "health". It also put an order total, amount=42.10 GBP, at 0.39 for "payment card or bank account details", which is enough to quarantine it. Asking instead about "a person's medical condition, symptoms or treatment" and "a payment card number or a bank account number" took both false alarms to 0.00 and left the real hits at 1.00. That's eight lines, not an evaluation. It's exactly the kind of thing a labelled set per question will find, so build one before you trust the thresholds.

Where to be careful. One GPU does ~57 reads a second, about 5 million lines a day flat out. That's plenty for free-text fields and WARN-and-above, but not for every DEBUG line in a large estate, so point it at the lines that can actually carry free text. It also tells you that a line holds personal data, not where. Redact the whole field, or hand the few flagged lines to a span extractor (an NER model like GLiNER, or a System 2 call) to cut precisely.

Build the model router

The code is in examples/worked/routing/. What it does and why is in the What part.

1Ask what the event needs, and map the answer to a tier in code. The most likely level picks the model, code work starts at Sonnet, and an unsure read errs one tier up.

ROUTE = {
    "actionable": {"type": "noul", "instructions": "Someone needs to act on or investigate this event."},
    "complexity": {
        "type": "score",
        "instructions": "How much analysis does a useful answer to this event need?",
        "criteria": [
            "routine: restate it in one line with the obvious next step",
            "simple: explain one known error and how to fix it",
            "moderate: diagnose one failing component from a stack trace or metrics",
            "hard: correlate several services or a timeline to find the root cause",
            "critical: security incident or possible data loss, needs careful high-stakes analysis",
        ],
    },
    "needs_code": {"type": "noul", "instructions": "A good answer needs someone to read or change application code."},
}
CHECK = {
    "adequate": {"type": "noul",
                 "instructions": "The analysis names a probable cause and a concrete next step for this event."},
}

LEVEL_TO_TIER = {0: "haiku", 1: "haiku", 2: "sonnet", 3: "opus", 4: "fable"}


def pick_tier(r):
    """Policy in code: most likely level -> tier, code work starts at Sonnet, unsure errs upwards."""
    cx = r.scores["complexity"]
    tier = TIERS.index(LEVEL_TO_TIER[int(most_likely(cx.probabilities))])
    if r.nouls["needs_code"].noul >= 0.5:
        tier = max(tier, TIERS.index("sonnet"))
    if cx.confidence < 0.4:
        tier = min(tier + 1, len(TIERS) - 1)
    return tier

2Wire the graph. Both System One reads are conditional edges. The reducers add up the models used and the cost across escalations.

class RouteState(TypedDict, total=False):
    event: str
    actionable: float
    tier: int
    answer: str
    adequate: float
    models: Annotated[list, operator.add]     # every model that ran, appended by each analyse
    cost: Annotated[float, operator.add]      # dollars, summed across escalations
    s1_tokens: Annotated[int, operator.add]


def build_router(s1, claude, adequate_at=0.7, actionable_at=0.3):
    def route(st):
        r = s1.system_one(st["event"], ROUTE)
        return {"actionable": r.nouls["actionable"].noul, "tier": pick_tier(r),
                "s1_tokens": r.usage.input_tokens or 0}

    def analyse(st):
        reply = claude(TIERS[st["tier"]], SYSTEM, st["event"])
        return {"answer": reply.text, "models": [reply.model], "cost": reply.cost}

    def check(st):
        r = s1.system_one({"event": st["event"], "analysis": st["answer"]}, CHECK)
        return {"adequate": r.nouls["adequate"].noul, "s1_tokens": r.usage.input_tokens or 0}

    def after_check(st):
        at_top = st["tier"] == len(TIERS) - 1
        return "done" if st["adequate"] >= adequate_at or at_top else "escalate"

    g = StateGraph(RouteState)
    g.add_node("route", route)
    g.add_node("analyse", analyse)
    g.add_node("check", check)
    g.add_node("escalate", lambda st: {"tier": st["tier"] + 1})
    g.add_node("drop", lambda st: {"answer": "(no model called)"})
    g.add_edge(START, "route")
    g.add_conditional_edges("route", lambda st: "analyse" if st["actionable"] >= actionable_at else "drop",
                            ["analyse", "drop"])
    g.add_edge("analyse", "check")
    g.add_conditional_edges("check", after_check, {"done": END, "escalate": "escalate"})
    g.add_edge("escalate", "analyse")
    g.add_edge("drop", END)
    return g.compile()

3Call Claude at the chosen tier. Opus 5 and Fable 5.1 use server-side fallbacks: "default", so a declined request is re-run on Anthropic's recommended fallback model rather than failing. Every call checks stop_reason before reading content.

class Claude:
    """claude(tier, system, prompt) -> Reply. Reads ANTHROPIC_API_KEY or an `ant auth login` profile."""

    def __init__(self, client=None, max_tokens=8000):
        self.client = client or anthropic.Anthropic()
        self.max_tokens = max_tokens

    def __call__(self, tier, system, prompt):
        request = dict(model=MODELS[tier], max_tokens=self.max_tokens, system=system,
                       messages=[{"role": "user", "content": prompt}])
        if tier in ("opus", "fable"):
            # a declined request is re-run server-side on Anthropic's recommended fallback model
            r = self.client.beta.messages.create(**request, betas=["server-side-fallback-2026-07-01"],
                                                 fallbacks="default")
        else:
            r = self.client.messages.create(**request)
        if r.stop_reason == "refusal":   # check before reading content: a refusal can arrive with none
            return Reply("", r.model, r.usage.input_tokens, r.usage.output_tokens, refused=True)
        text = "".join(b.text for b in r.content if b.type == "text")
        return Reply(text, r.model, r.usage.input_tokens, r.usage.output_tokens)

4Run it on a file, or on the live alert stream.

python examples/worked/routing/run_router.py examples/worked/routing/events.log

tail -F /var/log/alerts.log | python examples/worked/routing/run_router.py

Where to be careful. A misroute costs more than it saves if a Haiku answer to a real incident gets believed. The check read is what catches that, and it has to be measured too. Re-run a sample of each tier's traffic one tier up and track how often the check read disagrees. And log every routing decision with its distribution, so "why did this go to Haiku?" has an answer.

Caveats

• The answers are only as good as DiffusionGemma used this way. OpenJev's own README says so. Build a small labelled set for each decision and measure it before you trust it.
• Check calibration, don't assume it. Plot predicted P against observed accuracy for each question. If 0.9 doesn't mean right about 90% of the time, fit a temperature or move the threshold. Jev claims calibration from RLCD training. The open variants don't have RLCD.
• OpenJev relies on an unmerged vLLM pull request (#57250) pinned to a fork. Expect some churn until it lands upstream.
• Prompt injection still applies. The state is untrusted input. A typed answer limits the blast radius, but a crafted log line can still push a noul the wrong way. Keep a human or a second check on anything irreversible.
• The vendor benchmarks are vendor benchmarks. The 40–200× latency and 444× cost figures are TypeSafe's, on TypeSafe's workflows.

Further reading

• Introducing System One Models and Jev (TypeSafe)
• Jev documentation: primitives, confidence, patterns
• TypeSafe workflow evals
• OpenJev upstream and my fork
• vLLM PR #57250: structured reads for diffusion models
• DiffusionGemma 26B-A4B model card and the MLX 4-bit build
• Codiv: hosted OpenJev with a free tier
• Jev alternatives: a community list of open variants
• Thinking, Fast and Slow, the origin of System 1 and System 2
• LangGraph and Google ADK for graph-orchestrated agents
• Anthropic API pricing