File size: 6,375 Bytes
16d760d aff03ff 16d760d aff03ff 16d760d aff03ff 16d760d aff03ff 16d760d aff03ff 16d760d aff03ff 16d760d aff03ff 16d760d aff03ff 16d760d aff03ff 16d760d aff03ff 16d760d aff03ff 16d760d aff03ff 16d760d aff03ff 16d760d aff03ff 16d760d aff03ff 16d760d | 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 | """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]], # noqa: A002 -- OpenAI name
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: # never show the key
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
|