DecisionNode Edge Mini 0.9.2: point and box questions, the /v1/systemone route as hosted, a box calibration
0e77dfd verified | """POST /v1/systemone: the request and answer format public decision-model benchmarks use, read onto the runtime's own. | |
| translate(body) the format's pictures onto the runtime's, before the parse: | |
| - `assets`: {reference: {"type": "image", "format": "png" | "jpg" | "jpeg", "data_base64"}}. | |
| Every place a reference appears (as a whole string value or inside a longer string, in the | |
| state, a question's instructions or its criteria; never in an object key) is the picture | |
| "in its place". The model reads every picture of a request before the state, labelled | |
| `Picture <n> (<id>):`, so each reference is replaced in the text by `Picture <n>` and the | |
| asset becomes picture n. A picture is read once however many times it is referenced, and | |
| every question sees every picture. | |
| - chat content parts, `{"type": "image_url", "image_url": {"url": "data:..."}}` (or | |
| `"image_url": "data:..."`), anywhere in the state, the instructions or the criteria: the | |
| same, the part replaced by `Picture <n>`; an array made only of such parts and | |
| `{"type": "text", "text"}` parts is read as its text, the parts joined by a line break. | |
| - `images` as a list of data URLs. | |
| - `box` questions: in the picture the instructions name (`image_for_box`) when the request | |
| holds several of the format's pictures; answered in fractions (`normalised_boxes`, after | |
| the checks). A box with `max` keeps the runtime's list answer. | |
| The legend of a score whose levels held a reference echoes the levels as sent. | |
| box_answer(answer) a one-box answer in the format's shape: {"type": "box", "box": {"top", "bottom", "left", | |
| "right"}, "confidence"}, the edges as fractions of the picture's height and width, and | |
| `confidence` the probability that the box is right where the model folder has a box | |
| calibration (BoxCalibration), else `present`. | |
| mark_capacity(...) the words benchmark harnesses look for in an error ("maximum context length", "options per | |
| choice", "a score takes 2 to 10 levels"), so that a limit of the model is recorded as one. | |
| Choice and score probabilities on this route are normalised first and rounded to lists that sum to exactly 1.00 | |
| (answers.round_probabilities with exact_sum). Headers the format defines for pictures and long inputs are accepted and | |
| never required. A multipart body (`part` assets) is not taken: an asset with `part` is a 400 that says to send | |
| `data_base64`. | |
| """ | |
| from __future__ import annotations | |
| import base64 | |
| import binascii | |
| import hashlib | |
| import json | |
| import math | |
| import re | |
| from dataclasses import dataclass, field | |
| from typing import Any, Iterable | |
| from .answers import round2 | |
| from .request import IMAGE_MEDIA_TYPES, MAX_LEVELS, MAX_OPTIONS, RequestError | |
| ASSET_FORMATS = {"png": "image/png", "jpg": "image/jpeg", "jpeg": "image/jpeg"} # the format's Asset.format | |
| MAX_REFERENCE_BYTES = 1024 # one reference, UTF-8 | |
| MAX_REFERENCES_BYTES = 65536 # every reference of a request together | |
| MAX_NESTING = 128 # a body nested deeper is left as it is (the parse refuses it) | |
| PICTURE = "Picture {n}" # what the text reads in a reference's place: the model's own label of picture n | |
| BOX_CALIBRATION_FILE = "box_calibration.json" # in the model folder, optional | |
| BOX_CALIBRATION_FORMAT = "dn-box-calibration-1" | |
| BOX_CALIBRATION_FEATURES = ["logit_present", "logit_confidence"] | |
| _WHITESPACE = re.compile(r"[\t\n\x0b\x0c\r ]") | |
| _BACKTICKED = re.compile(r"`([^`\n]{1,512})`") | |
| _PATH_PART = re.compile(r"([^.\[\]]+)|\[(\d+)\]") | |
| class Translation: | |
| """What translate() did to a body, for the parse and the answer.""" | |
| images: list[dict] = field(default_factory=list) # the image objects it added, in picture order | |
| legends: dict[str, list] = field(default_factory=dict) # score question id -> its levels as sent | |
| boxes: list[str] = field(default_factory=list) # the box questions, answered in the format's shape | |
| references: int = 0 # the references replaced in the text | |
| # --------------------------------------------------------------------------- | |
| # Errors | |
| # --------------------------------------------------------------------------- | |
| def field_error(kind: str, loc: Iterable[Any], msg: str, value: Any, **extra: Any) -> dict: | |
| return {"type": kind, "loc": list(loc), "msg": msg, "input": value, **extra} | |
| def _invalid(entry: dict) -> RequestError: | |
| return RequestError(f"{'.'.join(str(part) for part in entry['loc'])}: {entry['msg']}", 422, errors=[entry]) | |
| def _asset_error(ref: str, problem: str) -> RequestError: | |
| shown = ref if len(ref) <= 80 else ref[:77] + "..." | |
| return RequestError(f"Asset {shown!r}: {problem}.") | |
| def _too_large(message: str) -> RequestError: | |
| return RequestError(message, 413, "request_too_large_error") | |
| # --------------------------------------------------------------------------- | |
| # Pictures | |
| # --------------------------------------------------------------------------- | |
| def _sniff(text: str) -> str | None: | |
| """The type of a base64 picture from its first bytes (PNG, JPEG, WebP), or None.""" | |
| head = text[:64] | |
| try: | |
| blob = base64.b64decode(head[: len(head) - len(head) % 4]) | |
| except (binascii.Error, ValueError): | |
| return None | |
| if blob.startswith(b"\x89PNG\r\n\x1a\n"): | |
| return "image/png" | |
| if blob.startswith(b"\xff\xd8\xff"): | |
| return "image/jpeg" | |
| if blob[:4] == b"RIFF" and blob[8:12] == b"WEBP": | |
| return "image/webp" | |
| return None | |
| def _clean_base64(text: str) -> str: | |
| """Base64 text as the decoder reads it: a `data:` prefix and any ASCII whitespace dropped, missing padding added.""" | |
| if text[:5].lower() == "data:": | |
| text = text.partition(",")[2] | |
| if _WHITESPACE.search(text): | |
| text = _WHITESPACE.sub("", text) | |
| if len(text) % 4 in (2, 3) and "=" not in text: | |
| text += "=" * (4 - len(text) % 4) | |
| return text | |
| def _asset_image(ref: str, asset: Any) -> dict: | |
| """The runtime's image object (without its id) for one asset. Raises RequestError.""" | |
| if asset.get("type") is not None and asset.get("type") != "image": | |
| raise _invalid(field_error("literal_error", ("body", "assets", ref, "type"), "Input should be 'image'", asset.get("type"), ctx={"expected": "'image'"})) | |
| if asset.get("part") is not None: | |
| raise _asset_error(ref, "multipart image parts are not taken by this server; send the image base64-encoded in data_base64 in a JSON body") | |
| data = asset.get("data_base64") | |
| if data is None: | |
| raise _asset_error(ref, "data_base64 is required") | |
| if not isinstance(data, str): | |
| raise _invalid(field_error("string_type", ("body", "assets", ref, "data_base64"), "Input should be a valid string", data)) | |
| fmt = asset.get("format") | |
| declared = None | |
| if fmt is not None: | |
| if not isinstance(fmt, str) or fmt not in ASSET_FORMATS: | |
| raise _asset_error(ref, "format must be png, jpg or jpeg") | |
| declared = ASSET_FORMATS[fmt] | |
| data = _clean_base64(data) | |
| if not data: | |
| raise _asset_error(ref, "data_base64 is empty") | |
| media = _sniff(data) or declared # the format is read from the bytes | |
| if media is None: | |
| raise _asset_error(ref, "the data is not a PNG or JPEG image") | |
| return {"media_type": media, "data": data} | |
| def _part_image(part: dict) -> dict: | |
| """The runtime's image object (without its id) for an `image_url` content part. Raises RequestError.""" | |
| url = part.get("image_url") | |
| if isinstance(url, dict): | |
| url = url.get("url") | |
| if not isinstance(url, str) or not url: | |
| raise RequestError("An image_url part needs a url: a data:image/...;base64 URL.") | |
| scheme = url[:8].lower() | |
| if scheme.startswith("data:"): | |
| head, comma, data = url.partition(",") | |
| fields = head[5:].split(";") | |
| if not comma or len(fields) < 2 or fields[-1].strip().lower() != "base64": | |
| raise RequestError("An image_url data URL must be data:<media type>;base64,<data>.") | |
| data = _clean_base64(data) | |
| declared = fields[0].strip().lower() | |
| declared = "image/jpeg" if declared == "image/jpg" else declared | |
| media = _sniff(data) or (declared if declared in IMAGE_MEDIA_TYPES else None) | |
| if media is None: | |
| raise RequestError("An image_url part must hold a PNG, JPEG or WebP image.") | |
| return {"media_type": media, "data": data} | |
| if scheme == "https://": | |
| raise RequestError("Image URLs are not fetched on this deployment: send the image as a base64 data URL.") | |
| raise RequestError("An image_url part must be a base64 data URL.") | |
| def _is_image_part(value: Any) -> bool: | |
| return isinstance(value, dict) and value.get("type") == "image_url" and "image_url" in value | |
| def _is_text_part(value: Any) -> bool: | |
| return isinstance(value, dict) and value.get("type") == "text" and isinstance(value.get("text"), str) and set(value) <= {"type", "text"} | |
| # --------------------------------------------------------------------------- | |
| # The walk | |
| # --------------------------------------------------------------------------- | |
| class _Walk: | |
| """Replaces references and image parts in the free-form parts of a request, numbering the pictures as it meets them.""" | |
| def __init__(self, refs: Iterable[str], first: int): | |
| self.refs = list(refs) | |
| self.first = first # the picture number of the first picture this walk adds, less one (the body's own images) | |
| self.order: list[tuple[str, Any]] = [] # ("ref", reference) or ("part", image object), in picture order | |
| self.by_ref: dict[str, int] = {} | |
| self.by_url: dict[str, int] = {} | |
| self.replaced = 0 | |
| ordered = sorted(self.refs, key=len, reverse=True) # at one position the longest reference wins | |
| self.pattern = re.compile("|".join(re.escape(ref) for ref in ordered)) if ordered else None | |
| self.whole = set(self.refs) | |
| def _number_of_ref(self, ref: str) -> int: | |
| if ref not in self.by_ref: | |
| self.order.append(("ref", ref)) | |
| self.by_ref[ref] = self.first + len(self.order) | |
| return self.by_ref[ref] | |
| def _number_of_part(self, part: dict) -> int: | |
| image = _part_image(part) | |
| key = image.get("data") or "" | |
| if key not in self.by_url: | |
| self.order.append(("part", image)) | |
| self.by_url[key] = self.first + len(self.order) | |
| return self.by_url[key] | |
| def text(self, value: str) -> str: | |
| if self.pattern is None: | |
| return value | |
| if value in self.whole: # a whole value first | |
| self.replaced += 1 | |
| return PICTURE.format(n=self._number_of_ref(value)) | |
| if not any(ref in value for ref in self.refs): | |
| return value | |
| def one(match: "re.Match[str]") -> str: | |
| self.replaced += 1 | |
| return PICTURE.format(n=self._number_of_ref(match.group(0))) | |
| return self.pattern.sub(one, value) | |
| def value(self, value: Any) -> Any: | |
| """`value` with every reference and image part replaced (a new object where anything changed).""" | |
| if isinstance(value, str): | |
| return self.text(value) | |
| if _is_image_part(value): | |
| self.replaced += 1 | |
| return PICTURE.format(n=self._number_of_part(value)) | |
| if isinstance(value, list): | |
| if value and any(_is_image_part(item) for item in value) and all(_is_image_part(item) or _is_text_part(item) for item in value): | |
| # a content-part array: its text, the pictures in their places | |
| return "\n".join(self.value(item) if _is_image_part(item) else self.text(item["text"]) for item in value) | |
| out = [self.value(item) for item in value] | |
| return out if any(a is not b for a, b in zip(out, value)) else value | |
| if isinstance(value, dict): | |
| out = {key: self.value(item) for key, item in value.items()} # keys stay text | |
| return out if any(out[key] is not value[key] for key in value) else value | |
| return value | |
| def _question(walk: _Walk, question: Any) -> Any: | |
| """A question with its instructions and criteria walked (its type and every other field as sent).""" | |
| if not isinstance(question, dict): | |
| return question | |
| out = dict(question) | |
| if "instructions" in out: | |
| out["instructions"] = walk.value(out["instructions"]) | |
| criteria = out.get("criteria") | |
| if isinstance(criteria, dict): | |
| out["criteria"] = {key: walk.value(item) for key, item in criteria.items()} # an option's name is a key: never replaced | |
| elif isinstance(criteria, list): | |
| out["criteria"] = [walk.value(item) for item in criteria] | |
| return out | |
| # --------------------------------------------------------------------------- | |
| # Box questions | |
| # --------------------------------------------------------------------------- | |
| def _resolve(state: Any, path: str) -> Any: | |
| """The value a backticked path names in the state (`photo`, `frames[2].frame`, `faces.happy`), or None.""" | |
| value = state | |
| for name, index in _PATH_PART.findall(path.strip()): | |
| if name: | |
| if not isinstance(value, dict) or name not in value: | |
| return None | |
| value = value[name] | |
| else: | |
| if not isinstance(value, list) or int(index) >= len(value): | |
| return None | |
| value = value[int(index)] | |
| return value | |
| def _first_ref(value: Any, refs: set[str]) -> str | None: | |
| """The first reference held whole by a string inside `value` (depth first, in order).""" | |
| if isinstance(value, str): | |
| return value if value in refs else None | |
| items = value.values() if isinstance(value, dict) else value if isinstance(value, list) else () | |
| for item in items: | |
| found = _first_ref(item, refs) | |
| if found is not None: | |
| return found | |
| return None | |
| def image_for_box(instructions: Any, state: Any, refs: set[str], numbers: dict[str, int], in_state: list[int], own: set[int], count: int) -> int: | |
| """The picture number a box question looks in when the request holds several pictures (the path in backticks names | |
| which one to search): the picture a backticked path of the instructions names in the state (a reference held there, | |
| whole); else the first picture of the state; else the first picture the instructions do not hold themselves (a | |
| picture in a box's instructions shows the target, it is not the picture to search); else picture 1.""" | |
| text = instructions if isinstance(instructions, str) else "" if instructions is None else str(instructions) | |
| for path in _BACKTICKED.findall(text): | |
| found = _first_ref(_resolve(state, path), refs) | |
| if found is not None and found in numbers: | |
| return numbers[found] | |
| if in_state: | |
| return in_state[0] | |
| for n in range(1, count + 1): | |
| if n not in own: | |
| return n | |
| return 1 | |
| def question_refs(question: dict, refs: Iterable[str]) -> set[str]: | |
| """The references a question's own instructions and criteria hold (whole or inside a string).""" | |
| found: set[str] = set() | |
| wanted = list(refs) | |
| def visit(value: Any) -> None: | |
| if isinstance(value, str): | |
| found.update(ref for ref in wanted if ref in value) | |
| elif isinstance(value, dict): | |
| for item in value.values(): | |
| visit(item) | |
| elif isinstance(value, list): | |
| for item in value: | |
| visit(item) | |
| visit(question.get("instructions")) | |
| visit(question.get("criteria")) | |
| return found | |
| # --------------------------------------------------------------------------- | |
| # The request | |
| # --------------------------------------------------------------------------- | |
| def _within_nesting(value: Any, limit: int = MAX_NESTING) -> bool: | |
| """At most `limit` containers inside each other (walked without recursion).""" | |
| stack = [(value, 0)] | |
| while stack: | |
| item, depth = stack.pop() | |
| if isinstance(item, (dict, list)): | |
| if depth + 1 > limit: | |
| return False | |
| stack.extend((child, depth + 1) for child in (item.values() if isinstance(item, dict) else item)) | |
| return True | |
| def _check_assets(assets: Any) -> dict: | |
| if not isinstance(assets, dict): | |
| raise _invalid(field_error("dict_type", ("body", "assets"), "Input should be a valid dictionary", assets)) | |
| if not assets: | |
| raise _invalid(field_error("too_short", ("body", "assets"), "Dictionary should have at least 1 item after validation, not 0", assets, ctx={"field_type": "Dictionary", "min_length": 1, "actual_length": 0})) | |
| total = 0 | |
| for ref, asset in assets.items(): | |
| if not isinstance(asset, dict): | |
| raise _invalid(field_error("model_attributes_type", ("body", "assets", ref), "Input should be a valid dictionary or object to extract fields from", asset)) | |
| if not ref.strip(): | |
| raise RequestError("An asset reference cannot be empty or only whitespace.") | |
| size = len(ref.encode("utf-8", "surrogatepass")) | |
| if size > MAX_REFERENCE_BYTES: | |
| raise _too_large(f"An asset reference is at most {MAX_REFERENCE_BYTES} bytes of UTF-8.") | |
| total += size | |
| if total > MAX_REFERENCES_BYTES: | |
| raise _too_large(f"The asset references of a request are at most {MAX_REFERENCES_BYTES} bytes of UTF-8 together.") | |
| return assets | |
| def _may_hold_parts(value: Any) -> bool: | |
| """Cheap: whether a value may hold an image_url part (any dict with that type somewhere), walked without recursion.""" | |
| stack = [value] | |
| while stack: | |
| item = stack.pop() | |
| if isinstance(item, dict): | |
| if item.get("type") == "image_url": | |
| return True | |
| stack.extend(child for child in item.values() if isinstance(child, (dict, list))) | |
| elif isinstance(item, list): | |
| stack.extend(child for child in item if isinstance(child, (dict, list))) | |
| return False | |
| def translate(body: Any, *, max_images: int) -> tuple[Any, Translation]: | |
| """A /v1/systemone body with the format's pictures on the runtime's (module docstring), and what was done. A body | |
| that is not an object with a question map is returned as it is (the parse refuses it). Raises RequestError for an | |
| asset or an image part the runtime cannot take.""" | |
| done = Translation() | |
| if not isinstance(body, dict) or not isinstance(body.get("questions"), dict): | |
| return body, done | |
| assets = body.get("assets") | |
| has_assets = assets is not None | |
| questions = body["questions"] | |
| has_parts = _may_hold_parts(body.get("state")) or any(_may_hold_parts(q.get("instructions")) or _may_hold_parts(q.get("criteria")) for q in questions.values() if isinstance(q, dict)) | |
| boxes = [qid for qid, q in questions.items() if isinstance(q, dict) and q.get("type") == "box"] | |
| url_images = isinstance(body.get("images"), list) and any(isinstance(image, str) for image in body["images"]) | |
| if not has_assets and not has_parts and not boxes and not url_images: | |
| if "assets" in body: # `"assets": null` equals no assets | |
| body = {name: value for name, value in body.items() if name != "assets"} | |
| return body, done | |
| if not _within_nesting(body): | |
| return body, done # the parse refuses it as it refuses any body that deep | |
| out = {name: value for name, value in body.items() if name != "assets"} | |
| refs: list[str] = list(_check_assets(assets)) if has_assets else [] | |
| if len(refs) > max_images: | |
| raise RequestError(f"Too many images: {len(refs)}, the limit is {max_images}.") | |
| own_images = body.get("images") if isinstance(body.get("images"), list) else [] | |
| if url_images: | |
| # `images` as a list of data URLs: each an image object, in the list's order | |
| if len(own_images) > max_images: | |
| raise RequestError(f"Too many images: {len(own_images)}, the limit is {max_images}.") | |
| listed = [] | |
| for i, image in enumerate(own_images): | |
| if isinstance(image, str): | |
| image = {"id": "image" if len(own_images) == 1 else f"image_{i + 1}", **_part_image({"image_url": image})} | |
| listed.append(image) | |
| own_images = listed | |
| out["images"] = listed | |
| walk = _Walk(refs, len(own_images)) | |
| state = body.get("state") | |
| out["state"] = walk.value(state) | |
| in_state = sorted(set(walk.by_ref.values()) | set(walk.by_url.values())) | |
| walked: dict[str, Any] = {} | |
| for qid, question in questions.items(): | |
| walked[qid] = _question(walk, question) | |
| if isinstance(question, dict) and question.get("type") == "score" and walked[qid].get("criteria") != question.get("criteria"): | |
| done.legends[str(qid)] = question.get("criteria") # the legend echoes the levels as sent | |
| unused = [ref for ref in refs if ref not in walk.by_ref] | |
| if unused: | |
| raise _asset_error(unused[0], "it is not referenced in the state, the instructions or the criteria") | |
| added = len(walk.order) | |
| total = len(own_images) + added | |
| if total > max_images: | |
| raise RequestError(f"Too many images: {total}, the limit is {max_images}.") | |
| taken = {image.get("id") for image in own_images if isinstance(image, dict)} | |
| images = [] | |
| for i, (kind, item) in enumerate(walk.order): | |
| n = len(own_images) + i + 1 | |
| image_id = "image" if total == 1 else f"image_{n}" | |
| while image_id in taken: | |
| image_id += "_" | |
| taken.add(image_id) | |
| image = _asset_image(item, assets[item]) if kind == "ref" else dict(item) | |
| images.append({"id": image_id, **image}) | |
| if images: | |
| out["images"] = list(own_images) + images | |
| done.images = images | |
| done.references = walk.replaced | |
| ids = [image.get("id") for image in out.get("images") or [] if isinstance(image, dict)] | |
| numbers = dict(walk.by_ref) | |
| for qid in boxes: | |
| # the picture is chosen only for the format's pictures (assets, image parts, data URLs), which no `image` field | |
| # can name; with image objects alone a box without `image` among several stays the 422 of the runtime's parse | |
| if walked[qid].get("image") is None and len(ids) > 1 and (added or url_images): | |
| question = dict(walked[qid]) | |
| own = {numbers[ref] for ref in question_refs(questions[qid], refs) if ref in numbers} | |
| n = image_for_box(questions[qid].get("instructions"), state, set(refs), numbers, in_state, own, len(ids)) | |
| question["image"] = ids[n - 1] | |
| walked[qid] = question | |
| done.boxes.append(str(qid)) | |
| out["questions"] = walked | |
| return out, done | |
| def normalised_boxes(specs: Iterable[Any]) -> None: | |
| """The box questions of a /v1/systemone request answered in fractions of the picture (the format's edges), set on the | |
| parsed specs (places.PlaceSpec) after their checks. A box with `max` keeps the units the caller sent.""" | |
| for spec in specs: | |
| if spec.shape == "box" and spec.max is None: | |
| spec.units = "normalized" | |
| # --------------------------------------------------------------------------- | |
| # The answer | |
| # --------------------------------------------------------------------------- | |
| class BoxCalibrationError(ValueError): | |
| """A box calibration file the runtime does not take (the server does not start on it).""" | |
| class BoxCalibration: | |
| """The `confidence` of a box on /v1/systemone: the probability that the box is right (an intersection over union of | |
| 0.5 or more with the true box), read from the one-box answer's own two numbers, `present` and `confidence`, as the | |
| answer writes them (2 decimals). A logistic map, fitted per model and kept beside it as box_calibration.json: | |
| {"format": "dn-box-calibration-1", "features": ["logit_present", "logit_confidence"], "clip": 0.005, | |
| "intercept": a, "coef": [b, c], "model": ..., "fit": {...}} | |
| p = sigmoid(a + b logit(present) + c logit(confidence)), each input clipped to [clip, 1 - clip] first, the result | |
| rounded to 2 decimals. Monotone: no coefficient below 0, so neither number can lower the probability by rising.""" | |
| intercept: float | |
| coef: tuple[float, float] | |
| clip: float | |
| sha256: str | |
| model: str = "" | |
| def load(cls, path: str, sha256: str = "") -> "BoxCalibration": | |
| """Read and check a calibration file; `sha256`, when given, must be the file's.""" | |
| try: | |
| with open(path, "rb") as f: | |
| data = f.read() | |
| except OSError as error: | |
| raise BoxCalibrationError(f"box calibration {path}: {error}") from None | |
| digest = hashlib.sha256(data).hexdigest() | |
| if sha256 and sha256.lower() != digest: | |
| raise BoxCalibrationError(f"box calibration {path}: sha256 {digest}, expected {sha256}") | |
| try: | |
| spec = json.loads(data) | |
| except ValueError as error: | |
| raise BoxCalibrationError(f"box calibration {path}: not JSON ({error})") from None | |
| if not isinstance(spec, dict) or spec.get("format") != BOX_CALIBRATION_FORMAT: | |
| raise BoxCalibrationError(f"box calibration {path}: the format is not {BOX_CALIBRATION_FORMAT}") | |
| if spec.get("features") != BOX_CALIBRATION_FEATURES: | |
| raise BoxCalibrationError(f"box calibration {path}: features must be {BOX_CALIBRATION_FEATURES}") | |
| coef, intercept, clip = spec.get("coef"), spec.get("intercept"), spec.get("clip") | |
| numbers = [intercept, clip] + (list(coef) if isinstance(coef, list) else []) | |
| if not isinstance(coef, list) or len(coef) != 2 or any(isinstance(x, bool) or not isinstance(x, (int, float)) or x != x or abs(x) == float("inf") for x in numbers): | |
| raise BoxCalibrationError(f"box calibration {path}: intercept, coef (two numbers) and clip must be finite numbers") | |
| if any(c < 0 for c in coef): | |
| raise BoxCalibrationError(f"box calibration {path}: a coefficient below 0 is not monotone") | |
| if not 0 < clip < 0.5: | |
| raise BoxCalibrationError(f"box calibration {path}: clip must be above 0 and below 0.5") | |
| return cls(float(intercept), (float(coef[0]), float(coef[1])), float(clip), digest, str(spec.get("model") or "")) | |
| def probability(self, present: float, confidence: float) -> float: | |
| """The unrounded probability that the box is right.""" | |
| def logit(p: float) -> float: | |
| p = min(1.0 - self.clip, max(self.clip, float(p))) | |
| return math.log(p / (1.0 - p)) | |
| z = self.intercept + self.coef[0] * logit(present) + self.coef[1] * logit(confidence) | |
| return 1.0 / (1.0 + math.exp(-z)) if z >= 0 else math.exp(z) / (1.0 + math.exp(z)) | |
| def __call__(self, present: float, confidence: float) -> float: | |
| return round2(self.probability(present, confidence)) | |
| def box_answer(answer: dict, calibration: BoxCalibration | None = None) -> dict: | |
| """A one-box answer (normalized units) in the format's shape. `confidence`: the calibrated probability that the box | |
| is right where there is a calibration, else `present`.""" | |
| return { | |
| "type": "box", | |
| "box": {"top": answer["y1"], "bottom": answer["y2"], "left": answer["x1"], "right": answer["x2"]}, | |
| "confidence": calibration(answer["present"], answer["confidence"]) if calibration is not None else answer["present"], | |
| } | |
| def box_answers(body: dict, calibration: BoxCalibration | None = None) -> dict: | |
| """The response body with every one-box answer in the format's shape (a list answer, `max`, stays the runtime's).""" | |
| answers = body.get("answers") | |
| if isinstance(answers, dict): | |
| for qid, answer in answers.items(): | |
| if isinstance(answer, dict) and answer.get("type") == "box" and "x1" in answer: | |
| answers[qid] = box_answer(answer, calibration) | |
| return body | |
| # --------------------------------------------------------------------------- | |
| # Errors with the capacity words | |
| # --------------------------------------------------------------------------- | |
| CONTEXT_MARKER = "maximum context length" | |
| OPTIONS_MARKER = "options per choice" | |
| LEVELS_MARKER = "a score takes 2 to 10 levels" | |
| TOO_MANY_LEVELS = f"Too many score levels. Must have at most {MAX_LEVELS} levels." | |
| def mark_capacity(error: RequestError) -> RequestError: | |
| """A refusal of /v1/systemone for a limit of the model, in the words and bodies benchmark harnesses read: | |
| over the length limit {"detail": {"error_type": "max_tokens_exceeded", "message": "The request does not fit | |
| the model's maximum context length (<the count and the limit>)."}} | |
| more than 255 options {"detail": "Too many choices. Must have at most 255 choices (255 options per choice)."} | |
| more than 10 levels {"detail": "Too many score levels. Must have at most 10 levels (a score takes 2 to 10 | |
| levels; one level is answered without the model)."} | |
| Every other refusal is returned as it is.""" | |
| message = str(error) | |
| if error.status != 400: | |
| return error | |
| if error.error_type == "max_tokens_exceeded": | |
| text = f"The request does not fit the model's {CONTEXT_MARKER} ({message})." | |
| return RequestError(text, 400, error.error_type, detail={"error_type": error.error_type, "message": text}) | |
| if message.startswith("Too many choices"): | |
| text = f"Too many choices. Must have at most {MAX_OPTIONS} choices ({MAX_OPTIONS} {OPTIONS_MARKER})." | |
| return RequestError(text, 400, error.error_type, detail=text) | |
| if message == TOO_MANY_LEVELS: | |
| text = f"Too many score levels. Must have at most {MAX_LEVELS} levels ({LEVELS_MARKER}; one level is answered without the model)." | |
| return RequestError(text, 400, error.error_type, detail=text) | |
| return error | |