File size: 6,891 Bytes
f8de5bd e95d5bf f8de5bd e95d5bf f8de5bd | 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 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 | """TypeSafe-compatible request/response shapes (POST /v1/systemone) mapped onto the single pointer primitive.
Noul -> 2 options [false, true]; answer = p(true)
Choice -> options 'name' or 'name: desc'; answer = argmax, probabilities by name, confidence
Score -> options = ordered level descriptions; answer = expected level, legend, probabilities by index
"""
import json
import re
from datetime import datetime
from typing import Any, Literal, Union
from pydantic import BaseModel, Field, model_validator
JSONContent = Union[str, dict, list, int, float, bool, None]
MAX_OPTIONS = 255
class Noul(BaseModel):
type: Literal["noul"]
instructions: JSONContent = None
criteria: dict[str, JSONContent] | None = None
class Choice(BaseModel):
type: Literal["choice"]
instructions: JSONContent = None
criteria: dict[str, JSONContent]
@model_validator(mode="after")
def _check(self):
if not 1 <= len(self.criteria) <= MAX_OPTIONS: raise ValueError(f"criteria must have 1..{MAX_OPTIONS} options")
return self
class Score(BaseModel):
type: Literal["score"]
instructions: JSONContent = None
criteria: list[JSONContent] = Field(min_length=1, max_length=MAX_OPTIONS)
Question = Union[Noul, Choice, Score]
class SystemOneRequest(BaseModel):
state: JSONContent
model: str = "drex-dlm"
questions: dict[str, Question] = Field(min_length=1)
def render(v: JSONContent, indent: int = 0) -> str:
"""Flatten str | object | array into text the model sees. Field names are kept as labels."""
pad = " " * indent
if v is None: return ""
if isinstance(v, (str, int, float, bool)): return str(v)
if isinstance(v, list): return "\n".join(f"{pad}- {render(x, indent + 1).lstrip()}" for x in v)
return "\n".join(f"{pad}{k}:\n{render(x, indent + 1)}" if isinstance(x, (dict, list)) else f"{pad}{k}: {render(x)}" for k, x in v.items())
def option_text(name: str, desc: JSONContent) -> str:
return name if desc is None or desc == "" else f"{name}: {render(desc)}"
MONTHS = "January|February|March|April|May|June|July|August|September|October|November|December"
_DATE = re.compile(rf"\b(?:{MONTHS}) \d{{1,2}}, \d{{4}}\b|\b\d{{4}}-\d{{2}}-\d{{2}}\b")
def date_facts(text: str) -> str:
"""Deterministic date arithmetic for the model: every pair of absolute dates found in `text`, as one sentence each
("August 3, 2026 is 12 days after July 22, 2026."). The model cannot subtract dates reliably (issue #8); it can use a
stated day count. Returns "" when fewer than two dates are found. Dates are listed in order of first appearance."""
found = []
for m in _DATE.finditer(text):
raw = m.group(0)
try: d = datetime.strptime(raw, "%B %d, %Y") if "," in raw else datetime.strptime(raw, "%Y-%m-%d")
except ValueError: continue
if raw not in [r for r, _ in found]: found.append((raw, d))
facts = []
for i in range(len(found)):
for j in range(i + 1, len(found)):
n = (found[j][1] - found[i][1]).days
facts.append(f"{found[j][0]} is {abs(n)} day{'s' if abs(n) != 1 else ''} {'after' if n > 0 else 'before'} {found[i][0]}." if n else f"{found[j][0]} is the same day as {found[i][0]}.")
return " ".join(facts)
def with_date_facts(state):
"""State with a `date_facts` field (object states) or an appended paragraph (string states) when two or more absolute
dates appear. Opt-in preprocessing (KEV_DATE_FACTS=1 in kev.serve, --date_facts in kev.benchmark)."""
facts = date_facts(render(state))
if not facts: return state
if isinstance(state, dict): return {**state, "date_facts": facts}
if isinstance(state, list): return state + [{"date_facts": facts}]
return f"{state}\n\ndate_facts: {facts}"
def question_keys(qtype: str, criteria) -> list[str]:
"""The keys a question's probabilities are reported under, in option order: the criteria names (choice),
["false", "true"] (noul), the level indices as strings (score). Labels, targets and anchors use the same keys."""
if qtype == "choice": return list(criteria)
if qtype == "noul": return ["false", "true"]
return [str(i) for i in range(len(criteria))]
def to_record(req: SystemOneRequest):
"""-> internal record for encode(), plus per-question metadata ({"id", "type", "keys", "legend" for score}) to map
probabilities back."""
qs, meta = [], []
for qid, q in req.questions.items():
m = {"id": qid, "type": q.type, "keys": question_keys(q.type, q.criteria)}
if q.type == "noul":
c = q.criteria or {}
opts = [option_text("no", c.get("false")), option_text("yes", c.get("true"))]
elif q.type == "choice":
opts = [option_text(k, v) for k, v in q.criteria.items()]
else:
opts = [render(x) for x in q.criteria]
m["legend"] = opts
qs.append({"instr": render(q.instructions), "options": opts, "label": 0}); meta.append(m)
return {"state": render(req.state), "questions": qs}, meta
def choice_confidence(p: list[float]) -> float:
K = len(p)
return 1.0 if K == 1 else (max(p) - 1 / K) / (1 - 1 / K)
def score_confidence(p: list[float]) -> float:
"""Approximation of TypeSafe's 'distance from the modal level' statistic (exact formula unpublished):
1 - E|level - mode| / (L - 1)."""
L = len(p); mode = max(range(L), key=lambda i: p[i])
return 1.0 if L == 1 else 1.0 - sum(pi * abs(i - mode) for i, pi in enumerate(p)) / (L - 1)
def round_prob(x: float) -> float:
"""Serialization precision for probabilities and derived scalars. 4 decimals keeps the sum of a rounded distribution
within TypeSafe's tolerance (|sum - 1| < 0.02) at the 255-option maximum: 255 * 0.00005 < 0.02."""
return round(float(x), 4)
def to_answers(probs: list[list[float]], meta: list[dict]) -> dict[str, Any]:
out = {}
for p, m in zip(probs, meta):
if m["type"] == "noul":
out[m["id"]] = {"type": "noul", "noul": round_prob(p[1])}
elif m["type"] == "choice":
dist = {k: round_prob(v) for k, v in zip(m["keys"], p)}
out[m["id"]] = {"type": "choice", "choice": m["keys"][max(range(len(p)), key=lambda i: p[i])], "confidence": round_prob(choice_confidence(p)), "probabilities": dist}
else:
score = sum(i * pi for i, pi in enumerate(p))
out[m["id"]] = {"type": "score", "score": round_prob(score), "legend": m["legend"], "probabilities": {str(i): round_prob(v) for i, v in enumerate(p)}, "confidence": round_prob(score_confidence(p))}
return out
def output_tokens(tok, answers: dict) -> int:
"""Billing-style figure: tokens of the serialised answers. Not a measure of generation (there is none)."""
return len(tok(json.dumps(answers), add_special_tokens=False).input_ids)
|