Coverage for src / local_deep_research / llm / providers / implementations / ollama.py: 89%
117 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-02-25 01:07 +0000
1"""Ollama LLM provider for Local Deep Research."""
3import requests
4from langchain_ollama import ChatOllama
5from loguru import logger
7from ....config.thread_settings import (
8 get_llm_setting_from_snapshot as get_setting_from_snapshot,
9)
10from ....utilities.url_utils import normalize_url
11from ...llm_registry import register_llm
12from ....security import safe_get
15class OllamaProvider:
16 """Ollama provider for Local Deep Research.
18 This is the Ollama local model provider.
19 """
21 provider_name = "Ollama"
22 default_model = "gemma:latest"
23 api_key_setting = "llm.ollama.api_key" # Optional API key for authenticated Ollama instances
24 url_setting = "llm.ollama.url" # URL setting for model listing
26 # Metadata for auto-discovery
27 provider_key = "OLLAMA"
28 company_name = "Ollama"
29 is_cloud = False
31 @classmethod
32 def _get_auth_headers(cls, api_key=None, settings_snapshot=None):
33 """Get authentication headers for Ollama API requests.
35 Args:
36 api_key: Optional API key to use (takes precedence)
37 settings_snapshot: Optional settings snapshot to get API key from
39 Returns:
40 Dict of headers, empty if no API key configured
41 """
42 headers = {}
44 # Use provided API key or get from settings
45 if api_key is None and settings_snapshot is not None:
46 api_key = get_setting_from_snapshot(
47 cls.api_key_setting,
48 "", # Empty string instead of None to avoid NoSettingsContextError
49 settings_snapshot=settings_snapshot,
50 )
52 if api_key:
53 # Support Bearer token authentication for proxied Ollama instances
54 headers["Authorization"] = f"Bearer {api_key}"
56 return headers
58 @classmethod
59 def list_models_for_api(cls, api_key=None, base_url=None):
60 """Get available models from Ollama.
62 Args:
63 api_key: Optional API key for authentication
64 base_url: Base URL for Ollama API (required)
66 Returns:
67 List of model dictionaries with 'value' and 'label' keys
68 """
69 from ....utilities.llm_utils import fetch_ollama_models
71 if not base_url:
72 logger.warning("Ollama URL not configured")
73 return []
75 base_url = normalize_url(base_url)
77 # Get authentication headers
78 headers = cls._get_auth_headers(api_key=api_key)
80 # Fetch models using centralized function
81 models = fetch_ollama_models(
82 base_url, timeout=2.0, auth_headers=headers
83 )
85 # Add provider info and format for LLM API
86 for model in models:
87 # Clean up the model name for display
88 model_name = model["value"]
89 display_name = model_name.replace(":latest", "").replace(":", " ")
90 model["label"] = f"{display_name} (Ollama)"
91 model["provider"] = "OLLAMA"
93 logger.info(f"Found {len(models)} Ollama models")
94 return models
96 @classmethod
97 def create_llm(cls, model_name=None, temperature=0.7, **kwargs):
98 """Factory function for Ollama LLMs.
100 Args:
101 model_name: Name of the model to use
102 temperature: Model temperature (0.0-1.0)
103 **kwargs: Additional arguments including settings_snapshot
105 Returns:
106 A configured ChatOllama instance
108 Raises:
109 ValueError: If Ollama is not available
110 """
111 settings_snapshot = kwargs.get("settings_snapshot")
113 # Use default model if none specified
114 if not model_name:
115 model_name = cls.default_model
117 # Use the configurable Ollama base URL
118 raw_base_url = get_setting_from_snapshot(
119 "llm.ollama.url",
120 None,
121 settings_snapshot=settings_snapshot,
122 )
123 if not raw_base_url:
124 raise ValueError(
125 "Ollama URL not configured. Please set llm.ollama.url in settings."
126 )
127 base_url = normalize_url(raw_base_url)
129 # Check if Ollama is available before trying to use it
130 if not cls.is_available(settings_snapshot):
131 logger.error(f"Ollama not available at {base_url}.")
132 raise ValueError(f"Ollama not available at {base_url}")
134 # Check if the requested model exists
135 try:
136 logger.info(f"Checking if model '{model_name}' exists in Ollama")
138 # Get authentication headers
139 headers = cls._get_auth_headers(settings_snapshot=settings_snapshot)
141 response = safe_get(
142 f"{base_url}/api/tags",
143 timeout=3.0,
144 headers=headers,
145 allow_localhost=True,
146 allow_private_ips=True,
147 )
148 if response.status_code == 200: 148 ↛ 179line 148 didn't jump to line 179 because the condition on line 148 was always true
149 # Handle both newer and older Ollama API formats
150 data = response.json()
151 models = []
152 if "models" in data: 152 ↛ 157line 152 didn't jump to line 157 because the condition on line 152 was always true
153 # Newer Ollama API
154 models = data.get("models", [])
155 else:
156 # Older Ollama API format
157 models = data
159 # Get list of model names
160 model_names = [m.get("name", "").lower() for m in models]
161 logger.info(
162 f"Available Ollama models: {', '.join(model_names[:5])}{' and more' if len(model_names) > 5 else ''}"
163 )
165 if model_name.lower() not in model_names: 165 ↛ 166line 165 didn't jump to line 166 because the condition on line 165 was never true
166 logger.error(
167 f"Model '{model_name}' not found in Ollama. Available models: {', '.join(model_names[:5])}"
168 )
169 raise ValueError(
170 f"Model '{model_name}' not found in Ollama"
171 )
172 except Exception:
173 logger.debug(
174 f"Error checking for model '{model_name}' in Ollama",
175 exc_info=True,
176 )
177 # Continue anyway, let ChatOllama handle potential errors
179 logger.info(
180 f"Creating ChatOllama with model={model_name}, base_url={base_url}"
181 )
183 # Build Ollama parameters
184 ollama_params = {
185 "model": model_name,
186 "base_url": base_url,
187 "temperature": temperature,
188 }
190 # Add authentication headers if configured
191 headers = cls._get_auth_headers(settings_snapshot=settings_snapshot)
192 if headers: 192 ↛ 194line 192 didn't jump to line 194 because the condition on line 192 was never true
193 # ChatOllama supports auth via headers parameter
194 ollama_params["headers"] = headers
196 # Get context window size from settings for local providers
197 context_window_size = get_setting_from_snapshot(
198 "llm.local_context_window_size",
199 4096,
200 settings_snapshot=settings_snapshot,
201 )
202 if context_window_size is not None: 202 ↛ 206line 202 didn't jump to line 206 because the condition on line 202 was always true
203 ollama_params["num_ctx"] = int(context_window_size)
205 # Add max_tokens if specified in settings and supported
206 if get_setting_from_snapshot( 206 ↛ 223line 206 didn't jump to line 223 because the condition on line 206 was always true
207 "llm.supports_max_tokens", True, settings_snapshot=settings_snapshot
208 ):
209 # Use 80% of context window to leave room for prompts
210 if context_window_size is not None: 210 ↛ 223line 210 didn't jump to line 223 because the condition on line 210 was always true
211 max_tokens = min(
212 int(
213 get_setting_from_snapshot(
214 "llm.max_tokens",
215 100000,
216 settings_snapshot=settings_snapshot,
217 )
218 ),
219 int(context_window_size * 0.8),
220 )
221 ollama_params["max_tokens"] = max_tokens
223 llm = ChatOllama(**ollama_params)
225 # Log the actual client configuration after creation
226 logger.debug(
227 f"ChatOllama created - base_url attribute: {getattr(llm, 'base_url', 'not found')}"
228 )
230 return llm
232 @classmethod
233 def is_available(cls, settings_snapshot=None):
234 """Check if Ollama is running.
236 Args:
237 settings_snapshot: Optional settings snapshot to use
239 Returns:
240 True if Ollama is available, False otherwise
241 """
242 try:
243 raw_base_url = get_setting_from_snapshot(
244 "llm.ollama.url",
245 None,
246 settings_snapshot=settings_snapshot,
247 )
248 if not raw_base_url:
249 logger.debug("Ollama URL not configured")
250 return False
251 base_url = normalize_url(raw_base_url)
252 logger.info(f"Checking Ollama availability at {base_url}/api/tags")
254 # Get authentication headers
255 headers = cls._get_auth_headers(settings_snapshot=settings_snapshot)
257 try:
258 response = safe_get(
259 f"{base_url}/api/tags",
260 timeout=3.0,
261 headers=headers,
262 allow_localhost=True,
263 allow_private_ips=True,
264 )
265 if response.status_code == 200:
266 logger.info(
267 f"Ollama is available. Status code: {response.status_code}"
268 )
269 # Log first 100 chars of response to debug
270 logger.info(f"Response preview: {str(response.text)[:100]}")
271 return True
272 else:
273 logger.warning(
274 f"Ollama API returned status code: {response.status_code}"
275 )
276 return False
277 except requests.exceptions.RequestException as req_error:
278 logger.warning(
279 f"Request error when checking Ollama: {req_error!s}"
280 )
281 return False
282 except Exception:
283 logger.warning(
284 "Unexpected error when checking Ollama", exc_info=True
285 )
286 return False
287 except Exception:
288 logger.warning("Error in is_ollama_available", exc_info=True)
289 return False
292# Keep the standalone functions for backward compatibility and registration
293def create_ollama_llm(model_name=None, temperature=0.7, **kwargs):
294 """Factory function for Ollama LLMs.
296 Args:
297 model_name: Name of the model to use (e.g., "llama2", "gemma", etc.)
298 temperature: Model temperature (0.0-1.0)
299 **kwargs: Additional arguments including settings_snapshot
301 Returns:
302 A configured ChatOllama instance
304 Raises:
305 ValueError: If Ollama is not available
306 """
307 return OllamaProvider.create_llm(model_name, temperature, **kwargs)
310def is_ollama_available(settings_snapshot=None):
311 """Check if Ollama is available.
313 Args:
314 settings_snapshot: Optional settings snapshot to use
316 Returns:
317 True if Ollama is running, False otherwise
318 """
319 return OllamaProvider.is_available(settings_snapshot)
322def register_ollama_provider():
323 """Register the Ollama provider with the LLM registry."""
324 register_llm("ollama", create_ollama_llm)
325 logger.info("Registered Ollama LLM provider")