| """The three question types, and the record every source is converted into. |
| |
| Every decision is one forward pass that scores a list of candidates. The three |
| types differ only in where the candidates come from: |
| |
| - choice: the caller supplies them. |
| - score: they are the points of a scale (`min`..`max`), in order. |
| - null: they are "yes" and "no"; the answer is P(yes). |
| |
| One head scores every candidate the same way, so what the model learns about |
| "which of these fits the state" transfers across all three. |
| |
| A record, one per JSONL line: |
| |
| {"type": "choice", "state": "...", "question": "...", |
| "options": ["a", "b", "c"], "label": 1, |
| "source": "banking77", "licence": "cc-by-4.0"} |
| |
| {"type": "score", "state": "...", "question": "...", |
| "scale": {"min": 1, "max": 5, "labels": {"1": "terrible", "5": "excellent"}}, |
| "label": 4, ...} # label is the scale value, not an index |
| |
| {"type": "null", "state": "...", "question": "The ticket is urgent.", |
| "label": true, ...} |
| """ |
|
|
| from __future__ import annotations |
|
|
| from dataclasses import dataclass, field |
|
|
| TYPES = ("choice", "score", "null") |
| NULL_OPTIONS = ("yes", "no") |
|
|
|
|
| @dataclass |
| class Decision: |
| type: str |
| state: str |
| question: str |
| options: list[str] = field(default_factory=list) |
| scale: dict | None = None |
| label: object = None |
| source: str = "" |
| licence: str = "" |
|
|
| def candidates(self) -> list[str]: |
| """The strings the head scores, in the order the model sees them.""" |
| if self.type == "choice": |
| return list(self.options) |
| if self.type == "null": |
| return list(NULL_OPTIONS) |
| lo, hi = int(self.scale["min"]), int(self.scale["max"]) |
| names = {str(k): v for k, v in (self.scale.get("labels") or {}).items()} |
| return [f"{v} ({names[str(v)]})" if str(v) in names else str(v) |
| for v in range(lo, hi + 1)] |
|
|
| def scale_values(self) -> list[int]: |
| return list(range(int(self.scale["min"]), int(self.scale["max"]) + 1)) |
|
|
| def target(self) -> int | None: |
| """The index of the right candidate, or None for an unlabelled record.""" |
| if self.label is None: |
| return None |
| if self.type == "choice": |
| return int(self.label) |
| if self.type == "null": |
| return 0 if bool(self.label) else 1 |
| return self.scale_values().index(int(round(float(self.label)))) |
|
|
| def validate(self) -> "Decision": |
| if self.type not in TYPES: |
| raise ValueError(f"type must be one of {TYPES}, not {self.type!r}") |
| if not self.question.strip(): |
| raise ValueError("question is empty") |
| if self.type == "choice": |
| if len(self.options) < 2: |
| raise ValueError("a choice needs at least two options") |
| if len(set(self.options)) != len(self.options): |
| raise ValueError("choice options must be distinct") |
| if self.type == "score": |
| if not self.scale or int(self.scale["max"]) <= int(self.scale["min"]): |
| raise ValueError("a score needs a scale with max > min") |
| if int(self.scale["max"]) - int(self.scale["min"]) > 20: |
| raise ValueError("a scale has at most 21 points") |
| t = self.target() |
| if t is not None and not 0 <= t < len(self.candidates()): |
| raise ValueError(f"label {self.label!r} is outside the candidates") |
| return self |
|
|
| @classmethod |
| def from_dict(cls, d: dict) -> "Decision": |
| return cls(type=d["type"], state=d.get("state", ""), question=d["question"], |
| options=list(d.get("options") or []), scale=d.get("scale"), |
| label=d.get("label"), source=d.get("source", ""), |
| licence=d.get("licence", "")).validate() |
|
|
| def to_dict(self) -> dict: |
| d = {"type": self.type, "state": self.state, "question": self.question, |
| "label": self.label, "source": self.source, "licence": self.licence} |
| if self.type == "choice": |
| d["options"] = self.options |
| if self.type == "score": |
| d["scale"] = self.scale |
| return d |
|
|
