Python SDK

jers-sdk 0.2 needs Python 3.10 or newer and nothing else; Pydantic is used only if you give it a Pydantic model. It is not on PyPI yet; until it is, call the HTTP API directly (every page shows the request).

export JERS_API_KEY=jj_live_...
export JERS_BASE_URL=https://api.getjers.com    # the default

The client

JersClient(api_key=None, base_url=None, timeout=120.0, max_retries=2, backoff_seconds=0.5): the key falls back to JERS_API_KEY, the gateway to JERS_BASE_URL and then https://api.getjers.com; timeout is seconds per call. It is safe to share between threads. AsyncJersClient takes the same keyword arguments. Use it as a context manager (with JersClient() as client:) or call client.close() when done.

Questions and answers

from jers import JersClient, Choice, Noul, Score

with JersClient() as client:
    r = client.system_one(
        state={"message": "I have asked three times now. Can I please talk to a real person?"},
        questions={
            "wants_human": Noul("Is the customer asking for a human agent?"),
            "intent": Choice("What does the customer want?", {"talk_to_human": "asks for a person", "status": "asks where something is", "other": "anything else"}),
            "frustration": Score("How frustrated is the customer?", ["calm", "annoyed", "angry"]),
        },
        model="jers-english")

r.answers["wants_human"].noul            # probability of yes, 0 to 1
r.answers["intent"].choice               # the most probable option name
r.answers["intent"].probabilities        # every option with its probability
r.answers["intent"].confidence           # 0 to 1: how concentrated the probabilities are
r.answers["frustration"].score           # probability-weighted level, 0 = the lowest
r.usage.answers, r.usage.cost, r.usage.balance

Question objects or plain dictionaries both work. The Quick start shows a measured answer.

Memory

client.remember("acme", "ACME holds an enterprise contract; refunds go to billing within one business day.")
r = client.system_one(state, questions, subject="acme", memory={"compare": True})
r.memory.lines_used, r.memory.lines_seen, r.memory.lines_dropped
r.memory.hits[0]["text"]
r.memory.without_memory["intent"].choice  # the answer without the memory
r.memory.changes["intent"]                # {"answer_changed": ..., ...}
client.forget("acme", "enterprise contract")["verified_forgotten"]
client.delete("acme")
client.memory("acme").lines_stored
client.memory("acme", lines=True).lines   # every stored line, word for word

Request options and what comes back

r = client.system_one(state, questions, model="jers-english", robust=True, windows=True,
                      derive={"over": "reading.t > limit"}, values={"limit": 75}, cache=True)
r.decision_id                              # for feedback
r.warnings                                 # [JersWarning(code, message, question)]
r.reading["intent"]                        # what the engine read of the state
r.usage.answers_billed, r.usage.state_truncated
r.answers["intent"].orders, r.answers["intent"].agreement   # with robust

robust=True asks each choice of up to 20 options in 3 option orders (2 for a two-option choice; {"orders": n} for 1 to 5) and averages them; agreement is the share of orders that picked the answer. windows=True, or {"combine": {"question_id": "max" | "mean" | "min"}}, reads a state that is cut in up to 16 overlapping windows. Robust orders, windows and the placebo pass are billed as answers; the compare pass is not. See Request options.

Pydantic models

A Literal or Enum field is a choice, a bool field a yes/no question, and an int or float field marked with Levels(...) a score. The field's description is the question; the answers come back as the model: an int score field gets the most likely level, a float the probability-weighted level, and a bool is true when the probability of yes is at least 0.5.

from typing import Annotated, Literal
from pydantic import BaseModel, Field
from jers import JersClient, Levels, Options

class Ticket(BaseModel):
    team: Annotated[Literal["billing", "support", "other"],
                    Options({"billing": "Charges and refunds", "support": "Product problems", "other": "None of these"})] = Field(
        description="Which team should handle this?")
    refund: bool = Field(description="Is the customer asking for money back?")
    urgency: Annotated[int, Levels("Can wait a week", "Should be handled today", "Blocking the customer now")] = Field(
        description="How urgent is this?")

r = JersClient().system_one("We were charged twice and need the money back today.", Ticket)
r.parsed.team, r.parsed.refund, r.parsed.urgency

Rules, labels, golden sets, batches, account

client.rules_add("mia", "session_result <= -30", "Mia's stop-loss is reached: she stops playing now.", ttl_seconds=3600)
client.rules("mia"); client.rules_delete("mia", rule_id)
client.feedback(r.decision_id, "intent", "status")
client.quality("intent"); client.calibration_fit(); client.calibration()   # the temperatures in use
client.golden_add([...]); client.golden(); client.golden_run("jers-english")
client.golden_delete(["refund-1"])          # these cases; [] deletes nothing; no argument deletes every case
client.decisions_delete()                   # every stored decision record and label of this tenant
job = client.batch_upload("requests.jsonl")  # or client.batch_create([{...}, ...])
client.batch_wait(job["id"]); client.batch_results(job["id"]); client.batch(job["id"]); client.batches()
client.models().available(); client.usage()
client.request("GET", "/v1/usage")          # any route, as parsed JSON

from jers import signup
signup("you@example.com", invite_code="...")   # no key needed; the key is shown once

golden_run answers every case in one request. If that takes longer than the client's timeout (120 s by default), the client raises ConnectionFailed and does not send it again while the gateway finishes and bills the run; give a large golden set a larger timeout.

Errors and retries

One class per status: AuthenticationError (401), PaymentRequiredError (402), PermissionDeniedError (403), NotFoundError (404), ConflictError (409), BadRequestError (413, 415, 422), RateLimitError (429, with retry_after), EngineError (502, 503), ServerError (500), ConnectionFailed. All inherit JersError and carry status and kind. The client keeps one connection open per thread, retries 429, 502 and 503 up to max_retries (default 2) with backoff, honouring Retry-After up to 30 seconds, and retries a connection that failed before the request was sent. A request that was sent and then failed is sent again only if it is a GET, so a decision is never billed twice.

Async

import asyncio
from jers import AsyncJersClient

async def main():
    async with AsyncJersClient() as client:
        a, b = await asyncio.gather(client.system_one(state_a, questions), client.system_one(state_b, questions))

asyncio.run(main())

Each call runs the synchronous client in a worker thread, so several requests can be in flight at once; the gateway accepts them concurrently and the engine answers them one at a time.