Skip to content

openai_compat_engines

openai_compat_engines

Data-driven registration of OpenAI-compatible inference engines.

Classes

OpenAICompatEngine 

OpenAICompatEngine(
    host: str | None = None,
    *,
    api_key: str | None = None,
    timeout: float = 600.0,
)

Bases: _OpenAICompatibleEngine

Generic engine for an explicitly-provided OpenAI-compatible endpoint.

Deliberately NOT registered in EngineRegistry: it is only ever constructed with an explicit host (e.g. diapason eval --base-url), so registering it would just add a useless localhost discovery probe and interact with the per-test registry wipe.

Source code in src/diapason/engine/_openai_compat.py 
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
def __init__(
    self,
    host: str | None = None,
    *,
    api_key: str | None = None,
    timeout: float = 600.0,
) -> None:
    import os

    # Sanitize the engine id for env-var lookup ("openai-compat" ->
    # "OPENAI_COMPAT_..."); shells cannot set hyphenated variable names.
    env_prefix = self.engine_id.upper().replace("-", "_")
    self._host = (
        host or os.environ.get(f"{env_prefix}_HOST") or self._default_host
    ).rstrip("/")
    # Bearer auth for endpoints started with e.g. ``vllm serve --api-key``.
    # Setting it on the client covers generate/stream/stream_full/
    # list_models/health alike; ``None`` keeps requests header-free.
    self._api_key = api_key or os.environ.get(f"{env_prefix}_API_KEY") or None
    headers = (
        {"Authorization": f"Bearer {self._api_key}"} if self._api_key else None
    )
    # Used by the shared async streaming plumbing (AsyncHTTPEngineMixin) so
    # the bounded request timeout is applied to streaming reads, not just
    # the synchronous methods (a wedged token read fails at ``timeout``
    # rather than hanging the caller for the httpx default).
    self._timeout = timeout
    self._headers = headers
    # Injection seam for tests: an ``httpx.MockTransport`` swapped in here lets
    # the async stream path be exercised with a mocked transport and no real
    # server. ``None`` in production so httpx uses its default networking.
    self._async_transport: httpx.AsyncBaseTransport | None = None
    self._client = httpx.Client(
        base_url=self._host, timeout=timeout, headers=headers
    )

Functions:

normalize_openai_base_url 

normalize_openai_base_url(url: str) -> str

Strip a single trailing /v1 segment from a user-supplied base URL.

Users habitually pass http://host:8000/v1 (the full OpenAI-compatible prefix); the engine's _api_prefix re-appends /v1 to every request path, so a trailing copy would double up as /v1/v1. Only a literal trailing /v1 is stripped — proxy/gateway path prefixes are preserved.

Source code in src/diapason/engine/openai_compat_engines.py 
31
32
33
34
35
36
37
38
39
40
41
42
def normalize_openai_base_url(url: str) -> str:
    """Strip a single trailing ``/v1`` segment from a user-supplied base URL.

    Users habitually pass ``http://host:8000/v1`` (the full OpenAI-compatible
    prefix); the engine's ``_api_prefix`` re-appends ``/v1`` to every request
    path, so a trailing copy would double up as ``/v1/v1``. Only a literal
    trailing ``/v1`` is stripped — proxy/gateway path prefixes are preserved.
    """
    base = url.rstrip("/")
    if base.endswith("/v1"):
        base = base[: -len("/v1")]
    return base