Coverage for src / local_deep_research / llm / providers / auto_discovery.py: 88%

1"""Auto-discovery system for OpenAI-compatible providers.""" 

2 

3import importlib 

4import inspect 

5from pathlib import Path 

6from typing import Dict, List, Optional 

7 

8from loguru import logger 

9 

10from .openai_base import OpenAICompatibleProvider 

11 

12 

13class ProviderInfo: 

14 """Information about a discovered provider.""" 

15 

16 def __init__(self, provider_class): 

17 self.provider_class = provider_class 

18 self.provider_key = getattr( 

19 provider_class, 

20 "provider_key", 

21 provider_class.__name__.replace("Provider", "").upper(), 

22 ) 

23 self.provider_name = provider_class.provider_name 

24 self.company_name = getattr( 

25 provider_class, "company_name", provider_class.provider_name 

26 ) 

27 self.is_cloud = getattr(provider_class, "is_cloud", True) 

28 # Handle providers that may not have requires_auth_for_models method 

29 if hasattr(provider_class, "requires_auth_for_models"): 

30 self.requires_auth_for_models = ( 

31 provider_class.requires_auth_for_models() 

32 ) 

33 else: 

34 # Default to True for providers without the method 

35 self.requires_auth_for_models = True 

36 

37 # Generate display name from attributes 

38 self.display_name = self._generate_display_name() 

39 

40 def _generate_display_name(self): 

41 """Generate a descriptive display name from provider attributes.""" 

42 # Start with the provider name 

43 name_parts = [self.provider_name] 

44 

45 # Add cloud/local indicator 

46 if self.is_cloud is True: 

47 name_parts.append("☁️ Cloud") 

48 elif self.is_cloud is False: 

49 name_parts.append("💻 Local") 

50 

51 return " ".join(name_parts) 

52 

53 def to_dict(self): 

54 """Convert to dictionary for API responses.""" 

55 return { 

56 "value": self.provider_key, 

57 "label": self.display_name, 

58 "is_cloud": self.is_cloud, 

59 } 

60 

61 

62class ProviderDiscovery: 

63 """Discovers and manages OpenAI-compatible providers.""" 

64 

65 _instance = None 

66 _providers: Dict[str, ProviderInfo] = {} 

67 

68 def __new__(cls): 

69 if cls._instance is None: 

70 cls._instance = super().__new__(cls) 

71 cls._instance._discovered = False 

72 return cls._instance 

73 

74 def discover_providers( 

75 self, force_refresh: bool = False 

76 ) -> Dict[str, ProviderInfo]: 

77 """Discover all providers in the providers directory. 

78 

79 Args: 

80 force_refresh: Force re-discovery even if already done 

81 

82 Returns: 

83 Dictionary mapping provider keys to ProviderInfo objects 

84 """ 

85 if self._discovered and not force_refresh: 

86 return self._providers 

87 

88 self._providers.clear() 

89 # Scan the implementations subdirectory for providers 

90 implementations_dir = Path(__file__).parent / "implementations" 

91 

92 if not implementations_dir.exists(): 92 ↛ 93line 92 didn't jump to line 93 because the condition on line 92 was never true

93 logger.warning( 

94 f"Implementations directory not found: {implementations_dir}" 

95 ) 

96 return self._providers 

97 

98 # Scan all Python files in the implementations directory 

99 logger.info(f"Scanning directory: {implementations_dir}") 

100 for file_path in implementations_dir.glob("*.py"): 

101 # Skip special files (like __init__.py) 

102 if file_path.name.startswith("_"): 102 ↛ 103line 102 didn't jump to line 103 because the condition on line 102 was never true

103 continue 

104 

105 module_name = file_path.stem 

106 logger.debug(f"Processing module: {module_name} from {file_path}") 

107 try: 

108 # Import the module from implementations subdirectory 

109 module = importlib.import_module( 

110 f".implementations.{module_name}", 

111 package="local_deep_research.llm.providers", 

112 ) 

113 

114 # Find all Provider classes (both OpenAICompatibleProvider and standalone) 

115 logger.debug( 

116 f"Inspecting module {module_name} for Provider classes" 

117 ) 

118 for name, obj in inspect.getmembers(module, inspect.isclass): 

119 if inspect.isclass(obj): 119 ↛ 124line 119 didn't jump to line 124 because the condition on line 119 was always true

120 logger.debug( 

121 f" Found class: {name}, bases: {obj.__bases__}" 

122 ) 

123 # Check if it's a Provider class (ends with "Provider" and has provider_name) 

124 if ( 

125 name.endswith("Provider") 

126 and hasattr(obj, "provider_name") 

127 and obj is not OpenAICompatibleProvider 

128 ): 

129 # Found a provider class 

130 provider_info = ProviderInfo(obj) 

131 self._providers[provider_info.provider_key] = ( 

132 provider_info 

133 ) 

134 

135 # Auto-register the provider 

136 register_func_name = f"register_{module_name}_provider" 

137 try: 

138 register_func = getattr(module, register_func_name) 

139 register_func() 

140 logger.info( 

141 f"Auto-registered provider: {provider_info.provider_key}" 

142 ) 

143 except AttributeError: 

144 logger.warning( 

145 f"Provider {provider_info.provider_key} from {module_name}.py " 

146 f"does not have a {register_func_name} function" 

147 ) 

148 

149 logger.info( 

150 f"Discovered provider: {provider_info.provider_key} from {module_name}.py" 

151 ) 

152 

153 except Exception: 

154 logger.exception(f"Error loading provider from {module_name}") 

155 

156 self._discovered = True 

157 logger.info(f"Discovered {len(self._providers)} providers") 

158 return self._providers 

159 

160 def get_provider_info(self, provider_key: str) -> Optional[ProviderInfo]: 

161 """Get information about a specific provider. 

162 

163 Args: 

164 provider_key: The provider key (e.g., 'IONOS', 'GOOGLE') 

165 

166 Returns: 

167 ProviderInfo object or None if not found 

168 """ 

169 if not self._discovered: 169 ↛ 170line 169 didn't jump to line 170 because the condition on line 169 was never true

170 self.discover_providers() 

171 return self._providers.get(provider_key.upper()) 

172 

173 def get_provider_options(self) -> List[Dict]: 

174 """Get list of provider options for UI dropdowns. 

175 

176 Returns: 

177 List of dictionaries with 'value' and 'label' keys 

178 """ 

179 if not self._discovered: 179 ↛ 180line 179 didn't jump to line 180 because the condition on line 179 was never true

180 self.discover_providers() 

181 

182 options = [] 

183 for provider_info in self._providers.values(): 

184 options.append(provider_info.to_dict()) 

185 

186 # Sort by label 

187 options.sort(key=lambda x: x["label"]) 

188 return options 

189 

190 def get_provider_class(self, provider_key: str): 

191 """Get the provider class for a given key. 

192 

193 Args: 

194 provider_key: The provider key (e.g., 'IONOS', 'GOOGLE') 

195 

196 Returns: 

197 Provider class or None if not found 

198 """ 

199 provider_info = self.get_provider_info(provider_key) 

200 return provider_info.provider_class if provider_info else None 

201 

202 

203# Global instance 

204provider_discovery = ProviderDiscovery() 

205 

206 

207def discover_providers(force_refresh: bool = False) -> Dict[str, ProviderInfo]: 

208 """Discover all available providers. 

209 

210 Args: 

211 force_refresh: Force re-discovery even if already done 

212 

213 Returns: 

214 Dictionary mapping provider keys to ProviderInfo objects 

215 """ 

216 return provider_discovery.discover_providers(force_refresh) 

217 

218 

219def get_discovered_provider_options() -> List[Dict]: 

220 """Get list of discovered provider options for UI dropdowns. 

221 

222 Returns: 

223 List of dictionaries with 'value' and 'label' keys 

224 """ 

225 return provider_discovery.get_provider_options() 

226 

227 

228def get_provider_class(provider_key: str): 

229 """Get the provider class for a given key. 

230 

231 Args: 

232 provider_key: The provider key (e.g., 'IONOS', 'GOOGLE') 

233 

234 Returns: 

235 Provider class or None if not found 

236 """ 

237 return provider_discovery.get_provider_class(provider_key)