Skip to content

Backends

Protocols

Backend

Bases: Protocol

What SystemOne needs from a backend.

Source code in system_one/backends/__init__.py 
26
27
28
29
30
31
32
33
34
class Backend(Protocol):
    """What `SystemOne` needs from a backend."""

    model: str
    """The model to ask when neither the call nor `SYSTEM_ONE_MODEL` names one."""

    def ask(self, request: SystemOneInput) -> SystemOneOutput: ...

    def close(self) -> None: ...

model instance-attribute 

model: str

The model to ask when neither the call nor SYSTEM_ONE_MODEL names one.

ask 

ask(request: SystemOneInput) -> SystemOneOutput
Source code in system_one/backends/__init__.py 
32
def ask(self, request: SystemOneInput) -> SystemOneOutput: ...

close 

close() -> None
Source code in system_one/backends/__init__.py 
34
def close(self) -> None: ...

AsyncBackend

Bases: Protocol

The same two methods as Backend, awaited.

Source code in system_one/backends/__init__.py 
37
38
39
40
41
42
43
44
class AsyncBackend(Protocol):
    """The same two methods as `Backend`, awaited."""

    model: str

    async def ask(self, request: SystemOneInput) -> SystemOneOutput: ...

    async def close(self) -> None: ...

model instance-attribute 

model: str

ask async 

ask(request: SystemOneInput) -> SystemOneOutput
Source code in system_one/backends/__init__.py 
42
async def ask(self, request: SystemOneInput) -> SystemOneOutput: ...

close async 

close() -> None
Source code in system_one/backends/__init__.py 
44
async def close(self) -> None: ...

create_backend 

create_backend(
    settings: Settings, config: BackendConfig | None = None
) -> Backend

The sync backend named by settings.backend.

Source code in system_one/backends/__init__.py 
73
74
75
76
77
78
79
80
81
def create_backend(settings: Settings, config: BackendConfig | None = None) -> Backend:
    """The sync backend named by `settings.backend`."""
    if settings.backend == "onnx":
        with _hint("onnx"):
            from system_one.backends.onnx import ONNXBackend
        return ONNXBackend(_resolve(settings, config, ONNXConfig))
    with _hint("http"):
        from system_one.backends.http import HTTPBackend
    return HTTPBackend(_resolve(settings, config, HTTPConfig))

create_async_backend 

create_async_backend(
    settings: Settings, config: BackendConfig | None = None
) -> AsyncBackend

The async backend named by settings.backend.

Source code in system_one/backends/__init__.py 
84
85
86
87
88
89
90
91
92
93
94
def create_async_backend(
    settings: Settings, config: BackendConfig | None = None
) -> AsyncBackend:
    """The async backend named by `settings.backend`."""
    if settings.backend == "onnx":
        with _hint("onnx"):
            from system_one.backends.onnx import AsyncONNXBackend
        return AsyncONNXBackend(_resolve(settings, config, ONNXConfig))
    with _hint("http"):
        from system_one.backends.http import AsyncHTTPBackend
    return AsyncHTTPBackend(_resolve(settings, config, HTTPConfig))

HTTP

HTTPBackend

Bases: BaseHTTPBackend

Talks to any vendor implementing the System One contract over HTTP.

Source code in system_one/backends/http.py 
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
148
class HTTPBackend(BaseHTTPBackend):
    """Talks to any vendor implementing the System One contract over HTTP."""

    def __init__(self, config: HTTPConfig, *, transport: Any = None) -> None:
        super().__init__(config, transport=transport)
        self._client = httpx2.Client(timeout=config.timeout, transport=transport)

    def ask(self, request: SystemOneInput) -> SystemOneOutput:
        response = self._send(self._ask_request(request))
        return SystemOneOutput.model_validate_json(response.content)

    def close(self) -> None:
        self._client.close()

    def _send(self, request: httpx2.Request) -> httpx2.Response:
        attempt = 0
        while True:
            try:
                return self._attempt(request)
            except SystemOneError as exc:
                delay = self._delay_or_reraise(exc, attempt)
                attempt += 1
            time.sleep(delay)

    def _attempt(self, request: httpx2.Request) -> httpx2.Response:
        try:
            response = self._client.send(request)
        except httpx2.HTTPError as exc:
            raise transport_error(request, exc) from exc
        if response.is_success:
            return response
        raise api_error(response)

ask 

ask(request: SystemOneInput) -> SystemOneOutput
Source code in system_one/backends/http.py 
124
125
126
def ask(self, request: SystemOneInput) -> SystemOneOutput:
    response = self._send(self._ask_request(request))
    return SystemOneOutput.model_validate_json(response.content)

close 

close() -> None
Source code in system_one/backends/http.py 
128
129
def close(self) -> None:
    self._client.close()

AsyncHTTPBackend

Bases: BaseHTTPBackend

The async counterpart of HTTPBackend, sharing its config and retry policy.

Source code in system_one/backends/http.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
class AsyncHTTPBackend(BaseHTTPBackend):
    """The async counterpart of `HTTPBackend`, sharing its config and retry policy."""

    def __init__(self, config: HTTPConfig, *, transport: Any = None) -> None:
        super().__init__(config, transport=transport)
        self._client = httpx2.AsyncClient(timeout=config.timeout, transport=transport)

    async def ask(self, request: SystemOneInput) -> SystemOneOutput:
        response = await self._send(self._ask_request(request))
        return SystemOneOutput.model_validate_json(response.content)

    async def close(self) -> None:
        await self._client.aclose()

    async def _send(self, request: httpx2.Request) -> httpx2.Response:
        attempt = 0
        while True:
            try:
                return await self._attempt(request)
            except SystemOneError as exc:
                delay = self._delay_or_reraise(exc, attempt)
                attempt += 1
            await asyncio.sleep(delay)

    async def _attempt(self, request: httpx2.Request) -> httpx2.Response:
        try:
            response = await self._client.send(request)
        except httpx2.HTTPError as exc:
            raise transport_error(request, exc) from exc
        if response.is_success:
            return response
        raise api_error(response)

ask async 

ask(request: SystemOneInput) -> SystemOneOutput
Source code in system_one/backends/http.py 
158
159
160
async def ask(self, request: SystemOneInput) -> SystemOneOutput:
    response = await self._send(self._ask_request(request))
    return SystemOneOutput.model_validate_json(response.content)

close async 

close() -> None
Source code in system_one/backends/http.py 
162
163
async def close(self) -> None:
    await self._client.aclose()

ONNX

ONNXBackend

Runs the exported graph in-process. No torch, no transformers, no network.

Source code in system_one/backends/onnx.py 
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
class ONNXBackend:
    """Runs the exported graph in-process. No torch, no transformers, no network."""

    def __init__(self, config: ONNXConfig) -> None:
        self.config = config
        self.model = config.model
        load_model(config.onnx_dir, self.model)

    def ask(self, request: SystemOneInput) -> SystemOneOutput:
        model = load_model(self.config.onnx_dir, request.model)
        items = build_items(model.tokenizer, model.config, request, model.max_options)
        logits = model.session.run(None, collate(items, model.pad_id))[0]
        return postprocess(logits, items, request, model.config)

    def close(self) -> None:
        """Sessions are shared through `load_model`, so there is nothing to release."""

config instance-attribute 

config = config

model instance-attribute 

model = config.model

ask 

ask(request: SystemOneInput) -> SystemOneOutput
Source code in system_one/backends/onnx.py 
389
390
391
392
393
def ask(self, request: SystemOneInput) -> SystemOneOutput:
    model = load_model(self.config.onnx_dir, request.model)
    items = build_items(model.tokenizer, model.config, request, model.max_options)
    logits = model.session.run(None, collate(items, model.pad_id))[0]
    return postprocess(logits, items, request, model.config)

close 

close() -> None

Sessions are shared through load_model, so there is nothing to release.

Source code in system_one/backends/onnx.py 
395
396
def close(self) -> None:
    """Sessions are shared through `load_model`, so there is nothing to release."""

AsyncONNXBackend

The sync backend on a worker thread — ONNX Runtime releases the GIL anyway.

Source code in system_one/backends/onnx.py 
399
400
401
402
403
404
405
406
407
408
409
410
class AsyncONNXBackend:
    """The sync backend on a worker thread — ONNX Runtime releases the GIL anyway."""

    def __init__(self, config: ONNXConfig) -> None:
        self.backend = ONNXBackend(config)
        self.model = self.backend.model

    async def ask(self, request: SystemOneInput) -> SystemOneOutput:
        return await asyncio.to_thread(self.backend.ask, request)

    async def close(self) -> None:
        """Sessions are shared through `load_model`, so there is nothing to release."""

backend instance-attribute 

backend = ONNXBackend(config)

model instance-attribute 

model = self.backend.model

ask async 

ask(request: SystemOneInput) -> SystemOneOutput
Source code in system_one/backends/onnx.py 
406
407
async def ask(self, request: SystemOneInput) -> SystemOneOutput:
    return await asyncio.to_thread(self.backend.ask, request)

close async 

close() -> None

Sessions are shared through load_model, so there is nothing to release.

Source code in system_one/backends/onnx.py 
409
410
async def close(self) -> None:
    """Sessions are shared through `load_model`, so there is nothing to release."""