best_engine_ai_helper.openai_api module

openai_api — speak OpenAI on the wire, choose the model locally.

From the outside this package becomes an OpenAI-compatible endpoint: point any OpenAI SDK at http://127.0.0.1:8000/v1, ask for chat completions, embeddings or images, and get them. Inside, the model is not a name the caller had to know — it is whatever this machine should actually run, chosen by the same ranking every other surface here uses.

That inversion is the whole point. The usual OpenAI-compatible server (vLLM, Ollama’s /v1, LM Studio) asks you which model to load; the hard part — which model fits this hardware, whether it is fast enough, whether a bigger one is worth the wait — is left to the person writing the config. Here the wire protocol is the familiar one and the choice is the product:

client.chat.completions.create(model=”best”, messages=[…])

best is a virtual model name. It resolves, per request, to the highest-ranked model that fits this machine’s memory budget for the kind the request needs — and a request carrying an image needs a VLM, so the same name resolves differently depending on what is in it. The grammar is small:

`best` / `auto`	Best text model; best vision model if the request carries images.
`best-llm`	Best text model, images refused rather than ignored.
`best-vlm`	Best vision-language model.
`best-embed`	Best embedding model (`/v1/embeddings`).
`best-image`	Best text-to-image model (`/v1/images/generations`).
`best:<task>`	Best model for that task, free text: `best:write SQL`, `best:read screenshots`. The task picks the benchmark axis, exactly as `report --task` does.
anything else	Passed through verbatim, so `qwen3:14b` still means `qwen3:14b`.

What is deliberately NOT faked

A compatibility layer earns trust by refusing rather than pretending:

Token counts are real where the backend reports them. Ollama returns prompt_eval_count/eval_count, and those are what usage carries. Where a backend reports nothing, the counts are a character-based estimate — flagged in system_fingerprint rather than passed off as measured.
Streaming is real streaming for a plain text request, over llm.chat_stream. For a request that also needs a schema or an image it degrades to one chunk, because the underlying stream carries neither, and saying so beats emitting a fake token-by-token drip.
Unsupported parameters are errors, not silence. n > 1, logprobs and friends return a 400 naming themselves. A parameter quietly ignored is worse than one refused: the caller believes they got what they asked for.
Remote image URLs are off by default. image_url with a data: URI always works. An http(s) URL makes the server fetch a caller-chosen address, which is a server-side request forgery primitive, so it needs BEST_OPENAI_FETCH_IMAGE_URLS=1 to be switched on.

Author

Warith Harchaoui <warithmetics@deraison.ai>

class best_engine_ai_helper.openai_api.ChatCompletionRequest(*, messages, model=None, temperature=0.2, max_tokens=None, stream=False, response_format=None, n=1, logprobs=None, tools=None, tool_choice=None, stop=None)[source]

Bases: BaseModel

Body for POST /v1/chat/completions.

Parameters:

messages (list[ChatMessage])
model (str | None)
temperature (float)
max_tokens (int | None)
stream (bool)
response_format (dict[str, Any] | None)
n (int)
logprobs (bool | None)
tools (list[dict[str, Any]] | None)
tool_choice (Any | None)
stop (str | list[str] | None)

logprobs: bool | None

max_tokens: int | None

messages: list[ChatMessage]

model: str | None

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

n: int

response_format: dict[str, Any] | None

stop: str | list[str] | None

stream: bool

temperature: float

tool_choice: Any | None

tools: list[dict[str, Any]] | None

class best_engine_ai_helper.openai_api.ChatMessage(*, role='user', content=None)[source]

Bases: BaseModel

One message in an OpenAI chat request.

Parameters:

role (str)
content (str | list[dict[str, Any]] | None)

content: str | list[dict[str, Any]] | None

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

role: str

class best_engine_ai_helper.openai_api.CompletionRequest(*, prompt, model=None, temperature=0.2, max_tokens=None, stream=False, n=1)[source]

Bases: BaseModel

Body for the legacy POST /v1/completions.

Parameters:

prompt (str | list[str])
model (str | None)
temperature (float)
max_tokens (int | None)
stream (bool)
n (int)

max_tokens: int | None

model: str | None

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

n: int

prompt: str | list[str]

stream: bool

temperature: float

class best_engine_ai_helper.openai_api.EmbeddingRequest(*, input, model=None, encoding_format='float')[source]

Bases: BaseModel

Body for POST /v1/embeddings.

Parameters:

input (str | list[str])
model (str | None)
encoding_format (Literal['float', 'base64'])

encoding_format: Literal['float', 'base64']

input: str | list[str]

model: str | None

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

best_engine_ai_helper.openai_api.FETCH_IMAGE_URLS_ENV = 'BEST_OPENAI_FETCH_IMAGE_URLS'

a server that fetches a caller-supplied address on request is an SSRF primitive, and this one usually runs on a developer’s machine with access to a private network a public client should not be able to probe.

Type:: Environment switch for fetching http(s) image URLs. Off by default

best_engine_ai_helper.openai_api.FINGERPRINT_ESTIMATED = 'best-engine-ai-helper:usage=estimated': Marks a response whose token counts are estimated rather than reported by the backend. system_fingerprint is the only field in the OpenAI response shape that can carry this honestly, which is why it is used for it.

class best_engine_ai_helper.openai_api.ImageRequest(*, prompt, model=None, n=1, size='auto', response_format='b64_json', steps=None, guidance=None, negative_prompt=None, seed=None)[source]

Bases: BaseModel

Body for POST /v1/images/generations.

Parameters:

prompt (str)
model (str | None)
n (int)
size (str)
response_format (Literal['b64_json', 'url'])
steps (int | None)
guidance (float | None)
negative_prompt (str | None)
seed (int | None)

guidance: float | None

model: str | None

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

n: int

negative_prompt: str | None

prompt: str

response_format: Literal['b64_json', 'url']

seed: int | None

size: str

steps: int | None

best_engine_ai_helper.openai_api.MODE_ENV = 'BEST_OPENAI_MODE': Environment switch for the default mode of a best* name that does not pin one. local (free, this machine) unless set to cloud (paid provider, with the local fallback the cloud descriptor already carries).

best_engine_ai_helper.openai_api.chat_completions(request)

OpenAI’s chat completions, with the model chosen for this machine.

Text and vision both land here: a request carrying image_url parts resolves to a VLM on its own, which is why best is usually the right name to send.

Parameters:: request (ChatCompletionRequest)
Return type:: Any

best_engine_ai_helper.openai_api.completions(request)

The legacy completions endpoint, for clients that still speak it.

Implemented by routing the prompt through the chat path, because every backend here is instruction-tuned: there is no separate raw-completion mode to expose, and pretending otherwise would just be a differently-shaped chat call.

Parameters:: request (CompletionRequest)
Return type:: Any

best_engine_ai_helper.openai_api.default_mode(env=None)[source]

The mode a best* name takes when it does not pin one.

Read from MODE_ENV rather than baked in, so the same build serves a free deployment and a paid one. Anything unrecognised is local, with a warning: the failure direction matters here, because the other one bills somebody.

Parameters:: env (dict[str, str] or None) – Environment to read; None uses os.environ. Injectable so this stays a pure function under test — the thing it reports on is exactly the kind of ambient state a test must not inherit from the host.
Returns:: 'local' or 'cloud'.
Return type:: str

Examples

>>> default_mode({})
'local'
>>> default_mode({"BEST_OPENAI_MODE": "cloud"})
'cloud'
>>> default_mode({"BEST_OPENAI_MODE": "CLOUD"})
'cloud'
>>> default_mode({"BEST_OPENAI_MODE": "nonsense"})   # unrecognised -> free
'local'

best_engine_ai_helper.openai_api.embeddings(request)

OpenAI’s embeddings endpoint, over the best local embedder.

best-embed resolves among the catalogue’s kind: embed entries, which are ranked on retrieval quality (MTEB) rather than on the general-knowledge axis a chat model is judged by — a distinction the usages catalogue already draws and this endpoint inherits.

Parameters:: request (EmbeddingRequest)
Return type:: dict[str, Any]

best_engine_ai_helper.openai_api.flatten_messages(messages)[source]

Collapse an OpenAI message list into what llm.chat takes.

chat has one prompt, one optional system preamble and a list of images, which is the shape every backend here actually supports. A multi-turn conversation is therefore rendered as a transcript — labelled by role, in order — rather than silently reduced to its last message, which is how a compatibility layer loses the context a caller carefully sent.

Parameters:: messages (list of dict) – OpenAI messages. Content may be a string or a list of parts ({"type": "text"} / {"type": "image_url"}).
Returns:: (system, prompt, images).
Return type:: tuple[str | None, str, list[bytes]]
Raises:: HTTPException – When the list is empty or carries no usable content.

Examples

>>> flatten_messages([{"role": "user", "content": "hello"}])
(None, 'hello', [])
>>> flatten_messages([
...     {"role": "system", "content": "Be brief."},
...     {"role": "user", "content": "hi"},
...     {"role": "assistant", "content": "hello"},
...     {"role": "user", "content": "and now?"},
... ])
('Be brief.', 'user: hi\nassistant: hello\nuser: and now?', [])
>>> system, prompt, images = flatten_messages([{
...     "role": "user",
...     "content": [
...         {"type": "text", "text": "what is this?"},
...         {"type": "image_url", "image_url": {"url": "data:image/png;base64,aGk="}},
...     ],
... }])
>>> system, prompt, images
(None, 'what is this?', [b'hi'])

best_engine_ai_helper.openai_api.images_generations(request)

OpenAI’s image endpoint, over a local diffusion pipeline.

size: "auto" (the default here) resolves to the chosen model’s native resolution, which is the one it is good at: a 512px model asked for 1024x1024 duplicates subjects rather than adding detail. Name a size explicitly to override that, and expect the quality to follow the model rather than the number.

response_format defaults to b64_json rather than OpenAI’s url, because a local server has nowhere to host a file; asking for url returns a data: URI, which every client that follows a URL can read.

Parameters:: request (ImageRequest)
Return type:: dict[str, Any]

best_engine_ai_helper.openai_api.install_error_handlers(app)[source]

Make every /v1 failure come back in OpenAI’s error envelope.

Raising HTTPException(detail={"error": ...}) is not enough on its own: FastAPI serialises the detail under a detail key, so the envelope ends up nested one level too deep and an SDK that reads error.message finds nothing. This unwraps it — and does the same for FastAPI’s own 422 validation errors, which an OpenAI client cannot parse at all in their native shape.

Scoped to /v1 on purpose. /api/* is this package’s own surface and its callers already expect FastAPI’s conventions; changing those to suit a compatibility layer would be the tail wagging the dog, so anything outside /v1 is handed straight back to the default handler.

Parameters:: app (Any) – The FastAPI application to register the handlers on.
Return type:: None

Examples

>>> from fastapi import FastAPI
>>> install_error_handlers(FastAPI()) is None
True

best_engine_ai_helper.openai_api.list_models()

List the virtual names and every catalogue model that fits this machine.

Both halves matter. The best* aliases are what a caller should use — they are the product — and the concrete ids are what makes the list useful in a client’s model picker: they are filtered to what this hardware can actually run, so the dropdown stops offering a 72B model to a laptop.

Returns:: OpenAI’s {"object": "list", "data": [...]}.
Return type:: dict[str, Any]

best_engine_ai_helper.openai_api.parse_model(name, *, has_images=False)[source]

Resolve a requested model name into what to actually run.

The name may pin a mode as an infix after best-: best-cloud-llm is the best paid text model, best-local-vlm the best free one that fits this machine. A name that pins nothing reports mode: None, meaning “the server decides” — default_mode() does, at call time. The mode is deliberately NOT resolved here: this function is pure, and a resolution that silently read the environment would make its own doctests depend on the host that happened to run them.

Parameters:

name (str or None) – The model field. None or empty is treated as best, because an OpenAI SDK always sends one and a caller who does not care should not have to invent a value.
has_images (bool) – Whether the request carries images. Only consulted for the auto names, where it is the difference between needing a VLM and not.

Returns:

virtual (bool), kind (llm/vlm/embed/image), task (free text or None), literal (the model name to pass through, or None when the choice is ours) and mode ('local'/'cloud', or None when the name did not say).

Return type:

dict[str, Any]

Examples

>>> parse_model("best") == {
...     'virtual': True, 'kind': 'llm', 'task': None, 'literal': None, 'mode': None}
True
>>> parse_model("best", has_images=True)["kind"]
'vlm'
>>> parse_model("best-image")["kind"]
'image'
>>> parse_model("best:write SQL for a warehouse")["task"]
'write SQL for a warehouse'
>>> parse_model("best-vlm:read screenshots") == {
...     'virtual': True, 'kind': 'vlm', 'task': 'read screenshots',
...     'literal': None, 'mode': None}
True

The mode infix, on a pinned kind and on an auto name alike:

>>> parse_model("best-cloud-llm")["mode"]
'cloud'
>>> parse_model("best-cloud-llm")["kind"]
'llm'
>>> parse_model("best-local-vlm") == {
...     'virtual': True, 'kind': 'vlm', 'task': None,
...     'literal': None, 'mode': 'local'}
True
>>> parse_model("best-cloud", has_images=True)["kind"]
'vlm'
>>> parse_model("best-cloud:write SQL")["task"]
'write SQL'

A concrete name still passes through untouched, colon and all:

>>> parse_model("qwen3:14b") == {
...     'virtual': False, 'kind': 'llm', 'task': None,
...     'literal': 'qwen3:14b', 'mode': None}
True
>>> parse_model(None)["virtual"]
True

best_engine_ai_helper.openai_api.reset_engine_cache()[source]

Forget every resolved engine, so the next request re-ranks.

Examples

>>> reset_engine_cache()
>>> _ENGINE_CACHE
{}

Return type:: None