Coverage for src/lilbee/providers/backends/liter_llm/backend.py: 100%
163 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +0000
1"""``LlmSdkBackend`` implementation backed by liter-llm."""
3from __future__ import annotations
5import logging
6from collections.abc import Callable, Iterator
7from typing import Any
9import httpx
11from lilbee.core.config import DEFAULT_HTTP_TIMEOUT
12from lilbee.providers import availability
13from lilbee.providers.backends.liter_llm.client import _client, _require_sdk
14from lilbee.providers.backends.liter_llm.config import _chat_fields, _route_model, _wire_base_url
15from lilbee.providers.backends.liter_llm.errors import (
16 _CHAT_FAILED,
17 _EMBED_FAILED,
18 _PROVIDER_NAME,
19 _RERANK_FAILED,
20 _provider_error,
21)
22from lilbee.providers.backends.liter_llm.events import (
23 _finish_reason,
24 _next_chunk,
25 _text,
26 _tool_call,
27 _tool_call_delta,
28 _usage,
29)
30from lilbee.providers.base import ProviderError, ProviderErrorKind
31from lilbee.providers.local_servers import (
32 OLLAMA,
33 LocalServerSpec,
34 detect_local_server,
35 openai_models_url,
36)
37from lilbee.providers.sdk_backend import (
38 CompletionRequest,
39 CompletionResult,
40 EmbeddingRequest,
41 EmbeddingResult,
42 RerankRequest,
43 RerankResult,
44 SdkToolCallDelta,
45 StreamChunk,
46 detect_backend_name,
47)
48from lilbee.runtime import asyncio_loop
50log = logging.getLogger(__name__)
52# The catalog's ``mode`` value for a chat model; embedding, image and audio models carry others.
53_CHAT_MODE = "chat"
54# lilbee provider key -> catalog provider id, where the two differ. The routing key
55# ``gemini`` is the catalog's ``google``. Every other lilbee provider uses the same id.
56_CATALOG_PROVIDER = {"gemini": "google"}
59class LiterLlmBackend:
60 """``LlmSdkBackend`` adapter backed by the ``liter-llm`` SDK."""
62 @property
63 def provider_name(self) -> str:
64 """Stable identifier used when wrapping errors in ``ProviderError``."""
65 return _PROVIDER_NAME
67 def active_backend_name(self, base_url: str) -> str:
68 """Return the display name of the backend ``base_url`` points at."""
69 return detect_backend_name(base_url)
71 def available(self) -> bool:
72 """Return True if the underlying SDK is installed."""
73 return availability.provider_sdk_available()
75 def supports_tools(self, _model_ref: str) -> bool:
76 """Optimistic: all SDK-routed refs report tool support.
78 A model that lacks a tool template just returns an empty
79 ``tool_calls`` array, which the dispatch handles as a normal
80 end-of-turn.
81 """
82 return True
84 def configure_logging(self, *, suppress_debug: bool) -> None:
85 """Do nothing: liter-llm prints no debug banner, so there is nothing to suppress."""
87 def complete(self, request: CompletionRequest) -> CompletionResult:
88 """Run a single-shot completion through ``liter_llm.chat``."""
89 liter_llm = _require_sdk()
90 model = request.ref.for_display()
91 try:
92 fields = _chat_fields(request, stream=False)
93 client = _client(
94 fields["model"], _wire_base_url(fields["model"], request.api_base), request.api_key
95 )
96 response = asyncio_loop.run(
97 liter_llm.chat(client, liter_llm.ChatCompletionRequest(**fields))
98 )
99 except Exception as exc:
100 raise _provider_error(_CHAT_FAILED, exc, model) from exc
101 choice = response.choices[0] if response.choices else None
102 message = choice.message if choice is not None else None
103 return CompletionResult(
104 content=_text(message.content if message is not None else None),
105 finish_reason=_finish_reason(choice.finish_reason) if choice is not None else None,
106 usage=_usage(response.usage),
107 model=response.model or None,
108 tool_calls=tuple(_tool_call(c) for c in (message.tool_calls or []))
109 if message is not None
110 else (),
111 )
113 def complete_stream(self, request: CompletionRequest) -> Iterator[StreamChunk]:
114 """Stream a completion through ``liter_llm.chat_stream``.
116 The first chunk is read before returning, so a connection or auth
117 failure raises here, as it does for a single-shot call.
118 """
119 liter_llm = _require_sdk()
120 model = request.ref.for_display()
121 try:
122 fields = _chat_fields(request, stream=True)
123 client = _client(
124 fields["model"], _wire_base_url(fields["model"], request.api_base), request.api_key
125 )
126 stream = liter_llm.chat_stream(client, liter_llm.ChatCompletionRequest(**fields))
127 first = asyncio_loop.run(_next_chunk(stream))
128 except Exception as exc:
129 raise _provider_error(_CHAT_FAILED, exc, model) from exc
130 return self._stream_chunks(stream, first, model)
132 @staticmethod
133 def _stream_chunks(stream: Any, first: Any, model: str) -> Iterator[StreamChunk]:
134 """Yield ``StreamChunk`` values from a liter-llm chunk stream.
136 Exceptions raised mid-iteration are classified into ``ProviderError``
137 so the semantic layer sees a consistent error type regardless of
138 where the SDK failed.
139 """
140 chunk = first
141 try:
142 while chunk is not None:
143 choice = chunk.choices[0] if chunk.choices else None
144 content = ""
145 finish_reason = None
146 tool_call_deltas: tuple[SdkToolCallDelta, ...] = ()
147 if choice is not None:
148 delta = choice.delta
149 content = delta.content or ""
150 finish_reason = _finish_reason(choice.finish_reason)
151 tool_call_deltas = tuple(_tool_call_delta(c) for c in delta.tool_calls or [])
152 usage = _usage(chunk.usage) if chunk.usage is not None else None
153 if content or finish_reason or tool_call_deltas or usage is not None:
154 yield StreamChunk(
155 content=content,
156 finish_reason=finish_reason,
157 tool_call_deltas=tool_call_deltas,
158 usage=usage,
159 )
160 chunk = asyncio_loop.run(_next_chunk(stream))
161 except Exception as exc:
162 raise _provider_error(_CHAT_FAILED, exc, model) from exc
164 def embed(self, request: EmbeddingRequest) -> EmbeddingResult:
165 """Embed inputs through ``liter_llm.embed``."""
166 liter_llm = _require_sdk()
167 model = _route_model(request.ref, request.api_base)
168 try:
169 client = _client(model, _wire_base_url(model, request.api_base), request.api_key)
170 response = asyncio_loop.run(
171 liter_llm.embed(
172 client, liter_llm.EmbeddingRequest(model=model, input=request.inputs)
173 )
174 )
175 except Exception as exc:
176 raise _provider_error(_EMBED_FAILED, exc, request.ref.for_display()) from exc
177 # Order by the response's ``index`` rather than arrival order: a proxy or
178 # gateway may return the batch out of order, and the consumer zips vectors
179 # to inputs positionally, so a reorder would silently mis-pair every chunk
180 # with the wrong vector.
181 ordered = sorted(response.data, key=lambda item: item.index)
182 # Reordering is not the only way the positional zip breaks. A gateway that
183 # drops an item, or repeats an index, yields a batch that still sorts
184 # cleanly but no longer corresponds one-to-one with the inputs. A
185 # spec-conforming response carries exactly one item per input, indexed
186 # 0..n-1, so anything else is refused rather than silently mis-paired.
187 expected = len(request.inputs)
188 if [item.index for item in ordered] != list(range(expected)):
189 raise ProviderError(
190 f"Embedding response does not match the request: expected {expected} "
191 f"vectors indexed 0-{expected - 1}, got {len(ordered)}. The endpoint "
192 "returned an incomplete or misindexed batch.",
193 provider=_PROVIDER_NAME,
194 kind=ProviderErrorKind.SERVER,
195 )
196 return EmbeddingResult(
197 vectors=[list(item.embedding) for item in ordered], model=response.model or None
198 )
200 def rerank(self, request: RerankRequest) -> RerankResult:
201 """Rerank documents via ``liter_llm.rerank`` (Cohere, Voyage, Jina, Together, HF TEI).
203 The SDK returns results sorted by relevance; we restore input
204 order via each result's ``index`` so scores line up with the
205 caller's ``candidates`` list.
206 """
207 if not request.candidates:
208 return RerankResult(scores=[])
209 liter_llm = _require_sdk()
210 model = _route_model(request.ref, request.api_base)
211 try:
212 client = _client(model, _wire_base_url(model, request.api_base), request.api_key)
213 response = asyncio_loop.run(
214 liter_llm.rerank(
215 client,
216 liter_llm.RerankRequest(
217 model=model, query=request.query, documents=request.candidates
218 ),
219 )
220 )
221 except Exception as exc:
222 raise _provider_error(_RERANK_FAILED, exc, request.ref.for_display()) from exc
223 scores = [0.0] * len(request.candidates)
224 for item in response.results:
225 scores[item.index] = float(item.relevance_score)
226 return RerankResult(scores=scores)
228 def list_models(self, *, spec: LocalServerSpec, base_url: str, api_key: str) -> list[str]:
229 """List models from Ollama (``/api/tags``) or an OpenAI-compatible ``/v1/models``.
231 *spec* is the configured server being listed, so the kind never depends on the
232 address. Ollama is listed without a key, as on the request path.
233 """
234 clean_base = base_url.rstrip("/")
235 if spec is OLLAMA:
236 return self._list_ollama_models(clean_base)
237 return self._list_openai_models(clean_base, api_key)
239 def list_chat_models(self, provider: str) -> list[str]:
240 """Return the bare chat-model names liter-llm's built-in catalog holds for *provider*.
242 The catalog is embedded, so the call needs no network and no key. Names
243 come back without the ``provider/`` prefix, sorted; the routing layer adds
244 the prefix back. A provider the catalog does not hold, or an empty one,
245 gives an empty list: the library reads an empty id as "every provider".
246 A library that is installed but cannot import also gives an empty list,
247 so the model pickers show no hosted models instead of failing.
248 """
249 catalog_id = _CATALOG_PROVIDER.get(provider, provider)
250 if not catalog_id:
251 return []
252 try:
253 sdk = _require_sdk()
254 except ProviderError:
255 log.debug("Provider library unavailable; no hosted chat models", exc_info=True)
256 return []
257 models = sdk.list_catalog_models(catalog_id)
258 return sorted({m.model for m in models if m.mode == _CHAT_MODE})
260 @staticmethod
261 def _list_ollama_models(base_url: str) -> list[str]:
262 """List models via the Ollama ``/api/tags`` endpoint."""
263 try:
264 resp = httpx.get(f"{base_url}/api/tags", timeout=DEFAULT_HTTP_TIMEOUT)
265 resp.raise_for_status()
266 data = resp.json()
267 return [m["name"] for m in data.get("models", [])]
268 except httpx.HTTPError as exc:
269 raise ProviderError(f"Cannot list models: {exc}", provider=_PROVIDER_NAME) from exc
271 @staticmethod
272 def _list_openai_models(base_url: str, api_key: str) -> list[str]:
273 """List models via an OpenAI-compatible ``/v1/models`` endpoint."""
274 headers: dict[str, str] = {}
275 if api_key:
276 headers["Authorization"] = f"Bearer {api_key}"
277 try:
278 resp = httpx.get(
279 openai_models_url(base_url), headers=headers, timeout=DEFAULT_HTTP_TIMEOUT
280 )
281 resp.raise_for_status()
282 data = resp.json()
283 return [m["id"] for m in data.get("data", [])]
284 except httpx.HTTPError:
285 log.debug("Failed to list models via /v1/models", exc_info=True)
286 return []
288 def pull_model(
289 self,
290 model: str,
291 *,
292 base_url: str,
293 on_progress: Callable[..., Any] | None = None,
294 ) -> None:
295 """Refuse to pull: local servers (Ollama, LM Studio) are read-only.
297 Their models are managed in their own app and surface here once
298 present, so lilbee never downloads them over the network.
299 """
300 spec = detect_local_server(base_url.rstrip("/"))
301 server = spec.display_name if spec is not None else "This server"
302 raise ProviderError(
303 f"{server} doesn't download models over the network. "
304 f"Add the model in its own app, then pick it here.",
305 provider=_PROVIDER_NAME,
306 )
308 def show_model(self, model: str, *, base_url: str) -> dict[str, Any] | None:
309 """Get model info via the Ollama ``/api/show`` endpoint.
311 Returns the raw ``parameters`` text and the ``capabilities`` list
312 (newer Ollama versions) so callers can check for vision support.
313 Returns ``None`` for servers without a metadata endpoint (LM Studio).
314 """
315 clean_base = base_url.rstrip("/")
316 spec = detect_local_server(clean_base)
317 if spec is None or not spec.supports_show:
318 return None
319 # Ollama's API uses bare model names; the routing-layer prefix has
320 # to come off before the request goes out.
321 ollama_name = model.removeprefix(OLLAMA.wire_prefix)
322 try:
323 resp = httpx.post(
324 f"{clean_base}/api/show",
325 json={"name": ollama_name},
326 timeout=DEFAULT_HTTP_TIMEOUT,
327 )
328 resp.raise_for_status()
329 data = resp.json()
330 except httpx.HTTPError:
331 return None
333 result: dict[str, Any] = {}
335 params = data.get("parameters", "")
336 if isinstance(params, str) and params:
337 result["parameters"] = params
338 elif params:
339 result["parameters"] = str(params)
341 capabilities = data.get("capabilities")
342 if isinstance(capabilities, list):
343 result["capabilities"] = capabilities
345 return result or None