Coverage for src/local_deep_research/web_search_engines/search_engine_factory.py: 90%
426 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1import inspect
2from typing import Any, Dict, Optional
4from ..security.egress.policy import PolicyDeniedError
5from ..security.log_sanitizer import scrub_error
6from ..security.module_whitelist import get_safe_module_class
7from ..security.secure_logging import logger
8from ..settings.exceptions import SettingsValueError
9from .engine_groups import is_collection_engine
10from .retriever_registry import retriever_registry
11from ..utilities.search_utilities import LANGUAGE_CODE_MAP
12from ..utilities.type_utils import (
13 _is_strict_scope,
14 resolve_snippets_only,
15 to_bool,
16)
17from .search_engine_base import BaseSearchEngine
18from .search_engines_config import search_config
20# Engines whose constructors accept the global ``search.time_period`` filter.
21# Used both by get_search()'s explicit forwarding and by
22# create_search_engine()'s settings-snapshot fallback so that direct callers
23# (langgraph-agent strategy tools, MCP server) honor the setting too.
24TIME_PERIOD_ENGINES = ("serpapi", "tavily", "wikinews")
26# Non-geographic region sentinels that should not be mapped to provider `location`
27_NON_LOCATION_REGIONS = frozenset({"wt-wt", "all", "none", "worldwide", ""})
29# Engines that accept a language code (ISO 639-1) in their ``language`` parameter.
30# Engines like OpenLibrary (MARC code or None) or Wayback (language name
31# for Justext) are excluded so that global language settings do not corrupt them.
32LANGUAGE_CODE_ENGINES = frozenset(
33 {"searxng", "mojeek", "wikipedia", "tinyfish", "scaleserp"}
34)
36# Engines that accept a language code (Google hl code) in their ``search_language`` parameter.
37# Engines like Brave, Google PSE, SerpAPI, Wikinews expect the language name ("English").
38SEARCH_LANGUAGE_CODE_ENGINES = frozenset({"serper"})
40# Engines that accept an ISO country code in their ``location`` parameter.
41# ScaleSerp expects a free-text place name ("United States") and is excluded.
42LOCATION_CODE_ENGINES = frozenset({"tinyfish"})
44# Parameters forwarded to BaseSearchEngine or accepted across multiple engines
45_BASE_FORWARDED_PARAMS = frozenset(
46 {
47 "search_snippets_only",
48 "max_filtered_results",
49 "max_results",
50 "safe_search",
51 "region",
52 "location",
53 "search_language",
54 "language",
55 "time_period",
56 "include_full_content",
57 "content_cache",
58 "progress_callback",
59 }
60)
63def _resolve_setting(
64 snapshot: Optional[Dict[str, Any]],
65 key: str,
66 default: Any = None,
67) -> Any:
68 """Safely extract a setting value from settings_snapshot (dict or scalar)."""
69 if not snapshot or key not in snapshot:
70 return default
71 raw = snapshot[key]
72 val = raw.get("value", default) if isinstance(raw, dict) else raw
73 return default if val is None else val
76def _normalize_language_code(language: str) -> Optional[str]:
77 """Normalize a language name or code to an ISO 639-1 code.
79 Returns None if the language name is unknown and cannot be converted to a valid code.
80 """
81 if not language:
82 return None
83 lang_clean = language.strip().lower()
84 if lang_clean in LANGUAGE_CODE_MAP:
85 return LANGUAGE_CODE_MAP[lang_clean]
86 # Pass through values that already look like ISO language codes (len <= 3 and alpha)
87 if len(lang_clean) <= 3 and lang_clean.isalpha():
88 return lang_clean
89 return None
92def _to_extraction_language_name(value: Any) -> Optional[str]:
93 """Convert a language name or ISO code to a justext stoplist name.
95 ``justext.get_stoplist`` expects names like ``English`` / ``German``,
96 not provider codes like ``en`` / ``de``.
97 """
98 if not isinstance(value, str):
99 return None
100 cleaned = value.strip()
101 if not cleaned: 101 ↛ 102line 101 didn't jump to line 102 because the condition on line 101 was never true
102 return None
103 lower = cleaned.lower()
104 if lower in LANGUAGE_CODE_MAP:
105 return lower.capitalize()
106 for name, code in LANGUAGE_CODE_MAP.items():
107 if code == lower:
108 return name.capitalize()
109 if len(cleaned) <= 3 and cleaned.isalpha():
110 return None
111 return cleaned
114def _resolve_extraction_language(
115 params: Dict[str, Any],
116 settings_snapshot: Optional[Dict[str, Any]],
117) -> Optional[str]:
118 """Resolve justext extraction language separately from provider codes."""
119 candidates: list[Any] = []
120 if "search_language" in params:
121 candidates.append(params.get("search_language"))
122 snapshot_lang = _resolve_setting(
123 settings_snapshot, "search.search_language"
124 )
125 if snapshot_lang is not None:
126 candidates.append(snapshot_lang)
127 if "language" in params:
128 candidates.append(params.get("language"))
129 for candidate in candidates:
130 converted = _to_extraction_language_name(candidate)
131 if converted:
132 return converted
133 return None
136_resolve_snippets_only = resolve_snippets_only
139def create_search_engine(
140 engine_name: str,
141 llm=None,
142 username: str | None = None,
143 settings_snapshot: Dict[str, Any] | None = None,
144 programmatic_mode: bool = False,
145 content_cache: Any = None,
146 **kwargs,
147) -> Optional[BaseSearchEngine]:
148 """
149 Create a search engine instance based on the engine name.
151 Args:
152 engine_name: Name of the search engine to create
153 llm: Language model instance (passed to engines that accept one)
154 programmatic_mode: If True, disables database operations and metrics tracking
155 content_cache: Optional shared ``ResearchContentCache`` for the whole
156 research run. Threaded into the engine (and its full-search
157 wrapper) so repeat URLs across quick phase, subsections and
158 subagents hit one cache. When None, engines fall back to the
159 cache bound to the current research via thread-local context.
160 **kwargs: Additional parameters to override defaults
162 Returns:
163 Initialized search engine instance or None if creation failed
164 """
165 # Debug logging
166 logger.info(
167 f"create_search_engine called with engine_name={engine_name} (type: {type(engine_name)})"
168 )
170 # Resolve the effective username for per-user registry lookups. Callers
171 # that thread it explicitly win; otherwise fall back to the ``_username``
172 # carried in the settings snapshot (injected by search_config.get_search /
173 # AdvancedSearchSystem). None resolves the shared namespace only.
174 if username is None: 174 ↛ 186line 174 didn't jump to line 186 because the condition on line 174 was always true
175 from ..search_system import username_from_snapshot
177 username = username_from_snapshot(settings_snapshot)
179 # Egress policy PEP runs AFTER retriever-registry + config lookup, so
180 # registered retrievers don't go through engine-name classification.
181 # The actual call site is right before engine instantiation, further
182 # down.
184 # Check if this is a registered retriever first (the caller's own
185 # retrievers resolve before shared/global ones).
186 retriever = retriever_registry.get(engine_name, username=username)
187 if retriever:
188 # Egress policy: gate the retriever against the active scope.
189 # Retrievers are classified at registration (is_local, default
190 # True). Snapshot REQUIRED — previously the policy block was
191 # gated on "if settings_snapshot:" which let snapshot=None
192 # callers instantiate retrievers ungated.
193 # NB: PolicyDeniedError is imported at MODULE level — importing it
194 # again here (function-local) would shadow it across the whole
195 # function and make the outer `except PolicyDeniedError` an unbound
196 # local on code paths that skip this branch.
197 from ..security.egress.policy import (
198 Decision,
199 context_from_snapshot,
200 evaluate_retriever,
201 resolve_run_primary_engine,
202 )
204 if not settings_snapshot: 204 ↛ 205line 204 didn't jump to line 205 because the condition on line 204 was never true
205 raise PolicyDeniedError(
206 Decision(False, "no_snapshot"),
207 target=engine_name,
208 )
210 try:
211 # STRICT must never let the request choose its own primary: a
212 # blank/missing saved primary fails closed (ValueError ->
213 # PolicyDeniedError below) instead of falling back to the
214 # engine being instantiated.
215 if _is_strict_scope(settings_snapshot):
216 primary = resolve_run_primary_engine(settings_snapshot)
217 else:
218 primary = resolve_run_primary_engine(
219 settings_snapshot, default=engine_name
220 )
221 _ctx = context_from_snapshot(
222 settings_snapshot,
223 primary,
224 username=username,
225 )
226 except PolicyDeniedError:
227 raise
228 except ValueError as exc:
229 raise PolicyDeniedError(
230 Decision(False, "invalid_policy_config"),
231 target=engine_name,
232 ) from exc
233 # Pass metadata from the SAME registry reference used for
234 # .get() above, so evaluate_retriever doesn't re-read a
235 # different (e.g. test-patched) global singleton.
236 try:
237 _meta = retriever_registry.get_metadata(
238 engine_name, username=username
239 )
240 except AttributeError:
241 _meta = None
242 _decision = evaluate_retriever(engine_name, _ctx, metadata=_meta)
243 if not _decision.allowed: 243 ↛ 244line 243 didn't jump to line 244 because the condition on line 243 was never true
244 logger.bind(policy_audit=True).warning(
245 "retriever denied by egress policy",
246 retriever=engine_name,
247 scope=_ctx.scope.value,
248 reason=_decision.reason,
249 )
250 raise PolicyDeniedError(_decision, target=engine_name)
252 logger.info(f"Using registered LangChain retriever: {engine_name}")
253 from .engines.search_engine_retriever import RetrieverSearchEngine
255 return RetrieverSearchEngine(
256 retriever=retriever,
257 name=engine_name,
258 max_results=kwargs.get("max_results", 10),
259 programmatic_mode=programmatic_mode,
260 content_cache=content_cache,
261 )
263 # Extract search engine configs from settings snapshot
264 if settings_snapshot:
265 config = search_config(
266 username=username, settings_snapshot=settings_snapshot
267 )
269 logger.debug(
270 f"Extracted search engines from snapshot: {list(config.keys())}"
271 )
272 else:
273 raise RuntimeError(
274 "settings_snapshot is required for search engine creation in threads"
275 )
277 if engine_name == "none":
278 # Reject the literal string "none". Historically this silently fell
279 # through to a meta engine and hit live networks — callers that
280 # wanted an offline pipeline were unknowingly doing real searches.
281 raise ValueError(
282 "search.tool='none' is not a valid engine. Register a LangChain "
283 "retriever via `retrievers={...}` (see "
284 "examples/llm_integration/mock_llm_example.py) or pick a real "
285 "engine."
286 )
288 if engine_name not in config:
289 # Deprecated path: some older callers pass UI display labels instead of
290 # config keys. Display labels look like "{icon} {base_name} ({category})"
291 # e.g. "🔬 OpenAlex (Scientific)". Only attempt (and warn about)
292 # this fallback when the name actually matches that shape — plain keys
293 # like "collection_<uuid>" never will, and the warning was pure noise.
294 looks_like_display_label = " (" in engine_name and engine_name.endswith(
295 ")"
296 )
297 if looks_like_display_label:
298 logger.warning(
299 f"Engine '{engine_name}' not found in config - attempting "
300 "display label fallback. This is deprecated; callers should "
301 "pass the config key directly."
302 )
304 # To avoid ReDoS, use string ops instead of regex.
305 # Pattern: icon, space, base_name, space, (category)
306 parts = engine_name.rsplit(" (", 1)
307 if len(parts) == 2: 307 ↛ 335line 307 didn't jump to line 335 because the condition on line 307 was always true
308 before_paren = parts[0]
309 space_idx = before_paren.find(" ")
310 if space_idx > 0:
311 base_name = before_paren[space_idx + 1 :].strip()
312 logger.info(
313 f"Extracted base name '{base_name}' from label "
314 f"'{engine_name}'"
315 )
317 for config_key, config_data in config.items():
318 if isinstance(config_data, dict): 318 ↛ 317line 318 didn't jump to line 317 because the condition on line 318 was always true
319 display_name = config_data.get(
320 "display_name", config_key
321 )
322 if display_name == base_name:
323 logger.info(
324 f"Matched label to config key: "
325 f"'{engine_name}' -> '{config_key}'"
326 )
327 engine_name = config_key
328 break
330 # If still not found, FAIL CLOSED.
331 # We raise a plain ValueError (not PolicyDeniedError) because
332 # "unknown engine name" is a config/wiring error, not a policy
333 # decision — keeping the exception type clean lets policy-aware
334 # callers tell them apart.
335 if engine_name not in config:
336 logger.bind(policy_audit=True).warning(
337 "unknown engine name rejected",
338 engine=engine_name,
339 available=list(config.keys()),
340 )
341 raise ValueError(
342 f"Unknown search engine '{engine_name}'. Available: "
343 f"{sorted(config.keys())}. The 'auto', 'meta', 'parallel' "
344 "and 'parallel_scientific' meta engines were removed — the "
345 "langgraph-agent strategy (the default) selects engines "
346 "dynamically; pick a concrete engine instead."
347 )
349 # Egress policy PEP: gate the resolved engine against the user's declared
350 # scope. Lazy import to break the circular dependency
351 # (security/egress/policy.py → web_search_engines/* → security/*).
352 # PolicyDeniedError comes from the module-level import (see the note
353 # in the retriever branch above — re-importing it here would shadow
354 # it function-wide and break the outer except clause).
355 from ..security.egress.policy import (
356 Decision,
357 context_from_snapshot,
358 evaluate_engine,
359 resolve_run_primary_engine,
360 )
362 # Resolve the run's primary via the shared helper so the scope this PEP
363 # enforces matches the one the LangGraph tool-list filter pre-filters with.
364 # Under STRICT a blank/missing saved primary fails closed instead of
365 # falling back to the engine being instantiated (which would make the
366 # STRICT check trivially true).
367 try:
368 if _is_strict_scope(settings_snapshot):
369 primary_engine = resolve_run_primary_engine(settings_snapshot)
370 else:
371 # (default: this engine when search.tool is unset).
372 primary_engine = resolve_run_primary_engine(
373 settings_snapshot, default=engine_name
374 )
375 ctx = context_from_snapshot(
376 settings_snapshot, primary_engine, username=username
377 )
378 except ValueError as exc:
379 raise PolicyDeniedError(
380 Decision(False, "invalid_policy_config"),
381 target=engine_name,
382 ) from exc
384 # Pass the engine config entry as metadata so a per-collection
385 # is_public classification is honored without a redundant DB lookup.
386 decision = evaluate_engine(
387 engine_name,
388 ctx,
389 settings_snapshot=settings_snapshot,
390 metadata=config.get(engine_name),
391 )
392 if not decision.allowed:
393 logger.bind(policy_audit=True).warning(
394 "engine denied by egress policy",
395 engine=engine_name,
396 scope=ctx.scope.value,
397 reason=decision.reason,
398 )
399 raise PolicyDeniedError(decision, target=engine_name)
401 # Get engine configuration
402 engine_config = config[engine_name]
403 default_params = engine_config.get("default_params", {}) or {}
405 # Set default max_results from config if not provided in kwargs
406 if "max_results" not in kwargs:
407 if settings_snapshot: 407 ↛ 416line 407 didn't jump to line 416 because the condition on line 407 was always true
408 raw_mr = _resolve_setting(
409 settings_snapshot, "search.max_results", 20
410 )
411 try:
412 max_results = int(raw_mr)
413 except (ValueError, TypeError):
414 max_results = 20
415 else:
416 max_results = 20
417 kwargs["max_results"] = max_results
419 # Route the global time filter to engines that understand it when the
420 # caller didn't pass one explicitly. get_search() forwards it as a kwarg,
421 # but direct callers (langgraph-agent strategy tools, the MCP server)
422 # only pass a settings_snapshot — without this fallback the user's
423 # Settings → Search → Time Period choice is silently ignored on those
424 # paths. Placed in kwargs so it takes precedence over per-engine
425 # default_params, matching get_search()'s behavior.
426 if (
427 engine_name in TIME_PERIOD_ENGINES
428 and "time_period" not in kwargs
429 and settings_snapshot
430 and "search.time_period" in settings_snapshot
431 ):
432 raw_time_period = _resolve_setting(
433 settings_snapshot, "search.time_period", "all"
434 )
435 if raw_time_period is not None: 435 ↛ 445line 435 didn't jump to line 445 because the condition on line 435 was always true
436 kwargs["time_period"] = str(raw_time_period)
438 # Route search.snippets_only to engines when the caller didn't pass one
439 # explicitly. get_search() forwards it as a kwarg, and langgraph-agent
440 # strategy tools explicitly force search_snippets_only=True, but other
441 # direct callers (e.g. MCP server) only pass a settings_snapshot — without
442 # this fallback the user's Settings → Search → Snippets Only choice is
443 # silently ignored on those paths and defaults to True. Placed in kwargs
444 # so it takes precedence over per-engine default_params.
445 if "search_snippets_only" not in kwargs and settings_snapshot:
446 resolved_snippets = _resolve_snippets_only(settings_snapshot)
447 if resolved_snippets is not None:
448 kwargs["search_snippets_only"] = resolved_snippets
450 # Mirror get_search(): wrap FullSearchResults when snippets-only is off
451 # and the caller did not pass use_full_search. Direct callers like the
452 # MCP server never set that flag (langgraph-agent tools explicitly set
453 # use_full_search=False), so without this fallback the engine fetches
454 # full content (search_snippets_only=False) but the wrapper path that
455 # actually hydrates ``full_content`` for searxng/serpapi/etc. never
456 # runs.
457 if "use_full_search" not in kwargs and "search_snippets_only" in kwargs:
458 kwargs["use_full_search"] = not kwargs["search_snippets_only"]
460 # Route search.max_filtered_results to engines when not explicitly passed.
461 if "max_filtered_results" not in kwargs and settings_snapshot:
462 raw_mfr = _resolve_setting(
463 settings_snapshot, "search.max_filtered_results"
464 )
465 if raw_mfr is not None:
466 try:
467 kwargs["max_filtered_results"] = int(raw_mfr)
468 except (ValueError, TypeError):
469 pass
471 # Route search.region to engines when not explicitly passed.
472 if "region" not in kwargs and settings_snapshot:
473 val_region = _resolve_setting(settings_snapshot, "search.region")
474 if val_region is not None:
475 kwargs["region"] = str(val_region)
477 # Route region -> location alias when location is not explicitly passed.
478 # Exclude non-country sentinels like 'wt-wt' / 'all' to avoid invalid location values.
479 # Only alias for engines whose location parameter expects an ISO country code.
480 if (
481 "location" not in kwargs
482 and "location" not in default_params
483 and engine_name in LOCATION_CODE_ENGINES
484 ):
485 effective_region = kwargs.get("region")
486 if ( 486 ↛ 491line 486 didn't jump to line 491 because the condition on line 486 was never true
487 not effective_region
488 and settings_snapshot
489 and "search.region" in settings_snapshot
490 ):
491 effective_region = _resolve_setting(
492 settings_snapshot, "search.region"
493 )
494 if (
495 effective_region
496 and isinstance(effective_region, str)
497 and effective_region.lower() not in _NON_LOCATION_REGIONS
498 and len(effective_region.strip()) == 2
499 and effective_region.strip().isalpha()
500 ):
501 kwargs["location"] = effective_region.strip().upper()
503 # Route search.safe_search to engines when not explicitly passed.
504 # Do not override a key the engine's default_params defines (e.g. SearXNG).
505 if (
506 "safe_search" not in kwargs
507 and "safe_search" not in default_params
508 and settings_snapshot
509 and "search.safe_search" in settings_snapshot
510 ):
511 raw_safe = _resolve_setting(settings_snapshot, "search.safe_search")
512 if raw_safe is not None: 512 ↛ 519line 512 didn't jump to line 519 because the condition on line 512 was always true
513 kwargs["safe_search"] = to_bool(raw_safe, default=True)
515 # Route search.search_language to engines when not explicitly passed.
516 # Only inject when actually present in snapshot (no hardcoded default),
517 # and do not override a key the engine's default_params defines.
518 # For Serper, normalize to Google hl code; for other engines, pass language name.
519 if (
520 "search_language" not in kwargs
521 and "search_language" not in default_params
522 and settings_snapshot
523 and "search.search_language" in settings_snapshot
524 ):
525 val_lang = _resolve_setting(settings_snapshot, "search.search_language")
526 if val_lang is not None: 526 ↛ 536line 526 didn't jump to line 536 because the condition on line 526 was always true
527 if engine_name in SEARCH_LANGUAGE_CODE_ENGINES:
528 norm_code = _normalize_language_code(str(val_lang))
529 if norm_code: 529 ↛ 536line 529 didn't jump to line 536 because the condition on line 529 was always true
530 kwargs["search_language"] = norm_code
531 else:
532 kwargs["search_language"] = str(val_lang)
534 # Route language code (ISO 639-1) for engines expecting code in 'language' param.
535 # Only alias when not defined in default_params, and only for engines whose parameter is a code.
536 if (
537 "language" not in kwargs
538 and "language" not in default_params
539 and engine_name in LANGUAGE_CODE_ENGINES
540 ):
541 lang_source = kwargs.get("search_language")
542 if ( 542 ↛ 547line 542 didn't jump to line 547 because the condition on line 542 was never true
543 not lang_source
544 and settings_snapshot
545 and "search.search_language" in settings_snapshot
546 ):
547 lang_source = _resolve_setting(
548 settings_snapshot, "search.search_language"
549 )
550 if lang_source and isinstance(lang_source, str):
551 norm_code = _normalize_language_code(lang_source)
552 if norm_code: 552 ↛ 555line 552 didn't jump to line 555 because the condition on line 552 was always true
553 kwargs["language"] = norm_code
554 # Check for API key requirements
555 requires_api_key = engine_config.get("requires_api_key", False)
557 # Bound here (not just inside ``if requires_api_key``) so the except
558 # handler below can pass the in-scope key to scrub_error() — without
559 # the hoist, the no-key path would hit a NameError when scrubbing.
560 api_key = None
562 if requires_api_key:
563 # Check the settings snapshot for the API key
564 api_key_path = f"search.engine.web.{engine_name}.api_key"
566 if settings_snapshot: 566 ↛ 577line 566 didn't jump to line 577 because the condition on line 566 was always true
567 api_key_setting = settings_snapshot.get(api_key_path)
569 if api_key_setting:
570 api_key = (
571 api_key_setting.get("value")
572 if isinstance(api_key_setting, dict)
573 else api_key_setting
574 )
576 # Still try to get from engine config if not found
577 if not api_key:
578 api_key = engine_config.get("api_key")
580 if not api_key:
581 logger.info(
582 f"Required API key for {engine_name} not found in settings."
583 )
584 return None
586 # Pass the API key in kwargs for engines that need it
587 if api_key: 587 ↛ 593line 587 didn't jump to line 593 because the condition on line 587 was always true
588 kwargs["api_key"] = api_key
590 # Warn about missing LLM but allow engine creation in degraded mode.
591 # All engines with requires_llm=True handle llm=None gracefully
592 # (e.g. skipping query optimization, using reliability-based sorting).
593 if engine_config.get("requires_llm", False) and not llm:
594 logger.warning(
595 f"Engine '{engine_name}' is configured with requires_llm=True but no LLM provided. "
596 f"Creating engine without LLM — some features (query optimization, relevance filtering) "
597 f"may be unavailable."
598 )
600 try:
601 # Load the engine class
602 module_path = engine_config["module_path"]
603 class_name = engine_config["class_name"]
605 engine_class = get_safe_module_class(module_path, class_name)
607 # Get the engine class's __init__ parameters to filter out unsupported ones
608 engine_init_signature = inspect.signature(engine_class.__init__)
609 engine_init_params = list(engine_init_signature.parameters.keys())
610 has_var_keyword = any(
611 p.kind == inspect.Parameter.VAR_KEYWORD
612 for p in engine_init_signature.parameters.values()
613 )
615 # Combine default parameters with provided ones
616 all_params = {**engine_config.get("default_params", {}), **kwargs}
617 # The shared research cache is a factory-level argument, not an
618 # engine default — inject it so engines that accept it (or **kwargs)
619 # receive the single cache for the whole research run.
620 if content_cache is not None and "content_cache" not in all_params:
621 all_params["content_cache"] = content_cache
623 # Filter out parameters that aren't accepted by the engine class
624 # Note: 'self' is always the first parameter of instance methods, so we skip it.
625 # If the class accepts **kwargs, parameters handled by BaseSearchEngine or
626 # common wrappers are preserved.
627 filtered_params = {
628 k: v for k, v in all_params.items() if k in engine_init_params[1:]
629 }
630 if has_var_keyword:
631 for k in _BASE_FORWARDED_PARAMS:
632 if k in all_params and k not in filtered_params:
633 filtered_params[k] = all_params[k]
635 # Always pass settings_snapshot if the engine accepts it
636 if "settings_snapshot" in engine_init_params[1:] and settings_snapshot:
637 filtered_params["settings_snapshot"] = settings_snapshot
639 # Pass programmatic_mode if the engine accepts it
640 if "programmatic_mode" in engine_init_params[1:]:
641 filtered_params["programmatic_mode"] = programmatic_mode
643 # Add LLM if required OR if provided and engine accepts it
644 if engine_config.get("requires_llm", False):
645 filtered_params["llm"] = llm
646 elif (
647 "llm" in engine_init_params[1:]
648 and llm
649 and "llm" not in filtered_params
650 ):
651 # If LLM was provided and engine accepts it, pass it through
652 filtered_params["llm"] = llm
653 logger.info(
654 f"Passing LLM to {engine_name} (engine accepts it and LLM was provided)"
655 )
657 # Add API key if required and not already in filtered_params
658 if (
659 engine_config.get("requires_api_key", False)
660 and "api_key" not in filtered_params
661 ):
662 # Use the api_key we got earlier from settings
663 if api_key: 663 ↛ 666line 663 didn't jump to line 666 because the condition on line 663 was always true
664 filtered_params["api_key"] = api_key
666 logger.info(
667 f"Creating {engine_name} with filtered parameters: {filtered_params.keys()}"
668 )
670 # Create the engine instance with filtered parameters
671 engine = engine_class(**filtered_params)
673 # Stamp the registry name so BaseSearchEngine._verify_egress_scope()
674 # can check egress scope at run time.
675 if isinstance(engine, BaseSearchEngine):
676 engine._engine_name = engine_name
678 # Most engine subclasses do not name ``programmatic_mode`` in their
679 # signature (or accept it via **kwargs without forwarding to
680 # ``super().__init__``), so the constructor often falls back to the
681 # BaseSearchEngine default of False even when the API caller asked
682 # for True. Apply the requested mode post-construction so the
683 # engine's rate tracker matches.
684 if isinstance(engine, BaseSearchEngine) and (
685 engine.programmatic_mode != programmatic_mode
686 ):
687 engine._configure_programmatic_mode(programmatic_mode)
689 # Same swallow: Wikipedia (and others) accept **kwargs but do not
690 # forward ``search_snippets_only`` to BaseSearchEngine, so run()
691 # would keep the True default and skip _get_full_content. Stamp it
692 # after construction whenever the factory resolved a value.
693 if (
694 isinstance(engine, BaseSearchEngine)
695 and "search_snippets_only" in all_params
696 ):
697 engine.search_snippets_only = bool(
698 all_params["search_snippets_only"]
699 )
701 # Thread the single research-scoped content cache into the engine
702 # (explicit wins; engines also fall back to the registry via
703 # thread-local search context). Covers subclasses that swallow
704 # **kwargs without forwarding to BaseSearchEngine.
705 if isinstance(engine, BaseSearchEngine) and content_cache is not None:
706 try:
707 engine.content_cache = content_cache
708 except Exception: # allow: silent-exception
709 pass
710 # Stamp the bulk-fetch liveness hook for engines built through
711 # this seam (pipeline-strategy full-search progress). The base
712 # engine itself does not fetch, but `_init_full_search` and
713 # `set_fetch_progress_callback` forward it to FullSearchResults.
714 if isinstance(engine, BaseSearchEngine) and (
715 "progress_callback" in all_params
716 ):
717 try:
718 engine.set_fetch_progress_callback(
719 all_params["progress_callback"]
720 )
721 except Exception:
722 logger.debug(
723 "Failed to stamp progress_callback on engine",
724 )
726 # Determine if this engine should use LLM relevance filtering
727 # Priority: per-engine setting > needs_llm_relevance_filter > global setting
728 #
729 # Rationale:
730 # - Engines with needs_llm_relevance_filter=True have poor native relevance ranking
731 # (keyword-only, no ML ranking) and benefit from LLM-based filtering
732 # - Well-ranked engines (Google, Brave) and semantic engines (Exa, Tavily)
733 # do not need this and should not waste LLM calls
734 # - The global skip_relevance_filter only affects unclassified engines
735 # - CrossEngineFilter still ranks combined results at the strategy level
736 should_filter = False
738 # Check for per-engine setting first (highest priority)
739 per_engine_key = f"search.engine.web.{engine_name}.default_params.enable_llm_relevance_filter"
740 if settings_snapshot and per_engine_key in settings_snapshot:
741 per_engine_setting = settings_snapshot[per_engine_key]
742 should_filter = (
743 per_engine_setting.get("value", False)
744 if isinstance(per_engine_setting, dict)
745 else per_engine_setting
746 )
747 logger.info(
748 f"Using per-engine setting for {engine_name}: "
749 f"enable_llm_relevance_filter={should_filter}"
750 )
751 else:
752 # Auto-detection based on engine attribute (medium priority)
753 if (
754 hasattr(engine_class, "needs_llm_relevance_filter")
755 and engine_class.needs_llm_relevance_filter
756 ):
757 should_filter = True
758 logger.info(
759 f"Auto-enabling LLM filtering for {engine_name} "
760 f"(needs_llm_relevance_filter=True)"
761 )
762 else:
763 # Global override only applies to engines without needs_llm_relevance_filter
764 if (
765 settings_snapshot
766 and "search.skip_relevance_filter" in settings_snapshot
767 ):
768 skip_filter_setting = settings_snapshot[
769 "search.skip_relevance_filter"
770 ]
771 skip_filter = (
772 skip_filter_setting.get("value", False)
773 if isinstance(skip_filter_setting, dict)
774 else skip_filter_setting
775 )
776 if skip_filter: 776 ↛ 784line 776 didn't jump to line 784 because the condition on line 776 was always true
777 should_filter = False
778 logger.debug(
779 f"Global skip_relevance_filter=True applied "
780 f"for {engine_name}"
781 )
783 # Apply the setting
784 if should_filter and hasattr(engine, "llm") and engine.llm:
785 engine.enable_llm_relevance_filter = True
786 logger.info(f"✓ Enabled LLM relevance filtering for {engine_name}")
787 elif should_filter:
788 logger.warning(
789 f"LLM relevance filtering requested for {engine_name} "
790 f"but no LLM is available — filtering skipped"
791 )
792 else:
793 logger.debug(f"LLM relevance filtering disabled for {engine_name}")
795 # Check if we need to wrap with full search capabilities. Two
796 # config flags identify engines that participate in the unified
797 # full-search pipeline:
798 # * ``supports_full_search`` — runtime-registered engines
799 # (retrievers, set in ``search_engines_config.py``)
800 # * ``full_search_module`` — registry engines, populated from
801 # ``engine_registry.EngineEntry.full_search_module``
802 # An explicit ``supports_full_search=False`` wins as a per-engine
803 # opt-out, letting users disable wrapping for any engine via
804 # settings even when the registry marks it as wrap-eligible.
805 if kwargs.get("use_full_search", False):
806 explicit_opt_out = (
807 "supports_full_search" in engine_config
808 and not engine_config["supports_full_search"]
809 )
810 if not explicit_opt_out and (
811 engine_config.get("supports_full_search", False)
812 or engine_config.get("full_search_module")
813 ):
814 return _create_full_search_wrapper(
815 engine_name,
816 engine,
817 engine_config,
818 llm,
819 kwargs,
820 username,
821 settings_snapshot,
822 content_cache=content_cache,
823 egress_context=ctx,
824 )
826 return engine # type: ignore[no-any-return]
828 except PolicyDeniedError:
829 # An engine __init__ (or a child engine / full-search wrapper it
830 # builds) may itself consult the PDP and raise. Re-raise so the
831 # denial surfaces to policy-aware callers and the audit trail,
832 # rather than being downgraded to a generic "failed to create"
833 # None — which would fail OPEN from an enforcement standpoint.
834 raise
835 except SettingsValueError:
836 raise
837 except Exception as e:
838 safe_msg = scrub_error(e, api_key)
839 logger.exception(
840 f"Failed to create search engine '{engine_name}' ({type(e).__name__}): {safe_msg}"
841 )
842 return None
845def _create_full_search_wrapper(
846 engine_name: str,
847 base_engine: BaseSearchEngine,
848 engine_config: Dict[str, Any],
849 llm,
850 params: Dict[str, Any],
851 username: str | None = None,
852 settings_snapshot: Dict[str, Any] | None = None,
853 content_cache: Any = None,
854 *,
855 egress_context: Any = None,
856) -> Optional[BaseSearchEngine]:
857 """Create a full search wrapper for the base engine if supported"""
858 try:
859 # Bound up here (re-populated below) so the except handler can
860 # always pull credential values out of it for scrub_error() —
861 # without the hoist, an exception before the dict is rebuilt would
862 # hit a NameError when trying to scrub the keys.
863 wrapper_params: Dict[str, Any] = {}
865 # Get full search class details from engine_config (already has
866 # registry-injected values from search_config()).
867 module_path = engine_config.get("full_search_module")
868 class_name = engine_config.get("full_search_class")
870 if not module_path or not class_name:
871 logger.warning(
872 f"Full search configuration missing for {engine_name}"
873 )
874 return base_engine
876 # Import the full search class
877 full_search_class = get_safe_module_class(module_path, class_name)
879 # Get the wrapper's __init__ parameters to filter out unsupported ones
880 wrapper_init_signature = inspect.signature(full_search_class.__init__)
881 wrapper_init_params = list(wrapper_init_signature.parameters.keys())[
882 1:
883 ] # Skip 'self'
885 # Extract relevant parameters for the full search wrapper
886 wrapper_params = {
887 k: v for k, v in params.items() if k in wrapper_init_params
888 }
890 # Special case for SerpAPI which needs the API key directly
891 if (
892 engine_name == "serpapi"
893 and "serpapi_api_key" in wrapper_init_params
894 ):
895 # Check settings snapshot for API key
896 serpapi_api_key = None
897 if settings_snapshot: 897 ↛ 907line 897 didn't jump to line 907 because the condition on line 897 was always true
898 serpapi_setting = settings_snapshot.get(
899 "search.engine.web.serpapi.api_key"
900 )
901 if serpapi_setting: 901 ↛ 907line 901 didn't jump to line 907 because the condition on line 901 was always true
902 serpapi_api_key = (
903 serpapi_setting.get("value")
904 if isinstance(serpapi_setting, dict)
905 else serpapi_setting
906 )
907 if serpapi_api_key: 907 ↛ 911line 907 didn't jump to line 911 because the condition on line 907 was always true
908 wrapper_params["serpapi_api_key"] = serpapi_api_key
910 # Map some parameter names to what the wrapper expects
911 if (
912 "language" in params
913 and "search_language" not in params
914 and "language" in wrapper_init_params
915 ):
916 wrapper_params["language"] = params["language"]
918 if (
919 "safesearch" not in wrapper_params
920 and "safe_search" in params
921 and "safesearch" in wrapper_init_params
922 ):
923 wrapper_params["safesearch"] = (
924 "active" if params["safe_search"] else "off"
925 )
927 # Special case for Brave which needs the API key directly
928 if engine_name == "brave" and "api_key" in wrapper_init_params:
929 # Check settings snapshot for API key
930 brave_api_key = None
931 if settings_snapshot:
932 brave_setting = settings_snapshot.get(
933 "search.engine.web.brave.api_key"
934 )
935 if brave_setting: 935 ↛ 942line 935 didn't jump to line 942 because the condition on line 935 was always true
936 brave_api_key = (
937 brave_setting.get("value")
938 if isinstance(brave_setting, dict)
939 else brave_setting
940 )
942 if brave_api_key:
943 wrapper_params["api_key"] = brave_api_key
945 # Map some parameter names to what the wrapper expects
946 if (
947 "language" in params
948 and "search_language" not in params
949 and "language" in wrapper_init_params
950 ):
951 wrapper_params["language"] = params["language"]
953 if (
954 "safesearch" not in wrapper_params
955 and "safe_search" in params
956 and "safesearch" in wrapper_init_params
957 ):
958 wrapper_params["safesearch"] = (
959 "moderate" if params["safe_search"] else "off"
960 )
962 # FullSearchResults.language is a justext stoplist name, not a
963 # provider ISO code. LANGUAGE_CODE_ENGINES share kwargs with the
964 # wrapper, so remap (or drop) before construction.
965 if "language" in wrapper_init_params:
966 extraction_language = _resolve_extraction_language(
967 params, settings_snapshot
968 )
969 if extraction_language:
970 wrapper_params["language"] = extraction_language
971 else:
972 wrapper_params.pop("language", None)
974 # Always include llm if it's a parameter
975 if "llm" in wrapper_init_params:
976 wrapper_params["llm"] = llm
978 # If the wrapper needs the base engine and has a parameter for it
979 if "web_search" in wrapper_init_params:
980 wrapper_params["web_search"] = base_engine
982 # The factory already validated this run and its primary engine
983 # against ``ctx``. Reuse that exact context for result-URL fetches.
984 # Without it FullSearchResults skips the per-URL egress scope gate,
985 # so keep the snippets-only engine instead of creating a wrapper.
986 if "egress_context" in wrapper_init_params:
987 if egress_context is None:
988 logger.bind(policy_audit=True).warning(
989 "full-content wrapper skipped: no run egress context",
990 engine=engine_name,
991 )
992 return base_engine
993 wrapper_params["egress_context"] = egress_context
995 # Always pass settings_snapshot if the wrapper accepts it.
996 # `params` is create_search_engine kwargs and never includes this
997 # separate factory argument, so signature filtering drops it otherwise.
998 if "settings_snapshot" in wrapper_init_params and settings_snapshot:
999 wrapper_params["settings_snapshot"] = settings_snapshot
1000 # Private result-fetch grant for the wrapper's SSRF validation and
1001 # download pipeline. Taken from the base engine's own resolution
1002 # only (``BaseSearchEngine.allow_private_result_fetch``, False
1003 # unless the engine resolved a grant, e.g. SearXNG from the instance
1004 # approval AND the env-only opt-in), never from caller params, so a
1005 # stray kwarg cannot open it for an engine that holds no grant
1006 # (issue #2477). The wrapper applies it only under the run's
1007 # ``unprotected`` scope (``FullSearchResults.
1008 # _effective_allow_private_ips``); under every other scope private
1009 # result URLs stay blocked.
1010 if "allow_private_ips" in wrapper_init_params:
1011 wrapper_params["allow_private_ips"] = (
1012 isinstance(base_engine, BaseSearchEngine)
1013 and base_engine.allow_private_result_fetch is True
1014 )
1016 # Thread the shared research cache into the full-search wrapper so
1017 # web_search full-content fetches share it with fetch_content and
1018 # the RAG engines.
1019 if "content_cache" in wrapper_init_params and content_cache is not None: 1019 ↛ 1020line 1019 didn't jump to line 1020 because the condition on line 1019 was never true
1020 wrapper_params["content_cache"] = content_cache
1022 logger.debug(
1023 f"Creating full search wrapper for {engine_name} with filtered parameters: {wrapper_params.keys()}"
1024 )
1026 # Create the full search wrapper with filtered parameters
1027 service: BaseSearchEngine = full_search_class(**wrapper_params)
1028 # Belt-and-suspenders for wrappers that swallow **kwargs without
1029 # forwarding content_cache to their base.
1030 if content_cache is not None: 1030 ↛ 1031line 1030 didn't jump to line 1031 because the condition on line 1030 was never true
1031 try:
1032 existing = getattr(service, "content_cache", None)
1033 if existing is None:
1034 service.content_cache = content_cache
1035 except Exception: # allow: silent-exception
1036 pass
1037 return service
1039 except SettingsValueError:
1040 raise
1041 except Exception as e:
1042 # Extract any credential values the wrapper was going to receive
1043 # so a constructor exception echoing them gets literal-redacted
1044 # (scrub_error skips None/falsy values and str-coerces the rest).
1045 safe_msg = scrub_error(
1046 e,
1047 wrapper_params.get("api_key"),
1048 wrapper_params.get("serpapi_api_key"),
1049 )
1050 logger.exception(
1051 f"Failed to create full search wrapper for {engine_name} ({type(e).__name__}): {safe_msg}"
1052 )
1053 return base_engine
1056def get_search(
1057 search_tool: str,
1058 llm_instance,
1059 max_results: int = 10,
1060 region: str = "us",
1061 time_period: str = "all",
1062 safe_search: bool = True,
1063 search_snippets_only: Optional[bool] = None,
1064 search_language: str = "English",
1065 max_filtered_results: Optional[int] = None,
1066 settings_snapshot: Dict[str, Any] | None = None,
1067 programmatic_mode: bool = False,
1068 content_cache: Any = None,
1069 progress_callback=None,
1070):
1071 """
1072 Get search tool instance based on the provided parameters.
1074 Args:
1075 search_tool: Name of the search engine to use
1076 llm_instance: Language model instance
1077 max_results: Maximum number of search results
1078 region: Search region/locale
1079 time_period: Time period for search results
1080 safe_search: Whether to enable safe search
1081 search_snippets_only: Whether to return just snippets (vs. full content).
1082 Defaults to None (defers to settings_snapshot or engine default).
1083 search_language: Language for search results
1084 max_filtered_results: Maximum number of results to keep after filtering
1085 programmatic_mode: If True, disables database operations and metrics tracking
1086 progress_callback: Optional ``(done, total)`` bulk-fetch liveness hook,
1087 forwarded to ``FullSearchResults`` when the engine is wrapped for
1088 full search. This is the pipeline-strategy live path for fetch
1089 progress (see #7193); ``None`` disables it.
1091 Returns:
1092 Initialized search engine instance
1093 """
1094 # Common parameters
1095 params = {
1096 "max_results": max_results,
1097 "llm": llm_instance, # Only used by engines that need it
1098 }
1100 # Add search_snippets_only if provided
1101 if search_snippets_only is not None:
1102 params["search_snippets_only"] = search_snippets_only
1104 # Add max_filtered_results if provided
1105 if max_filtered_results is not None:
1106 params["max_filtered_results"] = max_filtered_results
1108 # Forward the bulk-fetch liveness hook to the FullSearchResults
1109 # wrapper below (via create_search_engine kwargs → signature
1110 # filtering). Only matters when use_full_search resolves True.
1111 if progress_callback is not None:
1112 params["progress_callback"] = progress_callback
1114 # Resolve use_full_search for engines that support full search wrapping.
1115 # When search_snippets_only is not explicitly passed, defer to settings_snapshot
1116 # if present; otherwise default to False (matching BaseSearchEngine's snippet-only default).
1117 if search_snippets_only is not None:
1118 use_full_search = not search_snippets_only
1119 else:
1120 resolved_snippets = _resolve_snippets_only(settings_snapshot)
1121 use_full_search = (
1122 (not resolved_snippets) if resolved_snippets is not None else False
1123 )
1125 # Add engine-specific parameters
1126 if search_tool in [
1127 "serpapi",
1128 "google_pse",
1129 "brave",
1130 "mojeek",
1131 ]:
1132 params.update(
1133 {
1134 "region": region,
1135 "safe_search": safe_search,
1136 "use_full_search": use_full_search,
1137 }
1138 )
1140 if search_tool == "searxng":
1141 params["use_full_search"] = use_full_search
1143 if search_tool in ["serpapi", "brave", "google_pse", "wikinews"]:
1144 params["search_language"] = search_language
1146 if search_tool == "tinyfish": 1146 ↛ 1147line 1146 didn't jump to line 1147 because the condition on line 1146 was never true
1147 if (
1148 region
1149 and region.lower() not in _NON_LOCATION_REGIONS
1150 and len(region.strip()) == 2
1151 and region.strip().isalpha()
1152 ):
1153 params["location"] = region.strip().upper()
1154 norm_lang = _normalize_language_code(search_language)
1155 if norm_lang:
1156 params["language"] = norm_lang
1158 if search_tool == "sofya":
1159 # Sofya derives its request depth from search_snippets_only: the paid
1160 # "basic" depth (inline page content) is only requested when full
1161 # content is used. Handled via common params / snapshot fallback.
1162 pass
1164 if search_tool == "wikinews":
1165 params["adaptive_search"] = bool(
1166 (settings_snapshot or {})
1167 .get("search.engine.web.wikinews.adaptive_search", {})
1168 .get("value", True)
1169 )
1171 if is_collection_engine(search_tool):
1172 # BaseSearchEngine defaults this to True, so the RAG engines stay
1173 # snippet-only unless the setting is forwarded in params or snapshot.
1174 pass
1176 if search_tool in TIME_PERIOD_ENGINES:
1177 params["time_period"] = time_period
1179 # Create and return the search engine
1180 logger.info(
1181 f"Creating search engine for tool: {search_tool} (type: {type(search_tool)}) with params: {params.keys()}"
1182 )
1183 logger.info(
1184 f"About to call create_search_engine with search_tool={search_tool}, settings_snapshot type={type(settings_snapshot)}"
1185 )
1186 logger.info(
1187 f"Params being passed to create_search_engine: {list(params.keys()) if isinstance(params, dict) else type(params)}"
1188 )
1190 engine = create_search_engine(
1191 search_tool,
1192 settings_snapshot=settings_snapshot,
1193 programmatic_mode=programmatic_mode,
1194 content_cache=content_cache,
1195 **params,
1196 )
1198 # Add debugging to check if engine is None
1199 if engine is None:
1200 logger.error(
1201 f"Failed to create search engine for {search_tool} - returned None"
1202 )
1203 else:
1204 engine_type = type(engine).__name__
1205 logger.info(
1206 f"Successfully created search engine of type: {engine_type}"
1207 )
1208 # Check if the engine has run method
1209 if hasattr(engine, "run"): 1209 ↛ 1212line 1209 didn't jump to line 1212 because the condition on line 1209 was always true
1210 logger.info(f"Engine has 'run' method: {engine.run}")
1211 else:
1212 logger.error("Engine does NOT have 'run' method!")
1214 # Check engine availability flag / method
1215 avail = None
1216 if getattr(engine, "_is_available", None) is not None:
1217 avail = getattr(engine, "_is_available")
1218 elif hasattr(engine, "is_available"):
1219 if callable(engine.is_available): 1219 ↛ 1246line 1219 didn't jump to line 1246 because the condition on line 1219 was always true
1220 try:
1221 avail = engine.is_available(
1222 settings_snapshot=settings_snapshot
1223 )
1224 except TypeError:
1225 try:
1226 avail = engine.is_available()
1227 except SettingsValueError:
1228 raise
1229 except Exception as exc:
1230 logger.debug(
1231 "Engine availability fallback failed for {} ({})",
1232 engine_type,
1233 type(exc).__name__,
1234 )
1235 avail = None
1236 except SettingsValueError:
1237 raise
1238 except Exception as exc:
1239 logger.debug(
1240 "Engine availability check failed for {} ({})",
1241 engine_type,
1242 type(exc).__name__,
1243 )
1244 avail = None
1245 else:
1246 avail = engine.is_available
1247 if avail is not None:
1248 logger.info(f"Engine availability flag: {avail}")
1250 return engine