| """The System One API: named questions over a state in, typed answers out. |
| |
| model.system_one(state, {"refund": {"type": "noul", "instructions": "Is this a refund request?"}}) |
| # {"answers": {"refund": {"type": "noul", "noul": p}}, "usage": {"input_tokens": n, "output_tokens": 0}} |
| |
| A state is a string, any JSON value, or None with media alone; `media` holds its pictures, audio clips and videos |
| (`media`). A question is a dict in the Decision Index's schema (`type`, `instructions`, `criteria`; spoken |
| instructions a media dict) or one of `prompt`'s classes. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from collections.abc import Mapping, Sequence |
| from typing import Any |
|
|
| from .media import Media, media_list |
| from .prompt import ( |
| DEFAULT_STATE_STYLE, |
| Box, |
| Boxes, |
| Choice, |
| Estimate, |
| Noul, |
| Question, |
| Read, |
| as_question, |
| check_question, |
| state_text, |
| typed, |
| ) |
|
|
| |
| |
| |
| |
| SCORE_READOUTS = ("mean", "median", "argmax") |
| SCORE_READOUT = "median" |
| |
| |
| |
| |
| IDK_TEXTS = frozenset({"i don't know", "i do not know", "i don't know the answer", "i do not know the answer"}) |
|
|
|
|
| def abstain_option(criteria: Mapping[str, str | None] | None) -> str | None: |
| """The name of a choice's explicit "I don't know" option (the English `IDK_TEXTS`), or None.""" |
| if not isinstance(criteria, Mapping): |
| return None |
| for name, text in criteria.items(): |
| if str(text or name).strip().lower().replace("\u2019", "'").rstrip(".!") in IDK_TEXTS: |
| return name |
| return None |
|
|
|
|
| def choice_index(probs: Sequence[float], abstain: int | None = None, abstain_below: float | None = None) -> int: |
| """A choice's answer: the likeliest option; with `abstain` (the "I don't know" option's index) and |
| `abstain_below`, that option when the best other option holds less than `abstain_below` of the other options' |
| mass.""" |
| best = max(range(len(probs)), key=probs.__getitem__) |
| if abstain is None or abstain_below is None: |
| return best |
| others = [p for i, p in enumerate(probs) if i != abstain] |
| total = sum(others) |
| return abstain if total <= 0 or max(others) < abstain_below * total else best |
|
|
|
|
| def score_level(probs: Sequence[float], readout: str = SCORE_READOUT) -> int: |
| """A score question's answer as a level index under `readout` (`mean`: rounded, clamped to the levels).""" |
| total = sum(probs) or 1.0 |
| if readout == "mean": |
| return max(0, min(len(probs) - 1, int(round(sum(i * p / total for i, p in enumerate(probs)))))) |
| if readout == "median": |
| mass = 0.0 |
| for i, p in enumerate(probs): |
| mass += p / total |
| if mass >= 0.5: |
| return i |
| return len(probs) - 1 |
| if readout == "argmax": |
| return max(range(len(probs)), key=probs.__getitem__) |
| raise ValueError(f"score_readout {readout!r}: one of {SCORE_READOUTS}") |
|
|
|
|
| def answer(q: Question, probs: Sequence[float] | Sequence[Sequence[float]] | Mapping, score_readout: str = SCORE_READOUT, |
| abstain_below: float | None = None, context: str = "", calibration=None, edges: str = "median") -> dict: |
| """A noul's P(yes); a choice's pick (`choice_index`) and its probabilities; a score's level under |
| `score_readout` (`mean`: the expected level, unrounded); a typed answer's probabilities through `calibration`'s |
| tables (a read is found, a box drawn, an option chosen, at 0.5 after them; an estimate's quantiles read at the |
| levels its table gives). `edges`: how a box's edges are read (`box.EDGE_READS`, the checkpoint's `d1_box_edges`).""" |
| through = calibration.through if calibration is not None else lambda name, value: value |
| if isinstance(q, Box): |
| from .box import answer as box_answer |
|
|
| return box_answer(probs, through, edges) |
| if isinstance(q, Boxes): |
| from .box import answer_many |
|
|
| return answer_many(probs, through, edges) |
| if isinstance(q, Estimate): |
| value = q.grid.read(probs, None if calibration is None else lambda share: calibration.level("estimate", share)) |
| return {"type": "number", "number": value["median"], "interval": [value["p10"], value["p90"]], "unit": q.unit} |
| if isinstance(q, Read): |
| from .parsing import parse_date, parse_number, written_unit |
|
|
| found = through(f"found:{q.type}", probs["found"]) |
| present = found >= 0.5 |
| text = probs["text"] if present else None |
| span = probs["span"] if present else None |
| value, unit = text, q.unit |
| if text is not None and q.type == "number": |
| unit = unit or written_unit(text) |
| value = parse_number(text, context, unit=q.unit) |
| elif text is not None and q.type == "date": |
| value, _ = parse_date(text, context) |
| result = {"type": q.type, q.type: value, "span": span, "found": found} |
| if q.type != "text": |
| result["text"] = text |
| if present and "path" in probs: |
| result["path"] = probs["path"] |
| if q.type == "number": |
| result["unit"] = unit |
| return result |
| if isinstance(q, Noul): |
| return {"type": "noul", "noul": probs[0]} |
| if isinstance(q, Choice) and q.multiple: |
| probs = [through("multiple", p) for p in probs] |
| return {"type": "choice", "choices": [name for name, p in zip(q.criteria, probs) if p >= 0.5], |
| "probabilities": dict(zip(q.criteria, probs))} |
| best = max(range(len(probs)), key=probs.__getitem__) |
| if isinstance(q, Choice): |
| names = list(q.criteria) |
| idk = abstain_option(q.criteria) if abstain_below is not None else None |
| pick = choice_index(probs, None if idk is None else names.index(idk), abstain_below) |
| return {"type": "choice", "choice": names[pick], "confidence": probs[pick], |
| "probabilities": dict(zip(names, probs))} |
| level = (sum(i * p for i, p in enumerate(probs)) if score_readout == "mean" |
| else float(score_level(probs, score_readout))) |
| return {"type": "score", "score": level, "confidence": probs[best], |
| "probabilities": {str(i): p for i, p in enumerate(probs)}, |
| "legend": {str(i): text for i, text in enumerate(q.criteria)}} |
|
|
|
|
| def request_media(request: tuple) -> list[Media]: |
| """A `(state, questions)` or `(state, questions, media)` request's media.""" |
| return media_list(request[2] if len(request) > 2 else None) |
|
|
|
|
| class SystemOneApi: |
| """`system_one` and `system_one_batch`, and the plain probabilities under them, over a model's |
| `run(requests)`: each `(state, questions, media)` request's probabilities (in its questions' order) |
| and the tokens it read.""" |
|
|
| serves_typed = True |
|
|
| def ask(self, q: Mapping | Question) -> Question: |
| """A caller's question, refused with what to change, or when this engine cannot serve its type.""" |
| check_question(q, getattr(self, "tokenizer", None)) |
| q = as_question(q) |
| if typed(q) and not self.serves_typed: |
| raise ValueError(f"{type(self).__name__} serves noul, choice and score questions only") |
| return q |
|
|
| def named(self, questions: Mapping[str, Any]) -> dict[str, Question]: |
| """A request's named questions (`ask`), an error naming the question it is about.""" |
| if not isinstance(questions, Mapping): |
| raise ValueError(f'questions are a dict of name to question, {{"refund": {{"type": "noul", "instructions": ' |
| f'"Is it a refund request?"}}}}, not {type(questions).__name__}') |
| out = {} |
| for name, q in questions.items(): |
| try: |
| out[name] = self.ask(q) |
| except ValueError as e: |
| raise ValueError(f"question {name!r}: {e}") from None |
| return out |
|
|
| def answer(self, state: Any, q: Question, probs, score_readout: str = SCORE_READOUT, |
| abstain_below: float | None = None) -> dict: |
| """One question's answer as `system_one` gives it: a read's value parsed in its state, a typed answer's |
| probabilities through the served calibration's tables.""" |
| context = state_text(state, getattr(self, "state_style", DEFAULT_STATE_STYLE)) if isinstance(q, Read) else "" |
| return answer(q, probs, score_readout, abstain_below, context, getattr(self, "calibration", None), |
| getattr(getattr(self, "tokenizer", None), "d1_box_edges", "median")) |
|
|
| def run(self, requests: Sequence[tuple[Any, list[Question], Sequence[Media]]], |
| set_size: int | None = None) -> list[tuple[list[list[float]], int]]: |
| raise NotImplementedError |
|
|
| def probabilities(self, state: Any, questions: Sequence, media: Sequence | None = None) -> list[list[float]]: |
| """Each question's option probabilities; a multiple choice returns independent P(yes) per option.""" |
| return self.run([(state, [self.ask(q) for q in questions], media_list(media))])[0][0] |
|
|
| def system_one(self, state: Any, questions: Mapping[str, Any], media: Sequence | None = None, |
| score_readout: str = SCORE_READOUT, abstain_below: float | None = None, |
| set_size: int | None = None) -> dict: |
| """Named questions over one state, and its media if any (pictures, audio clips, videos), in one forward |
| pass.""" |
| return self.system_one_batch([(state, questions, media)], score_readout, abstain_below, set_size)[0] |
|
|
| def system_one_batch(self, requests: Sequence[tuple], score_readout: str = SCORE_READOUT, |
| abstain_below: float | None = None, set_size: int | None = None) -> list[dict]: |
| """Many `(state, questions)` or `(state, questions, media)` requests in one call, batched as the engine |
| reads them (`run`); a request with several questions reads its state once a pass. |
| `score_readout` (`SCORE_READOUTS`) sets how a score question answers; `abstain_below` (off by default) |
| when a choice with an "I don't know" option picks it (`choice_index`); `set_size` how many questions are |
| read together after the state, for this call (the model's trained size by default).""" |
| if score_readout not in SCORE_READOUTS: |
| raise ValueError(f"score_readout {score_readout!r}: one of {SCORE_READOUTS}") |
| if abstain_below is not None and not 0 < abstain_below <= 1: |
| raise ValueError(f"abstain_below {abstain_below!r}: a share in (0, 1]") |
| named = [(r[0], self.named(r[1]), request_media(r)) for r in requests] |
| done = self.run([(state, list(qs.values()), media) for state, qs, media in named], set_size=set_size) |
| return [{"answers": {n: self.answer(state, q, p, score_readout, abstain_below) for (n, q), p in zip(qs.items(), probs)}, |
| "usage": {"input_tokens": read, "output_tokens": 0}} |
| for (state, qs, _), (probs, read) in zip(named, done)] |
|
|
