llm/providers.py
Rama main · réplica de lectura del repositorio original.
# SPDX-License-Identifier: BUSL-1.1
# Copyright (c) 2024-2026 Ali Sadeghi Aghili
"""The OpenAI-compatible LLM transport shared by :mod:`llm.router` and :mod:`llm.wizard_llm`.
Before this module existed, provider switching lived only in
``llm/wizard_llm.py`` (built for the interactive setup wizard) while the
production engine was hardwired to Ollama with no way to swap providers at
all. Later, Phase 2 added :class:`OpenAIBackend` and ``AnthropicBackend``
here as two of several interchangeable hosted transports behind
:mod:`llm.router`.
That premise changed. The Ollama-specific transport and the Anthropic
Messages API transport are both gone: **OpenAI-compatible is the only
protocol this project speaks**, and ``base_url`` is what selects the
endpoint — a user's local ``gpt-oss`` server behind an OpenAI-compatible
API, a self-hosted vLLM/llama.cpp instance, or OpenAI's own hosted API are
all just an :class:`OpenAIBackend` with a different ``base_url``. Routing
across *multiple* such endpoints, with fallback chains and per-task
selection, is what :mod:`llm.router` and :mod:`llm.endpoints` are for —
this module only supplies the one transport class they route through, plus
:class:`MockBackend` for tests.
Because every real endpoint is now the same class, trust (whether an
endpoint may see schema/business-rule/row data) can no longer be decided
by class name — see :mod:`llm.trust` and :func:`llm.router.is_trusted_backend`
for why, and how it is decided instead.
"""
from __future__ import annotations
import json
import logging
import re
import time
from collections.abc import Iterable, Iterator
from typing import TYPE_CHECKING, Any
import requests
import config as cfg
from llm.base import LLMBackend
from llm.trust import default_trust_for_url
if TYPE_CHECKING:
# Annotation-only: llm.router imports llm.base, and llm.router resolves
# llm.providers lazily, so keep this out of the runtime import graph.
from llm.router import PromptSegments
logger = logging.getLogger(__name__)
_TIMEOUT: int = 120
_RETRIES: int = 3
_BACKOFF_BASE: int = 2
_FENCE_RE = re.compile(r"```(?:json)?\s*(.*?)\s*```", re.DOTALL | re.IGNORECASE)
_OUT_OF_SCOPE_SENTINEL = "OUT_OF_SCOPE"
#: ``finish_reason`` values an OpenAI-compatible ``/chat/completions``
#: response can carry that this project's contract already names 1:1 (see
#: ``observability.llm_status._VALID_FINISH_REASONS``). Anything else is
#: preserved, not discarded -- see :func:`_normalize_finish_reason`.
_KNOWN_FINISH_REASONS = frozenset({"stop", "length", "content_filter", "tool_calls"})
#: Field names, across OpenAI-compatible servers this project has seen
#: documented, that carry a model's reasoning/chain-of-thought text
#: separately from its final answer in ``content``. Never verified against
#: a live gpt-oss endpoint -- see the module docstring and
#: :func:`_extract_reasoning_text`.
_REASONING_FIELD_NAMES = ("reasoning_content", "reasoning", "thinking", "thought")
#: Literal markers that show up in ``content`` itself when a server's
#: OpenAI-compatible shim fails to separate a model's reasoning channel out
#: of its response -- gpt-oss's own "harmony" response format uses
#: ``<|channel|>``/``<|start|>``/``<|message|>`` control tokens, and several
#: other reasoning models (and the servers that front them) use a
#: ``<think>...</think>`` block. Kept intentionally narrow (literal,
#: well-known markers only) so this never misfires on an ordinary SQL
#: response -- see :func:`_content_carries_reasoning_markers`.
_REASONING_MARKER_RE = re.compile(
r"<\|channel\|>\s*analysis|<\|start\|>assistant<\|channel\|>|<think>", re.IGNORECASE,
)
#: Payload keys :func:`load_extra_body` refuses to let ``LLM_EXTRA_BODY``
#: set. Everything here decides *which model is asked what*, or is already
#: a first-class setting with its own validation and its own reporting in
#: the caller-facing ``llm`` status block: letting an environment variable
#: quietly redirect ``model``, rewrite ``messages``, or contradict the
#: ``temperature``/``seed`` that ``observability.llm_status`` reports would
#: make that block describe a request that was never sent.
#:
#: ``max_tokens`` is on the list for the same reason and one more: it is
#: the setting most likely to be reached for here (see
#: ``config.Settings.llm_extra_body_json``'s docstring on reasoning models),
#: and ``LLM_NUM_PREDICT`` is where it belongs -- two places setting one
#: value, with the status block reading only one of them, is exactly the
#: kind of disagreement this project fails loudly on elsewhere.
RESERVED_PAYLOAD_KEYS = frozenset({
"model", "messages", "temperature", "top_p", "seed", "max_tokens",
"stop", "stream", "n",
})
class ExtraBodyConfigError(ValueError):
"""``LLM_EXTRA_BODY`` is not a JSON object, or names a reserved key."""
def load_extra_body() -> dict[str, Any]:
"""Parse ``cfg.settings.llm_extra_body_json`` into request fields.
Read through ``cfg.settings`` on every call rather than cached, so
``config.override_settings()`` is visible immediately -- this project's
configuration contract (see ``config.py``'s module docstring).
Returns
-------
dict[str, Any]
The fields to merge into a chat-completions body. Empty for an
unset/blank value, which is the default and sends exactly what
this project sent before this setting existed.
Raises
------
ExtraBodyConfigError
If the value is present but is not valid JSON, is not a JSON
object, or names one of :data:`RESERVED_PAYLOAD_KEYS`. Every one
of these is a loud failure rather than a silently-ignored field:
an operator who set this to turn a model's reasoning off and got
no error has every reason to believe it worked, and the symptom
when it did not (an empty response, eventually) points nowhere
near the typo that caused it.
"""
raw = (cfg.settings.llm_extra_body_json or "").strip()
if not raw:
return {}
try:
parsed = json.loads(raw)
except json.JSONDecodeError as exc:
raise ExtraBodyConfigError(
f"LLM_EXTRA_BODY is not valid JSON: {exc}. It must be a JSON "
'object, e.g. {"chat_template_kwargs": {"enable_thinking": false}}'
) from exc
if not isinstance(parsed, dict):
raise ExtraBodyConfigError(
"LLM_EXTRA_BODY must be a JSON object of request fields, not "
f"{type(parsed).__name__} -- these are merged into the request "
"body, so there is nothing a list or a bare value could mean."
)
reserved = sorted(RESERVED_PAYLOAD_KEYS.intersection(parsed))
if reserved:
raise ExtraBodyConfigError(
f"LLM_EXTRA_BODY may not set {', '.join(repr(k) for k in reserved)} "
"-- those are set from their own settings and reported in the "
"llm status block, so overriding them here would make that block "
"describe a request that was never sent. Use LLM_NUM_PREDICT, "
"LLM_TEMPERATURE, LLM_TOP_P, LLM_SEED and LLM_STOP instead."
)
return parsed
def _normalize_finish_reason(raw_reason: Any) -> str:
"""Map a raw OpenAI-compatible ``finish_reason`` onto this project's contract.
``"stop"``/``"length"`` -- the two values the original four-value
contract (``stop | length | schema_violation | error``) already named --
pass through unchanged. ``"content_filter"`` (a moderation block) and
``"tool_calls"`` (the model tried to call a function instead of
answering) are real values several OpenAI-compatible servers emit that
the original contract didn't anticipate; this project's contract now
recognises them too (see ``observability/llm_status.py``'s
``_VALID_FINISH_REASONS`` and ``docs/api-contract-v2.md`` §6) rather
than collapsing them into the generic ``"error"`` bucket, which would
make a moderation block indistinguishable, in a week of audit logs,
from a dead endpoint.
Anything else -- a server-specific string this project has never seen,
or a missing/non-string ``finish_reason`` field entirely -- is
preserved behind an ``"other:"`` prefix instead of being discarded or
forced into ``"error"``: the whole point of deriving this value at all
is that an operator reading the log should see what the endpoint
actually said, not a value this module made up because it didn't
recognise the real one. A missing/absent value becomes ``"other:none"``
for the same reason: silently defaulting it to ``"stop"`` would claim a
complete, successful generation that was never confirmed.
Examples
--------
>>> _normalize_finish_reason("stop")
'stop'
>>> _normalize_finish_reason("length")
'length'
>>> _normalize_finish_reason("content_filter")
'content_filter'
>>> _normalize_finish_reason("tool_calls")
'tool_calls'
>>> _normalize_finish_reason("eos_token")
'other:eos_token'
>>> _normalize_finish_reason(None)
'other:none'
"""
if isinstance(raw_reason, str) and raw_reason in _KNOWN_FINISH_REASONS:
return raw_reason
label = raw_reason if isinstance(raw_reason, str) and raw_reason.strip() else "none"
return f"other:{label}"
def _extract_reasoning_text(message: Any) -> str:
"""Best-effort extraction of a model's reasoning text from *message*.
Several OpenAI-compatible servers expose a model's chain-of-thought
under a field separate from ``content`` -- most commonly
``reasoning_content`` (vLLM's DeepSeek-R1-style convention, also used by
some gpt-oss deployments), sometimes ``reasoning`` or
``thinking``/``thought`` on other community servers. This project has
never been tested against a live gpt-oss endpoint (see the module
docstring), so this check only reads well-known field names and never
guesses at new ones.
Used purely for *detection* -- see :meth:`OpenAIBackend.generate_with_meta`:
the real ``content`` field is always what is returned as the model's
answer. This function's result is never substituted for it; silently
stripping reasoning prose and hoping what remains is SQL would turn a
diagnosable protocol mismatch into a mysterious accuracy problem.
"""
if not isinstance(message, dict):
return ""
for key in _REASONING_FIELD_NAMES:
value = message.get(key)
if isinstance(value, str) and value.strip():
return value.strip()
return ""
def _content_carries_reasoning_markers(content: str) -> bool:
"""True when *content* itself contains raw reasoning-channel markup.
Some servers -- notably a gpt-oss deployment whose OpenAI-compatible
shim doesn't fully separate the model's own "harmony" response format
into distinct fields -- can leak internal channel markers
(``<|channel|>analysis``, ...) or a ``<think>...</think>`` block
straight into ``content`` instead of a separate reasoning field. Never
verified against a live endpoint (see the module docstring); kept
intentionally narrow (literal, well-known markers only) so it never
misfires on an ordinary SQL response.
"""
return bool(content) and bool(_REASONING_MARKER_RE.search(content))
#: Characters per token for the fallback in :func:`_reasoning_token_stats`
#: when a response carries reasoning text but no token counts at all. The
#: same rough ratio :func:`prompt_engine.static_prefix.estimate_tokens`
#: uses; repeated here because this module must not import
#: ``prompt_engine`` (it pulls in the whole knowledge base).
_CHARS_PER_TOKEN = 4
def _answer_text(message: Any) -> str:
"""The text of a response message's ``content``; ``""`` when there is none.
A reasoning model that is cut off while still thinking, or that puts
everything on its reasoning channel, comes back with ``"content": null``
(or no ``content`` key at all). ``str(None)`` is the four characters
``"None"``, which looked like an answer: it defeated
:func:`~observability.llm_status.is_truncated_empty_completion` (so the
``LLM_OUTPUT_TRUNCATED`` message never appeared) and would have counted
as answer tokens. Anything that is not a string is "no answer text".
Examples
--------
>>> _answer_text({"content": "SELECT 1"})
'SELECT 1'
>>> _answer_text({"content": None}), _answer_text({}), _answer_text(None)
('', '', '')
>>> _answer_text({"content": 7})
''
"""
content = message.get("content") if isinstance(message, dict) else None
return content if isinstance(content, str) else ""
def _int_or_none(value: Any) -> int | None:
"""*value* as an ``int`` when it is a real, non-boolean number, else ``None``.
Examples
--------
>>> _int_or_none(12), _int_or_none(12.0), _int_or_none("12"), _int_or_none(None)
(12, 12, None, None)
>>> _int_or_none(True) is None
True
"""
if isinstance(value, bool) or not isinstance(value, (int, float)):
return None
return int(value)
def _reasoning_token_stats(
body: dict[str, Any], reasoning_text: str, content: str,
) -> tuple[int | None, bool]:
"""How many of the completion's tokens were reasoning, and whether that is a guess.
Three cases, most to least trustworthy:
1. The server says: ``usage.completion_tokens_details.reasoning_tokens``
(OpenAI's own field, which some OpenAI-compatible servers fill in).
Returned as is, ``estimated=False`` -- even when it is ``0``.
2. The response carries reasoning text but no count. If
``usage.completion_tokens`` is known, it is split by the share of
characters that are reasoning (reasoning text and answer both
appear in those tokens, and the total is exact, so only the split
is approximate). Otherwise one token per four characters.
``estimated=True``.
3. No reasoning text and no count: ``(None, False)`` -- "not
measurable", not "zero". A non-reasoning model and a server that
hides its reasoning look the same here.
Only lengths are used. The reasoning text itself is never stored or
returned: it can quote the prompt, and the audit record that carries
this number must not hold row values.
Parameters
----------
body:
The ``/chat/completions`` response (or its reassembled stream).
reasoning_text:
The reasoning text from :func:`_extract_reasoning_text`, or ``""``.
content:
The answer text (``message.content``), or ``""``.
Returns
-------
tuple[int | None, bool]
``(reasoning_tokens, estimated)``.
Examples
--------
>>> _reasoning_token_stats(
... {"usage": {"completion_tokens": 90, "completion_tokens_details": {"reasoning_tokens": 80}}},
... "", "SELECT 1")
(80, False)
>>> _reasoning_token_stats({"usage": {"completion_tokens": 100}}, "a" * 300, "b" * 100)
(75, True)
>>> _reasoning_token_stats({}, "a" * 40, "")
(10, True)
>>> _reasoning_token_stats({"usage": {"completion_tokens": 20}}, "", "SELECT 1")
(None, False)
"""
usage = body.get("usage") if isinstance(body, dict) else None
usage = usage if isinstance(usage, dict) else {}
details = usage.get("completion_tokens_details")
if isinstance(details, dict):
reported = _int_or_none(details.get("reasoning_tokens"))
if reported is not None:
return max(reported, 0), False
if not reasoning_text:
return None, False
completion_tokens = _int_or_none(usage.get("completion_tokens"))
if completion_tokens and completion_tokens > 0:
total_chars = len(reasoning_text) + len(content)
return round(completion_tokens * len(reasoning_text) / total_chars), True
return max(1, len(reasoning_text) // _CHARS_PER_TOKEN), True
class StreamError(requests.RequestException):
"""A streamed ``/chat/completions`` response ended, or failed, mid-stream.
A :class:`requests.RequestException` so that
:meth:`OpenAIBackend.generate_with_meta` retries it like any other
transport failure.
"""
class _StreamAssembler:
"""Reassemble ``stream=true`` chunks into the body a non-streaming call returns.
The streamed protocol splits one response into ``chat.completion.chunk``
objects: ``choices[0].delta`` carries a piece of ``content`` (and, on
reasoning models, of a reasoning field), the last choice-bearing chunk
carries ``finish_reason``, and -- when the request set
``stream_options={"include_usage": true}`` -- one more chunk with an
empty ``choices`` list carries ``usage``. :meth:`body` puts those back
together as the ``chat.completion`` object the non-streaming request
would have returned (``choices[0].message.content``, the reasoning
field under the same name the server streamed it as, ``finish_reason``,
``usage``, ``system_fingerprint``), so everything downstream of the
transport -- reasoning detection, finish-reason mapping, the
``llm`` audit block -- reads the same bytes either way.
Timestamps are passed in by the caller (``perf_counter`` values), so the
class is a pure function of its input and testable on recorded chunks.
Examples
--------
>>> a = _StreamAssembler()
>>> a.feed({"id": "c1", "model": "m", "choices": [{"index": 0, "delta": {"role": "assistant"}}]}, 1.0)
>>> a.feed({"choices": [{"index": 0, "delta": {"content": "SEL"}}]}, 1.5)
>>> a.feed({"choices": [{"index": 0, "delta": {"content": "ECT 1"}, "finish_reason": "stop"}]}, 1.6)
>>> a.feed({"choices": [], "usage": {"prompt_tokens": 5, "completion_tokens": 2}}, 1.7)
>>> body = a.body()
>>> body["choices"][0]["message"]["content"], body["choices"][0]["finish_reason"]
('SELECT 1', 'stop')
>>> body["usage"]
{'prompt_tokens': 5, 'completion_tokens': 2}
>>> a.first_token_at
1.5
"""
def __init__(self) -> None:
self._id: Any = None
self._model: Any = None
self._created: Any = None
self._fingerprint: Any = None
self._role: str = "assistant"
self._content: list[str] = []
self._content_seen = False
self._reasoning: dict[str, list[str]] = {}
self._finish_reason: Any = None
self._finished = False
self._usage: dict[str, Any] | None = None
self.first_chunk_at: float | None = None
self.first_token_at: float | None = None
@property
def finished(self) -> bool:
"""True once a chunk carried a ``finish_reason``."""
return self._finished
def feed(self, chunk: dict[str, Any], t: float) -> None:
"""Fold one decoded chunk into the response.
Parameters
----------
chunk:
One decoded ``data:`` payload.
t:
When it was received (a ``perf_counter`` value).
Raises
------
StreamError
If the chunk is a server-sent error object.
"""
if not isinstance(chunk, dict):
raise StreamError("malformed stream chunk (not an object)")
error = chunk.get("error")
if error:
detail = error.get("type") or error.get("code") if isinstance(error, dict) else None
raise StreamError(f"endpoint reported an error mid-stream ({detail or 'unspecified'})")
if self.first_chunk_at is None:
self.first_chunk_at = t
for key, attr in (
("id", "_id"), ("model", "_model"), ("created", "_created"),
("system_fingerprint", "_fingerprint"),
):
if chunk.get(key) is not None and getattr(self, attr) is None:
setattr(self, attr, chunk[key])
usage = chunk.get("usage")
if isinstance(usage, dict) and usage:
self._usage = usage
choices = chunk.get("choices") or []
choice = next((c for c in choices if isinstance(c, dict) and c.get("index", 0) == 0), None)
if choice is None:
return
delta = choice.get("delta") or {}
if isinstance(delta.get("role"), str):
self._role = delta["role"]
produced = False
content = delta.get("content")
if content is not None:
self._content_seen = True
if content:
self._content.append(str(content))
produced = True
for key in _REASONING_FIELD_NAMES:
piece = delta.get(key)
if isinstance(piece, str):
self._reasoning.setdefault(key, []).append(piece)
produced = produced or bool(piece)
if produced and self.first_token_at is None:
self.first_token_at = t
if choice.get("finish_reason") is not None:
self._finish_reason = choice["finish_reason"]
self._finished = True
def body(self) -> dict[str, Any]:
"""The reassembled ``chat.completion`` body.
``message.content`` is ``None`` when no chunk ever carried a
``content`` field, which is what a non-streaming server returns for
a response that is all reasoning -- so the two paths treat that
response identically.
"""
message: dict[str, Any] = {
"role": self._role,
"content": "".join(self._content) if self._content_seen else None,
}
for key, parts in self._reasoning.items():
message[key] = "".join(parts)
body: dict[str, Any] = {
"id": self._id,
"object": "chat.completion",
"created": self._created,
"model": self._model,
"choices": [{"index": 0, "message": message, "finish_reason": self._finish_reason}],
}
if self._fingerprint is not None:
body["system_fingerprint"] = self._fingerprint
if self._usage is not None:
body["usage"] = self._usage
return body
def _iter_sse_events(lines: Iterable[bytes | str]) -> Iterator[dict[str, Any] | None]:
"""Decode server-sent-event lines into chunk objects; ``None`` marks ``[DONE]``.
Only ``data:`` lines matter; comments (``: keep-alive``), ``event:``
lines and blanks are skipped. Lines arrive as bytes and are decoded as
UTF-8 explicitly: ``requests`` would otherwise guess ISO-8859-1 for a
``text/event-stream`` response with no charset, which turns every
Persian character into mojibake.
Raises
------
StreamError
On a ``data:`` line that is not valid JSON.
Examples
--------
>>> list(_iter_sse_events([b": ping", b"", b'data: {"a": 1}', b"data: [DONE]"]))
[{'a': 1}, None]
"""
for raw in lines:
line = raw.decode("utf-8") if isinstance(raw, bytes) else raw
if not line.startswith("data:"):
continue
payload = line[5:].strip()
if not payload:
continue
if payload == "[DONE]":
yield None
return
try:
yield json.loads(payload)
except json.JSONDecodeError as exc:
raise StreamError("malformed stream chunk (invalid JSON)") from exc
def _strip_fences(text: str) -> str:
"""Remove the outermost markdown code fence from *text*, if present."""
m = _FENCE_RE.search(text)
return m.group(1).strip() if m else text.strip()
def _find_json_substring(text: str) -> str:
"""Return the substring starting from the first ``{`` or ``[`` character."""
for i, ch in enumerate(text):
if ch in ("{", "["):
return text[i:]
raise ValueError(f"No JSON object found in response. First 200 chars: {text[:200]!r}")
def parse_json_response(text: str) -> dict | list:
"""Best-effort JSON extraction: strip fences, find the start, parse.
Shared fallback for :meth:`LLMBackend.generate_structured`'s default
implementation (string-parse a plain-text response) when a provider
has no native constrained-decoding path.
Uses :class:`json.JSONDecoder.raw_decode` rather than a plain
``json.loads`` on the whole tail: a model that answers "here is the
JSON: {...} let me know if you need anything else" produces valid
JSON followed by trailing prose, which ``json.loads`` rejects outright
(``Extra data``) even though the object itself parsed fine.
``raw_decode`` parses just the first JSON value and ignores whatever
follows it.
Raises
------
ValueError
If no valid JSON can be extracted from *text*.
Examples
--------
>>> parse_json_response('```json\\n{"a": 1}\\n```')
{'a': 1}
>>> parse_json_response('here is the result: {"a": 1} done')
{'a': 1}
"""
cleaned = _strip_fences(text)
json_str = _find_json_substring(cleaned)
try:
obj, _end = json.JSONDecoder().raw_decode(json_str)
return obj
except json.JSONDecodeError as exc:
raise ValueError(f"JSON parse error: {exc}. Text: {json_str[:200]!r}") from exc
_JSON_SPACE = " \t\r\n"
def _after_space(text: str, pos: int) -> int:
"""Index of the first character at or after *pos* that is not JSON whitespace."""
while pos < len(text) and text[pos] in _JSON_SPACE:
pos += 1
return pos
def _complete_strings(text: str, pos: int, decoder: json.JSONDecoder) -> list[str]:
"""The complete strings of the JSON array whose first element starts at or after *pos*.
Stops at the closing bracket, at the first element that is cut off, and at
anything that is not a comma. Returns ``[]`` when an element is not a
string: a list that holds numbers or objects is not a flat list of text.
"""
items: list[str] = []
while True:
pos = _after_space(text, pos)
if pos >= len(text) or text[pos] == "]":
return items
try:
value, pos = decoder.raw_decode(text, pos)
except json.JSONDecodeError:
return items
if not isinstance(value, str):
return []
items.append(value)
pos = _after_space(text, pos)
if text[pos:pos + 1] != ",":
return items
pos += 1
def salvage_string_list(text: str, key: str | None = None) -> list[str]:
r"""The complete strings of a flat JSON list that the answer was cut off in.
A model that loops on the same few phrases until ``max_tokens`` stops it
leaves JSON that never closes (``{"aliases": ["a", "b", "a", "c``). Every
string before the cut is a whole, valid element, and is returned; the
element being written when the answer ended is not. The list is the
array itself, or, in an object, the array under *key* (the first array
when *key* is ``None``).
Only a **flat list of strings** is ever salvaged. A list with a number, an
object or another list in it gives ``[]``, and so does an answer with no
list: a cut-off SQL statement, a rule or a ``{"question": ..., "sql": ...}``
pair is not made usable by keeping its first half. Call this for answers
that are lists of words, such as table aliases, and nothing else.
Parameters
----------
text:
The raw answer: fenced or not, with or without prose in front.
key:
The key of the object member that holds the list; ``None`` takes the
first member that is a list.
Returns
-------
list[str]
The complete strings in order, repeats included (deduplicate with
:func:`core.wizard_drafts.tidy_aliases`); ``[]`` when there are none.
Raises
------
Nothing. This function never raises.
Examples
--------
>>> salvage_string_list('{"aliases": ["a", "b", "a", "c', "aliases")
['a', 'b', 'a']
>>> salvage_string_list('["x", "y", "z')
['x', 'y']
>>> salvage_string_list('```json\n{"aliases": ["one", "two"], "description": "cut off in')
['one', 'two']
>>> salvage_string_list(r'{"aliases": ["say \"hi\"", "café", "c', "aliases")
['say "hi"', 'café']
>>> salvage_string_list('{"aliases": ["معامله", "سفارش فروش", "قرارد')
['معامله', 'سفارش فروش']
>>> salvage_string_list('{"aliases": ["a", "b"]}'), salvage_string_list('{"aliases": []}')
(['a', 'b'], [])
>>> salvage_string_list('{"aliases": ["cut in the first strin')
[]
>>> salvage_string_list('{"aliases": ["a", {"x": 1}, "b"')
[]
>>> salvage_string_list('{"sql": "SELECT a FROM t WHERE x = ')
[]
>>> salvage_string_list('{"other": ["p"], "aliases": ["q"', "aliases")
['q']
>>> salvage_string_list("no json here")
[]
"""
try:
body = _find_json_substring(_strip_fences(text))
except ValueError:
return []
decoder = json.JSONDecoder()
if body[0] == "[":
return _complete_strings(body, 1, decoder)
pos = 1 # inside the object, after "{"
while True:
pos = _after_space(body, pos)
try:
name, pos = decoder.raw_decode(body, pos)
except json.JSONDecodeError:
return []
pos = _after_space(body, pos)
if not isinstance(name, str) or body[pos:pos + 1] != ":":
return []
pos = _after_space(body, pos + 1)
if body[pos:pos + 1] == "[" and (key is None or name == key):
return _complete_strings(body, pos + 1, decoder)
try:
_value, pos = decoder.raw_decode(body, pos)
except json.JSONDecodeError:
return []
pos = _after_space(body, pos)
if body[pos:pos + 1] != ",":
return []
pos += 1
class OpenAIBackend(LLMBackend):
"""The one real transport: an OpenAI-compatible ``/chat/completions`` endpoint.
Works unchanged against OpenAI's own hosted API, a self-hosted
``gpt-oss``/vLLM/llama.cpp server, or LM Studio — anything that speaks
the OpenAI chat-completions wire format, selected entirely by
*base_url*. Mirrors the production-grade contract the retired
Ollama-specific transport used to provide on its own: transport
retries with exponential back-off, the ``ValueError("OUT_OF_SCOPE")``
sentinel, and deterministic-decoding request fields — none of that is
specific to Ollama, so none of it should have been lost when Ollama
was.
Parameters
----------
model:
Model identifier, e.g. ``"gpt-4o-mini"`` or a local server's own
tag (``"gpt-oss-20b"``).
api_key:
Bearer token. Empty is valid — many self-hosted OpenAI-compatible
servers don't check it.
base_url:
API base URL. Defaults to ``https://api.openai.com/v1``.
trusted:
Whether this endpoint may see schema/business-rule/row data. When
``None`` (the default), resolved from *base_url* via
:func:`~llm.trust.default_trust_for_url` — loopback/private/``.local``
addresses are trusted by default, everything else (including the
factory-default ``https://api.openai.com/v1``) is not. An explicit
``True``/``False`` always wins over that default. See
:func:`~llm.router.is_trusted_backend`.
retries:
Number of transport-level retries with exponential back-off.
timeout:
Per-request timeout in seconds.
Examples
--------
>>> OpenAIBackend(model="m", api_key="k", base_url="http://localhost:8000/v1").trusted
True
>>> OpenAIBackend(model="m", api_key="k").trusted
False
>>> OpenAIBackend(model="m", api_key="k", base_url="http://localhost:8000/v1", trusted=False).trusted
False
"""
def __init__(
self,
model: str,
api_key: str,
base_url: str = "https://api.openai.com/v1",
*,
trusted: bool | None = None,
retries: int = _RETRIES,
timeout: int = _TIMEOUT,
) -> None:
self._model = model
self._api_key = api_key
self._base_url = base_url.rstrip("/")
self._trusted = trusted if trusted is not None else default_trust_for_url(self._base_url)
self._retries = retries
self._timeout = timeout
@classmethod
def from_settings(cls) -> "OpenAIBackend":
"""Build the trivial single-endpoint backend from :mod:`config`.
Reads ``cfg.settings.openai_base_url`` / ``openai_model`` /
``openai_api_key`` — the plain ``OPENAI_*`` variables, with no
``LLM_ENDPOINTS``/``LLM_ROUTES`` multi-endpoint configuration
involved. Used wherever a single, config-driven backend is enough
on its own: ``llm/wizard_llm.py``, ``app.py``'s REPL,
``eval/cli.py --live``. :meth:`~llm.router.LLMRouter.from_settings`
does NOT use this — it goes through :mod:`llm.endpoints` instead,
so a multi-endpoint deployment is honoured for production traffic
even though these simpler call sites only ever need the one
endpoint.
Examples
--------
>>> import config as cfg
>>> with cfg.override_settings(openai_model="m", openai_api_key="k"):
... backend = OpenAIBackend.from_settings()
>>> backend.name
'openai:m'
"""
return cls(
model=cfg.settings.openai_model,
api_key=cfg.settings.openai_api_key,
base_url=cfg.settings.openai_base_url,
)
@property
def name(self) -> str:
return f"openai:{self._model}"
@property
def trusted(self) -> bool:
return self._trusted
@property
def endpoint(self) -> str | None:
return self._base_url
@property
def _headers(self) -> dict[str, str]:
headers = {"Content-Type": "application/json"}
if self._api_key:
headers["Authorization"] = f"Bearer {self._api_key}"
return headers
def _build_payload(self, prompt: str) -> dict[str, Any]:
"""Build the ``/chat/completions`` request body for *prompt*.
Every sampling knob is read from :mod:`config` at call time, same
as the retired Ollama transport, and for the same reason: the same
question, run twice, must produce byte-identical SQL, which needs
``temperature=0``, ``top_p=1``, and a fixed ``seed`` sent on every
request. Unlike Ollama's own request shape (a nested ``options``
object), the OpenAI-compatible wire format sends these as
top-level sibling fields of ``model``/``messages``.
``seed`` is honoured by vLLM and llama.cpp; OpenAI's own hosted API
accepts the field but does not guarantee determinism from it (see
its docs on ``system_fingerprint``) — sent regardless, since it is
never harmful, and the caller-facing status block reports whether
the endpoint *claimed* to honour it rather than asserting
determinism this module cannot verify (see
``observability.llm_status.build_llm_status``'s ``seed_honored``).
Examples
--------
>>> backend = OpenAIBackend(model="m", api_key="k")
>>> payload = backend._build_payload("hello")
>>> payload["model"], payload["messages"]
('m', [{'role': 'user', 'content': 'hello'}])
>>> sorted(k for k in payload if k not in ("model", "messages"))
['max_tokens', 'seed', 'temperature', 'top_p']
"""
payload: dict[str, Any] = {
"model": self._model,
"messages": [{"role": "user", "content": prompt}],
"temperature": cfg.settings.llm_temperature,
"top_p": cfg.settings.llm_top_p,
"seed": cfg.settings.llm_seed,
"max_tokens": cfg.settings.llm_num_predict,
}
if cfg.settings.llm_stop:
payload["stop"] = list(cfg.settings.llm_stop)
# Merged last, but unable to reach anything above it:
# load_extra_body rejects every key set here (RESERVED_PAYLOAD_KEYS)
# rather than letting the merge win, so this can only ever ADD
# server-specific fields -- never quietly redirect the request.
payload.update(load_extra_body())
return payload
def _post_streaming(
self, payload: dict[str, Any], start: float,
) -> tuple[dict[str, Any], int, int | None]:
"""POST *payload* (already ``stream=true``) and reassemble the response.
Parameters
----------
payload:
The request body, with ``stream`` and ``stream_options`` set.
start:
``time.perf_counter()`` taken just before the request, so the
time to first token is measured from the same origin as
``total_ms``.
Returns
-------
tuple[dict[str, Any], int, int | None]
``(body, http_status, ttft_ms)`` -- *body* is the
``chat.completion`` object :class:`_StreamAssembler` rebuilt,
*ttft_ms* the time to the first content or reasoning token (or
to the first chunk of any kind when the response had no token
at all).
Raises
------
requests.Timeout
If the whole call takes longer than the backend's timeout (the
per-read timeout alone would let a slow but steady stream run
for ever, which the non-streaming request cannot).
StreamError
If the stream carries an error object, a malformed chunk, or
ends before a ``finish_reason`` or ``[DONE]``.
requests.HTTPError
For a non-2xx status, exactly as the non-streaming call.
"""
resp = requests.post(
f"{self._base_url}/chat/completions",
headers=self._headers,
json=payload,
timeout=self._timeout,
stream=True,
)
try:
resp.raise_for_status()
assembler = _StreamAssembler()
done = False
# chunk_size=64, not requests' default of 512: on a response that
# is not chunk-encoded (close-delimited), a read of N bytes
# blocks until N have arrived, which would hold back the first
# token's timestamp -- the one number this path exists to
# measure -- by however long the next tokens take. An SSE event
# is longer than 64 bytes, so the first one is never held back;
# byte-at-a-time (1) fixes the same thing but costs ~7x the CPU
# (measured: 343 vs 65 ms per 2000-event response, against 53 at
# 512). A chunk-encoded response, which is what vLLM and
# llama.cpp send, yields as each piece arrives at any size.
for event in _iter_sse_events(resp.iter_lines(chunk_size=64)):
now = time.perf_counter()
if now - start > self._timeout:
raise requests.Timeout(
f"streamed response exceeded the {self._timeout}s timeout"
)
if event is None:
done = True
break
assembler.feed(event, now)
if not (done or assembler.finished):
raise StreamError("stream ended before the response was complete")
first = assembler.first_token_at
if first is None:
first = assembler.first_chunk_at
ttft_ms = round((first - start) * 1000) if first is not None else None
return assembler.body(), resp.status_code, ttft_ms
finally:
resp.close()
def warm_prefix(self, prefix: str, *, timeout: float | None = None) -> dict[str, Any]:
"""Send *prefix* alone as a one-token request, to prime the server's prefix cache.
The body is the one :meth:`_build_payload` builds for a real request
(model, sampling fields, ``LLM_EXTRA_BODY``), with ``max_tokens``
forced to ``1`` and ``temperature`` to ``0``, so the server
computes -- and a server with prefix caching keeps -- the KV cache
for exactly the tokens a real request starts with. One attempt, no
retries, and always non-streaming: a warm-up that fails is simply
replaced by the first real question.
Parameters
----------
prefix:
The static prompt prefix, byte for byte what the real requests
start with (:func:`llm.router.build_prompt_segments`).
timeout:
HTTP timeout in seconds; the backend's own when ``None``.
Returns
-------
dict[str, Any]
``{"prompt_tokens", "cached_tokens", "endpoint_status"}``. The
first two are ``None`` when the server does not report them
(``cached_tokens`` needs vLLM's ``--enable-prompt-tokens-details``).
Never the prompt or the response text.
Raises
------
requests.RequestException
Any transport failure, a timeout, or a non-2xx status.
ExtraBodyConfigError
If ``LLM_EXTRA_BODY`` is invalid.
Examples
--------
>>> from unittest.mock import MagicMock, patch
>>> reply = MagicMock(status_code=200)
>>> reply.json.return_value = {"usage": {"prompt_tokens": 4600}}
>>> with patch("llm.providers.requests.post", return_value=reply) as post:
... info = OpenAIBackend(model="m", api_key="k").warm_prefix("PREFIX")
>>> post.call_args.kwargs["json"]["max_tokens"], post.call_args.kwargs["json"]["messages"]
(1, [{'role': 'user', 'content': 'PREFIX'}])
>>> info
{'prompt_tokens': 4600, 'cached_tokens': None, 'endpoint_status': 200}
"""
payload = self._build_payload(prefix)
payload["max_tokens"] = 1
payload["temperature"] = 0
resp = requests.post(
f"{self._base_url}/chat/completions",
headers=self._headers,
json=payload,
timeout=timeout if timeout is not None else self._timeout,
)
resp.raise_for_status()
body = resp.json()
usage = body.get("usage") if isinstance(body, dict) else None
usage = usage if isinstance(usage, dict) else {}
details = usage.get("prompt_tokens_details")
details = details if isinstance(details, dict) else {}
return {
"prompt_tokens": _int_or_none(usage.get("prompt_tokens")),
"cached_tokens": _int_or_none(details.get("cached_tokens")),
"endpoint_status": resp.status_code,
}
def generate(self, prompt: str, *, max_tokens: int | None = None) -> str:
"""POST *prompt* and return the raw response string.
Thin wrapper over :meth:`generate_with_meta` that discards the
metadata half of its return value — see that method for the full
retry/error contract, which is identical here.
"""
raw, _meta = self.generate_with_meta(prompt, max_tokens=max_tokens)
return raw
def generate_with_meta(
self, prompt: str, *, max_tokens: int | None = None,
) -> tuple[str, dict[str, Any]]:
"""POST *prompt*; return ``(raw_text, meta)``.
*max_tokens* is a per-call ceiling on the completion length that
replaces ``LLM_NUM_PREDICT`` for this request only; ``None`` -- what the SQL path always passes --
sends the setting unchanged.
*meta* carries the raw response body (undiscarded) plus call-level
facts the body itself cannot express:
* ``"raw"`` — the full ``resp.json()`` dict, including ``usage``
(``prompt_tokens`` / ``completion_tokens``) and
``system_fingerprint`` where the endpoint returns one — exactly
what ``observability/llm_status.py::build_llm_status`` expects
as its ``raw`` argument.
* ``"endpoint_status"`` — the HTTP status code of the attempt that
produced this result.
* ``"attempts"`` — which transport attempt (1-based) succeeded.
* ``"total_ms"`` — wall-clock time of the successful attempt, in
milliseconds, measured here (never inferred or guessed) since an
OpenAI-compatible response carries no server-side timing of its
own for :func:`~observability.llm_status.build_llm_status` to
read.
* ``"finish_reason"`` — the response's ``choices[0].finish_reason``,
mapped onto this project's contract by
:func:`_normalize_finish_reason`: ``"stop"``/``"length"`` pass
through, ``"content_filter"``/``"tool_calls"`` are recognised
too, and anything else is preserved as ``"other:<raw>"`` rather
than being discarded. Never hardcoded — this is what lets a
caller tell a truncated response (``"length"``) from a genuinely
complete one (``"stop"``) instead of both reading as success.
* ``"reasoning_detected"`` — ``True`` when the response appears to
carry a model's reasoning/chain-of-thought text (see
:func:`_extract_reasoning_text` and
:func:`_content_carries_reasoning_markers`) rather than, or in
addition to, a final answer. Detection only; ``raw`` is always
``content`` itself, never the reasoning text.
* ``"reasoning_tokens"`` / ``"reasoning_tokens_estimated"`` — how
many completion tokens were reasoning, and whether that figure is
the server's own (``usage.completion_tokens_details.reasoning_tokens``)
or an estimate from the reasoning text's length -- see
:func:`_reasoning_token_stats`. ``None`` / ``False`` when the
response shows no reasoning at all.
* ``"ttft_ms"`` / ``"generation_ms"`` — only with
``LLM_STREAM_TIMINGS=true``: time to the first generated token
(queue plus prefill) and the time from there to the end of the
stream; ``ttft_ms + generation_ms == total_ms``. ``None`` for a
non-streaming call, which cannot see inside its own wait.
With ``LLM_STREAM_TIMINGS=true`` the request is sent with
``stream=true`` and ``stream_options={"include_usage": true}`` and
the chunks are reassembled by :class:`_StreamAssembler` into the
body a non-streaming call would have returned, so ``raw`` and every
field above are derived from the same shape either way. The timeout
then bounds the whole call (checked as chunks arrive) as well as the
wait for each chunk.
On the ``OUT_OF_SCOPE`` sentinel, *meta* is attached to the raised
``ValueError`` as an ``llm_meta`` attribute (rather than lost),
since that is still a genuine, successful model response — just
one the caller must not treat as a corrected-SQL attempt.
Raises
------
ValueError("OUT_OF_SCOPE")
Passed through from the model sentinel; carries ``.llm_meta``.
requests.Timeout
Propagated immediately — not retried. Caller maps to ModelTimeoutError.
RuntimeError
When the endpoint is unreachable, or every retry's response
body was unparsable, after all retries.
"""
payload = self._build_payload(prompt)
if max_tokens is not None:
if max_tokens < 1:
raise ValueError(f"max_tokens must be a positive integer, got {max_tokens!r}")
# This request only: the next call is built from LLM_NUM_PREDICT again.
payload["max_tokens"] = max_tokens
streaming = bool(cfg.settings.llm_stream_timings)
if streaming:
payload["stream"] = True
payload["stream_options"] = {"include_usage": True}
last_exc: Exception | None = None
for attempt in range(1, self._retries + 1):
# perf_counter, not monotonic: monotonic() is GetTickCount64-backed
# on Windows before Python 3.13 (~15.6ms resolution), which would
# quantise -- or zero out -- total_ms for any fast response.
start = time.perf_counter()
try:
ttft_ms: int | None = None
if streaming:
body, status_code, ttft_ms = self._post_streaming(payload, start)
else:
resp = requests.post(
f"{self._base_url}/chat/completions",
headers=self._headers,
json=payload,
timeout=self._timeout,
)
resp.raise_for_status()
body = resp.json()
status_code = resp.status_code
choice: dict[str, Any] = (body.get("choices") or [{}])[0]
message: dict[str, Any] = choice.get("message") or {}
raw: str = _answer_text(message).strip()
logger.debug("OpenAI raw (attempt %d): %.300s", attempt, raw)
# Reasoning-channel detection (see module docstring's note
# on the gpt-oss deployment target and the helpers above):
# never substituted for `raw` -- only ever used to flag the
# response as suspect, in `meta`, so a downstream "No
# SELECT / CTE found" rejection reads as a protocol
# mismatch instead of the model looking incompetent at SQL.
reasoning_text = _extract_reasoning_text(message)
reasoning_detected = bool(reasoning_text) or _content_carries_reasoning_markers(raw)
if reasoning_detected:
logger.warning(
"OpenAI response (attempt %d) appears to carry reasoning-channel "
"text rather than a final answer; excerpt: %.200s",
attempt, reasoning_text or raw,
)
reasoning_tokens, reasoning_estimated = _reasoning_token_stats(
body, reasoning_text, raw,
)
total_ms = round((time.perf_counter() - start) * 1000)
meta: dict[str, Any] = {
"raw": body,
"endpoint_status": status_code,
"attempts": attempt,
"total_ms": total_ms,
"finish_reason": _normalize_finish_reason(choice.get("finish_reason")),
"reasoning_detected": reasoning_detected,
"reasoning_tokens": reasoning_tokens,
"reasoning_tokens_estimated": reasoning_estimated,
"ttft_ms": ttft_ms,
"generation_ms": (
max(total_ms - ttft_ms, 0) if ttft_ms is not None else None
),
}
if raw.strip().upper() == _OUT_OF_SCOPE_SENTINEL:
exc = ValueError(_OUT_OF_SCOPE_SENTINEL)
exc.llm_meta = meta # type: ignore[attr-defined]
raise exc
return raw, meta
except requests.Timeout:
# Timeout is a hard failure -- propagate immediately, do not retry.
raise
except requests.RequestException as exc:
# Mirrors the retired Ollama transport's discrimination:
# requests.exceptions.JSONDecodeError subclasses BOTH
# ValueError and RequestException, so a truncated/non-JSON
# body from a flaky endpoint is caught here, not by a
# `except ValueError` clause (there is none) that would
# otherwise also catch -- and mask -- the OUT_OF_SCOPE
# sentinel raised two lines above (a plain ValueError, not
# a RequestException, so it never reaches this branch).
last_exc = exc
wait = _BACKOFF_BASE ** (attempt - 1)
logger.warning(
"OpenAI attempt %d/%d failed: %s -- retrying in %ds",
attempt, self._retries, exc, wait,
)
if attempt < self._retries:
time.sleep(wait)
raise RuntimeError(
f"OpenAI-compatible endpoint {self._base_url!r} unreachable "
f"after {self._retries} retries: {last_exc}"
)
def generate_structured(self, segments: "PromptSegments", schema: dict) -> tuple[dict, dict[str, Any]]:
"""Constrained decoding via ``response_format: json_schema`` (strict mode)."""
structured_payload: dict[str, Any] = {
"model": self._model,
"messages": [{"role": "user", "content": segments.flatten()}],
"temperature": cfg.settings.llm_temperature,
"top_p": cfg.settings.llm_top_p,
"seed": cfg.settings.llm_seed,
"response_format": {
"type": "json_schema",
"json_schema": {"name": "result", "schema": schema, "strict": True},
},
}
# Same merge as the unstructured path. A constrained decode is
# still a decode: a reasoning model asked for JSON reasons first
# and can be cut off mid-object, so an operator who turned
# reasoning off for one path has not turned it off at all unless
# it applies here too.
structured_payload.update(load_extra_body())
resp = requests.post(
f"{self._base_url}/chat/completions",
headers=self._headers,
json=structured_payload,
timeout=self._timeout,
)
resp.raise_for_status()
body = resp.json()
choice: dict[str, Any] = (body.get("choices") or [{}])[0]
text = _answer_text(choice.get("message"))
# A schema-constrained decode can still be cut off by max_tokens
# (a half-emitted JSON object) -- derive finish_reason the same way
# generate_with_meta does rather than assuming "stop" here too, for
# the same reason: json.loads(text) below would already raise on a
# truncated body, but if a future caller catches that and falls
# back to *meta*, it must not read "structured_output: True" next
# to a finish_reason that silently claims a clean completion.
meta = {
"raw": body,
"endpoint_status": resp.status_code,
"attempts": 1,
"structured_output": True,
"finish_reason": _normalize_finish_reason(choice.get("finish_reason")),
}
return json.loads(text), meta
def test_connection(self) -> bool:
try:
r = requests.get(f"{self._base_url}/models", headers=self._headers, timeout=5)
return r.status_code == 200
except Exception: # noqa: BLE001
return False
class MockBackend(LLMBackend):
"""Deterministic stub backend — no network, for CI / the router's own tests.
Trusted by default (inherits :attr:`~llm.base.LLMBackend.trusted`'s
``True`` default): a stub with no real endpoint is not what the
remote-provider governance gate exists to catch — see that property's
docstring.
"""
def __init__(self, response: str = "SELECT 1", structured: dict | None = None) -> None:
self._response = response
self._structured = structured if structured is not None else {}
@property
def name(self) -> str:
return "mock:stub"
def generate(self, prompt: str) -> str: # noqa: ARG002
return self._response
def generate_with_meta(
self, prompt: str, *, max_tokens: int | None = None, # noqa: ARG002
) -> tuple[str, dict[str, Any]]:
return self._response, {"raw": {}, "endpoint_status": 200, "attempts": 1}
def generate_structured(self, segments: "PromptSegments", schema: dict) -> tuple[dict, dict[str, Any]]: # noqa: ARG002
return dict(self._structured), {"raw": {}, "endpoint_status": 200, "attempts": 1, "structured_output": True}
def test_connection(self) -> bool:
return True