TypeSafe AI

Run TypeSafe’s System One model (Jev) inside durable Flyte tasks.

Jev answers typed questions in parallel instead of generating text, and returns calibrated confidence with every answer. This plugin gives those answers a shape that crosses a Flyte task boundary — Choice, Score and Noul — and a way to ask a whole battery of them in a single request.

        
    
import enum
from dataclasses import dataclass, field

import flyte
from flyteplugins.typesafe_ai import Choice, Noul, Score, ask

env = flyte.TaskEnvironment(
    "triage",
    secrets=[flyte.Secret(key="TYPESAFE_API_KEY", as_env_var="TYPESAFE_API_KEY")],
)


class Intent(enum.Enum):
    '''Which intent best fits the ticket?'''

    REFUND = "refund"
    '''they want money back'''
    DELIVERY = "delivery"
    '''they are asking where their order is'''


class Severity(enum.IntEnum):
    '''How badly is this customer affected?'''

    NONE = 0
    '''no impact; a question or a comment'''
    MINOR = 1
    '''inconvenient, but they can carry on'''
    SERIOUS = 2
    '''they are blocked'''


@dataclass
class Triage:
    # The enums above document themselves, so these fields need no metadata at all.
    intent: Choice[Intent]
    severity: Score[Severity]
    # A Noul has no vocabulary to document itself with, so it needs both.
    hostile: Noul = field(
        metadata={"question": "Is the customer hostile?", "criteria": {"true": "insults or threats", "false": "civil"}}
    )


@env.task
async def triage(ticket: str) -> Triage:
    return await ask(Triage, {"ticket": ticket})

Class	Description
`CallInfo`	What one `system_one` call cost, plus the calibration a plain field dropped.
`Choice`	One pick from a fixed vocabulary, carrying the calibration it came with.
`Noul`	Truthfulness in 0..1.
`Score`	A position on a rubric.

Exception	Description
`BatteryError`	The questions cannot be compiled.
`MissingAPIKey`	Raised at the point of use, in the task that actually needs the key.

Method	Description
`ask()`	Answer a battery, a single question, or a mapping of them – in one call.
`ask_with_info()`	Answer everything in one call, and report what the call cost.
`client()`	An `AsyncTypeSafeClient`, or a message that says exactly what to do.
`compile_questions()`	Build the SDK’s question objects from any accepted form.
`thresholds()`	Where each `bool` field cuts its 0..1 answer.

Property	Type	Description
`API_KEY_ENV`	`str`
`CRITERIA_KEY`	`str`
`QUESTION_KEY`	`str`
`THRESHOLD_KEY`	`str`

Methods

ask()

        
    
def ask(
    askable: Askable,
    state: Any,
    model: Optional[str] = None,
    client: Any = None,
    threshold: Optional[float] = None,
) -> Any

Answer a battery, a single question, or a mapping of them – in one call.

        
    
triage = await ask(Triage, {"ticket": text})                  # -> Triage
intent = await ask(Choice[Intent], {"ticket": text})          # -> Choice[Intent]
both = await ask({"intent": Choice[Intent], "hot": Noul}, s)  # -> dict

Parameter	Type	Description
`askable`	`Askable`
`state`	`Any`
`model`	`Optional[str]`
`client`	`Any`
`threshold`	`Optional[float]`

ask_with_info()

        
    
def ask_with_info(
    askable: Askable,
    state: Any,
    model: Optional[str] = None,
    client: Any = None,
    threshold: Optional[float] = None,
) -> Tuple[Any, CallInfo]

Answer everything in one call, and report what the call cost.

threshold cuts every bool field that does not declare its own. It is checked before the request, so a battery missing one fails without spending a call.

Parameter	Type	Description
`askable`	`Askable`
`state`	`Any`
`model`	`Optional[str]`
`client`	`Any`
`threshold`	`Optional[float]`

client()

        
    
def client(
    api_key: Optional[str] = None,
    model: Optional[str] = None,
    **kwargs: Any,
)

An AsyncTypeSafeClient, or a message that says exactly what to do.

Without this, a missing key surfaces as a 401 from inside the vendor SDK, which tells you nothing about Flyte secrets.

Parameter	Type	Description
`api_key`	`Optional[str]`
`model`	`Optional[str]`
`**kwargs`	`Any`

compile_questions()

        
    
def compile_questions(
    askable: Askable,
) -> Dict[str, Any]

Build the SDK’s question objects from any accepted form.

Parameter	Type	Description
`askable`	`Askable`

thresholds()

        
    
def thresholds(
    specs: list[_Spec],
    default: Optional[float] = None,
) -> Dict[str, float]

Where each bool field cuts its 0..1 answer.

A bool has to be cut somewhere, and that decision belongs in the caller’s code rather than in this plugin, so there is no implicit default: the field says it, the call says it, or this raises. Requiring it also makes a misspelled metadata key loud instead of silently meaning 0.5.

Parameter	Type	Description
`specs`	`list[_Spec]`
`default`	`Optional[float]`

TypeSafe AI

Directory

Classes

Errors

Methods

Variables

Methods

ask()

ask_with_info()

client()

compile_questions()

thresholds()