cloudraker-milliseconds

decision-machine-1 decides about text. It answers a statement yes or no, picks a label, walks a label tree, rates the text on a scale, quotes an answer with its offsets, fills a JSON Schema, finds entities, and checks a value. It never generates prose. Each call takes milliseconds.

This SDK adds the typed vocabulary. It keeps every wire name, unwraps three envelopes, validates your arguments before it spends a token, retries the right failures, and raises one exception tree you can catch by class.

pip install cloudraker-milliseconds
export MS_API_KEY=sk-ms-...        # get a key at https://console.milliseconds.ai

What Python can and cannot infer

Python has no mapped types and no conditional return types. It cannot read your label names out of a dict display. It can solve a TypeVar from an annotated constant. One annotation buys the whole chain.

from typing import Final, Literal, Mapping

from milliseconds import DecisionMachine

Intent = Literal["billing", "shipping", "account"]

LABELS: Final[Mapping[Intent, str]] = {
    "billing": "payments, invoices, charges and refunds",
    "shipping": "delivery, tracking and packages",
    "account": "login, passwords and profile settings",
}

dm = DecisionMachine()  # reads MS_API_KEY

r = dm.classify("I was charged twice.", LABELS)
r.label  # Intent. A match statement over it is exhaustive.
r.scores["billing"]  # ok. r.scores["refunds"] is a type error.
r.probability  # float
r.confidence  # float. 1 = one clear winner, 0 = flat.

Without the annotation you get ClassifyResult[str]. Nothing breaks. You lose only the names. Keep every label set in one constants file and the annotation lands where the constants already live.

classify_tree returns label: str. Python has no expression that reads literals out of a nested dict.

The eight capabilities

statement = dm.yes_no(
    "Fix this today.",
    "The customer expresses urgency.",
    when_true="Time pressure, ASAP, losing money",
)
statement.answer  # bool
statement.probability  # float

label = dm.classify("I was charged twice.", LABELS)

tree = dm.classify_tree(
    "I want my money back.",
    {
        "billing": {
            "description": "payments, invoices, charges, refunds and subscriptions",
            "labels": {
                "refund_request": "the customer asks for money back",
                "subscription_change": "the customer wants to upgrade or cancel a plan",
            },
        },
        "shipping": "delivery, tracking, lost or damaged parcels",
    },
)
tree.path  # winning label per level, top to bottom
tree.label  # the deepest label

mood = dm.rate(
    "I am done with this company.",
    ["Calm", "Annoyed", "Angry", "Threatening to leave"],
)
mood.score  # 0 to len(scale) - 1. Route on this.
mood.level  # the most likely index. It flips on 0.001.

who = dm.answer("Apple announced the M5 today.", "Who announced the product?")
if who.span is not None:
    start, end = who.span

people = dm.entities("Ada met Grace in Paris.", {"person": "a human name", "place": "a city"})
[e.text for e in people]

check = dm.verify("Invoice 4471, total 120.00 EUR.", "invoice_number", 4471)
check.matches  # bool
check.found  # what the text actually says

Describe every label. The label text is the instruction, and the model reads it literally. Described labels score measurably better than bare names.

Annotate a tree you keep in a constant, as you annotate a label set. A bare TAXONOMY = {...} infers a wider type, and classify_tree then refuses it.

from typing import Final

from milliseconds import Tree

TAXONOMY: Final[Tree] = {
    "billing": {
        "description": "payments, invoices, charges, refunds and subscriptions",
        "labels": {
            "refund_request": "the customer asks for money back",
            "subscription_change": "the customer wants to upgrade or cancel a plan",
        },
    },
    "shipping": "delivery, tracking, lost or damaged parcels",
}

walked = dm.classify_tree("I want my money back.", TAXONOMY)
walked.label

Batching

Pass a list of texts for a batch. The reply follows your request, never the other way.

tickets = ["I was charged twice.", "Where is my parcel?"]

many = dm.classify(tickets, LABELS)  # Results[ClassifyResult[Intent]]
many[0].label
many.usage.input_tokens  # the usage of the one call

grid = dm.yes_no(tickets, ["The text mentions a price.", "The customer is angry."])
grid[0][1].answer  # text 0, statement 1

The limits are 32 texts per call, and 20,000 characters per text. The SDK never splits a batch for you. Splitting costs money and changes failure modes, so you decide.

Extraction

Four schema shapes work: a plain dict JSON Schema, a TypedDict, a dataclass, and a pydantic v2 model. The SDK never imports pydantic. It calls model_json_schema() by duck typing.

Declare every field | None. A missing value comes back as None.

from typing import Literal, TypedDict


class Invoice(TypedDict):
    invoice_number: str | None
    total: float | None
    currency: Literal["USD", "EUR"] | None


data = dm.extract("Invoice 4471, total 120.00 EUR.", Invoice)
data["total"]  # float | None

A plain dict carries descriptions, which raise accuracy:

schema = {
    "type": "object",
    "properties": {
        "invoice_number": {"description": "the identifier printed on the invoice"},
        "total": {"type": "number", "description": "the amount due including tax"},
        "tags": {"type": "array", "items": {"type": "string"}},
    },
}
invoice = dm.extract("Invoice 4471, total 120.00 EUR.", schema)

Four degradations are real, and no Python annotation can hide them:

• a missing value is None;
• an array of objects always comes back [];
• an array of scalars comes back as a list of strings;
• an enum is not checked on the server, so a value outside your Literal can arrive.

Usage and rate limits

Every result carries the usage of the call that produced it. A single-text extract and post() are the two exceptions. Both return your own object, which has no place for the usage. Send a one-text batch to reach it: dm.extract([text], Invoice).usage.

r = dm.classify("I was charged twice.", LABELS)
r.usage.input_chars
r.usage.input_tokens  # what this call bills
r.usage.inference_ms  # model time, not wall clock
r.usage.headers["x-input-tokens"]  # every response header stays reachable

limits = r.usage.rate_limit  # RateLimit | None
if limits is not None:
    limits.remaining_requests
    limits.reset_requests  # '5m0s'

The rate-limit numbers come from the previous request at that Cloudflare colo. The server accounts after the response. Read them as a trailing gauge. Never build admission control on them.

Errors and retries

from milliseconds import (
    AuthenticationError,
    InvalidRequestError,
    MillisecondsError,
    OverloadedError,
    QuotaExceededError,
    RateLimitError,
)

try:
    r = dm.classify("I was charged twice.", LABELS)
except RateLimitError as e:
    print(e.retry_after, e.attempts)
except QuotaExceededError:
    print("add credits at https://console.milliseconds.ai")
except MillisecondsError as e:
    print(e.code, e.status, e.api_message)

The SDK retries 429 rate_limit_exceeded, 502 runner_error, 529 overloaded, and transport failures. Every capability is a pure function, so a retry is always safe. It never retries 400, 401 or 429 insufficient_quota. A timer retry cannot fix a spent quota.

Pass max_retries=0 to turn retries off. max_retries and timeout also work per call: dm.classify(text, LABELS, max_retries=5, timeout=10.0).

Some checks run before any HTTP call. They raise InvalidRequestError with code client_error and status 0. Nothing was sent, so no token was billed.

try:
    dm.classify("I was charged twice.", ["billing"])
except InvalidRequestError as e:
    print(e.code)  # client_error
    print(e.api_message)  # labels has 1 entry. classify needs 2 to 64.

What the SDK changes, and nothing else

Every other field keeps its exact wire name, snake_case included: when_true, input_chars, inference_ms, probability, scores, start, end.

dm.post() reaches the untouched body:

raw = dm.post("/v1/decision-machine-1/classify", {"text": "hi", "labels": ["a", "b"]})
raw["label"]

Async, and your own pool

import asyncio

from milliseconds import AsyncDecisionMachine


async def main() -> None:
    async with AsyncDecisionMachine() as adm:
        r = await adm.classify("I was charged twice.", LABELS)
        print(r.label)


asyncio.run(main())

Both clients share one transport module, so the retries, the error parsing and the header parsing cannot drift. Pass http_client= to bring your own httpx.Client or httpx.AsyncClient. The SDK closes only a pool it opened itself.

Gotchas

• yes-no and answer are the two endpoints with no text refinement on the server. A body with no text returns 200 and {"results": []}. The SDK rejects that body instead.
• A labels or types dict has no size limit on the server. The 2-to-64 rule binds the list form only, and the SDK checks the same way.
• classify_tree re-sends the text at every level. The per-level input_chars therefore do not sum to usage.input_chars, which counts one pass over the body.
• The SDK reports x-input-tokens. It never estimates a cost.

Links

• Docs: https://docs.milliseconds.ai
• Console and keys: https://console.milliseconds.ai
• The TypeScript SDK and the dm1 CLI: @cloudraker/milliseconds