Coverage for src/local_deep_research/llm/providers/implementations/ollama.py: 97%

1"""Ollama LLM provider for Local Deep Research.""" 

2 

3import requests 

4from langchain_ollama import ChatOllama 

5from loguru import logger 

6 

7from ....config.thread_settings import get_setting_from_snapshot 

8from ....utilities.url_utils import normalize_url 

9from ....security import safe_get 

10from ..base import BaseLLMProvider 

11 

12 

13class OllamaProvider(BaseLLMProvider): 

14 """Ollama provider for Local Deep Research. 

15 

16 This is the Ollama local model provider. 

17 """ 

18 

19 provider_name = "Ollama" 

20 default_model = "" 

21 api_key_setting = "llm.ollama.api_key" # Optional API key for authenticated Ollama instances 

22 url_setting = "llm.ollama.url" # URL setting for model listing 

23 

24 # Metadata for auto-discovery 

25 provider_key = "OLLAMA" 

26 company_name = "Ollama" 

27 is_cloud = False 

28 

29 @classmethod 

30 def _get_auth_headers(cls, api_key=None, settings_snapshot=None): 

31 """Get authentication headers for Ollama API requests. 

32 

33 Args: 

34 api_key: Optional API key to use (takes precedence) 

35 settings_snapshot: Optional settings snapshot to get API key from 

36 

37 Returns: 

38 Dict of headers, empty if no API key configured 

39 """ 

40 headers = {} 

41 

42 # Use provided API key or get from settings 

43 if api_key is None and settings_snapshot is not None: 

44 api_key = get_setting_from_snapshot( 

45 cls.api_key_setting, 

46 "", # Empty string instead of None to avoid NoSettingsContextError 

47 settings_snapshot=settings_snapshot, 

48 ) 

49 

50 if api_key: 

51 # Support Bearer token authentication for proxied Ollama instances 

52 headers["Authorization"] = f"Bearer {api_key}" 

53 

54 return headers 

55 

56 @classmethod 

57 def list_models_for_api(cls, api_key=None, base_url=None): 

58 """Get available models from Ollama. 

59 

60 Args: 

61 api_key: Optional API key for authentication 

62 base_url: Base URL for Ollama API (required) 

63 

64 Returns: 

65 List of model dictionaries with 'value' and 'label' keys 

66 """ 

67 from ....utilities.llm_utils import fetch_ollama_models 

68 

69 if not base_url: 

70 logger.warning("Ollama URL not configured") 

71 return [] 

72 

73 base_url = normalize_url(base_url) 

74 

75 # Get authentication headers 

76 headers = cls._get_auth_headers(api_key=api_key) 

77 

78 # Fetch models using centralized function 

79 models = fetch_ollama_models( 

80 base_url, timeout=2.0, auth_headers=headers 

81 ) 

82 

83 # Add provider info and format for LLM API 

84 for model in models: 

85 # Clean up the model name for display 

86 model_name = model["value"] 

87 display_name = model_name.replace(":latest", "").replace(":", " ") 

88 model["label"] = f"{display_name} (Ollama)" 

89 model["provider"] = "OLLAMA" 

90 

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

92 return models 

93 

94 @classmethod 

95 def create_llm(cls, model_name=None, temperature=0.7, **kwargs): 

96 """Factory function for Ollama LLMs. 

97 

98 Args: 

99 model_name: Name of the model to use 

100 temperature: Model temperature (0.0-1.0) 

101 **kwargs: Additional arguments including settings_snapshot 

102 

103 Returns: 

104 A configured ChatOllama instance 

105 

106 Raises: 

107 ValueError: If Ollama is not available 

108 """ 

109 settings_snapshot = kwargs.get("settings_snapshot") 

110 

111 # Defense-in-depth: callers using the central get_llm() already get a 

112 # clear ValueError when llm.model is unset. This second check covers 

113 # direct callers of OllamaProvider.create_llm() (programmatic API, 

114 # custom registrations) so they don't get a confusing langchain 

115 # error about a model that wasn't actually requested. 

116 if not model_name or not model_name.strip(): 

117 logger.error("Ollama model name not provided to create_llm()") 

118 raise ValueError( 

119 "Ollama model not configured. Please set llm.model in " 

120 "settings (e.g. 'llama3.1:8b', 'qwen2.5:14b')." 

121 ) 

122 

123 # Use the configurable Ollama base URL 

124 raw_base_url = get_setting_from_snapshot( 

125 "llm.ollama.url", 

126 None, 

127 settings_snapshot=settings_snapshot, 

128 ) 

129 if not raw_base_url: 

130 raise ValueError( 

131 "Ollama URL not configured. Please set llm.ollama.url in settings." 

132 ) 

133 base_url = normalize_url(raw_base_url) 

134 

135 logger.info( 

136 f"Creating ChatOllama with model={model_name}, base_url={base_url}" 

137 ) 

138 

139 # Build Ollama parameters 

140 ollama_params = { 

141 "model": model_name, 

142 "base_url": base_url, 

143 "temperature": temperature, 

144 } 

145 

146 # Add authentication headers if configured 

147 headers = cls._get_auth_headers(settings_snapshot=settings_snapshot) 

148 if headers: 148 ↛ 150line 148 didn't jump to line 150 because the condition on line 148 was never true

149 # ChatOllama supports auth via headers parameter 

150 ollama_params["headers"] = headers 

151 

152 # Get context window size from settings for local providers 

153 context_window_size = get_setting_from_snapshot( 

154 "llm.local_context_window_size", 

155 4096, 

156 settings_snapshot=settings_snapshot, 

157 ) 

158 if context_window_size is not None: 

159 ollama_params["num_ctx"] = int(context_window_size) 

160 

161 # Add max_tokens if specified in settings and supported 

162 if get_setting_from_snapshot( 162 ↛ 179line 162 didn't jump to line 179 because the condition on line 162 was always true

163 "llm.supports_max_tokens", True, settings_snapshot=settings_snapshot 

164 ): 

165 # Use 80% of context window to leave room for prompts 

166 if context_window_size is not None: 

167 max_tokens = min( 

168 int( 

169 get_setting_from_snapshot( 

170 "llm.max_tokens", 

171 100000, 

172 settings_snapshot=settings_snapshot, 

173 ) 

174 ), 

175 int(context_window_size * 0.8), 

176 ) 

177 ollama_params["max_tokens"] = max_tokens 

178 

179 llm = ChatOllama(**ollama_params) 

180 

181 # Log the actual client configuration after creation 

182 logger.debug( 

183 f"ChatOllama created - base_url attribute: {getattr(llm, 'base_url', 'not found')}" 

184 ) 

185 

186 return llm 

187 

188 @classmethod 

189 def is_available(cls, settings_snapshot=None): 

190 """Check if Ollama is running. 

191 

192 Args: 

193 settings_snapshot: Optional settings snapshot to use 

194 

195 Returns: 

196 True if Ollama is available, False otherwise 

197 """ 

198 try: 

199 raw_base_url = get_setting_from_snapshot( 

200 "llm.ollama.url", 

201 None, 

202 settings_snapshot=settings_snapshot, 

203 ) 

204 if not raw_base_url: 

205 logger.debug("Ollama URL not configured") 

206 return False 

207 base_url = normalize_url(raw_base_url) 

208 logger.info(f"Checking Ollama availability at {base_url}/api/tags") 

209 

210 # Get authentication headers 

211 headers = cls._get_auth_headers(settings_snapshot=settings_snapshot) 

212 

213 try: 

214 response = safe_get( 

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

216 timeout=3, 

217 headers=headers, 

218 allow_localhost=True, 

219 allow_private_ips=True, 

220 ) 

221 if response.status_code == 200: 

222 logger.info( 

223 f"Ollama is available. Status code: {response.status_code}" 

224 ) 

225 # Log first 100 chars of response to debug 

226 logger.info(f"Response preview: {str(response.text)[:100]}") 

227 return True 

228 logger.warning( 

229 f"Ollama API returned status code: {response.status_code}" 

230 ) 

231 return False 

232 except requests.exceptions.RequestException: 

233 logger.warning("Request error when checking Ollama") 

234 return False 

235 except Exception: 

236 logger.warning("Unexpected error when checking Ollama") 

237 return False 

238 except Exception: 

239 logger.warning("Error in OllamaProvider.is_available") 

240 return False 

241 

242 @classmethod 

243 def requires_auth_for_models(cls): 

244 """Ollama is local and does not need auth to list models.""" 

245 return False