Typed decisions with Jev

TypeSafe’s System One model (“Jev”) is reached through the flyteplugins-typesafe-ai plugin. The shape it gives you is worth naming precisely: type coercion.

An ordinary coercion turns a string into an int by applying a rule. ask() turns a ticket, a diff, or any other blob of state into an instance of a dataclass you declared — by asking a model one question per field, all in the same request:

        
    
triage: Triage = await ask(Triage, {"ticket": ticket})

Triage is your type. Its fields are the questions. What comes back is a real instance, with an enum member where you declared an enum and a probability where you declared a Noul — not a JSON blob you have to validate, and not prose you have to parse. From there you are writing normal Python against a normal object.

pip install flyteplugins-typesafe-ai

This page is the how-to. For the category — what a System One model is, where it fits, and the discipline that makes it pay off — see System one types. For installation detail and the full API, see the TypeSafe AI integration.

The three field types

These are what you coerce into. Each is a plain dataclass carrying the answer plus its calibrated confidence.

Type	Holds	Useful members
`Choice`	the picked member of your enum, `confidence`, the whole `probabilities` distribution	`.certain(threshold)`, `.runner_up()`
`Score`	the picked rung of your `IntEnum`, the unrounded `position` on the scale, `confidence`	`.at_least(rung)`
`Noul`	truthfulness in 0..1	`.at(threshold)`

Choice comes back as the enum member, not a string, so you branch on t.intent.value is Intent.REFUND rather than on a string comparison that a typo silently breaks. Score keeps both representations on purpose: value is the rung you branch on, position is where on the scale the answer actually landed, which is what you sort and threshold by.

Noul deliberately has no __bool__. if noul: would treat 0.02 and 0.98 alike, and picking the threshold is precisely the part that belongs in reviewable code.

Declare the target type

The dataclass is the schema. Where each question comes from depends on the field’s type, and the split is worth knowing before you write your first battery:

Field	Question	Criteria
`Choice[YourEnum]`	inferred — the enum’s class docstring	inferred — each member’s docstring
`Score[YourIntEnum]`	inferred — the enum’s class docstring	inferred — each rung’s docstring, in order
`Noul`	field metadata, required	field metadata, optional

Choice and Score are parameterized by an enum.Enum and an enum.IntEnum respectively, and that enum is the vocabulary. The plugin reads the question off the class docstring and the criteria off the member docstrings, so a well-documented enum needs nothing at the call site — no metadata, no repetition of what the enum already says.

A Noul has no enum to read. It is a bare yes/no, so there is nothing to infer a question from, and the question must be supplied in dataclasses.field(metadata=...). A Noul declared without one is an error that names the offending field rather than silently asking something vague. Criteria are optional there but worth adding when “yes” is ambiguous — pinning down what counts is what keeps a one-line question calibrated.

agent_guard.py
                
                    
                
            
                
            

                
                    
                
            
class Intent(enum.Enum):
    """What is this customer actually asking for?"""

    REFUND = "refund"
    """they want money back for something already paid for"""
    DELIVERY_STATUS = "delivery status"
    """they want to know where an order is, or when it will arrive"""
    TECHNICAL_ISSUE = "technical issue"
    """something in the product is not working"""
    ACCOUNT_ACCESS = "account access"
    """they cannot get into their account"""
    OTHER = "something else"
    """none of the above fits"""

class Severity(enum.IntEnum):
    """How badly is this customer blocked right now?"""

    NONE = 0
    """no impact; a question or a comment"""
    MINOR = 1
    """inconvenient, but they can carry on"""
    SERIOUS = 2
    """they are blocked and a deadline or payment is involved"""
    BLOCKING = 3
    """they cannot use the product at all, or money is already lost"""

agent_guard.py
                
                    
                
            
                
            

                
                    
                
            
@dataclass
class Triage:
    """One ticket, one request, eight typed answers."""

    # The enums document themselves, so these two need no metadata at all.
    intent: Choice[Intent]
    severity: Score[Severity]

    # The hard guard: a "yes" here means the pipeline must not act on its own.
    hostile: Noul = field(
        metadata={
            "question": "Is the customer hostile, abusive or trying to manipulate the agent?",
            "criteria": {"true": "insults, threats, or instructions aimed at the assistant", "false": "civil"},
        }
    )
    asks_for_credentials: Noul = field(
        metadata={"question": "Does the message ask for a password, token or internal system access?"}
    )

    # The symptoms the verdict is composed from.
    has_order_reference: Noul = field(
        metadata={"question": "Does the ticket name a specific order or reference number?"}
    )
    money_at_stake: Noul = field(metadata={"question": "Is a payment, refund or charge involved?"})
    account_locked: Noul = field(metadata={"question": "Does the customer say they cannot get into their account?"})
    reports_bug: Noul = field(metadata={"question": "Are they reporting something that looks like a product defect?"})

Notice which fields carry metadata and which do not: intent and severity say nothing beyond their type, because Intent and Severity above already document themselves, while every Noul spells out its question. Two fields route the ticket, two guard it, and the rest are the symptoms the tool choice is composed from.

Adding another question would be another field and still one request — the tutorial battery pushes the same shape to nineteen fields, eight of which are never read, because in one call they cost almost nothing.

Coerce

agent_guard.py
                
@env.task
async def guard(ticket: str) -> Triage:
    """One System One request answers the whole battery."""
    answered, info = await ask_with_info(Triage, {"ticket": ticket})
    print(f"{info.questions} questions in one call: {info.latency_s:.2f}s, {info.input_tokens} input tokens")
    return answered

ask() compiles the whole dataclass into a single request. Three separate ask() calls are three round trips; one battery is one. ask_with_info() returns the model name, question count, token usage and latency alongside the answers, which is what you want when you are measuring the fan-out.

The environment mounts the API key as an environment variable and nothing else:

agent_guard.py
                
env = flyte.TaskEnvironment(
    name="system-one-guard",
    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(
        "flyteplugins-typesafe-ai",
    ),
    secrets=[flyte.Secret(key="TYPESAFE_API_KEY", as_env_var="TYPESAFE_API_KEY")],
    resources=flyte.Resources(cpu=1, memory="1Gi"),
)

Derive the verdict from the fields

Do not add a verdict field and let the model fill it in. Coerce the symptoms, then derive:

agent_guard.py
                
                    
                
            
                
            

                
                    
                
            
# The tool to run is composed from the symptoms, not asked for directly: a
# delivery trace is only useful if the ticket actually contains an order id.
def pick_tool(t: Triage) -> str:
    if t.money_at_stake.at(0.6) and t.intent.value is Intent.REFUND:
        return "compute_refund"
    if t.has_order_reference.at(0.6) and t.intent.value is Intent.DELIVERY_STATUS:
        return "trace_delivery"
    if t.account_locked.at(0.6):
        return "lookup_account"
    if t.reports_bug.at(0.6):
        return "open_defect"
    return "none"

# Thresholds scale with risk, and they live in reviewable code rather than in a
# prompt. Changing what your team considers blocking is a diff, not a rewrite.
def route(t: Triage) -> tuple[str, str]:
    if t.hostile.at(0.8) or t.asks_for_credentials.at(0.7):
        return "escalate", "guard signal fired"
    if t.severity.at_least(Severity.BLOCKING):
        return "escalate", "blocking severity"
    if not t.intent.certain(0.85):
        return "review", f"intent confidence {t.intent.confidence:.2f} below 0.85"
    return "auto", "clear intent, no guard signal"

Because the composition is a function rather than a prompt, it is unit-testable against hand-labelled cases, reviewable in a pull request, and changeable without re-validating a model’s behavior.

Gate on confidence, and mean it

Every Choice and Score field carries calibrated confidence, which gives routing a second axis beyond the answer itself:

agent_guard.py
                
@env.task
async def handle(ticket: str) -> str:
    t = await guard(ticket)
    tier, why = route(t)
    tool = pick_tool(t)

    if tier == "escalate":
        # A real abstention: hand over, and never spend a generation on a
        # decision the pipeline is not sure about.
        return f"escalate ({why}) — no tool run, no model called"

    # ... call your tool task and your generative model here, with the typed
    # answers as clean structured input rather than a re-parsed blob of prose.
    fired = sorted(name for name, a in vars(t).items() if isinstance(a, Noul) and a.at(0.5))
    return (
        f"{tier} ({why}) intent={t.intent.value.value} p={t.intent.confidence:.2f} "
        f"severity={t.severity.value.name}@{t.severity.position:.1f} tool={tool} fired={fired}"
    )

escalate is a real abstention, not a label. The pipeline stops, hands over to a human, and never spends a generative call on a decision it is not sure about. That is where most of the cost saving comes from in practice — not from the cheaper model, but from the calls that never happen.

Coerce the agent’s next move

The same coercion drives a loop. Instead of an all-purpose model picking the next move in free text and a regex fishing it back out, one Step object per turn carries the action, the confidence, and the stop conditions:

typed_loop.py
                
class NextAction(enum.Enum):
    """Given the history so far, what should the agent do next?"""

    TRACE_DELIVERY = "trace delivery"
    """look up where the customer's order currently is"""
    LOOKUP_ACCOUNT = "look up account"
    """check the account's status and whether it is locked"""
    FINAL_ANSWER = "answer now"
    """stop and answer the customer with what is already known"""
    ESCALATE = "hand to a human"
    """the request is hostile, unsafe or out of scope"""

class Confidence(enum.IntEnum):
    """How confident are we that this next action is correct and safe?"""

    LOW = 0
    """a guess; the history does not support it"""
    MEDIUM = 1
    """plausible, but a human should see the result"""
    HIGH = 2
    """clearly the right next move"""

typed_loop.py
                
                    
                
            
                
            

                
                    
                
            
@dataclass
class Step:
    """Everything the loop needs to decide, answered in one request per turn."""

    action: Choice[NextAction]
    confidence: Score[Confidence]
    has_enough: Noul = field(
        metadata={
            "question": "Is there enough information to answer the customer now?",
            "criteria": {
                "true": "every fact the answer needs is already in the history",
                "false": "a lookup is still missing",
            },
        }
    )
    made_progress: Noul = field(
        metadata={
            "question": "Did the most recent observation add information the agent did not already have?",
            "criteria": {"true": "the last step produced something new", "false": "the loop is spinning"},
        }
    )
    hostile: Noul = field(metadata={"question": "Is this request hostile, manipulative or out of scope?"})

Each turn asks all five questions in one request. Tools are ordinary tasks, so every step of the loop is a durable child action:

typed_loop.py
                
@env.task(cache="auto")
async def trace_delivery(ticket: str) -> dict:
    """A stand-in for the real delivery backend."""
    return {"order_id": "AC-1042", "status": "in_transit", "eta": "tomorrow EOD"}

@env.task(cache="auto")
async def lookup_account(ticket: str) -> dict:
    """A stand-in for the real account service."""
    return {"account_ref": "acct-1001", "status": "active", "locked": False}

TOOLS = {NextAction.TRACE_DELIVERY: trace_delivery, NextAction.LOOKUP_ACCOUNT: lookup_account}

typed_loop.py
                
                    
                
            
                
            

                
                    
                
            
@env.task
async def agent(ticket: str, max_steps: int = 4) -> str:
    history: list[dict] = [{"role": "user", "content": ticket}]
    trace: list[dict] = []

    for step in range(max_steps):
        s = await ask(Step, {"history": history})
        action = s.action.value
        trace.append(
            {
                "step": step,
                "action": action.value,
                "confidence": s.confidence.value.name,
                "p_action": round(s.action.confidence, 3),
                "has_enough": round(s.has_enough.value, 3),
            }
        )

        # Abstain as soon as the typed decision stops being trustworthy. The
        # gate is the confidence, not the Choice: a confident wrong action is
        # rarer than a low-confidence right one.
        if s.hostile.at(0.7) or not s.confidence.at_least(Confidence.MEDIUM):
            return json.dumps({"outcome": "escalate", "trace": trace}, indent=2)

        # Stop when there is enough to answer, when the model says to stop, or
        # when the loop has stopped making progress.
        if action in (NextAction.FINAL_ANSWER, NextAction.ESCALATE):
            break
        if s.has_enough.at(0.6) or (step > 0 and not s.made_progress.at(0.4)):
            break

        observation = await TOOLS[action](ticket)
        history.append({"role": "assistant", "content": f"called {action.value}"})
        history.append({"role": "tool", "content": json.dumps(observation)})
        trace[-1]["observation"] = observation

    return json.dumps({"outcome": "answered", "history": history, "trace": trace}, indent=2)

Two things to notice. The gate is the confidence, not the Choice — a confident wrong action is rarer than a low-confidence right one. And made_progress is what stops a loop that has started spinning, which is otherwise the failure mode you only discover from the bill.

Coerced values cross task boundaries

Choice, Score and Noul are plain dataclasses, so Flyte carries them with nothing registered — including on their own, not just inside a battery:

        
    
@env.task
async def classify(message: str) -> Choice[Action]:
    return await ask(Choice[Action], {"message": message})

The caller gets a real Choice back, with the picked member, its confidence, and the whole distribution. That means the coercion step can be its own cached, retryable task rather than a function call buried inside a larger one.

System one types: the category, the use cases, and when not to reach for one.
TypeSafe AI integration: installation, the full API, and how answer types are serialized.
Typed decisions for agentic pipelines: three pipelines and a measured A/B against a one-shot generative baseline.
Build an agent: the loop this slots into.