Coverage for src/local_deep_research/web_search_engines/engines/search_engine_serpapi.py: 100%
41 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +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: Retained for backward compatibility; full-content
56 retrieval is handled via the factory wrapper.
57 max_filtered_results: Maximum number of results to keep after filtering
58 settings_snapshot: Settings snapshot for thread context
59 **kwargs: Additional parameters (ignored but accepted for compatibility)
60 """
61 # Initialize the BaseSearchEngine with LLM, max_filtered_results, and max_results
62 super().__init__(
63 llm=llm,
64 max_filtered_results=max_filtered_results,
65 max_results=max_results,
66 include_full_content=include_full_content,
67 settings_snapshot=settings_snapshot,
68 )
70 # Set up language code mapping
71 if language_code_mapping is None:
72 from ...utilities.search_utilities import LANGUAGE_CODE_MAP
74 language_code_mapping = LANGUAGE_CODE_MAP
76 # Get API key - check params, settings, or env vars
77 serpapi_api_key = self._resolve_api_key(
78 api_key,
79 "search.engine.web.serpapi.api_key",
80 engine_name="SerpAPI",
81 settings_snapshot=settings_snapshot,
82 )
83 # Store for error-message redaction (BaseSearchEngine._scrub_error
84 # reads self.api_key via _secret_attrs). SerpAPIWrapper sends the key
85 # as a URL param the regex pass already catches, but storing it gives
86 # the literal-redaction pass a belt-and-suspenders for any other shape.
87 self.api_key = serpapi_api_key
89 # Get language code
90 language_code = language_code_mapping.get(search_language.lower(), "en")
92 # Initialize SerpAPI wrapper
93 params = {
94 "engine": "google",
95 "hl": language_code,
96 "gl": region,
97 "safe": "active" if safe_search else "off",
98 # Google's "num" tops out at 100 results per request; cap so a
99 # large user-supplied max_results can't request an unbounded page.
100 # Use the base-class-normalized self.max_results (positive int)
101 # rather than the raw arg, which could be None/str.
102 "num": min(self.max_results, 100),
103 }
104 # Google's tbs validator accepts qdr:d/w/m/y (h is not one of LDR's
105 # canonical time_period codes). Only forward recognized codes; "all",
106 # any unrecognized value (typo, None, "") and non-string types omit
107 # the param entirely so the search is unfiltered rather than
108 # silently malformed (e.g. "qdr:all") — the isinstance guard also
109 # keeps unhashable values from raising on the frozenset lookup.
110 if isinstance(time_period, str) and time_period in VALID_TIME_PERIODS:
111 params["tbs"] = f"qdr:{time_period}"
112 self.engine = SerpAPIWrapper(
113 serpapi_api_key=serpapi_api_key,
114 params=params,
115 )
117 # Full-content retrieval is the factory wrapper's job — the
118 # wrapper fetches pages via ``batch_fetch_and_extract`` and
119 # populates ``full_content`` on each result; the inner engine
120 # only emits snippets from ``_get_previews``.
122 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
123 """
124 Get preview information from SerpAPI.
126 Args:
127 query: The search query
129 Returns:
130 List of preview dictionaries
131 """
132 logger.info("Getting search results from SerpAPI")
134 try:
135 # Get search results from SerpAPI
136 organic_results = self.engine.results(query).get(
137 "organic_results", []
138 )
140 # Format results as previews
141 previews: list[dict[str, Any]] = []
142 for result in organic_results:
143 preview = {
144 "id": result.get(
145 "position", len(previews)
146 ), # Use position as ID
147 "title": result.get("title", ""),
148 "link": result.get("link", ""),
149 "snippet": result.get("snippet", ""),
150 "displayed_link": result.get("displayed_link", ""),
151 "position": result.get("position"),
152 }
154 # Store full SerpAPI result for later
155 preview["_full_result"] = result
157 previews.append(preview)
159 # Store the previews for potential full content retrieval
160 self._search_results = previews
162 return previews
164 except RateLimitError:
165 raise
166 except Exception as e:
167 safe_msg = self._scrub_error(e)
168 logger.warning(f"Error getting SerpAPI results: {safe_msg}")
169 self._raise_if_rate_limit(e)
170 return []