Coverage for src / local_deep_research / utilities / llm_utils.py: 85%
109 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-01-11 00:51 +0000
1# utilities/llm_utils.py
2"""
3LLM utilities for Local Deep Research.
5This module provides utility functions for working with language models
6when the user's llm_config.py is missing or incomplete.
7"""
9from loguru import logger
10from typing import Any, Optional, Dict
12from ..config.thread_settings import get_setting_from_snapshot
15def get_ollama_base_url(
16 settings_snapshot: Optional[Dict[str, Any]] = None,
17) -> str:
18 """
19 Get Ollama base URL from settings with normalization.
21 Checks both embeddings.ollama.url and llm.ollama.url settings,
22 falling back to http://localhost:11434.
24 Args:
25 settings_snapshot: Optional settings snapshot
27 Returns:
28 Normalized Ollama base URL
29 """
30 from .url_utils import normalize_url
32 raw_base_url = get_setting_from_snapshot(
33 "embeddings.ollama.url",
34 default=get_setting_from_snapshot(
35 "llm.ollama.url", # Fall back to LLM setting
36 default="http://localhost:11434",
37 settings_snapshot=settings_snapshot,
38 ),
39 settings_snapshot=settings_snapshot,
40 )
41 return (
42 normalize_url(raw_base_url)
43 if raw_base_url
44 else "http://localhost:11434"
45 )
48def get_server_url(settings_snapshot: Optional[Dict[str, Any]] = None) -> str:
49 """
50 Get server URL from settings with fallback logic.
52 Checks multiple sources in order:
53 1. Direct server_url in settings snapshot
54 2. system.server_url in settings
55 3. Constructs from web.host, web.port, and web.use_https
56 4. Fallback to http://127.0.0.1:5000/
58 Args:
59 settings_snapshot: Optional settings snapshot
61 Returns:
62 Server URL with trailing slash
63 """
64 from loguru import logger
66 server_url = None
68 if settings_snapshot:
69 # Try to get server URL from research metadata first (where we added it)
70 server_url = settings_snapshot.get("server_url")
72 # If not found, try system settings
73 if not server_url:
74 system_settings = settings_snapshot.get("system", {})
75 server_url = system_settings.get("server_url")
77 # If not found, try web.host and web.port settings
78 if not server_url:
79 host = get_setting_from_snapshot(
80 "web.host", settings_snapshot, "127.0.0.1"
81 )
82 port = get_setting_from_snapshot(
83 "web.port", settings_snapshot, 5000
84 )
85 use_https = get_setting_from_snapshot(
86 "web.use_https", settings_snapshot, True
87 )
89 # Use localhost for 0.0.0.0 bindings as that's what users will use
90 if host == "0.0.0.0":
91 host = "127.0.0.1"
93 scheme = "https" if use_https else "http"
94 server_url = f"{scheme}://{host}:{port}/"
96 # Fallback to default if still not found
97 if not server_url:
98 server_url = "http://127.0.0.1:5000/"
99 logger.warning("Could not determine server URL, using default")
101 return server_url
104def fetch_ollama_models(
105 base_url: str,
106 timeout: float = 3.0,
107 auth_headers: Optional[Dict[str, str]] = None,
108) -> list[Dict[str, str]]:
109 """
110 Fetch available models from Ollama API.
112 Centralized function to avoid duplication between LLM and embedding providers.
114 Args:
115 base_url: Ollama base URL (should be normalized)
116 timeout: Request timeout in seconds
117 auth_headers: Optional authentication headers
119 Returns:
120 List of model dicts with 'value' (model name) and 'label' (display name) keys.
121 Returns empty list on error.
122 """
123 from loguru import logger
124 from ..security import safe_get
126 models = []
128 try:
129 response = safe_get(
130 f"{base_url}/api/tags",
131 timeout=timeout,
132 headers=auth_headers or {},
133 allow_localhost=True,
134 allow_private_ips=True,
135 )
137 if response.status_code == 200:
138 data = response.json()
140 # Handle both newer and older Ollama API formats
141 ollama_models = (
142 data.get("models", []) if isinstance(data, dict) else data
143 )
145 for model_data in ollama_models:
146 model_name = model_data.get("name", "")
147 if model_name:
148 models.append({"value": model_name, "label": model_name})
150 logger.info(f"Found {len(models)} Ollama models")
151 else:
152 logger.warning(
153 f"Failed to fetch Ollama models: HTTP {response.status_code}"
154 )
156 except Exception:
157 logger.exception("Error fetching Ollama models")
159 return models
162def get_model(
163 model_name: Optional[str] = None,
164 model_type: Optional[str] = None,
165 temperature: Optional[float] = None,
166 **kwargs,
167) -> Any:
168 """
169 Get a language model instance as fallback when llm_config.get_llm is not available.
171 Args:
172 model_name: Name of the model to use
173 model_type: Type of the model provider
174 temperature: Model temperature
175 **kwargs: Additional parameters
177 Returns:
178 LangChain language model instance
179 """
180 # Get default values from kwargs or use reasonable defaults
181 model_name = model_name or kwargs.get("DEFAULT_MODEL", "mistral")
182 model_type = model_type or kwargs.get("DEFAULT_MODEL_TYPE", "ollama")
183 temperature = temperature or kwargs.get("DEFAULT_TEMPERATURE", 0.7)
184 max_tokens = kwargs.get("max_tokens", kwargs.get("MAX_TOKENS", 30000))
186 # Common parameters
187 common_params = {
188 "temperature": temperature,
189 "max_tokens": max_tokens,
190 }
192 # Add additional kwargs
193 for key, value in kwargs.items():
194 if key not in [
195 "DEFAULT_MODEL",
196 "DEFAULT_MODEL_TYPE",
197 "DEFAULT_TEMPERATURE",
198 "MAX_TOKENS",
199 ]:
200 common_params[key] = value
202 # Try to load the model based on type
203 if model_type == "ollama":
204 try:
205 from langchain_ollama import ChatOllama
207 return ChatOllama(model=model_name, **common_params)
208 except ImportError:
209 try:
210 from langchain_community.llms import Ollama
212 return Ollama(model=model_name, **common_params)
213 except ImportError:
214 logger.exception(
215 "Neither langchain_ollama nor langchain_community.llms.Ollama available"
216 )
217 raise
219 elif model_type == "openai":
220 try:
221 from langchain_openai import ChatOpenAI
223 api_key = get_setting_from_snapshot("llm.openai.api_key")
224 if not api_key:
225 raise ValueError("OpenAI API key not found in settings")
226 return ChatOpenAI(
227 model=model_name, api_key=api_key, **common_params
228 )
229 except ImportError:
230 logger.exception("langchain_openai not available")
231 raise
233 elif model_type == "anthropic":
234 try:
235 from langchain_anthropic import ChatAnthropic
237 api_key = get_setting_from_snapshot("llm.anthropic.api_key")
238 if not api_key:
239 raise ValueError("Anthropic API key not found in settings")
240 return ChatAnthropic(
241 model=model_name, anthropic_api_key=api_key, **common_params
242 )
243 except ImportError:
244 logger.exception("langchain_anthropic not available")
245 raise
247 elif model_type == "openai_endpoint":
248 try:
249 from langchain_openai import ChatOpenAI
251 api_key = get_setting_from_snapshot("llm.openai_endpoint.api_key")
252 if not api_key:
253 raise ValueError(
254 "OpenAI endpoint API key not found in settings"
255 )
257 endpoint_url = kwargs.get(
258 "OPENAI_ENDPOINT_URL",
259 get_setting_from_snapshot(
260 "llm.openai_endpoint.url", "https://openrouter.ai/api/v1"
261 ),
262 )
264 if model_name is None and not kwargs.get( 264 ↛ 267line 264 didn't jump to line 267 because the condition on line 264 was never true
265 "OPENAI_ENDPOINT_REQUIRES_MODEL", True
266 ):
267 return ChatOpenAI(
268 api_key=api_key,
269 openai_api_base=endpoint_url,
270 **common_params,
271 )
272 else:
273 return ChatOpenAI(
274 model=model_name,
275 api_key=api_key,
276 openai_api_base=endpoint_url,
277 **common_params,
278 )
279 except ImportError:
280 logger.exception("langchain_openai not available")
281 raise
283 # Default fallback
284 try:
285 from langchain_ollama import ChatOllama
287 logger.warning(
288 f"Unknown model type '{model_type}', defaulting to Ollama"
289 )
290 return ChatOllama(model=model_name, **common_params)
291 except (ImportError, Exception):
292 logger.exception("Failed to load any model")
294 # Last resort: create a dummy model
295 try:
296 from langchain_community.llms.fake import FakeListLLM
298 return FakeListLLM(
299 responses=[
300 "No language models are available. Please install Ollama or set up API keys."
301 ]
302 )
303 except ImportError:
304 raise ValueError(
305 "No language models available and could not create dummy model"
306 )