Backends

Protocols

Backend

Bases: Protocol

What SystemOne needs from a backend.

Source code in system_one/backends/__init__.py

class Backend(Protocol):
    """What `SystemOne` needs from a backend."""

    model: str
    """The model to ask when the call names none."""

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

model instance-attribute

model: str

The model to ask when the call names none.

ask

ask(request: SystemOneInput) -> SystemOneOutput

Source code in system_one/backends/__init__.py

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

AsyncBackend

Bases: Protocol

The same ask as Backend, awaited.

Source code in system_one/backends/__init__.py

class AsyncBackend(Protocol):
    """The same `ask` as `Backend`, awaited."""

    model: str

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

model instance-attribute

model: str

ask async

ask(request: SystemOneInput) -> SystemOneOutput

Source code in system_one/backends/__init__.py

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

create_backend

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

The sync backend named by settings.backend; overrides patch its config.

Source code in system_one/backends/__init__.py

def create_backend(
    settings: Settings, config: BackendConfig | None = None, **overrides: Any
) -> Backend:
    """The sync backend named by `settings.backend`; `overrides` patch its config."""
    if settings.backend == "stub":
        from system_one.backends.stub import StubBackend

        return StubBackend(_resolve(settings, config, StubConfig, overrides))
    if settings.backend == "onnx":
        with _hint("onnx"):
            from system_one.backends.onnx import ONNXBackend
        return ONNXBackend(_resolve(settings, config, ONNXConfig, overrides))
    with _hint("http"):
        from system_one.backends.http import HTTPBackend
    return HTTPBackend(_resolve(settings, config, HTTPConfig, overrides))

create_async_backend

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

The async backend named by settings.backend; overrides patch its config.

Source code in system_one/backends/__init__.py

def create_async_backend(
    settings: Settings, config: BackendConfig | None = None, **overrides: Any
) -> AsyncBackend:
    """The async backend named by `settings.backend`; `overrides` patch its config."""
    if settings.backend == "stub":
        from system_one.backends.stub import AsyncStubBackend

        return AsyncStubBackend(_resolve(settings, config, StubConfig, overrides))
    if settings.backend == "onnx":
        with _hint("onnx"):
            from system_one.backends.onnx import AsyncONNXBackend
        return AsyncONNXBackend(_resolve(settings, config, ONNXConfig, overrides))
    with _hint("http"):
        from system_one.backends.http import AsyncHTTPBackend
    return AsyncHTTPBackend(_resolve(settings, config, HTTPConfig, overrides))

HTTP

HTTPBackend

Bases: BaseHTTPBackend

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

Source code in system_one/backends/http.py

class HTTPBackend(BaseHTTPBackend):
    """Talks to any vendor implementing the System One contract over HTTP."""

    def __init__(
        self, config: HTTPConfig, *, client: httpx2.Client | None = None
    ) -> None:
        super().__init__(config)
        if client is None:
            client = httpx2.Client(timeout=config.timeout)
        self._client = client

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

    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

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

AsyncHTTPBackend

Bases: BaseHTTPBackend

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

Source code in system_one/backends/http.py

class AsyncHTTPBackend(BaseHTTPBackend):
    """The async counterpart of `HTTPBackend`, sharing its config and retry policy."""

    def __init__(
        self, config: HTTPConfig, *, client: httpx2.AsyncClient | None = None
    ) -> None:
        super().__init__(config)
        if client is None:
            client = httpx2.AsyncClient(timeout=config.timeout)
        self._client = client

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

    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

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

ONNX

ONNXBackend

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

Source code in system_one/backends/onnx.py

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.resolved_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)

config instance-attribute

config = config

model instance-attribute

model = config.resolved_model

ask

ask(request: SystemOneInput) -> SystemOneOutput

Source code in system_one/backends/onnx.py

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)

AsyncONNXBackend

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

Source code in system_one/backends/onnx.py

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)

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

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