| """HTTP API contracts on top of :func:`deg.infer.fill` (SPEC.md §9). |
| |
| Three request shapes are translated into one deg fill (one schema, one context read) and |
| the fill response is translated back: |
| |
| * ``POST /v1/decisions``: the OpenAI Decisions API (public beta, 2026-10-06). Field names |
| and answer objects mirror ``openai-python`` 3.26.1 ``types/decision.py`` and |
| ``types/decision_create_params.py`` and the guide at |
| https://developers.openai.com/api/docs/guides/decisions: ``model``, ``input`` (string or |
| user messages of ``input_text`` parts), ``questions`` (``predicate`` / ``choice`` / |
| ``score``, each with an optional ``name``) -> ``{answers: [...], model, usage}`` with |
| answers in question order. |
| * ``POST /v1/systemone``: TypeSafe Jev / SystemOne (``state`` + a map of ``noul`` / |
| ``choice`` / ``score`` questions). Shapes, limits and error bodies follow live |
| ``api.typesafe.ai`` responses from ``jev-1.13.0`` (2026-10-09): answers keyed by question |
| id in request order, ``{"type": "noul", "noul": p}``, choice ``{type, choice, confidence, |
| probabilities{option: p}}``, score ``{type, score, confidence, legend{"i": level}, |
| probabilities{"i": p}}``, usage ``{input_tokens, output_tokens}``; 400 ``{"detail": |
| {"error_type": "api_usage_error", "message": ...}}`` for semantic errors and 422 |
| ``{"detail": [{type, loc, msg, input}]}`` for missing fields. |
| * ``POST /v1/fill``: deg's own contract (:mod:`deg.request`), returned unchanged. |
| |
| Mapping (the same in both decision APIs): ``predicate`` = ``noul`` -> ``boolean`` |
| (decide.noul); ``choice`` -> ``enum`` / ``oneOf`` of ``const`` with descriptions |
| (decide.choice); ``score`` -> ``integer`` levels ``0..L-1`` (decide.score) with level labels |
| (and descriptions, ``label: description``) as ``x-levels``, so each becomes that level's |
| option line ``i: label`` (the jev training layout); the description keeps only the |
| instructions. One-option choices and one-level scores are |
| answered without the model (probability 1), as Jev does. |
| |
| deg extensions, opt-in (stock clients never send them): |
| |
| * ``extract``: a value copied from the input. ``value_type`` is ``string`` (default), |
| ``number``, ``integer``, ``date``, ``date-time``, ``email``, ``uri`` or ``uuid``; optional |
| ``pattern``, ``min_length``, ``max_length`` (string types). Always nullable: the answer is |
| ``{type: "extract", value, span: [start, end) | null, confidence}`` plus ``null_reason`` |
| when ``value`` is null. ``span`` indexes code points of the rendered input. |
| * ``generate``: free text (the only new tokens), with optional ``max_tokens`` (that |
| question's own budget, ``x-max-tokens``; questions in one request may differ), |
| ``min_length``, ``max_length``, ``pattern``; greedy unless ``options.temperature`` > 0 |
| (then ``options.seed`` is required). Generated after every other question and conditioned |
| on their answers. Answer: ``{type: "generate", text, output_tokens}``. |
| * a request-level ``options`` object: ``temperature``, ``seed``, ``date_order``, |
| ``default_offset``, ``array_decode``, ``null_margin`` (as in ``/v1/fill``). |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| from collections.abc import Callable |
| from dataclasses import dataclass, field |
| from typing import Any |
|
|
| from deg.request import MODEL_ALIASES, FillOptions, parse_options |
| from deg.response import MODEL_ID |
| from deg.schema import SchemaError, compile_schema |
|
|
| MAX_CHOICES = 255 |
| MAX_LEVELS = 10 |
| OPENAI_COMPAT_MODELS = frozenset({"gpt-6-luna"}) |
| JEV_COMPAT_MODELS = frozenset({"jev-latest", "jev-preview", "clef-flash"}) |
| EXTRACT_TYPES = { |
| "string": ("string", None), |
| "number": ("number", None), |
| "integer": ("integer", None), |
| "date": ("string", "date"), |
| "date-time": ("string", "date-time"), |
| "email": ("string", "email"), |
| "uri": ("string", "uri"), |
| "uuid": ("string", "uuid"), |
| } |
| REQUEST_OPTION_KEYS = frozenset({"temperature", "seed", "date_order", "default_offset", |
| "array_decode", "null_margin"}) |
| DECIDE_TYPES = ("predicate", "choice", "score") |
| EXTENSION_TYPES = ("extract", "generate") |
|
|
| FillFn = Callable[[Any, dict[str, Any], FillOptions], dict[str, Any]] |
|
|
|
|
| class ApiError(Exception): |
| """A client (4xx) or server (5xx) error with the API's own body shape.""" |
|
|
| def __init__(self, status: int, body: dict[str, Any], message: str) -> None: |
| super().__init__(message) |
| self.status = status |
| self.body = body |
|
|
|
|
| |
|
|
|
|
| @dataclass(frozen=True) |
| class Question: |
| """One question in either decision API, in deg terms.""" |
|
|
| key: str |
| kind: str |
| instructions: str |
| name: str | None = None |
| options: tuple[Any, ...] = () |
| descriptions: tuple[str | None, ...] = () |
| labels: tuple[str, ...] = () |
| true_criterion: str | None = None |
| false_criterion: str | None = None |
| extra: dict[str, Any] = field(default_factory=dict) |
|
|
| @property |
| def trivial(self) -> bool: |
| """A one-option choice or one-level score: answered with probability 1.""" |
| return (self.kind == "choice" and len(self.options) == 1) or ( |
| self.kind == "score" and len(self.labels) == 1) |
|
|
| def field_schema(self) -> dict[str, Any]: |
| if self.kind == "noul": |
| desc = self.instructions |
| if self.true_criterion: |
| desc += f"\nYes means: {self.true_criterion}" |
| if self.false_criterion: |
| desc += f"\nNo means: {self.false_criterion}" |
| return {"type": "boolean", "description": desc} |
| if self.kind == "choice": |
| if any(self.descriptions): |
| return {"description": self.instructions, "oneOf": [ |
| {"const": v, **({"description": d} if d else {})} |
| for v, d in zip(self.options, self.descriptions)]} |
| return {"description": self.instructions, "enum": list(self.options)} |
| if self.kind == "score": |
| return {"type": "integer", "minimum": 0, "maximum": len(self.labels) - 1, |
| "description": self.instructions, |
| "x-levels": {str(i): lab + (f": {d}" if d else "") |
| for i, (lab, d) in enumerate(zip(self.labels, |
| self.descriptions))}} |
| if self.kind == "extract": |
| json_type, fmt = EXTRACT_TYPES[self.extra.get("value_type", "string")] |
| out: dict[str, Any] = {"type": [json_type, "null"], "description": self.instructions} |
| if fmt: |
| out["format"] = fmt |
| for src, dst in (("pattern", "pattern"), ("min_length", "minLength"), |
| ("max_length", "maxLength")): |
| if src in self.extra: |
| out[dst] = self.extra[src] |
| return out |
| out = {"type": "string", "x-generate": True, "description": self.instructions} |
| for src, dst in (("pattern", "pattern"), ("min_length", "minLength"), |
| ("max_length", "maxLength"), ("max_tokens", "x-max-tokens")): |
| if src in self.extra: |
| out[dst] = self.extra[src] |
| return out |
|
|
|
|
| @dataclass(frozen=True) |
| class Job: |
| """One fill to run: ``schema`` is None when every question is trivial.""" |
|
|
| context: Any |
| schema: dict[str, Any] | None |
| options: FillOptions |
| questions: tuple[Question, ...] |
|
|
| @property |
| def question_types(self) -> tuple[str, ...]: |
| return tuple(q.kind for q in self.questions) |
|
|
|
|
| def _is_int(value: Any) -> bool: |
| return isinstance(value, int) and not isinstance(value, bool) |
|
|
|
|
| def _constraint_errors(kind: str, raw: dict[str, Any], add: Callable[[str, str], None] |
| ) -> dict[str, Any]: |
| """Validate extract/generate keys; returns the constraint dict (``extra``).""" |
| extra: dict[str, Any] = {} |
| if kind == "extract": |
| vt = raw.get("value_type", "string") |
| if vt not in EXTRACT_TYPES: |
| add("value_type", "value_type must be one of " + ", ".join(EXTRACT_TYPES)) |
| return extra |
| extra["value_type"] = vt |
| string_like = EXTRACT_TYPES[vt][0] == "string" and EXTRACT_TYPES[vt][1] is None |
| else: |
| string_like = True |
| if "max_tokens" in raw: |
| if not _is_int(raw["max_tokens"]) or raw["max_tokens"] < 1: |
| add("max_tokens", "max_tokens must be a positive integer") |
| else: |
| extra["max_tokens"] = raw["max_tokens"] |
| for key in ("min_length", "max_length"): |
| if key in raw: |
| if not string_like: |
| add(key, f"{key} applies only to string values") |
| elif not _is_int(raw[key]) or raw[key] < 0: |
| add(key, f"{key} must be a non-negative integer") |
| else: |
| extra[key] = raw[key] |
| if "pattern" in raw: |
| if not string_like: |
| add("pattern", "pattern applies only to string values") |
| elif not isinstance(raw["pattern"], str): |
| add("pattern", "pattern must be a string") |
| else: |
| extra["pattern"] = raw["pattern"] |
| return extra |
|
|
|
|
| def _request_options(raw: Any, add: Callable[[str, str], None]) -> FillOptions: |
| if raw is None: |
| return FillOptions() |
| if not isinstance(raw, dict): |
| add("options", "options must be an object") |
| return FillOptions() |
| errors: list[dict[str, Any]] = [] |
| for key in raw: |
| if key not in REQUEST_OPTION_KEYS: |
| errors.append({"loc": ["options", key], "msg": "unknown option"}) |
| opts = parse_options({k: v for k, v in raw.items() if k in REQUEST_OPTION_KEYS}, errors) |
| for e in errors: |
| add(".".join(map(str, e["loc"][-2:] if e["loc"][0] == "body" else e["loc"])), e["msg"]) |
| return opts |
|
|
|
|
| def build_job(context: Any, questions: list[Question], options: FillOptions |
| ) -> tuple[Job, list[tuple[str, str]]]: |
| """The fill schema for these questions; schema compile errors come back as (loc, msg).""" |
| live = [q for q in questions if not q.trivial] |
| if not live: |
| return Job(context, None, options, tuple(questions)), [] |
| schema = {"type": "object", |
| "properties": {q.key: q.field_schema() for q in live}, |
| "required": [q.key for q in live], |
| "additionalProperties": False} |
| try: |
| compile_schema(schema) |
| except SchemaError as exc: |
| return Job(context, None, options, tuple(questions)), [ |
| (".".join(map(str, e["loc"])), e["msg"]) for e in exc.errors] |
| return Job(context, schema, options, tuple(questions)), [] |
|
|
|
|
| def _pointer(key: str) -> str: |
| return "/" + key.replace("~", "~0").replace("/", "~1") |
|
|
|
|
| def _expected(probabilities: list[float]) -> float: |
| return sum(i * p for i, p in enumerate(probabilities)) |
|
|
|
|
| @dataclass(frozen=True) |
| class Outcome: |
| """A question's deg outcome, API-neutral.""" |
|
|
| kind: str |
| value: Any = None |
| probability: float | None = None |
| probabilities: tuple[float, ...] = () |
| confidence: float | None = None |
| span: list[int] | None = None |
| null_reason: str | None = None |
| output_tokens: int = 0 |
|
|
|
|
| def outcomes(job: Job, response: dict[str, Any] | None) -> list[Outcome]: |
| """Per-question outcomes from a fill response (``None`` when nothing was filled).""" |
| out = [] |
| for q in job.questions: |
| if q.trivial: |
| value = q.options[0] if q.kind == "choice" else 0 |
| out.append(Outcome(q.kind, value, probabilities=(1.0,), confidence=1.0)) |
| continue |
| assert response is not None |
| meta = response["meta"][_pointer(q.key)] |
| value = response["value"].get(q.key) |
| if q.kind == "noul": |
| out.append(Outcome("noul", value, probability=float(meta["probability"]))) |
| elif q.kind in ("choice", "score"): |
| out.append(Outcome(q.kind, value, probabilities=tuple(meta["probabilities"]), |
| confidence=float(meta["confidence"]))) |
| elif q.kind == "extract": |
| out.append(Outcome("extract", value, confidence=float(meta["confidence"]), |
| span=meta.get("span"), null_reason=meta.get("null_reason"))) |
| else: |
| out.append(Outcome("generate", value, |
| output_tokens=int(meta.get("generated_tokens", 0)))) |
| return out |
|
|
|
|
| def _usage_tokens(response: dict[str, Any] | None) -> tuple[int, int]: |
| if response is None: |
| return 0, 0 |
| usage = response.get("usage") or {} |
| return int(usage.get("prompt_tokens", 0)), int(usage.get("generated_tokens", 0)) |
|
|
|
|
| def _extension_answer(o: Outcome) -> dict[str, Any]: |
| if o.kind == "extract": |
| ans = {"type": "extract", "value": o.value, "span": o.span, "confidence": o.confidence} |
| if o.value is None: |
| ans["null_reason"] = o.null_reason or "not_found" |
| return ans |
| return {"type": "generate", "text": o.value, "output_tokens": o.output_tokens} |
|
|
|
|
| def _text(value: Any) -> str: |
| return value if isinstance(value, str) else json.dumps(value, ensure_ascii=False) |
|
|
|
|
| |
|
|
|
|
| def openai_error(status: int, message: str, param: str | None = None, |
| kind: str = "invalid_request_error", code: str | None = None) -> ApiError: |
| return ApiError(status, {"error": {"message": message, "type": kind, "param": param, |
| "code": code}}, message) |
|
|
|
|
| def _openai_input(raw: Any, add: Callable[[str, str], None]) -> str: |
| """Text the model reads: the string, or the user messages' text parts joined by blank |
| lines. Images are refused (deg is text-only).""" |
| if isinstance(raw, str): |
| if not raw.strip(): |
| add("input", "input must be a non-empty string") |
| return raw |
| if not isinstance(raw, list) or not raw: |
| add("input", "input must be a string or a non-empty array of user messages") |
| return "" |
| texts: list[str] = [] |
| for i, msg in enumerate(raw): |
| loc = f"input[{i}]" |
| if not isinstance(msg, dict): |
| add(loc, "each input item must be a message object") |
| continue |
| for key in msg: |
| if key not in ("role", "content", "type"): |
| add(f"{loc}.{key}", f"Unrecognized message field: {key}") |
| if msg.get("type", "message") != "message": |
| add(f"{loc}.type", "only message items are supported") |
| if msg.get("role") != "user": |
| add(f"{loc}.role", "only user messages are supported") |
| content = msg.get("content") |
| if isinstance(content, str): |
| texts.append(content) |
| continue |
| if not isinstance(content, list) or not content: |
| add(f"{loc}.content", "content must be a string or a non-empty array of parts") |
| continue |
| for j, part in enumerate(content): |
| ploc = f"{loc}.content[{j}]" |
| ptype = part.get("type") if isinstance(part, dict) else None |
| if ptype == "input_text" and isinstance(part.get("text"), str): |
| texts.append(part["text"]) |
| elif ptype == "input_image": |
| add(ploc, "input_image is not supported: this deg model reads text only") |
| else: |
| add(ploc, "each part must be an input_text part with a string text") |
| text = "\n\n".join(texts) |
| if not text.strip(): |
| add("input", "input has no text") |
| return text |
|
|
|
|
| def _openai_question(i: int, raw: Any, add: Callable[[str, str], None]) -> Question | None: |
| loc = f"questions[{i}]" |
| if not isinstance(raw, dict): |
| add(loc, "each question must be an object") |
| return None |
| kind = raw.get("type") |
| allowed = {"predicate": {"type", "name", "instructions"}, |
| "choice": {"type", "name", "instructions", "choices"}, |
| "score": {"type", "name", "instructions", "levels"}, |
| "extract": {"type", "name", "instructions", "value_type", "pattern", |
| "min_length", "max_length"}, |
| "generate": {"type", "name", "instructions", "max_tokens", "pattern", |
| "min_length", "max_length"}}.get(kind) |
| if allowed is None: |
| add(f"{loc}.type", "type must be one of predicate, choice, score (deg extensions: " |
| "extract, generate)") |
| return None |
| n = 0 |
|
|
| def qadd(key: str, msg: str) -> None: |
| nonlocal n |
| n += 1 |
| add(f"{loc}.{key}", msg) |
|
|
| for key in raw: |
| if key not in allowed: |
| qadd(key, f"Unrecognized question field for type {kind}: {key}") |
| name = raw.get("name") |
| if name is not None and (not isinstance(name, str) or not name): |
| qadd("name", "name must be a non-empty string") |
| name = None |
| instructions = raw.get("instructions") |
| if not isinstance(instructions, str) or not instructions.strip(): |
| qadd("instructions", "instructions must be a non-empty string") |
| key = name if name is not None else f"question_{i}" |
| if kind == "predicate": |
| q = Question(key, "noul", instructions or "", name) |
| elif kind == "choice": |
| choices = raw.get("choices") |
| if not isinstance(choices, list) or not choices: |
| qadd("choices", "choices must be a non-empty array") |
| return None |
| if len(choices) > MAX_CHOICES: |
| qadd("choices", f"Too many choices. Must have at most {MAX_CHOICES} choices.") |
| values, descs, seen = [], [], set() |
| for j, ch in enumerate(choices): |
| cloc = f"choices[{j}]" |
| if not isinstance(ch, dict) or set(ch) - {"value", "description"}: |
| qadd(cloc, "each choice must be {value, description?}") |
| continue |
| value, desc = ch.get("value"), ch.get("description") |
| if not isinstance(value, (str, bool)) or value == "": |
| qadd(f"{cloc}.value", "value must be a non-empty string or a boolean") |
| continue |
| if desc is not None and not isinstance(desc, str): |
| qadd(f"{cloc}.description", "description must be a string") |
| typed = (type(value).__name__, value) |
| if typed in seen: |
| qadd(f"{cloc}.value", f"duplicate choice value {value!r}") |
| seen.add(typed) |
| values.append(value) |
| descs.append(desc or None) |
| q = Question(key, "choice", instructions or "", name, tuple(values), tuple(descs)) |
| elif kind == "score": |
| levels = raw.get("levels") |
| if not isinstance(levels, list) or not levels: |
| qadd("levels", "levels must be a non-empty array") |
| return None |
| if len(levels) > MAX_LEVELS: |
| qadd("levels", f"Too many score levels. Must have at most {MAX_LEVELS} levels.") |
| labels, descs = [], [] |
| for j, lv in enumerate(levels): |
| lloc = f"levels[{j}]" |
| if (not isinstance(lv, dict) or set(lv) - {"label", "description"} |
| or not isinstance(lv.get("label"), str) or not lv["label"].strip() |
| or not isinstance(lv.get("description", ""), str)): |
| qadd(lloc, "each level must be {label: non-empty string, description?: string}") |
| continue |
| labels.append(lv["label"]) |
| descs.append(lv.get("description") or None) |
| q = Question(key, "score", instructions or "", name, descriptions=tuple(descs), |
| labels=tuple(labels)) |
| else: |
| extra = _constraint_errors(kind, raw, qadd) |
| q = Question(key, kind, instructions or "", name, extra=extra) |
| return None if n else q |
|
|
|
|
| def parse_decisions(payload: Any, accepted_models: frozenset[str] = MODEL_ALIASES |
| | OPENAI_COMPAT_MODELS) -> Job: |
| """Validate an OpenAI Decisions request. Raises :class:`ApiError` (400).""" |
| if not isinstance(payload, dict): |
| raise openai_error(400, "The request body must be a JSON object.") |
| errors: list[tuple[str, str]] = [] |
|
|
| def add(param: str, msg: str) -> None: |
| errors.append((param, msg)) |
|
|
| for key in payload: |
| if key not in ("model", "input", "questions", "safety_identifier", "options"): |
| add(key, f"Unrecognized request argument supplied: {key}") |
| model = payload.get("model") |
| if not isinstance(model, str): |
| add("model", "Missing required parameter: 'model'.") |
| elif model not in accepted_models: |
| add("model", f"The model `{model}` does not exist or you do not have access to it.") |
| if "input" not in payload: |
| add("input", "Missing required parameter: 'input'.") |
| text = "" |
| else: |
| text = _openai_input(payload["input"], add) |
| sid = payload.get("safety_identifier") |
| if sid is not None and not isinstance(sid, str): |
| add("safety_identifier", "safety_identifier must be a string") |
| options = _request_options(payload.get("options"), add) |
| raw_questions = payload.get("questions") |
| questions: list[Question] = [] |
| if not isinstance(raw_questions, list) or not raw_questions: |
| add("questions", "questions must be a non-empty array") |
| else: |
| for i, raw in enumerate(raw_questions): |
| q = _openai_question(i, raw, add) |
| if q is not None: |
| questions.append(q) |
| names = [q.get("name") for q in raw_questions if isinstance(q, dict) and q.get("name")] |
| dup = sorted({n for n in names if isinstance(n, str) and names.count(n) > 1}) |
| if dup: |
| add("questions", f"question names must be unique: {dup}") |
| keys = [q.key for q in questions] |
| if len(set(keys)) != len(keys) and not dup: |
| add("questions", "a question name collides with an unnamed question's key " |
| "(question_<index>)") |
| job = None |
| if not errors: |
| job, schema_errors = build_job(text, questions, options) |
| for loc, msg in schema_errors: |
| add("questions", f"{loc}: {msg}") |
| if errors: |
| param, _ = errors[0] |
| raise openai_error(400, "; ".join(f"{p}: {m}" for p, m in errors), param) |
| assert job is not None |
| return job |
|
|
|
|
| def decisions_response(job: Job, response: dict[str, Any] | None, |
| model: str = MODEL_ID) -> dict[str, Any]: |
| """The ``Decision`` object (openai-python ``types/decision.py``).""" |
| answers = [] |
| for q, o in zip(job.questions, outcomes(job, response), strict=True): |
| if o.kind == "noul": |
| answers.append({"type": "predicate", "name": q.name, "probability": o.probability}) |
| elif o.kind == "choice": |
| answers.append({"type": "choice", "name": q.name, "choice": o.value, |
| "probabilities": [{"value": v, "probability": p} |
| for v, p in zip(q.options, o.probabilities)], |
| "confidence": o.confidence}) |
| elif o.kind == "score": |
| probs = list(o.probabilities) |
| answers.append({"type": "score", "name": q.name, "score": _expected(probs), |
| "probabilities": [{"value": i, "label": lab, "probability": p} |
| for i, (lab, p) in enumerate(zip(q.labels, probs))], |
| "confidence": o.confidence}) |
| else: |
| ext = _extension_answer(o) |
| answers.append({"type": ext.pop("type"), "name": q.name, **ext}) |
| inp, out = _usage_tokens(response) |
| return {"answers": answers, "model": model, |
| "usage": {"input_tokens": inp, |
| "input_tokens_details": {"cache_write_tokens": 0, "cached_tokens": 0}, |
| "output_tokens": out, |
| "output_tokens_details": {"reasoning_tokens": 0}, |
| "total_tokens": inp + out}} |
|
|
|
|
| |
|
|
|
|
| def jev_error(message: str, status: int = 400, error_type: str = "api_usage_error", |
| *, plain: bool = False) -> ApiError: |
| """Jev's 400 body; ``plain`` = a bare string ``detail`` (live Jev's limit errors).""" |
| detail: Any = message if plain else {"error_type": error_type, "message": message} |
| return ApiError(status, {"detail": detail}, message) |
|
|
|
|
| def jev_validation_error(items: list[dict[str, Any]]) -> ApiError: |
| return ApiError(422, {"detail": items}, "; ".join(i["msg"] for i in items)) |
|
|
|
|
| def _jev_structured(value: Any) -> bool: |
| if isinstance(value, str): |
| return bool(value.strip()) |
| if isinstance(value, (dict, list)) and value: |
| try: |
| json.dumps(value, allow_nan=False) |
| except (TypeError, ValueError): |
| return False |
| return True |
| return False |
|
|
|
|
| def _jev_question(qid: str, raw: Any) -> Question: |
| """One SystemOne question; raises :class:`ApiError` (Jev bodies).""" |
| loc = ["body", "questions", qid] |
| if not isinstance(raw, dict): |
| raise jev_validation_error([{"type": "model_type", "loc": loc, |
| "msg": "Input should be a valid dictionary", "input": raw}]) |
| kind = raw.get("type") |
| if kind not in ("noul", "choice", "score", "extract", "generate"): |
| raise jev_error("Invalid request.") |
| if "instructions" not in raw: |
| raise jev_validation_error([{"type": "missing", "loc": [*loc, "instructions"], |
| "msg": "Field required", "input": raw}]) |
| instructions = raw["instructions"] |
| if not _jev_structured(instructions): |
| raise jev_error("Invalid request.") |
| text = _text(instructions) |
| criteria = raw.get("criteria") |
| if kind == "noul": |
| true_c = false_c = None |
| if criteria is not None: |
| if not isinstance(criteria, dict) or set(criteria) - {"true", "false"} or not all( |
| v is None or isinstance(v, str) for v in criteria.values()): |
| raise jev_error("Invalid request.") |
| true_c, false_c = criteria.get("true"), criteria.get("false") |
| return Question(qid, "noul", text, true_criterion=true_c, false_criterion=false_c) |
| if kind == "choice": |
| if not isinstance(criteria, dict) or not criteria: |
| raise jev_error("Invalid request.") |
| if len(criteria) > MAX_CHOICES: |
| raise jev_error(f"Too many choices. Must have at most {MAX_CHOICES} choices.", |
| plain=True) |
| if any(not k.strip() or (d is not None and not isinstance(d, str)) |
| for k, d in criteria.items()): |
| raise jev_error("Invalid request.") |
| return Question(qid, "choice", text, options=tuple(criteria), |
| descriptions=tuple(criteria.values())) |
| if kind == "score": |
| if not isinstance(criteria, list) or not criteria: |
| raise jev_error("Invalid request.") |
| if len(criteria) > MAX_LEVELS: |
| raise jev_error(f"Too many score levels. Must have at most {MAX_LEVELS} levels.", |
| plain=True) |
| if any(not isinstance(c, str) or not c.strip() for c in criteria): |
| raise jev_error("Invalid request.") |
| return Question(qid, "score", text, descriptions=(None,) * len(criteria), |
| labels=tuple(criteria)) |
| errors: list[str] = [] |
| extra = _constraint_errors(kind, raw, lambda k, m: errors.append(f"{qid}.{k}: {m}")) |
| if errors: |
| raise jev_error("; ".join(errors)) |
| return Question(qid, kind, text, extra=extra) |
|
|
|
|
| def parse_systemone(payload: Any, accepted_models: frozenset[str] = MODEL_ALIASES |
| | JEV_COMPAT_MODELS) -> Job: |
| """Validate a SystemOne request (live Jev semantics). Raises :class:`ApiError`.""" |
| if not isinstance(payload, dict): |
| raise jev_validation_error([{"type": "model_attributes_type", "loc": ["body"], |
| "msg": "Input should be a valid dictionary or object to " |
| "extract fields from", "input": payload}]) |
| missing = [{"type": "missing", "loc": ["body", k], "msg": "Field required", "input": payload} |
| for k in ("model", "state", "questions") if k not in payload] |
| if missing: |
| raise jev_validation_error(missing) |
| questions = payload["questions"] |
| if not isinstance(questions, dict): |
| raise jev_validation_error([{"type": "dict_type", "loc": ["body", "questions"], |
| "msg": "Input should be a valid dictionary", |
| "input": questions}]) |
| if not questions: |
| raise jev_validation_error([{ |
| "type": "too_short", "loc": ["body", "questions"], |
| "msg": "Dictionary should have at least 1 item after validation, not 0", |
| "input": {}, "ctx": {"field_type": "Dictionary", "min_length": 1, |
| "actual_length": 0}}]) |
| if set(payload) - {"model", "state", "questions", "options"}: |
| raise jev_error("Invalid request.") |
| model = payload["model"] |
| if not isinstance(model, str) or model not in accepted_models: |
| raise jev_error(f"Unknown model: {model}") |
| if not _jev_structured(payload["state"]): |
| raise jev_error("Invalid request.") |
| parsed = [_jev_question(qid, raw) for qid, raw in questions.items()] |
| if any(not qid for qid in questions): |
| raise jev_error("Invalid request.") |
| errors: list[str] = [] |
| options = _request_options(payload.get("options"), lambda k, m: errors.append(f"{k}: {m}")) |
| if errors: |
| raise jev_error("; ".join(errors)) |
| job, schema_errors = build_job(payload["state"], parsed, options) |
| if schema_errors: |
| raise jev_error("; ".join(f"{loc}: {msg}" for loc, msg in schema_errors)) |
| return job |
|
|
|
|
| def systemone_response(job: Job, response: dict[str, Any] | None, |
| model: str = MODEL_ID) -> dict[str, Any]: |
| answers: dict[str, Any] = {} |
| for q, o in zip(job.questions, outcomes(job, response), strict=True): |
| if o.kind == "noul": |
| answers[q.key] = {"type": "noul", "noul": o.probability} |
| elif o.kind == "choice": |
| answers[q.key] = {"type": "choice", "choice": o.value, "confidence": o.confidence, |
| "probabilities": dict(zip(q.options, o.probabilities))} |
| elif o.kind == "score": |
| probs = list(o.probabilities) |
| answers[q.key] = {"type": "score", "score": _expected(probs), |
| "confidence": o.confidence, |
| "legend": {str(i): lab for i, lab in enumerate(q.labels)}, |
| "probabilities": {str(i): p for i, p in enumerate(probs)}} |
| else: |
| answers[q.key] = _extension_answer(o) |
| inp, out = _usage_tokens(response) |
| return {"model": model, "answers": answers, |
| "usage": {"input_tokens": inp, "output_tokens": out}} |
|
|
|
|
