Coverage for src/local_deep_research/llm/providers/base.py: 100%
46 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1"""Base class for LLM providers."""
3from ...security.egress.classification import Exposure
5# Single placeholder for ChatOpenAI(api_key=...) when no real key is
6# configured (unauthenticated local LM Studio, llama.cpp, etc.). Subclasses
7# MUST NOT override this constant; per-provider drift here was the reason
8# we previously had three different placeholder strings.
9OPTIONAL_API_KEY_PLACEHOLDER = "not-required" # noqa: S105
12def normalize_provider(provider):
13 """Normalize provider name to lowercase canonical form.
15 All provider comparisons in route/service code should use
16 this function to ensure consistent casing.
17 """
18 return provider.lower() if provider else None
21class BaseLLMProvider:
22 """Base class for all LLM providers.
24 Defines the minimum interface that all providers must satisfy.
25 Subclasses should override the methods below as needed.
27 **Class attribute contract:**
29 - ``api_key_setting`` — settings key for the provider's API key
30 (e.g. ``"llm.openai.api_key"``), or ``None`` for providers that have
31 no API key concept at all.
32 - ``api_key_optional`` — when ``True``, a missing/whitespace-only key
33 returns ``None`` from ``resolve_api_key()`` instead of raising.
34 Construction-time callers can then substitute
35 ``OPTIONAL_API_KEY_PLACEHOLDER`` (typical for local providers like
36 LM Studio that may or may not have auth enabled).
37 - ``provider_name`` — display name used in error messages and logs.
38 Subclasses MUST override this away from the ``"unknown"`` default
39 or auto-discovery will skip them.
41 **Helper methods (use these instead of reading settings directly):**
43 - ``resolve_api_key(snapshot)`` → ``str | None``. Returns the
44 stripped key, ``None`` for optional providers when missing, or
45 raises ``ValueError`` for required providers when missing.
46 - ``resolve_api_key_or_placeholder(snapshot)`` → ``str``. Same as
47 above but substitutes the placeholder when optional+missing — use
48 this when constructing ``ChatOpenAI(api_key=...)``.
49 - ``build_bearer_header(snapshot)`` → ``dict``. Returns
50 ``{"Authorization": "Bearer <key>"}`` or ``{}``. Use this when
51 sending Authorization headers to a provider's REST API.
52 - ``has_api_key(snapshot)`` → ``bool``. Use in ``is_available()``
53 overrides to fail-closed when a required key is missing.
55 **`create_llm()` contract:** subclass implementations return a *bare*
56 LangChain ``BaseChatModel``. Wrapping (rate limiting, token counting,
57 think-tag stripping) is applied by ``llm_config.get_llm()`` after the
58 provider class returns. Do NOT wrap inside ``create_llm()``.
59 """
61 # Settings key for this provider's API key. None means the provider
62 # has no API key concept at all.
63 api_key_setting: str | None = None
65 # Display name used in error messages and logs. Subclasses set this.
66 provider_name: str = "unknown"
68 # When True, a missing/whitespace-only key returns None (callers may
69 # substitute OPTIONAL_API_KEY_PLACEHOLDER) instead of raising.
70 api_key_optional: bool = False
72 # Egress exposure (ADR-0007): declares whether this is an exposing
73 # inference sink (cloud = exposing, local = contained). Declarative default
74 # only — the egress resolver classifies the provider's ACTUAL endpoint at
75 # run time (run_classification.llm_label via evaluate_llm_endpoint), so a
76 # URL-configurable provider is refined then; this attribute documents intent
77 # and is asserted consistent by test.
78 egress_exposure = Exposure.EXPOSING
80 @classmethod
81 def create_llm(cls, model_name=None, temperature=0.7, **kwargs):
82 """Create and return a LangChain chat model instance.
84 Subclasses MUST override this method. The returned LLM is BARE —
85 wrapping (rate limiting, token counting, think-tag stripping) is
86 applied by ``llm_config.get_llm()`` after this returns.
88 Args:
89 model_name: Name of the model to use
90 temperature: Model temperature (0.0-1.0)
91 **kwargs: Additional arguments including settings_snapshot
93 Returns:
94 A configured BaseChatModel instance
96 Raises:
97 NotImplementedError: If not overridden by subclass
98 """
99 raise NotImplementedError(f"{cls.__name__} must implement create_llm()")
101 @classmethod
102 def is_available(cls, settings_snapshot=None):
103 """Check if this provider is available.
105 Returns False by default (fail-closed). Subclasses MUST override
106 this method to implement their own availability logic.
107 """
108 return False
110 @classmethod
111 def requires_auth_for_models(cls):
112 """Whether auth is needed to list models.
114 Returns True by default. Override in subclasses that allow
115 unauthenticated model listing (e.g., local providers).
116 """
117 return True
119 @classmethod
120 def resolve_api_key(cls, settings_snapshot=None) -> str | None:
121 """Read this provider's API key from settings, normalized.
123 Returns:
124 The stripped key if non-empty; None if optional and missing
125 or whitespace-only, or if api_key_setting is None.
126 Raises:
127 ValueError if required (api_key_optional=False) and missing.
128 """
129 from ...config.thread_settings import get_setting_from_snapshot
131 if not cls.api_key_setting:
132 return None
133 raw = get_setting_from_snapshot(
134 cls.api_key_setting, "", settings_snapshot=settings_snapshot
135 )
136 key = str(raw or "").strip()
137 if key:
138 return key
139 if cls.api_key_optional:
140 return None
141 raise ValueError(
142 f"{cls.provider_name} API key not configured. "
143 f"Please set {cls.api_key_setting} in settings."
144 )
146 @classmethod
147 def resolve_api_key_or_placeholder(cls, settings_snapshot=None) -> str:
148 """For ``ChatOpenAI(api_key=...)``-style construction.
150 Always returns a string. Real key when set; otherwise the unified
151 ``OPTIONAL_API_KEY_PLACEHOLDER`` so the langchain client doesn't
152 reject the construction call.
153 """
154 key = cls.resolve_api_key(settings_snapshot)
155 return key if key else OPTIONAL_API_KEY_PLACEHOLDER
157 @classmethod
158 def build_bearer_header(cls, settings_snapshot=None) -> dict[str, str]:
159 """Build an ``Authorization: Bearer`` header from the resolved key.
161 Returns an empty dict when no real key is configured (so callers
162 can spread it conditionally). Required-key providers that raise
163 from ``resolve_api_key`` are treated as "no header" here too —
164 the construction-time error happens elsewhere.
165 """
166 try:
167 key = cls.resolve_api_key(settings_snapshot)
168 except ValueError:
169 return {}
170 return {"Authorization": f"Bearer {key}"} if key else {}
172 @classmethod
173 def has_api_key(cls, settings_snapshot=None) -> bool:
174 """True if a real key is configured.
176 Returns False for missing/whitespace keys (required providers raise
177 ValueError; we swallow it). Also returns False on broader settings
178 errors so callers like ``is_available()`` fail closed instead of
179 propagating settings infrastructure failures.
180 """
181 try:
182 return cls.resolve_api_key(settings_snapshot) is not None
183 except Exception:
184 return False