| """A local HTTP server for a DecisionNode Edge model, on the standard library only. |
| |
| python -m decisionnode_edge serve path/to/model [--host 127.0.0.1] [--port 8000] [--api-key FILE] [--backend ...] |
| |
| POST /v1/systemone the /v1/systemone format used by public decision-model leaderboards: |
| {"model", "state", "questions", "assets"}; noul, choice and score questions; a noul is |
| answered {"type": "noul", "noul": p}, a choice and a score as on /v1/decide |
| POST /v1/decide the DecisionNode API's call: the runtime's own request and answers (EdgeModel.decide) |
| GET /v1/models {"models": [{"name", "description", "release_date"}]} |
| GET /healthz {"status": "ok", "model", "backend"} (no key needed) |
| |
| Pictures: `images` as the runtime takes them, base64 `data` only (a local `path` is refused over HTTP: a caller must |
| never make the server read its files), and on /v1/systemone the `assets` map of that format, {"<ref>": {"type": "image", |
| "format": "png" | "jpg" | "jpeg", "data_base64": ...}}, each asset a picture whose id is its reference (the state names |
| it), in the map's order. |
| |
| Errors carry a `detail` body: {"detail": {"error_type", "message"}} (a 422 as a list of field errors). A |
| request over the model's length limit says "maximum context length", a choice over 255 options "options per choice", |
| the words leaderboard harnesses look for to count a request as beyond the model rather than as a failure. |
| |
| The model answers one request at a time (a lock); the server listens on the loopback address unless told otherwise. |
| `model` is required on /v1/systemone, as on the hosted API, and any name is accepted: the answer names the model that |
| answered. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import hmac |
| import json |
| import logging |
| import sys |
| import threading |
| import time |
| import traceback |
| from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer |
| from typing import Any |
|
|
| from .request import RequestError |
|
|
| MAX_BODY_BYTES = 64 << 20 |
| SYSTEMONE_TYPES = ("noul", "choice", "score") |
| ASSET_MEDIA_TYPES = {"png": "image/png", "jpg": "image/jpeg", "jpeg": "image/jpeg"} |
| AUTH_MESSAGE = "Cannot authenticate with the server. Please check your API key and try again." |
| ROUTES = {"/v1/systemone": "POST", "/v1/decide": "POST", "/v1/models": "GET", "/healthz": "GET"} |
|
|
| log = logging.getLogger("decisionnode_edge.server") |
|
|
|
|
| class HTTPError(Exception): |
| def __init__(self, status: int, error_type: str, message: str, headers: dict | None = None): |
| super().__init__(message) |
| self.status, self.error_type, self.message, self.headers = status, error_type, message, headers or {} |
|
|
|
|
| def error_body(status: int, error_type: str, message: str) -> dict: |
| if status == 422: |
| return {"detail": [{"type": "value_error", "loc": ["body"], "msg": message, "input": None}]} |
| return {"detail": {"error_type": error_type, "message": message}} |
|
|
|
|
| def _images_over_http(images: Any) -> list: |
| if images is None: |
| return [] |
| if not isinstance(images, list): |
| raise RequestError("images must be a list", 422) |
| for image in images: |
| if isinstance(image, dict) and image.get("path") is not None: |
| raise RequestError("Over HTTP a picture is sent as base64 `data` (or an asset), never as a local `path`.") |
| return list(images) |
|
|
|
|
| def _asset_images(assets: Any) -> list[dict]: |
| """The /v1/systemone `assets` map as the runtime's images: the reference is the picture's id.""" |
| if assets is None: |
| return [] |
| if not isinstance(assets, dict): |
| raise RequestError("assets must be an object of references", 422) |
| out = [] |
| for ref, asset in assets.items(): |
| if not isinstance(asset, dict) or asset.get("type", "image") not in ("image", None): |
| raise RequestError(f"Asset {ref!r}: only image assets are taken.") |
| media_type = ASSET_MEDIA_TYPES.get(str(asset.get("format", "")).lower()) |
| if media_type is None: |
| raise RequestError(f"Asset {ref!r}: format must be one of png, jpg, jpeg.") |
| if not isinstance(asset.get("data_base64"), str) or not asset["data_base64"]: |
| raise RequestError(f"Asset {ref!r} needs data_base64.", 422) |
| out.append({"id": ref, "data": asset["data_base64"], "media_type": media_type}) |
| return out |
|
|
|
|
| class Service: |
| """The routes, without the socket: handle(method, path, headers, body) gives (status, JSON body, extra headers).""" |
|
|
| def __init__(self, model: Any, api_key: str | None = None, description: str | None = None, release_date: str | None = None): |
| self.model = model |
| self.api_key = api_key |
| self.name = model.config["name"] |
| self.description = description or model.config.get("description") or f"{self.name}: typed, calibrated decisions, run locally" |
| self.release_date = release_date or model.config.get("release_date") |
| self.lock = threading.Lock() |
|
|
| def handle(self, method: str, path: str, headers: dict[str, str], body: bytes) -> tuple[int, dict, dict]: |
| try: |
| return 200, self._route(method, path.split("?", 1)[0], headers, body), {} |
| except HTTPError as error: |
| return error.status, error_body(error.status, error.error_type, error.message), error.headers |
| except RequestError as error: |
| return error.status, error_body(error.status, error.error_type, str(error)), {} |
| except Exception: |
| log.error("the model could not answer:\n%s", traceback.format_exc()) |
| return 500, error_body(500, "internal_error", "The model could not answer this request."), {} |
|
|
| def _route(self, method: str, path: str, headers: dict[str, str], body: bytes) -> dict: |
| if path not in ROUTES: |
| raise HTTPError(404, "not_found_error", "Not found.") |
| if method != ROUTES[path]: |
| raise HTTPError(405, "api_usage_error", "Method not allowed.", {"Allow": ROUTES[path]}) |
| if path == "/healthz": |
| return {"status": "ok", "model": self.name, "backend": self.model.backend} |
| self._authenticate(headers) |
| if path == "/v1/models": |
| return {"models": [{"name": self.name, "description": self.description, "release_date": self.release_date}]} |
| try: |
| request = json.loads(body) |
| except (UnicodeDecodeError, json.JSONDecodeError) as error: |
| raise RequestError(f"the body is not JSON: {error}", 422) from None |
| if path == "/v1/decide": |
| if isinstance(request, dict) and request.get("images") is not None: |
| request = {**request, "images": _images_over_http(request["images"])} |
| with self.lock: |
| return self.model.decide(request) |
| return self._systemone(request) |
|
|
| def _authenticate(self, headers: dict[str, str]) -> None: |
| if self.api_key is None: |
| return |
| given = headers.get("authorization", "") |
| scheme, _, token = given.partition(" ") |
| if scheme.lower() != "bearer" or not hmac.compare_digest(token.strip().encode(), self.api_key.encode()): |
| raise HTTPError(401, "authentication_error", AUTH_MESSAGE, {"WWW-Authenticate": "Bearer"}) |
|
|
| def _systemone(self, request: Any) -> dict: |
| if not isinstance(request, dict): |
| raise RequestError("the request must be a JSON object", 422) |
| if not isinstance(request.get("model"), str): |
| raise RequestError("model: Field required, a string", 422) |
| questions = request.get("questions") |
| if isinstance(questions, dict): |
| for qid, question in questions.items(): |
| kind = question.get("type") if isinstance(question, dict) else None |
| if kind == "number": |
| raise RequestError(f"question {qid!r}: the number type is not available on /v1/systemone; send it to /v1/decide") |
| if isinstance(question, dict) and "type" in question and kind not in SYSTEMONE_TYPES: |
| raise RequestError(f"question {qid!r}: unknown question type {kind!r}") |
| images = _images_over_http(request.get("images")) + _asset_images(request.get("assets")) |
| body = {"state": request.get("state"), "questions": questions} |
| if images: |
| body["images"] = images |
| with self.lock: |
| response = self.model.decide(body) |
| answers = {} |
| for qid, answer in response["answers"].items(): |
| answers[qid] = {"type": "noul", "noul": answer["truth"]} if answer["type"] == "truth" else answer |
| return {"model": response["model"], "answers": answers, "usage": response["usage"]} |
|
|
|
|
| class Handler(BaseHTTPRequestHandler): |
| protocol_version = "HTTP/1.1" |
| disable_nagle_algorithm = True |
| server_version = "decisionnode-edge" |
| service: Service = None |
|
|
| def _answer(self, method: str) -> None: |
| started = time.perf_counter() |
| body = b"" |
| if method == "POST": |
| length = self.headers.get("Content-Length") |
| if length is None or not length.isdigit(): |
| return self._send(411, error_body(411, "api_usage_error", "Content-Length is required."), {"Connection": "close"}, started) |
| if int(length) > MAX_BODY_BYTES: |
| return self._send(413, error_body(413, "api_usage_error", "Request body too large."), {"Connection": "close"}, started) |
| body = self.rfile.read(int(length)) |
| headers = {k.lower(): v for k, v in self.headers.items()} |
| status, payload, extra = self.service.handle(method, self.path, headers, body) |
| self._send(status, payload, extra, started) |
|
|
| def _send(self, status: int, payload: dict, extra: dict, started: float) -> None: |
| data = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8") |
| self.send_response(status) |
| self.send_header("Content-Type", "application/json") |
| self.send_header("Content-Length", str(len(data))) |
| self.send_header("X-Latency-Ms", f"{(time.perf_counter() - started) * 1000:.2f}") |
| for name, value in extra.items(): |
| self.send_header(name, value) |
| self.end_headers() |
| self.wfile.write(data) |
| if extra.get("Connection") == "close": |
| self.close_connection = True |
|
|
| def do_GET(self) -> None: |
| self._answer("GET") |
|
|
| def do_POST(self) -> None: |
| self._answer("POST") |
|
|
| def do_PUT(self) -> None: |
| self._answer("PUT") |
|
|
| def do_DELETE(self) -> None: |
| self._answer("DELETE") |
|
|
| def log_message(self, format: str, *args: Any) -> None: |
| log.info("%s %s", self.address_string(), format % args) |
|
|
|
|
| def make_server(model: Any, host: str = "127.0.0.1", port: int = 8000, api_key: str | None = None, **service: Any) -> ThreadingHTTPServer: |
| handler = type("EdgeHandler", (Handler,), {"service": Service(model, api_key=api_key, **service)}) |
| server = ThreadingHTTPServer((host, port), handler) |
| server.daemon_threads = True |
| return server |
|
|
|
|
| def main(argv: list[str] | None = None) -> int: |
| from . import EdgeModel |
|
|
| parser = argparse.ArgumentParser(prog="python -m decisionnode_edge serve", description="Serve a DecisionNode Edge model over HTTP: /v1/systemone, /v1/decide, /v1/models.") |
| parser.add_argument("model", help="the model directory") |
| parser.add_argument("--host", default="127.0.0.1", help="the address to listen on (default 127.0.0.1, this machine only)") |
| parser.add_argument("--port", type=int, default=8000) |
| parser.add_argument("--api-key", default=None, help="a file holding the key callers must send as `Authorization: Bearer <key>`") |
| parser.add_argument("--device", default=None) |
| parser.add_argument("--dtype", default=None) |
| parser.add_argument("--backend", default=None, help="auto, torch, vllm, openvino, mlx or llamacpp (default: auto)") |
| parser.add_argument("--artifacts", default=None, help="the directory python -m decisionnode_edge.build wrote") |
| parser.add_argument("--description", default=None, help="the description GET /v1/models gives") |
| parser.add_argument("--release-date", default=None, help="the release date GET /v1/models gives (YYYY-MM-DD)") |
| args = parser.parse_args(argv) |
| logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(message)s") |
| api_key = None |
| if args.api_key: |
| with open(args.api_key, encoding="utf-8") as f: |
| api_key = f.read().strip() |
| if not api_key: |
| raise SystemExit(f"{args.api_key} is empty") |
| model = EdgeModel.load(args.model, device=args.device, dtype=args.dtype, backend=args.backend, artifacts=args.artifacts) |
| model.decide({"state": "The color is red.", "questions": {"warmup": {"type": "choice", "instructions": "Which color is named?", "criteria": {"red": "red", "blue": "blue"}}}}) |
| server = make_server(model, args.host, args.port, api_key, description=args.description, release_date=args.release_date) |
| log.info("serving %s (%s backend) on http://%s:%d%s", model.config["name"], model.backend, args.host, server.server_address[1], ", key required" if api_key else "") |
| try: |
| server.serve_forever() |
| except KeyboardInterrupt: |
| pass |
| finally: |
| server.server_close() |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| sys.exit(main()) |
|
|
