Coverage for src/local_deep_research/llm/providers/auto_discovery.py: 98%
111 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""Auto-discovery system for OpenAI-compatible providers."""
3import importlib
4import inspect
5from pathlib import Path
6from typing import Dict, List, Optional
8from ...security.secure_logging import logger
9from .base import BaseLLMProvider, normalize_provider
10from .openai_base import OpenAICompatibleProvider
11from ..llm_registry import register_llm
12from ...security.log_sanitizer import scrub_error
15class ProviderInfo:
16 """Information about a discovered provider."""
18 def __init__(self, provider_class):
19 self.provider_class = provider_class
20 self.provider_key = getattr(
21 provider_class,
22 "provider_key",
23 provider_class.__name__.replace("Provider", "").upper(),
24 )
25 self.provider_name = provider_class.provider_name
26 self.company_name = getattr(
27 provider_class, "company_name", provider_class.provider_name
28 )
29 self.is_cloud = getattr(provider_class, "is_cloud", True)
30 self.requires_auth_for_models = (
31 provider_class.requires_auth_for_models()
32 )
34 # Generate display name from attributes
35 self.display_name = self._generate_display_name()
37 def _generate_display_name(self):
38 """Generate a descriptive display name from provider attributes."""
39 # Start with the provider name
40 name_parts = [self.provider_name]
42 # Add cloud/local indicator
43 if self.is_cloud is True:
44 name_parts.append("☁️ Cloud")
45 elif self.is_cloud is False:
46 name_parts.append("💻 Local")
48 return " ".join(name_parts)
50 def to_dict(self):
51 """Convert to dictionary for API responses."""
52 return {
53 "value": self.provider_key,
54 "label": self.display_name,
55 "is_cloud": self.is_cloud,
56 }
59class ProviderDiscovery:
60 """Discovers and manages OpenAI-compatible providers."""
62 _instance = None
63 _providers: Dict[str, ProviderInfo] = {}
64 _discovered: bool = False
66 def __new__(cls):
67 if cls._instance is None:
68 cls._instance = super().__new__(cls)
69 cls._instance._discovered = False
70 return cls._instance
72 def discover_providers(
73 self, force_refresh: bool = False
74 ) -> Dict[str, ProviderInfo]:
75 """Discover all providers in the providers directory.
77 Args:
78 force_refresh: Force re-discovery even if already done
80 Returns:
81 Dictionary mapping provider keys to ProviderInfo objects
82 """
83 if self._discovered and not force_refresh:
84 return self._providers
86 self._providers.clear()
87 # Scan the implementations subdirectory for providers
88 implementations_dir = Path(__file__).parent / "implementations"
90 if not implementations_dir.exists():
91 logger.warning(
92 f"Implementations directory not found: {implementations_dir}"
93 )
94 return self._providers
96 # Scan all Python files in the implementations directory
97 logger.info(f"Scanning directory: {implementations_dir}")
98 # nosemgrep: semgrep.rules.path-traversal-risk, reason: Provider modules are discovered beneath the installed package __file__ directory
99 for file_path in implementations_dir.glob("*.py"):
100 # Skip special files (like __init__.py)
101 if file_path.name.startswith("_"):
102 continue
104 module_name = file_path.stem
105 logger.debug(f"Processing module: {module_name} from {file_path}")
106 try:
107 # Import the module from implementations subdirectory
108 module = importlib.import_module(
109 f".implementations.{module_name}",
110 package="local_deep_research.llm.providers",
111 )
113 # Find all Provider classes (both OpenAICompatibleProvider and standalone)
114 logger.debug(
115 f"Inspecting module {module_name} for Provider classes"
116 )
117 for name, obj in inspect.getmembers(module, inspect.isclass):
118 if inspect.isclass(obj): 118 ↛ 126line 118 didn't jump to line 126 because the condition on line 118 was always true
119 logger.debug(
120 f" Found class: {name}, bases: {obj.__bases__}"
121 )
122 # Check if it's a Provider class with a real provider_name.
123 # BaseLLMProvider now sets a default "unknown" so all
124 # subclasses inherit the attribute; skip ones that
125 # haven't customized it (incomplete fixture classes etc.).
126 if (
127 name.endswith("Provider")
128 and getattr(obj, "provider_name", "unknown")
129 not in (None, "", "unknown")
130 and issubclass(obj, BaseLLMProvider)
131 and obj is not OpenAICompatibleProvider
132 and obj is not BaseLLMProvider
133 # Only register classes DEFINED in this module, not ones
134 # merely imported into its namespace (e.g.
135 # custom_anthropic_endpoint imports AnthropicProvider as
136 # its base). Without this, the imported base would be
137 # re-registered from every module that imports it.
138 and obj.__module__ == module.__name__
139 ):
140 # Found a provider class
141 provider_info = ProviderInfo(obj)
142 self._providers[provider_info.provider_key] = (
143 provider_info
144 )
146 # Auto-register the provider directly using the class
147 register_llm(
148 normalize_provider(provider_info.provider_key),
149 obj.create_llm,
150 )
151 logger.info(
152 f"Auto-registered provider: {provider_info.provider_key}"
153 )
155 logger.info(
156 f"Discovered provider: {provider_info.provider_key} from {module_name}.py"
157 )
159 except Exception as e:
160 safe_msg = scrub_error(e)
161 logger.exception(
162 f"Error loading provider from {module_name} ({type(e).__name__}): {safe_msg}"
163 )
165 self._discovered = True
166 logger.info(f"Discovered {len(self._providers)} providers")
167 return self._providers
169 def get_provider_info(self, provider_key: str) -> Optional[ProviderInfo]:
170 """Get information about a specific provider.
172 Args:
173 provider_key: The provider key (e.g., 'IONOS', 'GOOGLE')
175 Returns:
176 ProviderInfo object or None if not found
177 """
178 if not self._discovered:
179 self.discover_providers()
180 return self._providers.get(provider_key.upper())
182 def get_provider_options(self) -> List[Dict]:
183 """Get list of provider options for UI dropdowns.
185 Returns:
186 List of dictionaries with 'value' and 'label' keys
187 """
188 if not self._discovered:
189 self.discover_providers()
191 options = []
192 for provider_info in self._providers.values():
193 options.append(provider_info.to_dict())
195 # Sort by label
196 options.sort(key=lambda x: x["label"])
197 return options
199 def get_available_provider_options(
200 self, settings_snapshot=None
201 ) -> List[Dict]:
202 """Get list of available provider options, filtered by availability.
204 Filters out providers that are not available (e.g., missing API keys).
205 Useful for contexts where only usable providers should be shown
206 (e.g., starting a research). For settings/configuration UIs, prefer
207 get_provider_options() so users can discover and configure new providers.
209 Args:
210 settings_snapshot: Settings snapshot for checking provider availability.
211 Should be provided to correctly check cloud provider API keys.
213 Returns:
214 List of dictionaries with 'value' and 'label' keys
215 """
216 if not self._discovered: 216 ↛ 217line 216 didn't jump to line 217 because the condition on line 216 was never true
217 self.discover_providers()
219 options = []
220 for provider_info in self._providers.values():
221 if not provider_info.provider_class.is_available(
222 settings_snapshot=settings_snapshot
223 ):
224 logger.debug(
225 f"Provider {provider_info.provider_key} filtered out "
226 f"(not available)"
227 )
228 continue
229 options.append(provider_info.to_dict())
231 if not options:
232 logger.warning(
233 "No auto-discovered providers passed availability filter. "
234 "Check that API keys are configured for cloud providers."
235 )
237 # Sort by label
238 options.sort(key=lambda x: x["label"])
239 return options
241 def get_provider_class(self, provider_key: str):
242 """Get the provider class for a given key.
244 Args:
245 provider_key: The provider key (e.g., 'IONOS', 'GOOGLE')
247 Returns:
248 Provider class or None if not found
249 """
250 provider_info = self.get_provider_info(provider_key)
251 return provider_info.provider_class if provider_info else None
254# Global instance
255provider_discovery = ProviderDiscovery()
258def discover_providers(force_refresh: bool = False) -> Dict[str, ProviderInfo]:
259 """Discover all available providers.
261 Args:
262 force_refresh: Force re-discovery even if already done
264 Returns:
265 Dictionary mapping provider keys to ProviderInfo objects
266 """
267 return provider_discovery.discover_providers(force_refresh)
270def get_discovered_provider_options() -> List[Dict]:
271 """Get list of discovered provider options for UI dropdowns.
273 Returns:
274 List of dictionaries with 'value' and 'label' keys
275 """
276 return provider_discovery.get_provider_options()
279def get_available_discovered_provider_options(
280 settings_snapshot=None,
281) -> List[Dict]:
282 """Get list of available provider options, filtered by availability.
284 Only returns providers that pass is_available() check. Useful for
285 contexts where only usable providers matter (e.g., starting a research).
286 For settings/configuration UIs, use get_discovered_provider_options()
287 instead so users can discover and configure new providers.
289 Args:
290 settings_snapshot: Settings snapshot for checking provider availability
292 Returns:
293 List of dictionaries with 'value' and 'label' keys
294 """
295 return provider_discovery.get_available_provider_options(settings_snapshot)
298def get_provider_class(provider_key: str):
299 """Get the provider class for a given key.
301 Args:
302 provider_key: The provider key (e.g., 'IONOS', 'GOOGLE')
304 Returns:
305 Provider class or None if not found
306 """
307 return provider_discovery.get_provider_class(provider_key)
310_IMPLEMENTATIONS_PACKAGE = ".".join((__package__, "implementations"))
313def is_builtin_provider_factory(factory) -> bool:
314 """Whether *factory* is a built-in provider's ``create_llm``.
316 ``discover_providers()`` registers the ``create_llm`` classmethod of
317 each provider class defined in ``llm/providers/implementations``, and
318 every one of them builds a new model on each call. A factory registered
319 any other way, through ``register_llm()`` or the ``llms=`` argument of
320 the programmatic API, can return a model that other researches share,
321 such as a cached or singleton model.
322 """
323 provider_class = getattr(factory, "__self__", None)
324 if not (
325 isinstance(provider_class, type)
326 and issubclass(provider_class, BaseLLMProvider)
327 and provider_class.__module__.startswith(f"{_IMPLEMENTATIONS_PACKAGE}.")
328 ):
329 return False
330 return getattr(factory, "__func__", None) is getattr(
331 provider_class.create_llm, "__func__", None
332 )