pipelines.decider

Typed, calibrated decisions via the shared /v1/systemone (System One / Jev) API.

source

Decider

def Decider(
    base_url:Optional[str]=None, # Base URL of a decision server (e.g., "http://localhost:11434" for Ollama, "https://api.typesafe.ai" for Jev, "https://api.openai.com" for OpenAI). Falls back to the `TYPESAFE_BASE_URL` environment variable.
    model:Optional[str]=None, # Model name/route (e.g., "nimble", "jev-latest", "gpt-6-luna"). Falls back to the `TYPESAFE_DEFAULT_MODEL` environment variable.
    api_key:Optional[str]=None, # Bearer API key. Falls back to the `TYPESAFE_API_KEY` environment variable. Use "ollama" (or any placeholder) for a local Ollama server.
    protocol:str='systemone', # Wire protocol: "systemone" (legacy TypeSafe/Ollama `/v1/systemone`, the default) or "decisions" (OpenAI `/v1/decisions`). Falls back to the `DECIDER_PROTOCOL` environment variable.
    timeout:int=120, # Per-request timeout in seconds.
    max_retries:int=3, # Number of retries for retryable statuses (429/5xx) using exponential backoff with jitter.
    verify:Union[bool, str, NoneType]=None, # TLS verification passed to `requests`. A path to a custom CA bundle (e.g., a corporate root CA behind a TLS-intercepting firewall), or `True`/`False`. When `None` (default), verification is left to `requests` (which honors the `REQUESTS_CA_BUNDLE`/`CURL_CA_BUNDLE` env vars, then `certifi`). Also falls back to the `TYPESAFE_CA_BUNDLE` environment variable.
    cert:Union[str, Tuple[str, str], NoneType]=None, # Optional client-side certificate for mutual TLS, passed to `requests` as `cert`: either a path to a combined cert+key PEM, or a `(cert_path, key_path)` tuple. Required by some corporate proxies.
    **kwargs
):

Decider sends application state* plus one or more typed questions to a decision model and returns typed, calibrated answers*.

It speaks two interchangeable HTTP contracts, selected via protocol:

  • "systemone" (default): the legacy TypeSafe/Ollama /v1/systemone format. Use this for Ollama (which is /v1/systemone-only) and existing TypeSafe/Jev deployments.
  • "decisions": OpenAI’s /v1/decisions format (e.g., gpt-6-luna).

Requests are built for the chosen protocol and responses are normalized to the same SystemOneResponse/_Answer shape, so downstream code (.noul, .choice, .score, .probabilities) is identical either way.

source

SystemOneResponse

def SystemOneResponse(
    data:Dict[str, Any], protocol:str='systemone'
):

Parsed response from a decision endpoint (/v1/systemone or /v1/decisions).

Attributes:

  • model: The model that produced the answers.
  • answers: A dict mapping each question key/name to an _Answer.
  • usage: Token-usage dict (the billing basis).
  • raw: The raw decoded JSON response.

The /v1/decisions response returns answers as a list of objects each carrying a name; this is normalized to the same {name: _Answer} map used by System One so downstream code is identical across protocols.

source

SystemOneError

def SystemOneError(
    message:str, status_code:Optional[int]=None
):

Raised when the System One endpoint returns an error response.

source

Decider.ask

def ask(
    state:Union[str, dict, list], # The content/context to evaluate (string, or object/array serialized per the API).
    *,
    nouls:Optional[Sequence[str]]=None, # Yes/no question instructions (each a string). Answered as a probability in [0, 1].
    choices:Optional[Sequence[Tuple[str, Sequence[str]]]]=None, # `(question, [option, ...])` pairs (pick one option).
    scores:Optional[Sequence[Tuple[str, Sequence[str]]]]=None, # `(question, [level, ...])` pairs; levels in ascending order (2-10 levels).
    questions:Optional[Dict[str, Dict[str, Any]]]=None, # Advanced: a raw `questions` map (per the API spec). Mutually exclusive with `nouls`/`choices`/`scores`.
)->__main__.SystemOneResponse:

Ask one or more typed questions about state and return a SystemOneResponse.

Provide questions either via the convenience arguments (nouls/choices/scores, auto-keyed noul_0, choice_0, …) or via a raw questions map for full control over keys and criteria descriptions.

Returns:

  • A SystemOneResponse whose .answers[key] exposes .choice, .noul, .score, .probabilities, and .confidence.

source

Decider.noul

def noul(
    state:Union[str, dict, list], question:str
)->float:

Ask a single yes/no question and return its probability (0.0 - 1.0).

Example:

d.noul("I was charged twice. Please refund.",
       "Does the customer ask for a refund?")
# 0.98

source

Decider.choice

def choice(
    state:Union[str, dict, list], question:str, options:Sequence[str]
)->Dict[str, float]:

Ask a single multiple-choice question and return a dict of {option: probability}.

Example:

d.choice("I was charged twice. Please refund.",
         "Which team should handle this?",
         ["billing", "technical", "other"])
# {'billing': 0.98, 'technical': 0.01, 'other': 0.01}

source

Decider.score

def score(
    state:Union[str, dict, list], question:str, levels:Sequence[str]
)->float:

Ask a single ordinal-score question (levels ascending) and return the continuous score.

Example:

d.score("I was charged twice. Please refund.",
        "How urgent is this ticket?",
        ["Routine", "Soon", "Urgent"])
# 1.40  (between "Soon" (1) and "Urgent" (2))

source

Decider.classify

def classify(
    X:Union[str, List[str]], # A single text or a list of texts.
    labels:Sequence[str], # Candidate labels (the choice options).
    question:str='Which category best describes the following content?', # The classification prompt.
    return_proba:bool=False, # If True, return `{label: probability}` dict(s) instead of the single top label.
):

Zero-shot classification: map each text in X to one of labels.

A thin convenience wrapper over a single choice question. A list is returned when X is a list; a scalar when X is a string.

Example:

d.classify("I was charged twice. Please refund.",
           ["billing", "technical", "other"])
# 'billing'

d.classify(["I was charged twice.", "The app crashes on login."],
           ["billing", "technical"])
# ['billing', 'technical']

d.classify("I was charged twice.", ["billing", "technical"],
           return_proba=True)
# {'billing': 0.95, 'technical': 0.05}