File size: 3,478 Bytes
da1fe36 | 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 | """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 pictures alone. A question is a dict in the Decision
Index's schema (`type`, `instructions`, `criteria`) or one of `prompt`'s classes.
"""
from __future__ import annotations
from collections.abc import Mapping, Sequence
from typing import Any
from .prompt import Choice, Noul, Question, as_question
def answer(q: Question, probs: Sequence[float]) -> dict:
"""A noul's P(yes); a choice's pick and its probabilities; a score's expected level."""
if isinstance(q, Noul):
return {"type": "noul", "noul": probs[0]}
best = max(range(len(probs)), key=probs.__getitem__)
if isinstance(q, Choice):
names = list(q.criteria)
return {"type": "choice", "choice": names[best], "confidence": probs[best],
"probabilities": dict(zip(names, probs))}
return {"type": "score", "score": sum(i * p for i, p in enumerate(probs)), "confidence": probs[best],
"probabilities": {str(i): p for i, p in enumerate(probs)},
"legend": {str(i): text for i, text in enumerate(q.criteria)}}
class SystemOneApi:
"""`system_one` and `system_one_batch`, and the plain probabilities under them, over a model's
`run(requests)`: each `(state, questions, images)` request's probabilities (in its questions' order)
and the tokens it read."""
def run(self, requests: Sequence[tuple[Any, list[Question], Sequence]]) -> list[tuple[list[list[float]], int]]:
raise NotImplementedError
def probabilities(self, state: Any, questions: Sequence, images: Sequence | None = None) -> list[list[float]]:
"""Each question's distribution over its options (`yes`, `no` for a noul), in one pass."""
return self.run([(state, [as_question(q) for q in questions], images or ())])[0][0]
def probabilities_batch(self, requests: Sequence[tuple]) -> list[list[list[float]]]:
"""`probabilities` for many `(state, questions[, images])` requests, packed as `system_one_batch`."""
reqs = [(r[0], [as_question(q) for q in r[1]], (r[2] if len(r) > 2 else None) or ()) for r in requests]
return [probs for probs, _ in self.run(reqs)]
def system_one(self, state: Any, questions: Mapping[str, Any], images: Sequence | None = None) -> dict:
"""Named questions over one state, and its pictures if any, in one forward pass."""
return self.system_one_batch([(state, questions, images)])[0]
def system_one_batch(self, requests: Sequence[tuple]) -> list[dict]:
"""Many `(state, questions)` or `(state, questions, images)` requests. Single-question requests
are packed together with no padding; a request with several questions reads its state once."""
named = [(r[0], {n: as_question(q) for n, q in r[1].items()}, (r[2] if len(r) > 2 else None) or ())
for r in requests]
done = self.run([(state, list(qs.values()), images) for state, qs, images in named])
return [{"answers": {n: answer(q, p) for (n, q), p in zip(qs.items(), probs)},
"usage": {"input_tokens": read, "output_tokens": 0}}
for (_, qs, _), (probs, read) in zip(named, done)]
|