Coverage for src / local_deep_research / llm / providers / openai_base.py: 97%
120 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-14 23:55 +0000
1"""Base OpenAI-compatible endpoint provider for Local Deep Research."""
3from langchain_openai import ChatOpenAI
4from loguru import logger
6from ...config.thread_settings import (
7 get_setting_from_snapshot,
8 NoSettingsContextError,
9)
10from ...utilities.url_utils import normalize_url
11from .base import BaseLLMProvider
14class OpenAICompatibleProvider(BaseLLMProvider):
15 """Base class for OpenAI-compatible API providers.
17 This class provides a common implementation for any service that offers
18 an OpenAI-compatible API endpoint (Google, OpenRouter, Groq, Together, etc.)
19 """
21 # Override these in subclasses
22 provider_name = "openai_endpoint" # Name used in logs
23 api_key_setting = "llm.openai_endpoint.api_key" # Settings key for API key
24 url_setting = None # Settings key for URL (e.g., "llm.lmstudio.url")
25 default_base_url = "https://api.openai.com/v1" # Default endpoint URL
26 default_model = "gpt-3.5-turbo" # Default model if none specified
28 @classmethod
29 def create_llm(cls, model_name=None, temperature=0.7, **kwargs):
30 """Factory function for OpenAI-compatible LLMs.
32 Args:
33 model_name: Name of the model to use
34 temperature: Model temperature (0.0-1.0)
35 **kwargs: Additional arguments including settings_snapshot
37 Returns:
38 A configured ChatOpenAI instance
40 Raises:
41 ValueError: If API key is not configured
42 """
43 settings_snapshot = kwargs.get("settings_snapshot")
45 # Get API key from settings (if provider requires one)
46 if cls.api_key_setting:
47 api_key = get_setting_from_snapshot(
48 cls.api_key_setting,
49 default=None,
50 settings_snapshot=settings_snapshot,
51 )
53 if not api_key:
54 logger.error(
55 f"{cls.provider_name} API key not found in settings"
56 )
57 raise ValueError(
58 f"{cls.provider_name} API key not configured. "
59 f"Please set {cls.api_key_setting} in settings."
60 )
61 else:
62 # Provider doesn't require API key (e.g., LM Studio)
63 api_key = kwargs.get("api_key", "dummy-key")
65 # Use default model if none specified
66 if not model_name:
67 model_name = cls.default_model
69 # Get endpoint URL (can be overridden in kwargs for flexibility)
70 base_url = kwargs.get("base_url", cls.default_base_url)
71 base_url = normalize_url(base_url) if base_url else cls.default_base_url
73 # Build parameters for OpenAI client
74 llm_params = {
75 "model": model_name,
76 "api_key": api_key,
77 "base_url": base_url,
78 "temperature": temperature,
79 }
81 # Add max_tokens if specified in settings
82 try:
83 max_tokens = get_setting_from_snapshot(
84 "llm.max_tokens",
85 default=None,
86 settings_snapshot=settings_snapshot,
87 )
88 if max_tokens:
89 llm_params["max_tokens"] = int(max_tokens)
90 except NoSettingsContextError:
91 pass # Optional parameter
93 # Add streaming if specified
94 try:
95 streaming = get_setting_from_snapshot(
96 "llm.streaming",
97 default=None,
98 settings_snapshot=settings_snapshot,
99 )
100 if streaming is not None:
101 llm_params["streaming"] = streaming
102 except NoSettingsContextError:
103 pass # Optional parameter
105 # Add max_retries if specified
106 try:
107 max_retries = get_setting_from_snapshot(
108 "llm.max_retries",
109 default=None,
110 settings_snapshot=settings_snapshot,
111 )
112 if max_retries is not None:
113 llm_params["max_retries"] = max_retries
114 except NoSettingsContextError:
115 pass # Optional parameter
117 # Add request_timeout if specified
118 try:
119 request_timeout = get_setting_from_snapshot(
120 "llm.request_timeout",
121 default=None,
122 settings_snapshot=settings_snapshot,
123 )
124 if request_timeout is not None:
125 llm_params["request_timeout"] = request_timeout
126 except NoSettingsContextError:
127 pass # Optional parameter
129 logger.info(
130 f"Creating {cls.provider_name} LLM with model: {model_name}, "
131 f"temperature: {temperature}, endpoint: {base_url}"
132 )
134 return ChatOpenAI(**llm_params)
136 @classmethod
137 def _create_llm_instance(cls, model_name=None, temperature=0.7, **kwargs):
138 """Internal method to create LLM instance with provided parameters.
140 This bypasses API key checking for providers that handle auth differently.
141 """
142 settings_snapshot = kwargs.get("settings_snapshot")
144 # Use default model if none specified
145 if not model_name:
146 model_name = cls.default_model
148 # Get endpoint URL (can be overridden in kwargs for flexibility)
149 base_url = kwargs.get("base_url", cls.default_base_url)
150 base_url = normalize_url(base_url) if base_url else cls.default_base_url
152 # Get API key from kwargs (caller is responsible for providing it)
153 api_key = kwargs.get("api_key", "dummy-key")
155 # Build parameters for OpenAI client
156 llm_params = {
157 "model": model_name,
158 "api_key": api_key,
159 "base_url": base_url,
160 "temperature": temperature,
161 }
163 # Add optional parameters (same as in create_llm)
164 try:
165 max_tokens = get_setting_from_snapshot(
166 "llm.max_tokens",
167 default=None,
168 settings_snapshot=settings_snapshot,
169 )
170 if max_tokens: 170 ↛ 171line 170 didn't jump to line 171 because the condition on line 170 was never true
171 llm_params["max_tokens"] = int(max_tokens)
172 except NoSettingsContextError:
173 pass
175 return ChatOpenAI(**llm_params)
177 @classmethod
178 def is_available(cls, settings_snapshot=None):
179 """Check if this provider is available.
181 Args:
182 settings_snapshot: Optional settings snapshot to use
184 Returns:
185 True if API key is configured (or not needed), False otherwise
186 """
187 try:
188 # If provider doesn't require API key, it's available
189 if not cls.api_key_setting:
190 return True
192 # Check if API key is configured
193 api_key = get_setting_from_snapshot(
194 cls.api_key_setting,
195 default=None,
196 settings_snapshot=settings_snapshot,
197 )
198 return bool(api_key and str(api_key).strip())
199 except Exception:
200 return False
202 @classmethod
203 def requires_auth_for_models(cls):
204 """Check if this provider requires authentication for listing models.
206 Override in subclasses that don't require auth.
208 Returns:
209 True if authentication is required, False otherwise
210 """
211 return True
213 # Resolves base URL from settings; called by list_models().
214 @classmethod
215 def _get_base_url_for_models(cls, settings_snapshot=None):
216 """Get the base URL to use for listing models.
218 Reads from url_setting if defined, otherwise uses default_base_url.
220 Args:
221 settings_snapshot: Optional settings snapshot dict
223 Returns:
224 The base URL string to use for model listing
225 """
226 if cls.url_setting:
227 # Use get_setting_from_snapshot which handles both settings_snapshot
228 # and thread-local context, with proper fallback
229 url = get_setting_from_snapshot(
230 cls.url_setting,
231 default=None,
232 settings_snapshot=settings_snapshot,
233 )
234 if url: 234 ↛ 237line 234 didn't jump to line 237 because the condition on line 234 was always true
235 return url.rstrip("/")
237 return cls.default_base_url
239 @classmethod
240 def list_models_for_api(cls, api_key=None, base_url=None):
241 """List available models for API endpoint use.
243 This method is designed to be called from Flask routes.
245 Args:
246 api_key: Optional API key (if None and required, returns empty list)
247 base_url: Optional base URL to use (if None, uses cls.default_base_url)
249 Returns:
250 List of model dictionaries with 'value' and 'label' keys
251 """
252 try:
253 # Check if auth is required
254 if cls.requires_auth_for_models():
255 if not api_key:
256 logger.debug(
257 f"{cls.provider_name} requires API key for model listing"
258 )
259 return []
260 else:
261 # Use a dummy key for providers that don't require auth
262 api_key = api_key or "dummy-key-for-models-list"
264 from openai import OpenAI
266 # Use provided base_url or fall back to class default
267 if not base_url:
268 base_url = cls.default_base_url
270 # Create OpenAI client (uses library defaults for timeout)
271 client = OpenAI(api_key=api_key, base_url=base_url)
273 # Fetch models
274 logger.debug(
275 f"Fetching models from {cls.provider_name} at {base_url}"
276 )
277 models_response = client.models.list()
279 models = []
280 for model in models_response.data:
281 if model.id: 281 ↛ 280line 281 didn't jump to line 280 because the condition on line 281 was always true
282 models.append(
283 {
284 "value": model.id,
285 "label": model.id,
286 }
287 )
289 logger.info(f"Found {len(models)} models from {cls.provider_name}")
290 return models
292 except Exception:
293 # Use warning level since connection failures are expected
294 # when the provider is not running (e.g., LM Studio not started)
295 logger.warning(f"Could not list models from {cls.provider_name}")
296 return []
298 # High-level settings-aware wrapper around list_models_for_api().
299 # Documented in docs/developing/EXTENDING.md as the provider interface
300 # for custom providers.
301 @classmethod
302 def list_models(cls, settings_snapshot=None):
303 """List available models from this provider.
305 Args:
306 settings_snapshot: Optional settings snapshot to use
308 Returns:
309 List of model dictionaries with 'value' and 'label' keys
310 """
311 try:
312 # Get API key from settings if auth is required
313 api_key = None
314 if cls.requires_auth_for_models(): 314 ↛ 322line 314 didn't jump to line 322 because the condition on line 314 was always true
315 api_key = get_setting_from_snapshot(
316 cls.api_key_setting,
317 default=None,
318 settings_snapshot=settings_snapshot,
319 )
321 # Get base URL from settings if provider has configurable URL
322 base_url = cls._get_base_url_for_models(settings_snapshot)
324 return cls.list_models_for_api(api_key, base_url)
326 except Exception:
327 logger.exception(f"Error listing models from {cls.provider_name}")
328 return []