Pluto-1 private release candidate: dual-encoder ensemble, calibrated Typed Decisions specialist
2778af7 verified | """Typed decision schema shared by training, inference, serving and evaluation.""" | |
| from __future__ import annotations | |
| import json | |
| from dataclasses import dataclass, field | |
| from typing import Any, Iterable | |
| KINDS = ("choice", "noul", "score") | |
| KIND_TO_ID = {k: i for i, k in enumerate(KINDS)} | |
| MAX_OPTIONS = 192 | |
| class SchemaError(ValueError): | |
| """Raised when a decision request is malformed.""" | |
| class Question: | |
| name: str | |
| kind: str | |
| instructions: str | |
| keys: list[str] | |
| descriptions: list[str] = field(default_factory=list) | |
| def n_options(self) -> int: | |
| return len(self.keys) | |
| def _coerce_criteria(kind: str, criteria: Any) -> tuple[list[str], list[str]]: | |
| if kind == "noul": | |
| return ["false", "true"], ["false", "true"] | |
| if criteria is None: | |
| raise SchemaError(f"{kind!r} questions require 'criteria'") | |
| if isinstance(criteria, dict): | |
| keys = [str(k) for k in criteria.keys()] | |
| descs = [str(v) for v in criteria.values()] | |
| if len(keys) < 2: | |
| raise SchemaError("a question needs at least two options") | |
| return keys, descs | |
| if isinstance(criteria, (list, tuple)): | |
| if len(criteria) < 2: | |
| raise SchemaError("a question needs at least two options") | |
| labels = [str(x) for x in criteria] | |
| if kind == "choice": | |
| return labels, labels | |
| return [str(i) for i in range(len(labels))], labels | |
| raise SchemaError("'criteria' must be a mapping or a list") | |
| def parse_question(name: str, spec: Any) -> Question: | |
| if not isinstance(spec, dict): | |
| raise SchemaError(f"question {name!r} must be an object") | |
| kind = str(spec.get("type", "")).lower() | |
| if kind not in KINDS: | |
| raise SchemaError(f"question {name!r} has unknown type {kind!r}") | |
| instructions = str(spec.get("instructions") or spec.get("question") or "").strip() | |
| if not instructions: | |
| raise SchemaError(f"question {name!r} is missing 'instructions'") | |
| keys, descs = _coerce_criteria(kind, spec.get("criteria")) | |
| if kind == "score" and not 2 <= len(keys) <= 12: | |
| raise SchemaError("score questions support 2-12 ordered levels") | |
| if len(keys) > MAX_OPTIONS: | |
| raise SchemaError(f"too many options ({len(keys)} > {MAX_OPTIONS})") | |
| return Question(str(name), kind, instructions, keys, descs) | |
| def parse_questions(raw: Any) -> list[Question]: | |
| if not isinstance(raw, dict) or not raw: | |
| raise SchemaError("'questions' must be a non-empty object keyed by question name") | |
| return [parse_question(k, v) for k, v in raw.items()] | |
| def normalize_state(state: Any) -> str: | |
| if isinstance(state, str): | |
| return state.strip() | |
| if isinstance(state, (dict, list)): | |
| return json.dumps(state, ensure_ascii=False, indent=2) | |
| raise SchemaError("'state' must be a string, object or array") | |
| def load_questions(path: str) -> Any: | |
| with open(path, "r", encoding="utf-8") as f: | |
| return json.load(f) | |