Coverage for src / local_deep_research / utilities / llm_utils.py: 92%

1# utilities/llm_utils.py 

2""" 

3LLM utilities for Local Deep Research. 

4 

5This module provides utility functions for working with language models 

6when the user's llm_config.py is missing or incomplete. 

7""" 

8 

9from loguru import logger 

10from typing import Any, Optional, Dict 

11 

12from ..config.constants import DEFAULT_OLLAMA_URL 

13from ..config.thread_settings import get_setting_from_snapshot 

14 

15 

16def _close_base_llm(llm): 

17 """Close per-instance HTTP clients on a raw LLM. Internal use only. 

18 

19 Only ChatOllama creates per-instance httpx.Client objects. 

20 ChatAnthropic and ChatOpenAI use @lru_cache'd shared httpx clients 

21 that must NOT be closed. 

22 """ 

23 # If the llm is another wrapper with its own close(), delegate 

24 if hasattr(type(llm), "close"): 

25 llm.close() 

26 return 

27 # Otherwise introspect for Ollama's per-instance httpx client 

28 ollama_client = getattr(llm, "_client", None) 

29 if ollama_client is None: 

30 return 

31 if not type(ollama_client).__module__.startswith("ollama"): 

32 return 

33 httpx_client = getattr(ollama_client, "_client", None) 

34 if httpx_client is not None and hasattr(httpx_client, "close"): 

35 httpx_client.close() 

36 

37 

38def get_ollama_base_url( 

39 settings_snapshot: Optional[Dict[str, Any]] = None, 

40) -> str: 

41 """ 

42 Get Ollama base URL from settings with normalization. 

43 

44 Checks both embeddings.ollama.url and llm.ollama.url settings, 

45 falling back to http://localhost:11434. 

46 

47 Args: 

48 settings_snapshot: Optional settings snapshot 

49 

50 Returns: 

51 Normalized Ollama base URL 

52 """ 

53 from .url_utils import normalize_url 

54 

55 raw_base_url = get_setting_from_snapshot( 

56 "embeddings.ollama.url", 

57 default=get_setting_from_snapshot( 

58 "llm.ollama.url", # Fall back to LLM setting 

59 default=DEFAULT_OLLAMA_URL, 

60 settings_snapshot=settings_snapshot, 

61 ), 

62 settings_snapshot=settings_snapshot, 

63 ) 

64 return normalize_url(raw_base_url) if raw_base_url else DEFAULT_OLLAMA_URL 

65 

66 

67def get_server_url(settings_snapshot: Optional[Dict[str, Any]] = None) -> str: 

68 """ 

69 Get server URL from settings with fallback logic. 

70 

71 Checks multiple sources in order: 

72 1. Direct server_url in settings snapshot 

73 2. system.server_url in settings 

74 3. Constructs from web.host, web.port, and web.use_https 

75 4. Fallback to http://127.0.0.1:5000/ 

76 

77 Args: 

78 settings_snapshot: Optional settings snapshot 

79 

80 Returns: 

81 Server URL with trailing slash 

82 """ 

83 from loguru import logger 

84 

85 server_url = None 

86 

87 if settings_snapshot: 

88 # Try to get server URL from research metadata first (where we added it) 

89 server_url = settings_snapshot.get("server_url") 

90 

91 # If not found, try system settings 

92 if not server_url: 

93 system_settings = settings_snapshot.get("system", {}) 

94 server_url = system_settings.get("server_url") 

95 

96 # If not found, try web.host and web.port settings 

97 if not server_url: 

98 host = get_setting_from_snapshot( 

99 "web.host", settings_snapshot, "127.0.0.1" 

100 ) 

101 port = get_setting_from_snapshot( 

102 "web.port", settings_snapshot, 5000 

103 ) 

104 use_https = get_setting_from_snapshot( 

105 "web.use_https", settings_snapshot, True 

106 ) 

107 

108 # Use localhost for 0.0.0.0 bindings as that's what users will use 

109 if host == "0.0.0.0": 

110 host = "127.0.0.1" 

111 

112 scheme = "https" if use_https else "http" 

113 server_url = f"{scheme}://{host}:{port}/" 

114 

115 # Fallback to default if still not found 

116 if not server_url: 

117 server_url = "http://127.0.0.1:5000/" 

118 logger.warning("Could not determine server URL, using default") 

119 

120 return server_url 

121 

122 

123def fetch_ollama_models( 

124 base_url: str, 

125 timeout: float = 3.0, 

126 auth_headers: Optional[Dict[str, str]] = None, 

127) -> list[Dict[str, str]]: 

128 """ 

129 Fetch available models from Ollama API. 

130 

131 Centralized function to avoid duplication between LLM and embedding providers. 

132 

133 Args: 

134 base_url: Ollama base URL (should be normalized) 

135 timeout: Request timeout in seconds 

136 auth_headers: Optional authentication headers 

137 

138 Returns: 

139 List of model dicts with 'value' (model name) and 'label' (display name) keys. 

140 Returns empty list on error. 

141 """ 

142 from loguru import logger 

143 from ..security import safe_get 

144 

145 models = [] 

146 

147 try: 

148 response = safe_get( 

149 f"{base_url}/api/tags", 

150 timeout=timeout, 

151 headers=auth_headers or {}, 

152 allow_localhost=True, 

153 allow_private_ips=True, 

154 ) 

155 

156 if response.status_code == 200: 

157 data = response.json() 

158 

159 # Handle both newer and older Ollama API formats 

160 ollama_models = ( 

161 data.get("models", []) if isinstance(data, dict) else data 

162 ) 

163 

164 for model_data in ollama_models: 

165 model_name = model_data.get("name", "") 

166 if model_name: 

167 models.append({"value": model_name, "label": model_name}) 

168 

169 logger.info(f"Found {len(models)} Ollama models") 

170 else: 

171 logger.warning( 

172 f"Failed to fetch Ollama models: HTTP {response.status_code}" 

173 ) 

174 

175 except Exception: 

176 logger.exception("Error fetching Ollama models") 

177 

178 return models 

179 

180 

181def get_model( 

182 model_name: Optional[str] = None, 

183 model_type: Optional[str] = None, 

184 temperature: Optional[float] = None, 

185 **kwargs, 

186) -> Any: 

187 """ 

188 Get a language model instance as fallback when llm_config.get_llm is not available. 

189 

190 Args: 

191 model_name: Name of the model to use 

192 model_type: Type of the model provider 

193 temperature: Model temperature 

194 **kwargs: Additional parameters 

195 

196 Returns: 

197 LangChain language model instance 

198 """ 

199 # Get default values from kwargs or use reasonable defaults 

200 model_name = model_name or kwargs.get("DEFAULT_MODEL", "mistral") 

201 model_type = model_type or kwargs.get("DEFAULT_MODEL_TYPE", "ollama") 

202 temperature = temperature or kwargs.get("DEFAULT_TEMPERATURE", 0.7) 

203 max_tokens = kwargs.get("max_tokens", kwargs.get("MAX_TOKENS", 30000)) 

204 

205 # Common parameters 

206 common_params = { 

207 "temperature": temperature, 

208 "max_tokens": max_tokens, 

209 } 

210 

211 # Add additional kwargs 

212 common_params.update( 

213 { 

214 key: value 

215 for key, value in kwargs.items() 

216 if key 

217 not in [ 

218 "DEFAULT_MODEL", 

219 "DEFAULT_MODEL_TYPE", 

220 "DEFAULT_TEMPERATURE", 

221 "MAX_TOKENS", 

222 ] 

223 } 

224 ) 

225 

226 # Try to load the model based on type 

227 if model_type == "ollama": 

228 try: 

229 from langchain_ollama import ChatOllama 

230 

231 return ChatOllama(model=model_name, **common_params) 

232 except ImportError: 

233 try: 

234 from langchain_community.llms import Ollama 

235 

236 return Ollama(model=model_name, **common_params) 

237 except ImportError: 

238 logger.exception( 

239 "Neither langchain_ollama nor langchain_community.llms.Ollama available" 

240 ) 

241 raise 

242 

243 elif model_type == "openai": 

244 try: 

245 from langchain_openai import ChatOpenAI 

246 

247 api_key = get_setting_from_snapshot("llm.openai.api_key") 

248 if not api_key: 

249 raise ValueError("OpenAI API key not found in settings") 

250 return ChatOpenAI( 

251 model=model_name, api_key=api_key, **common_params 

252 ) 

253 except ImportError: 

254 logger.exception("langchain_openai not available") 

255 raise 

256 

257 elif model_type == "anthropic": 

258 try: 

259 from langchain_anthropic import ChatAnthropic 

260 

261 api_key = get_setting_from_snapshot("llm.anthropic.api_key") 

262 if not api_key: 

263 raise ValueError("Anthropic API key not found in settings") 

264 return ChatAnthropic( 

265 model=model_name, anthropic_api_key=api_key, **common_params 

266 ) 

267 except ImportError: 

268 logger.exception("langchain_anthropic not available") 

269 raise 

270 

271 elif model_type == "openai_endpoint": 

272 try: 

273 from langchain_openai import ChatOpenAI 

274 

275 api_key = get_setting_from_snapshot("llm.openai_endpoint.api_key") 

276 if not api_key: 

277 raise ValueError( 

278 "OpenAI endpoint API key not found in settings" 

279 ) 

280 

281 endpoint_url = kwargs.get( 

282 "OPENAI_ENDPOINT_URL", 

283 get_setting_from_snapshot( 

284 "llm.openai_endpoint.url", "https://openrouter.ai/api/v1" 

285 ), 

286 ) 

287 

288 if model_name is None and not kwargs.get( 

289 "OPENAI_ENDPOINT_REQUIRES_MODEL", True 

290 ): 

291 return ChatOpenAI( # type: ignore[call-arg] 

292 api_key=api_key, 

293 openai_api_base=endpoint_url, 

294 **common_params, 

295 ) 

296 return ChatOpenAI( # type: ignore[call-arg] 

297 model=model_name, 

298 api_key=api_key, 

299 openai_api_base=endpoint_url, 

300 **common_params, 

301 ) 

302 except ImportError: 

303 logger.exception("langchain_openai not available") 

304 raise 

305 

306 # Default fallback 

307 try: 

308 from langchain_ollama import ChatOllama 

309 

310 logger.warning( 

311 f"Unknown model type '{model_type}', defaulting to Ollama" 

312 ) 

313 return ChatOllama(model=model_name, **common_params) 

314 except (ImportError, Exception): 

315 logger.exception("Failed to load any model") 

316 

317 raise ValueError( 

318 "No language models available. Please install Ollama or set up API keys." 

319 )