Coverage for src/lilbee/providers/sdk_backend.py: 100%
82 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-14 11:46 +0000
1"""Protocol and value types for SDK-backed LLM backends.
3A backend hides one third-party SDK. The ``SdkLLMProvider`` speaks to
4backends exclusively through the ``LlmSdkBackend`` Protocol and the
5value types defined here, so SDK response objects never leak outside
6the adapter.
8This module is intentionally dependency-free (no SDK imports, no
9lilbee provider imports beyond the shared base types).
10"""
12from __future__ import annotations
14import os
15from collections.abc import Callable, Iterator
16from dataclasses import dataclass, field
17from typing import TYPE_CHECKING, Any, Protocol
19# Display name for the active backend the SDK is talking to. The
20# adapter's own identity is exposed separately via provider_name.
21from lilbee.providers.backend_names import BackendName
22from lilbee.providers.local_servers import detect_local_server
24if TYPE_CHECKING:
25 # circular: sdk_backend -> model_ref -> types -> sdk_backend (annotation-only)
26 from lilbee.providers.model_ref import ProviderModelRef
28# Single source of truth for per-provider API key configuration.
29# Maps (provider_name, config_field, env_var, display_label). Backend-agnostic:
30# OpenAI-compatible SDKs all read these env vars at call time. Tuple order
31# is the canonical display order downstream consumers (TUI grouping, catalog
32# sections) honor when surfacing providers.
33PROVIDER_KEYS: tuple[tuple[str, str, str, str], ...] = (
34 ("openrouter", "openrouter_api_key", "OPENROUTER_API_KEY", "OpenRouter"),
35 ("gemini", "gemini_api_key", "GEMINI_API_KEY", "Gemini"),
36 ("anthropic", "anthropic_api_key", "ANTHROPIC_API_KEY", "Anthropic"),
37 ("openai", "openai_api_key", "OPENAI_API_KEY", "OpenAI"),
38 ("mistral", "mistral_api_key", "MISTRAL_API_KEY", "Mistral"),
39 ("deepseek", "deepseek_api_key", "DEEPSEEK_API_KEY", "DeepSeek"),
40)
42# Provider name -> cfg attribute holding that provider's API key.
43PROVIDER_API_KEY_FIELD: dict[str, str] = {prov: field for prov, field, *_ in PROVIDER_KEYS}
46# Provider name -> the SDK's own env var (read at call time by the backend).
47PROVIDER_API_KEY_ENV: dict[str, str] = {prov: env for prov, _field, env, *_ in PROVIDER_KEYS}
50def get_provider_api_key(provider: str) -> str | None:
51 """Return the configured API key for *provider*, or ``None`` if unknown / unset.
53 *provider* is the lowercase routing key from a parsed model ref (e.g.
54 ``"openai"``). Returns ``None`` for unknown providers AND for known
55 providers whose key is unconfigured; callers can distinguish via
56 :data:`PROVIDER_API_KEY_FIELD`. Reads only the lilbee config field; use
57 :func:`provider_has_key` to also honor the SDK's own env var.
58 """
59 from lilbee.core.config import cfg
61 field = PROVIDER_API_KEY_FIELD.get(provider.lower())
62 if field is None:
63 return None
64 value = getattr(cfg, field)
65 return value or None
68def provider_has_key(provider: str) -> bool:
69 """True if *provider* has a key via its standard env var or the lilbee config field."""
70 env_var = PROVIDER_API_KEY_ENV.get(provider.lower())
71 if env_var and os.environ.get(env_var):
72 return True
73 return get_provider_api_key(provider) is not None
76# Hosted API providers identified by URL substring. Local OpenAI-compatible
77# servers (Ollama, LM Studio) are matched ahead of this table via the
78# local-servers registry, so they are not listed here.
79_REMOTE_API_URL_PATTERNS: tuple[tuple[str, BackendName], ...] = (
80 ("openrouter", BackendName.OPENROUTER),
81 ("openai", BackendName.OPENAI),
82 ("anthropic", BackendName.ANTHROPIC),
83 ("googleapis", BackendName.GEMINI),
84 ("gemini", BackendName.GEMINI),
85 ("mistral", BackendName.MISTRAL),
86 ("deepseek", BackendName.DEEPSEEK),
87)
90def detect_backend_name(base_url: str) -> BackendName:
91 """Return the display name of the backend behind ``base_url``.
93 Adapter-agnostic; any SDK implementation can delegate to this helper.
94 Checks the local-server registry (Ollama, LM Studio) first, then the
95 hosted-API URL patterns, and falls back to ``BackendName.REMOTE``.
96 """
97 local = detect_local_server(base_url)
98 if local is not None:
99 return local.display_name
100 url_lower = base_url.lower()
101 for pattern, name in _REMOTE_API_URL_PATTERNS:
102 if pattern in url_lower:
103 return name
104 return BackendName.REMOTE
107@dataclass(frozen=True)
108class SdkToolCall:
109 """One tool call extracted from a non-streaming SDK chat response."""
111 id: str
112 name: str
113 arguments: str
116@dataclass(frozen=True)
117class SdkToolCallDelta:
118 """One streaming tool-call delta from an SDK chat chunk.
120 ``id`` and ``name`` arrive on the opener; ``arguments_delta`` accumulates
121 across subsequent chunks at the same ``index``. Mirrors the per-frame
122 shape that the dispatch's stream translator already understands.
123 """
125 index: int
126 id: str | None = None
127 name: str | None = None
128 arguments_delta: str | None = None
131@dataclass(frozen=True)
132class CompletionResult:
133 """Single-shot chat completion result returned by a backend."""
135 content: str
136 finish_reason: str | None = None
137 model: str | None = None
138 tool_calls: tuple[SdkToolCall, ...] = ()
141@dataclass(frozen=True)
142class StreamChunk:
143 """One delta yielded during a streaming chat completion."""
145 content: str
146 finish_reason: str | None = None
147 tool_call_deltas: tuple[SdkToolCallDelta, ...] = ()
150@dataclass(frozen=True)
151class EmbeddingResult:
152 """Embedding vectors returned by a backend for a batch of inputs."""
154 vectors: list[list[float]]
155 model: str | None = None
158@dataclass(frozen=True)
159class CompletionRequest:
160 """Backend-agnostic request for a single completion call.
162 ``ref`` carries the parsed model reference; the adapter converts it
163 to the wire format its SDK expects. ``messages`` is the raw lilbee
164 message list (may contain ``images`` bytes); the adapter formats it
165 for its SDK. ``api_base`` is populated for local/Ollama deployments
166 and omitted for API-hosted models.
167 """
169 ref: ProviderModelRef
170 messages: list[dict[str, Any]]
171 options: dict[str, Any] = field(default_factory=dict)
172 api_base: str | None = None
173 api_key: str | None = None
176@dataclass(frozen=True)
177class EmbeddingRequest:
178 """Backend-agnostic request for an embedding call."""
180 ref: ProviderModelRef
181 inputs: list[str]
182 api_base: str | None = None
183 api_key: str | None = None
186@dataclass(frozen=True)
187class RerankRequest:
188 """Backend-agnostic rerank request."""
190 ref: ProviderModelRef
191 query: str
192 candidates: list[str]
193 api_base: str | None = None
194 api_key: str | None = None
197@dataclass(frozen=True)
198class RerankResult:
199 """Rerank scores returned by a backend, one per candidate in input order."""
201 scores: list[float]
202 model: str | None = None
205class LlmSdkBackend(Protocol):
206 """Protocol every LLM SDK adapter must satisfy.
208 The provider calls these methods through the Protocol only; SDK
209 response objects never cross the seam. Methods with a natural
210 "not supported" signal are documented below.
212 Lifecycle: ``available()`` is the cheap install check called before
213 any other method; ``configure_logging`` runs once at first use.
214 ``complete`` / ``complete_stream`` / ``embed`` are the hot-path
215 operations. ``list_models`` / ``list_chat_models`` / ``pull_model``
216 / ``show_model`` are catalog helpers and may raise
217 ``NotImplementedError`` or return empty values when unsupported.
219 Error contract: implementations must raise only ``ProviderError`` or
220 ``NotImplementedError`` from any method. ``SdkLLMProvider`` wraps any
221 other exception at the seam; adapters should translate SDK-specific
222 errors (httpx errors, third-party SDK exceptions) into
223 ``ProviderError`` so the provider can pass them through.
224 """
226 @property
227 def provider_name(self) -> str:
228 """Stable identifier used when wrapping errors in ``ProviderError``."""
229 ...
231 def active_backend_name(self, base_url: str) -> str:
232 """Return the display name of the backend the adapter is talking to.
234 ``"Ollama"`` for an Ollama URL, ``"OpenAI"`` for an OpenAI URL,
235 etc.; unknown URLs fall back to ``"Remote"``. The adapter's own
236 identity is exposed separately through ``provider_name``.
237 """
238 ...
240 def available(self) -> bool:
241 """Return True when the underlying SDK is importable."""
242 ...
244 def configure_logging(self, *, suppress_debug: bool) -> None:
245 """Apply backend-level logging toggles (best-effort no-op if unsupported)."""
246 ...
248 def complete(self, request: CompletionRequest) -> CompletionResult:
249 """Run a single-shot chat completion."""
250 ...
252 def complete_stream(self, request: CompletionRequest) -> Iterator[StreamChunk]:
253 """Run a streaming chat completion, yielding content chunks."""
254 ...
256 def embed(self, request: EmbeddingRequest) -> EmbeddingResult:
257 """Embed a batch of inputs, returning one vector per input."""
258 ...
260 def rerank(self, request: RerankRequest) -> RerankResult:
261 """Score *candidates* against *query*, returning one float per candidate.
263 Raise ``NotImplementedError`` if the backend has no rerank API.
264 An empty ``request.candidates`` returns ``RerankResult([])``
265 without an SDK call.
266 """
267 ...
269 def list_models(self, *, base_url: str, api_key: str) -> list[str]:
270 """List model identifiers visible to the backend. Return [] if unsupported."""
271 ...
273 def list_chat_models(self, provider: str) -> list[str]:
274 """List chat-mode models from the SDK's catalog for *provider*.
276 Returns the unfiltered upstream catalog. Backends without a
277 notion of frontier providers return ``[]``.
279 Unlike ``list_models``, this is a static pricing/capability table,
280 not a runtime HTTP probe.
281 """
282 ...
284 def pull_model(
285 self,
286 model: str,
287 *,
288 base_url: str,
289 on_progress: Callable[..., Any] | None = None,
290 ) -> None:
291 """Pull a model. Raise NotImplementedError if unsupported."""
292 ...
294 def show_model(self, model: str, *, base_url: str) -> dict[str, Any] | None:
295 """Return model metadata dict or None if unsupported / not found."""
296 ...
298 def supports_tools(self, model_ref: str) -> bool:
299 """Return True iff the backend can route tool calls for *model_ref*."""
300 ...