Semantic Operators

pip install "semantic-operators[typesafe]"

Write a semantic judgment once, run it on any System One model, and get a "don't know" instead of a guess when the model isn't sure.

System One models are small, fast models that classify instead of generating text, as opposed to a chat LLM. You ask them typed questions (yes/no, pick one, rate on a scale) about text or data, and they return answers with probabilities. Jev was the first; Laya is an open-weight, Jev-compatible alternative you can run locally. More are coming.

A provider is the service or runtime you talk to; the model is a setting.

Install

pip install "semantic-operators[typesafe]"        # TypeSafe (hosted Jev)
pip install "semantic-operators[laya]"            # Laya (local; pulls in torch)
pip install "semantic-operators[typesafe,laya]"   # both

The core alone (pip install semantic-operators) has no dependencies. Add mcp to the extras for the MCP server ([typesafe,mcp]), or install the semop command on its own with uv tool install "semantic-operators[typesafe,mcp]".

The whole idea

A System One model is asked named questions about a piece of state and returns an answer with probabilities for each. There are three kinds of question:

Every Answer also carries probabilities (a dict, in the question's option/level order), confidence (the probability of its own answer), raw (the provider's own answer object), and call (which model answered, below).

A provider is anything with one method:

def ask(self, state, questions: dict[str, Question]) -> dict[str, Answer]

That's the entire abstraction.

Quick start

echo "TYPESAFE_API_KEY=..." > .env
uv run --env-file .env --extra typesafe python examples/hello.py

from typesafe_sdk import TypeSafeClient
from semantic_operators import Boolean, Choice, Score
from semantic_operators.providers.typesafe import TypeSafe

with TypeSafeClient() as client:          # you create and own the SDK client
    provider = TypeSafe(client)           # model defaults to "jev-latest"
    answers = provider.ask(
        "I was charged twice and I'm furious.",
        {
            "is_complaint": Boolean("Is the customer complaining?"),
            "department": Choice("Which team should handle this?",
                                 {"billing": "Payments, refunds", "other": "Anything else"}),
            "urgency": Score("How urgent is this?", ["low", "medium", "high"]),
        },
    )

answers["department"].value           # "billing"
answers["department"].probabilities   # {"billing": 0.97, "other": 0.03}

Swapping to Laya changes only how the provider is built:

import laya
from semantic_operators.providers.laya import Laya

provider = Laya(laya.load("convaiinnovations/laya"))   # or Laya(laya.Router())
answers = provider.ask(state, questions)                # same questions, same Answer type

Compare both side by side:

uv run --env-file .env --extra typesafe --extra laya python examples/compare.py

Command line and MCP

The semop command answers one JSON request, from a file or stdin:

echo '{
  "state": "The payment failed and now I cannot sign in.",
  "questions": [
    {"name": "is_complaint", "type": "boolean", "instructions": "Is it a complaint?"},
    {"name": "department", "type": "choice", "instructions": "Which team?",
     "options": [{"name": "billing", "description": "Charges and refunds"}, "technical"],
     "min_confidence": 0.8},
    {"name": "urgency", "type": "score", "instructions": "How urgent?",
     "levels": ["low", "medium", "high"]}
  ]
}' | semop ask --provider typesafe --pretty

It prints each answer's value (null when undecided), decided, confidence, and probabilities, rounded to 4 significant digits, plus the call (provider, model, tokens). Exit code 0 means answered, including "don't know"; 1, the provider failed or timed out; 2, the request was invalid (the error says where, e.g. questions[1].options). --timeout SECONDS limits each call. The TypeSafe provider reads TYPESAFE_API_KEY from the environment.

semop mcp runs the same thing as an MCP server over stdio, with one tool, ask, that takes the same request and returns the same response. Register it with Claude Code:

claude mcp add semop -e TYPESAFE_API_KEY="$TYPESAFE_API_KEY" -- semop mcp --provider typesafe
claude mcp add semop-local -- semop mcp --provider laya --timeout 120     # local model

• The provider is fixed when the server starts. Nothing an agent sends can change it, so a local-only server can't be switched to a paid hosted API from a tool call.
• The tool description tells the agent where answers come from (and whether the data leaves the machine), to treat "don't know" as unresolved rather than as no, and not to reword a question or lower min_confidence to get the answer it wants.
• A local model loads on the first call (a few seconds) and is reused after that.

Named operators

This is where the library gets its name. An operator is a semantic judgment defined once, with a name, and used anywhere:

from semantic_operators import Boolean, Choice, Score
from semantic_operators.operators import Operator, apply

is_complaint = Operator("is_complaint", Boolean("Is the customer complaining?"))
urgency = Operator("urgency", Score("How urgent is this?", ["low", "medium", "high"]))

is_complaint(provider, message).value                        # one operator, one call
answers = apply(provider, message, [is_complaint, urgency])  # several, still one call
answers["urgency"].value

• Combine freely. System One models answer many questions in one pass, so apply asks any set of operators in a single provider call. Names must be unique.
• Provider-neutral. An operator doesn't hold a provider; you pass one in, so the same operator runs on TypeSafe, Laya, or anything else.
• Wording is part of the operator. It changes the answers (see the benchmark), so keep operators in code, under version control, and benchmark them as they are. operators.questions([...]) turns them into the dict bench.run takes.
• Async: await op.call_async(provider, state) and await apply_async(...).

examples/triage.py builds ticket triage from three operators, sending anything the model isn't sure about to a person. Add --laya to run the same code locally.

Reranking

Your retrieval (search, vector index, database) finds candidates; a System One model reorders them by how well each one answers the query. Each candidate is scored on its own against a relevance rubric, one call each, and plain code sorts the results:

from semantic_operators import Score
from semantic_operators.rerank import rerank, rerank_async, reweighted

relevance = Score("How useful is the document for answering the query?",
                  ["no useful information", "on topic but doesn't answer", "partly answers",
                   "answers with minor gaps", "fully answers"])

ranking = rerank(provider, relevance, query="How do I reset my password?",
                 candidates={"doc-1": {"title": ..., "text": ...}, ...})  # retrieval order
for r in ranking.top(10):
    r.id, r.score, r.answer.probabilities

• Each call sees only {"query", "document"} (plus "context" if you pass one). Ids and positions are never sent, so a score doesn't depend on the other candidates.
• Ranked by expected rubric level, not confidence. Ties keep retrieval order, and scores aren't normalized across documents: all can be relevant, or none.
• Nothing gets a made-up score. Undecided or failed candidates go to unscored. top() refuses a partial ranking unless you pass allow_partial=True, and refuses scores from more than one model (ranking.models), since a moving alias such as jev-latest can change mid-run.
• Your own level weights: reweighted(ranking, [0, 10, 40, 80, 100]) re-sorts from the probabilities already returned, with no new calls. Uneven weights can change the order, so evaluate them first.
• Async: await rerank_async(..., concurrency=8) keeps up to 8 calls in flight.

bench.ndcg(order, grades) scores an ordering against graded relevance labels. examples/rerank.py reranks three hand-graded searches and compares NDCG before and after; on those, TypeSafe went from 0.56 (retrieval order) to 1.00 and Laya to 0.84.

Escalation

Ask a fast or local model first, and a stronger one only about what it wasn't sure of:

from semantic_operators.cascade import cascade

provider = cascade(Laya(model), TypeSafe(client), escalate_below=0.8)   # still a provider
answers = provider.ask(message, questions)
answers["department"].call.provider    # "Laya", or "TypeSafe" if it was escalated

• Every question goes to the first provider. Answers with confidence under escalate_below (or undecided) are re-asked of the next provider, all in one call, and so on down the list. The last provider's answer stands.
• Two thresholds, two decisions: escalate_below decides when to ask a stronger model; a question's min_confidence decides when the final answer is a "don't know".
• It escalates when a model is unsure, never when it fails: a ProviderError propagates, because quietly routing around an outage would hide it.
• It's a provider, so operators, rerank, and benchmarks work unchanged. cascade_async does the same for async providers.

It's only as good as the first model's confidence, and a message costs a call to the stronger model if any of its questions escalates. benchmarks/run_cascade.py measures both. On the support suite, Laya first with escalate_below=0.9 came within one answer of TypeSafe alone, but still needed TypeSafe for 20 of 20 messages: Laya was unsure of something in almost every one, and confidently wrong on others. The mechanism works; whether a pairing pays off is a benchmark question.

"Don't know" answers

A model that's split, or not sure enough, should say so rather than guess. Give any question a min_confidence; below it, the answer comes back undecided (value is None, decided is False), with its probabilities kept:

department = Choice("Which team?", {"billing": None, "technical": None},
                    min_confidence=0.8)
answer = provider.ask(message, {"department": department})["department"]

if answer.decided:
    route(answer.value)
else:
    send_to_a_human(answer.probabilities)   # still shows what it was leaning toward

An exact tie (a Boolean at 0.5, two options equally likely) is always undecided. Confidence is the model's own view, not a guarantee. benchmarks/run_confidence.py checks whether it means anything: on the support suite, TypeSafe's urgency answers at min_confidence=0.8 were right 10 of 10 times (answering half the cases), while Laya's were right 4 of 7.

Which model answered

jev-latest moves over time, so every answer records what its provider call reported:

answers["department"].call
# Call(provider='TypeSafe', model='jev-1.13.0', input_tokens=291, output_tokens=20)

All answers from one call share one Call. model is exactly what the provider reported, which can differ from what you asked for (above, jev-latest). Laya reports a fixed agent name (laya-rl-agent), not which checkpoint answered. Anything a provider doesn't report is None.

Errors

Every provider raises one error type, whatever went wrong underneath (network, bad key, rate limit, a model failure, an unexpected response):

from semantic_operators import ProviderError

try:
    answers = provider.ask(state, questions)
except ProviderError as error:
    error.provider     # "TypeSafe" or "Laya"
    error.__cause__    # the original exception, for details

Provider output is checked before it becomes an Answer: probabilities must be finite, between 0 and 1, sum to 1 (allowing for the providers' rounding), and agree with the answer. A malformed response raises ProviderError rather than looking like a confident answer. Questions check themselves too: a Choice needs at least two distinct options, a Score at least two distinct levels, and a bad definition raises ValueError.

Retries belong to the client you build. The TypeSafe SDK retries by default; to have every failure reach you (for example, when something above you does its own retrying), turn that off (timeout here is the SDK's total budget across retries):

TypeSafeClient(retry=RetryPolicy(max_retries=0, timeout=10.0))

Timeouts

Give any async provider a time limit per call:

from semantic_operators import ProviderTimeout, with_timeout

provider = with_timeout(AsyncTypeSafe(client), 2.0)   # still a provider
try:
    answers = await provider.ask(state, questions)
except ProviderTimeout:        # also a ProviderError and a TimeoutError
    ...

Because it's just a provider, the limit carries through to operators, apply_async, and rerank_async (a candidate that runs out of time is simply unscored). For a limit on a whole batch of work, wrap it in asyncio.timeout(...).

Stopping to wait isn't always stopping the work: a hosted call may still finish (and be billed) on the server, and a local model finishes its current prediction in the background while later calls wait their turn. Sync code has no neutral timeout: set it on the client you build (TypeSafeClient(timeout=5), which also comes back as ProviderTimeout). A local model can't be interrupted mid-prediction.

Benchmark

bench.run(provider, questions, cases) asks each labeled case all questions in one call and reports, per question, accuracy (Score values are rounded to the nearest level) and p(correct), the average probability the provider gave the right answer, plus latency and every miss. Undecided answers are counted separately (answered), not as misses. bench.at_min_confidence(report, questions, cases, 0.8) re-scores a run at a stricter threshold without asking the provider again.

uv run --env-file .env --extra typesafe --extra laya python benchmarks/run.py

benchmarks/support_tickets.py holds 20 hand-written, hand-labeled support messages and the same 3 questions in three wordings. bench.stability(reports) reports how often a provider's decision stays the same when only the wording changes (labels play no part). It's a smoke test, not a verdict: small, authored, one person's labels.

Async

Every provider has an async twin with the same contract, await provider.ask(...):

from typesafe_sdk import AsyncTypeSafeClient
from semantic_operators.providers.typesafe import AsyncTypeSafe
from semantic_operators.providers.laya import AsyncLaya

async with AsyncTypeSafeClient() as client:
    answers = await AsyncTypeSafe(client).ask(state, questions)

AsyncLaya runs the local model in a worker thread, one call at a time. Concurrency speeds up a hosted API (many requests in flight), not a single local model. bench.run_async(provider, questions, cases, concurrency=8) benchmarks async providers:

uv run --env-file .env --extra typesafe --extra laya python benchmarks/run_async.py

Layout

src/semantic_operators/
  types.py          Boolean, Choice, Score, Answer, Call, make_answer: our vocabulary
  provider.py       Provider and AsyncProvider (one method each)
  errors.py         ProviderError, the one error every provider raises
  providers/typesafe.py  translates to/from the TypeSafe SDK
  providers/laya.py translates to/from the laya package
  operators.py      (higher layer) named operators: define once, combine in one call
  rerank.py         (higher layer) rerank search results by relevance
  cascade.py        (higher layer) escalate unsure answers to a stronger provider
  bench.py          (higher layer) run labeled cases through a provider, score them;
                    ndcg for rankings
  interfaces/       (application) the semop CLI and MCP server: wire.py (the JSON
                    shape), backends.py (providers from flags), cli.py, mcp_server.py
examples/
  hello.py          one real call to Jev
  compare.py        the same questions through TypeSafe and Laya
  triage.py         ticket triage built from named operators
  rerank.py         rerank three searches, NDCG before and after
benchmarks/
  support_tickets.py  20 labeled messages + the questions
  run.py              runs the suite through TypeSafe and Laya
  run_async.py        concurrency, and both providers at once
  run_confidence.py   the "don't know" trade-off at several min_confidence levels
  run_cascade.py      Laya first, TypeSafe for what Laya wasn't sure of
tests/                offline tests (uv run --extra typesafe --extra mcp pytest); CI runs them
ROADMAP.md            where this is headed

Layers

Semantic Operators is built in layers inside one package:

1. Base layer: a clean, provider-neutral abstraction over System One models: types.py, provider.py, errors.py, providers/.

2. Higher layers: built only on the base layer: operators.py (named operators), rerank.py (reranking), cascade.py (escalation), and bench.py (benchmarking).

3. Application: interfaces/, the semop CLI and MCP server. It uses the library the way your own code does, and it's the only part that reads the environment.

The base layer never imports from a higher layer, and nothing in the library imports interfaces/, so the base could later be split out as its own package without changing how it's used.

Rules

• The library never reads API keys or environment variables. You build the client. (The semop application does, since it builds the client for you.)
• The core has no dependencies. Each provider's SDK is an optional extra ([typesafe], [laya]).
• Our names, not the provider's: Boolean, not noul.

Not here yet (on purpose)

Operators are just named questions for now. Combining them is where this is headed: flows built from operators, as plain Python functions with a trace. Also planned: suites as data files, and more MCP tools (reranking, your named operators). See ROADMAP.md.

License

MIT