Coverage for src/local_deep_research/web_search_engines/engines/search_engine_serpapi.py: 100%
42 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1from ...security.secure_logging import logger
2from typing import Any, Dict, List, Optional
4from langchain_community.utilities import SerpAPIWrapper
5from langchain_core.language_models import BaseLLM
7from ..rate_limiting import RateLimitError
8from ..search_engine_base import (
9 VALID_TIME_PERIODS,
10 BaseSearchEngine,
11 Exposure,
12 Sensitivity,
13)
16class SerpAPISearchEngine(BaseSearchEngine):
17 """Google search engine implementation using SerpAPI with two-phase approach"""
19 # Mark as public search engine
20 is_public = True
21 egress_sensitivity = Sensitivity.NON_SENSITIVE
22 egress_exposure = Exposure.EXPOSING
23 # Mark as generic search engine (general web search via Google)
24 is_generic = True
26 def __init__(
27 self,
28 max_results: int = 10,
29 region: str = "us",
30 time_period: str = "y",
31 safe_search: bool = True,
32 search_language: str = "English",
33 api_key: Optional[str] = None,
34 language_code_mapping: Optional[Dict[str, str]] = None,
35 llm: Optional[BaseLLM] = None,
36 include_full_content: bool = False,
37 max_filtered_results: Optional[int] = None,
38 settings_snapshot: Optional[Dict[str, Any]] = None,
39 **kwargs,
40 ):
41 """
42 Initialize the SerpAPI search engine.
44 Args:
45 max_results: Maximum number of search results
46 region: Region code for search results
47 time_period: Time period filter (d/w/m/y); forwarded to Google's
48 ``tbs`` param as qdr:{d,w,m,y}. ``all`` (or any unrecognized
49 value) omits the filter.
50 safe_search: Whether to enable safe search
51 search_language: Language for search results
52 api_key: SerpAPI API key (can also be set via LDR_SEARCH_ENGINE_WEB_SERPAPI_API_KEY env var or in UI settings)
53 language_code_mapping: Mapping from language names to codes
54 llm: Language model for relevance filtering
55 include_full_content: Whether to include full webpage content in results
56 max_filtered_results: Maximum number of results to keep after filtering
57 settings_snapshot: Settings snapshot for thread context
58 **kwargs: Additional parameters (ignored but accepted for compatibility)
59 """
60 # Initialize the BaseSearchEngine with LLM, max_filtered_results, and max_results
61 super().__init__(
62 llm=llm,
63 max_filtered_results=max_filtered_results,
64 max_results=max_results,
65 include_full_content=include_full_content,
66 settings_snapshot=settings_snapshot,
67 )
69 # Set up language code mapping
70 if language_code_mapping is None:
71 from ...utilities.search_utilities import LANGUAGE_CODE_MAP
73 language_code_mapping = LANGUAGE_CODE_MAP
75 # Get API key - check params, settings, or env vars
76 serpapi_api_key = self._resolve_api_key(
77 api_key,
78 "search.engine.web.serpapi.api_key",
79 engine_name="SerpAPI",
80 settings_snapshot=settings_snapshot,
81 )
82 # Store for error-message redaction (BaseSearchEngine._scrub_error
83 # reads self.api_key via _secret_attrs). SerpAPIWrapper sends the key
84 # as a URL param the regex pass already catches, but storing it gives
85 # the literal-redaction pass a belt-and-suspenders for any other shape.
86 self.api_key = serpapi_api_key
88 # Get language code
89 language_code = language_code_mapping.get(search_language.lower(), "en")
91 # Initialize SerpAPI wrapper
92 params = {
93 "engine": "google",
94 "hl": language_code,
95 "gl": region,
96 "safe": "active" if safe_search else "off",
97 # Google's "num" tops out at 100 results per request; cap so a
98 # large user-supplied max_results can't request an unbounded page.
99 # Use the base-class-normalized self.max_results (positive int)
100 # rather than the raw arg, which could be None/str.
101 "num": min(self.max_results, 100),
102 }
103 # Google's tbs validator accepts qdr:d/w/m/y (h is not one of LDR's
104 # canonical time_period codes). Only forward recognized codes; "all",
105 # any unrecognized value (typo, None, "") and non-string types omit
106 # the param entirely so the search is unfiltered rather than
107 # silently malformed (e.g. "qdr:all") — the isinstance guard also
108 # keeps unhashable values from raising on the frozenset lookup.
109 if isinstance(time_period, str) and time_period in VALID_TIME_PERIODS:
110 params["tbs"] = f"qdr:{time_period}"
111 self.engine = SerpAPIWrapper(
112 serpapi_api_key=serpapi_api_key,
113 params=params,
114 )
116 # If full content is requested, initialize FullSearchResults
117 self._init_full_search(
118 web_search=self.engine,
119 language=search_language,
120 max_results=max_results,
121 region=region,
122 time_period=time_period,
123 safe_search="Moderate" if safe_search else "Off",
124 )
126 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
127 """
128 Get preview information from SerpAPI.
130 Args:
131 query: The search query
133 Returns:
134 List of preview dictionaries
135 """
136 logger.info("Getting search results from SerpAPI")
138 try:
139 # Get search results from SerpAPI
140 organic_results = self.engine.results(query).get(
141 "organic_results", []
142 )
144 # Format results as previews
145 previews: list[dict[str, Any]] = []
146 for result in organic_results:
147 preview = {
148 "id": result.get(
149 "position", len(previews)
150 ), # Use position as ID
151 "title": result.get("title", ""),
152 "link": result.get("link", ""),
153 "snippet": result.get("snippet", ""),
154 "displayed_link": result.get("displayed_link", ""),
155 "position": result.get("position"),
156 }
158 # Store full SerpAPI result for later
159 preview["_full_result"] = result
161 previews.append(preview)
163 # Store the previews for potential full content retrieval
164 self._search_results = previews
166 return previews
168 except RateLimitError:
169 raise
170 except Exception as e:
171 safe_msg = self._scrub_error(e)
172 logger.warning(f"Error getting SerpAPI results: {safe_msg}")
173 self._raise_if_rate_limit(e)
174 return []