nectec/pathumma-crossborder-guardrail

1"""Request / response types.2 3Field names deliberately mirror the TypeSafe `POST /v1/systemone` contract so that4code written against the TypeSafe SDK can be pointed at OpenThai-SystemOne unchanged:5 6    state      : str | dict | list        -- the thing to judge7    questions  : {id: Choice|Score|Noul}  -- typed questions8    answers    : {id: ChoiceAnswer|ScoreAnswer|NoulAnswer}9"""10from __future__ import annotations11 12from typing import Any, Dict, List, Literal, Optional, Union13 14from pydantic import BaseModel, Field, field_validator, model_validator15 16MAX_OPTIONS = 255  # single-stage cardinality limit (slots 0..254); slot 255 = abstain17MIN_SCORE_LEVELS = 218MAX_SCORE_LEVELS = 1019 20 21class Noul(BaseModel):22    """A yes/no question. Returns p(yes)."""23 24    type: Literal["noul"] = "noul"25    instructions: str26    criteria: Optional[Dict[str, Optional[str]]] = None  # {"true": "...", "false": "..."}27 28    @field_validator("criteria")29    @classmethod30    def _check_criteria(cls, v):31        if v is None:32            return v33        extra = set(v) - {"true", "false"}34        if extra:35            raise ValueError(f"noul criteria keys must be 'true'/'false', got {sorted(extra)}")36        return v37 38 39class Choice(BaseModel):40    """Pick one option. `criteria` maps option name -> description (or null)."""41 42    type: Literal["choice"] = "choice"43    instructions: str44    criteria: Dict[str, Optional[str]]45 46    @field_validator("criteria")47    @classmethod48    def _check_criteria(cls, v):49        if len(v) < 1:50            raise ValueError("choice needs at least one option")51        if len(v) > MAX_OPTIONS:52            raise ValueError(f"choice supports at most {MAX_OPTIONS} options in one stage")53        for k in v:54            if not str(k).strip():55                raise ValueError("option names must be non-empty")56        return v57 58 59class Score(BaseModel):60    """Rate the state against ordered levels; `criteria[i]` describes level i (low -> high)."""61 62    type: Literal["score"] = "score"63    instructions: str64    criteria: List[str]65 66    @field_validator("criteria")67    @classmethod68    def _check_criteria(cls, v):69        if not (MIN_SCORE_LEVELS <= len(v) <= MAX_SCORE_LEVELS):70            raise ValueError(f"score needs {MIN_SCORE_LEVELS}..{MAX_SCORE_LEVELS} levels")71        return v72 73 74Question = Union[Noul, Choice, Score]75 76 77class NoulAnswer(BaseModel):78    type: Literal["noul"] = "noul"79    noul: float80 81 82class ChoiceAnswer(BaseModel):83    type: Literal["choice"] = "choice"84    choice: str85    probabilities: Dict[str, float]86    confidence: float87    abstain: Optional[float] = None  # OpenThai extension: p(none of the options); not in TypeSafe88 89 90class ScoreAnswer(BaseModel):91    type: Literal["score"] = "score"92    score: float93    legend: Dict[int, str]94    probabilities: Dict[str, float]95    confidence: float96 97 98Answer = Union[NoulAnswer, ChoiceAnswer, ScoreAnswer]99 100 101class Usage(BaseModel):102    input_tokens: int103    output_tokens: int = 0104    permutations: int = 1  # OpenThai extension: number of option orders averaged (order-invariant mode)105 106 107class SystemOneRequest(BaseModel):108    state: Union[str, Dict[str, Any], List[Any]]109    model: str = "openthai-systemone"110    questions: Dict[str, Question] = Field(discriminator=None)111    # OpenThai extensions. order_invariant=True averages the answer over several option orders (removes position112    # bias, ~2x latency); None = automatic (on for choice questions with > 10 options); permutations overrides the count.113    order_invariant: Optional[bool] = None114    permutations: Optional[int] = Field(default=None, ge=1, le=32)115 116    @model_validator(mode="after")117    def _non_empty(self):118        if not self.questions:119            raise ValueError("at least one question is required")120        return self121 122 123class SystemOneResponse(BaseModel):124    model: str125    answers: Dict[str, Answer]126    usage: Usage127 128 129def parse_question(obj: Union[Question, Dict[str, Any]]) -> Question:130    """Accept a dict (raw JSON) or an already-typed question."""131    if isinstance(obj, (Noul, Choice, Score)):132        return obj133    t = obj.get("type")134    if t == "noul":135        return Noul(**obj)136    if t == "choice":137        return Choice(**obj)138    if t == "score":139        return Score(**obj)140    raise ValueError(f"unknown question type: {t!r}")141