In this tutorial, we work with Jev, TypeSafe AI’s first System One model, which does not generate text at all: we send it a piece of program state and a set of typed questions, and it returns choices, scores, and yes/no probabilities that our code can branch on directly. We install the official Python SDK, make a first call that uses all three question primitives at once, and look at how the shape of the state changes what the model can know. We then recompute the published confidence statistic from the returned probabilities, measure what batching ten questions into one call buys over ten separate calls, and build the patterns the API is designed for: confidence-gated routing, composite scoring with the weights kept in code, typed function calling, and counting done the way the model can actually do it. We close with the production shape: Pydantic response models, an async client fanned out with asyncio, retry policies, typed errors, and a running ledger that prices the whole notebook.
import os
import sys
import json
import time
import asyncio
import traceback
import subprocess
from getpass import getpass
RESULTS = {}
LEDGER = {"calls": 0, "input_tokens": 0, "output_tokens": 0}
USD_PER_MILLION_INPUT_TOKENS = 0.042 # Jev list price; output tokens are free
def banner(title):
print("\n" + "=" * 78)
print(title)
print("=" * 78)
def section(name):
def wrap(fn):
def run(*a, **kw):
banner(name)
try:
out = fn(*a, **kw)
RESULTS[name] = out if isinstance(out, str) else "ok"
return out
except Exception as e:
RESULTS[name] = f"SKIPPED / FAILED -> {type(e).__name__}: {e}"
print(f"\n[!] {name} did not complete: {type(e).__name__}: {e}")
traceback.print_exc(limit=3)
return None
return run
return wrap
banner("0. Install the SDK, load the API key, list the models")
subprocess.run([sys.executable, "-m", "pip", "install", "-q", "typesafe-sdk==0.7.0"], check=True)
import typesafe_sdk
from typesafe_sdk import Choice, Noul, Score, TypeSafeClient
def load_api_key():
key = os.environ.get("TYPESAFE_API_KEY", "").strip()
if not key:
try:
from google.colab import userdata # Colab: key stored under the Secrets tab
key = (userdata.get("TYPESAFE_API_KEY") or "").strip()
except Exception:
key = ""
return key or getpass("TypeSafe API key (console.typesafe.ai/keys): ").strip()
os.environ["TYPESAFE_API_KEY"] = load_api_key()
client = TypeSafeClient() # reads TYPESAFE_API_KEY, defaults to jev-latest
print(f" typesafe-sdk {typesafe_sdk.__version__} | Python {sys.version.split()[0]}")
print(" models available to this key:")
for m in client.models.list().models:
print(f" {m.name:<14s} released {m.release_date} {m.description}")
def ask(state, questions, **kw):
"""One System One call, timed, with its tokens added to the running ledger."""
t0 = time.perf_counter()
response = client.system_one(state, questions, **kw)
ms = (time.perf_counter() - t0) * 1e3
LEDGER["calls"] += 1
LEDGER["input_tokens"] += response.usage.input_tokens or 0
LEDGER["output_tokens"] += response.usage.output_tokens or 0
return response, ms
We install typesafe-sdk, pinned to the version this notebook was written against, and load the API key from the environment, from Colab’s Secrets tab, or from a hidden prompt, so it never appears in the notebook. TypeSafeClient reads TYPESAFE_API_KEY on its own and defaults to the jev-latest alias; listing the models shows which names and pinned versions the key can use. The small ask helper wraps system_one so that every call in the rest of the notebook is timed and its token usage lands in a ledger we total at the end.
TICKET = {
"ticket": {
"subject": "Duplicate charge",
"messages": [
{"from": "customer", "text": "I was charged twice for order A-104. This is the second time "
"this year. Please refund the duplicate today."},
{"from": "support", "text": "We are checking the charges."},
],
},
"order": {"id": "A-104", "charges": [{"amount_usd": 49, "status": "captured"},
{"amount_usd": 49, "status": "captured"}]},
"refund_policy": "Duplicate charges are eligible for a full refund within 30 days.",
}
@section("1. Three primitives, one call: Choice, Score, Noul")
def three_primitives():
response, ms = ask(TICKET, {
"department": Choice(
instructions="Which team should handle this ticket",
criteria={"billing": "Payment, refund or subscription issues",
"technical": "Bugs, outages or integration problems",
"sales": "Pricing, plans or account upgrades"},
),
"frustration": Score(
instructions="How frustrated the customer appears in `ticket.messages[0].text`",
criteria=["Calm, just stating facts", "Frustrated but civil", "Very angry, strong language"],
),
"refund_requested": Noul(instructions="The customer is explicitly asking for a refund"),
"policy_supports": Noul(instructions="The stated `refund_policy` covers this situation"),
})
dept = response.choices["department"]
print(f" department -> {dept.choice!r} confidence {dept.confidence:.3f}")
print(f" probabilities {({k: round(v, 3) for k, v in dept.probabilities.items()})}")
fr = response.scores["frustration"]
print(f" frustration -> score {fr.score:.3f} on 0..{len(fr.legend) - 1} confidence {fr.confidence:.3f}")
for level, text in fr.legend.items():
print(f" {level}: p={fr.probabilities[level]:.3f} {text}")
print(f" refund_requested -> noul {response.nouls['refund_requested'].noul:.3f}")
print(f" policy_supports -> noul {response.nouls['policy_supports'].noul:.3f}")
print(f"\n answered by {response.model} in {ms:.0f} ms "
f"input tokens {response.usage.input_tokens}, output tokens {response.usage.output_tokens}")
return f"{dept.choice}, frustration {fr.score:.2f}, refund {response.nouls['refund_requested'].noul:.2f}"
three_primitives()
A System One request has two parts: state, which is any text, JSON object or array describing the situation, and a dictionary of named questions. Choice selects one label from the criteria we define and returns a probability for every label; Score places the state on an ordered rubric and returns the probability-weighted level, so it can land between two levels; Noul returns a single probability that a statement is true. The question names are ours and never reach the model, which is why the instructions carry the full meaning and can point at nested fields with backticked paths. All four questions are evaluated in one request, in parallel and in isolation from one another, and the response reports the pinned model version that answered and the tokens it billed.
@section("2. State is program state: the same question over a string and over named fields")
def state_shapes():
question = {"eligible": Noul(
instructions="The customer is eligible for a refund under the company's written policy",
criteria={"true": "A policy is present and it covers the customer's situation",
"false": "No policy is given, or the policy does not cover the situation"},
)}
bare = "I was charged twice for order A-104. Please refund the duplicate."
as_list = [m["text"] for m in TICKET["ticket"]["messages"]]
shapes = [("string: the message only", bare),
("array : the conversation", as_list),
("object: ticket + order + policy", TICKET)]
print(f" {'state shape':<34s} {'noul':>6s} input tokens ms")
seen = {}
for label, state in shapes:
response, ms = ask(state, question)
seen[label] = response.nouls["eligible"].noul
print(f" {label:<34s} {seen[label]:6.3f} {response.usage.input_tokens:12d} {ms:5.0f}")
print("\n Only the object carries the policy and the two captured charges; the question")
print(" is identical in all three calls, so any movement comes from the state.")
return "noul by state shape: " + ", ".join(f"{v:.2f}" for v in seen.values())
state_shapes()
State is the only thing the model knows, so we ask one question, whether the customer is eligible for a refund under the company’s written policy, over three shapes of state. A bare string contains the complaint and nothing else; an array adds the conversation; the JSON object adds the order with its two captured charges and the refund policy itself. The question never changes, so whatever difference appears in the returned probability is attributable to the state, and the token column shows what the extra context costs. Named fields are the documented recommendation whenever the context has several parts, because the instructions can then refer to them by name.
def confidence_from(probabilities):
"""TypeSafe's published statistic: (count x peak - 1) / (count - 1)."""
p = list(probabilities.values())
return (len(p) * max(p) - 1) / (len(p) - 1)
@section("3. Confidence is a statistic of the distribution, and you can recompute it")
def confidence_math():
tone = Choice(instructions="What is the tone of the message",
criteria={"angry": "Upset or hostile", "calm": "Neutral or polite", "excited": "Enthusiastic or eager"})
urgency = Score(instructions="How soon this needs attention",
criteria=["Can wait", "Needs attention this week", "Needs attention today"])
messages = {
"clear ": "This is the third outage this week and nobody answers. Fix it NOW or I cancel today.",
"ambiguous": "Well. That was certainly an experience. Let me know when you get a chance.",
}
print(f" {'message':<10s} {'choice':<8s} {'API conf':>8s} {'recomputed':>11s} "
f"{'score':>6s} {'sum(level*p)':>13s} {'API conf':>9s}")
worst = 1.0
for label, text in messages.items():
response, _ = ask(text, {"tone": tone, "urgency": urgency})
t, u = response.choices["tone"], response.scores["urgency"]
expected = sum(level * p for level, p in u.probabilities.items())
print(f" {label:<10s} {t.choice:<8s} {t.confidence:8.3f} {confidence_from(t.probabilities):11.3f} "
f"{u.score:6.3f} {expected:13.3f} {u.confidence:9.3f}")
worst = min(worst, t.confidence)
print("\n A Noul has no confidence field: its value already is the probability of yes,")
print(" so 0.5 means undecided, not medium.")
return f"lowest tone confidence {worst:.2f}"
confidence_math()
TypeSafe documents confidence as a statistic computed from the distribution that the answer already contains: the number of options times the peak probability, minus one, divided by the number of options minus one. We recompute it from a Choice’s probabilities and compare it with the confidence field, and we recompute the Score as the sum of each level times its probability. Running a blunt message and a deliberately vague one through the same two questions shows how the distribution, and therefore the confidence, responds to ambiguity. A Noul carries no confidence field at all, since its value already is the probability of yes, and a value near 0.5 means undecided rather than moderate.
POSTMORTEM = """Incident 2291 - checkout latency, 14 March. At 09:12 UTC the payments gateway began timing out
for roughly 18 percent of checkout requests in the EU region. The on-call engineer was paged at 09:15 and
acknowledged at 09:21. Initial suspicion fell on the new fraud-scoring service deployed the previous evening,
and it was rolled back at 09:40 with no improvement. At 10:05 the database team found that a connection pool
limit had been lowered from 400 to 40 by an automated configuration sync, which had silently overwritten a
manual override. The limit was restored at 10:11 and error rates returned to baseline by 10:19. Customer
impact: 3,420 failed checkouts and an estimated 61,000 USD in delayed revenue; no data was lost and no
customer data was exposed. Customers were not notified during the incident; the status page was updated at
10:30, after recovery. Follow-ups: alert on pool saturation, require review for configuration-sync overrides,
and add the status page update to the first fifteen minutes of the on-call checklist."""
FANOUT = {
"root_cause": Choice(instructions="What was the root cause of the incident",
criteria={"bad_deploy": "A faulty code or service deployment",
"config_change": "An incorrect configuration value",
"capacity": "Organic traffic exceeded provisioned capacity",
"third_party": "A failure at an external vendor",
"unknown": "The text does not establish a cause"}),
"detected_by": Choice(instructions="How the incident was first detected",
criteria={"alerting": "Automated monitoring or paging", "customer": "Customer reports",
"employee": "An employee noticed by chance", "unclear": "Not stated"}),
"severity": Score(instructions="Severity of customer impact",
criteria=["No customer-visible impact", "Minor degradation for a few customers",
"A core flow failed for a meaningful share of customers",
"Full outage of a core flow for most customers"]),
"comms_quality": Score(instructions="Quality of customer communication during the incident",
criteria=["Customers were informed promptly while it was happening",
"Customers were informed, but late",
"Customers were only informed after recovery, or never"]),
"data_exposed": Noul(instructions="Customer data was exposed or leaked"),
"rollback_helped": Noul(instructions="Rolling back the fraud-scoring service resolved the incident"),
"human_error": Noul(instructions="A person making a manual mistake directly caused the incident"),
"has_followups": Noul(instructions="The text lists concrete follow-up actions"),
"revenue_lost": Noul(instructions="Revenue was permanently lost, as opposed to delayed"),
"eu_only": Noul(instructions="The impact was limited to the EU region"),
}
def value_of(answer):
for field in ("choice", "score", "noul"): # a score of 0.0 is a real value, not a miss
if hasattr(answer, field):
return getattr(answer, field)
@section("4. Speculative fan-out: ten questions in one call versus ten calls")
def fan_out():
batched, batched_ms = ask({"postmortem": POSTMORTEM}, FANOUT)
batched_tokens = batched.usage.input_tokens
seq_ms, seq_tokens, agree = 0.0, 0, 0
print(f" {'question':<16s} {'one call':>10s} {'own call':>10s}")
for name, q in FANOUT.items():
single, ms = ask({"postmortem": POSTMORTEM}, {name: q})
seq_ms, seq_tokens = seq_ms + ms, seq_tokens + single.usage.input_tokens
a, b = value_of(batched.answers[name]), value_of(single.answers[name])
same = a == b if isinstance(a, str) else abs(a - b) < 0.05
agree += same
fmt = (lambda v: f"{v:>10s}") if isinstance(a, str) else (lambda v: f"{v:10.3f}")
print(f" {name:<16s} {fmt(a)} {fmt(b)} {'same' if same el