File size: 3,001 Bytes
2778af7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 | """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."""
@dataclass
class Question:
name: str
kind: str
instructions: str
keys: list[str]
descriptions: list[str] = field(default_factory=list)
@property
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)
|