| """Python SDK for a deg server (stdlib only). |
| |
| ```python |
| from deg.client import Deg |
| |
| client = Deg("http://127.0.0.1:8790") # base_url; "/v1" is appended if missing |
| decision = client.decisions.create( # OpenAI Decisions API signature |
| model="deg-latest", |
| input="Order AB-1234 arrived broken. Please refund me.", |
| questions=[ |
| {"type": "choice", "name": "event", "instructions": "Which event is this?", |
| "choices": [{"value": "refund_request"}, {"value": "shipping_issue"}]}, |
| {"type": "extract", "name": "order_id", "instructions": "The order id."}, |
| {"type": "generate", "name": "note", "instructions": "One-line note for the agent.", |
| "max_length": 200}, |
| ], |
| ) |
| decision.answers[0].choice, decision.answers[1].value, decision.answers[2].text |
| |
| client.systemone(state="...", questions={"urgent": {"type": "noul", "instructions": "..."}}) |
| client.fill(context="...", schema={...}) # deg /v1/fill |
| ``` |
| |
| Responses are :class:`Record` objects: plain ``dict`` subclasses whose keys are also |
| attributes (``decision.answers[0].probability``), so they serialize as the wire JSON. |
| HTTP errors raise :class:`DegAPIError` with ``status`` and the decoded ``body``. |
| |
| The stock ``openai`` SDK (>= 3.26) also works against the same server: |
| ``OpenAI(base_url="http://127.0.0.1:8790/v1", api_key="unused").decisions.create(...)``. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| import os |
| import urllib.error |
| import urllib.request |
| from collections.abc import Iterable, Mapping |
| from typing import Any |
|
|
| DEFAULT_BASE_URL = "http://127.0.0.1:8790/v1" |
| DEFAULT_MODEL = "deg-latest" |
|
|
|
|
| class DegAPIError(RuntimeError): |
| def __init__(self, status: int, body: Any) -> None: |
| super().__init__(f"deg HTTP {status}: {json.dumps(body)[:500]}") |
| self.status = status |
| self.body = body |
|
|
|
|
| class Record(dict): |
| """A response object: a dict whose keys are also attributes (recursively).""" |
|
|
| def __getattr__(self, name: str) -> Any: |
| try: |
| return self[name] |
| except KeyError: |
| raise AttributeError(name) from None |
|
|
| @classmethod |
| def wrap(cls, value: Any) -> Any: |
| if isinstance(value, dict): |
| return cls({k: cls.wrap(v) for k, v in value.items()}) |
| if isinstance(value, list): |
| return [cls.wrap(v) for v in value] |
| return value |
|
|
|
|
| class Decisions: |
| def __init__(self, client: Deg) -> None: |
| self._client = client |
|
|
| def create(self, *, input: str | Iterable[Mapping[str, Any]], |
| questions: Iterable[Mapping[str, Any]], model: str = DEFAULT_MODEL, |
| safety_identifier: str | None = None, options: Mapping[str, Any] | None = None, |
| timeout: float | None = None) -> Record: |
| """``POST /v1/decisions``, the OpenAI ``client.decisions.create`` signature (plus |
| deg's request ``options``).""" |
| body: dict[str, Any] = {"model": model, |
| "input": input if isinstance(input, str) else list(input), |
| "questions": list(questions)} |
| if safety_identifier is not None: |
| body["safety_identifier"] = safety_identifier |
| if options is not None: |
| body["options"] = dict(options) |
| return self._client._post("/decisions", body, timeout) |
|
|
|
|
| class Deg: |
| def __init__(self, base_url: str | None = None, *, api_key: str | None = None, |
| timeout: float = 300.0) -> None: |
| url = (base_url or os.environ.get("DEG_BASE_URL") or DEFAULT_BASE_URL).rstrip("/") |
| self.base_url = url if url.endswith("/v1") else url + "/v1" |
| self._api_key = api_key if api_key is not None else os.environ.get("DEG_API_KEY") |
| self.timeout = timeout |
| self.decisions = Decisions(self) |
|
|
| def __repr__(self) -> str: |
| return f"Deg({self.base_url!r})" |
|
|
| def systemone(self, state: Any, questions: Mapping[str, Mapping[str, Any]], *, |
| model: str = DEFAULT_MODEL, options: Mapping[str, Any] | None = None, |
| timeout: float | None = None) -> Record: |
| """``POST /v1/systemone`` (TypeSafe Jev request: state + questions by id).""" |
| body: dict[str, Any] = {"model": model, "state": state, "questions": dict(questions)} |
| if options is not None: |
| body["options"] = dict(options) |
| return self._post("/systemone", body, timeout) |
|
|
| def fill(self, context: Any, schema: Mapping[str, Any], *, model: str = DEFAULT_MODEL, |
| options: Mapping[str, Any] | None = None, timeout: float | None = None) -> Record: |
| """``POST /v1/fill``: ``{model, value, meta, usage, plan}`` (SPEC.md §8).""" |
| body: dict[str, Any] = {"context": context, "schema": dict(schema), "model": model} |
| if options is not None: |
| body["options"] = dict(options) |
| return self._post("/fill", body, timeout) |
|
|
| def models(self) -> Record: |
| return self._request("GET", "/models", None, None) |
|
|
| def health(self) -> Record: |
| root = self.base_url[: -len("/v1")] |
| return self._request("GET", "/health", None, None, base=root) |
|
|
| def _post(self, path: str, body: dict[str, Any], timeout: float | None) -> Record: |
| return self._request("POST", path, body, timeout) |
|
|
| def _request(self, method: str, path: str, body: Any, timeout: float | None, |
| base: str | None = None) -> Record: |
| headers = {"Content-Type": "application/json", "Accept": "application/json"} |
| if self._api_key: |
| headers["Authorization"] = f"Bearer {self._api_key}" |
| data = None if body is None else json.dumps(body, ensure_ascii=False).encode("utf-8") |
| req = urllib.request.Request((base or self.base_url) + path, data=data, |
| headers=headers, method=method) |
| try: |
| with urllib.request.urlopen(req, timeout=timeout or self.timeout) as resp: |
| return Record.wrap(json.loads(resp.read())) |
| except urllib.error.HTTPError as exc: |
| raw = exc.read().decode("utf-8", "replace") |
| try: |
| err: Any = json.loads(raw) |
| except json.JSONDecodeError: |
| err = raw |
| raise DegAPIError(exc.code, err) from None |
|
|
