Coverage for src/local_deep_research/web_search_engines/engines/search_engine_serper.py: 99%

1from ...security.secure_logging import logger 

2from typing import Any, Dict, List, Optional 

3import requests 

4 

5from langchain_core.language_models import BaseLLM 

6 

7from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity 

8from ..rate_limiting import RateLimitError 

9from ...security import safe_post 

10 

11 

12class SerperSearchEngine(BaseSearchEngine): 

13 """Google search engine implementation using Serper API with two-phase approach""" 

14 

15 # Mark as public search engine 

16 is_public = True 

17 egress_sensitivity = Sensitivity.NON_SENSITIVE 

18 egress_exposure = Exposure.EXPOSING 

19 # Mark as generic search engine (general web search via Google) 

20 is_generic = True 

21 

22 # Class constants 

23 BASE_URL = "https://google.serper.dev/search" 

24 DEFAULT_TIMEOUT = 30 

25 DEFAULT_REGION = "us" 

26 DEFAULT_LANGUAGE = "en" 

27 

28 def __init__( 

29 self, 

30 max_results: int = 10, 

31 region: str = "us", 

32 time_period: Optional[str] = None, 

33 safe_search: bool = True, 

34 search_language: str = "en", 

35 api_key: Optional[str] = None, 

36 llm: Optional[BaseLLM] = None, 

37 include_full_content: bool = False, 

38 max_filtered_results: Optional[int] = None, 

39 settings_snapshot: Optional[Dict[str, Any]] = None, 

40 **kwargs, 

41 ): 

42 """ 

43 Initialize the Serper search engine. 

44 

45 Args: 

46 max_results: Maximum number of search results (default 10) 

47 region: Country code for localized results (e.g., 'us', 'gb', 'fr') 

48 time_period: Time filter for results ('day', 'week', 'month', 'year', or None for all time) 

49 safe_search: Whether to enable safe search 

50 search_language: Language code for results (e.g., 'en', 'es', 'fr') 

51 api_key: Serper API key (can also be set in settings) 

52 llm: Language model for relevance filtering 

53 include_full_content: Retained for backward compatibility; full-content 

54 retrieval is handled via the factory wrapper. 

55 max_filtered_results: Maximum number of results to keep after filtering 

56 settings_snapshot: Settings snapshot for thread context 

57 **kwargs: Additional parameters (ignored but accepted for compatibility) 

58 """ 

59 # Initialize the BaseSearchEngine with LLM, max_filtered_results, and max_results 

60 super().__init__( 

61 llm=llm, 

62 max_filtered_results=max_filtered_results, 

63 max_results=max_results, 

64 include_full_content=include_full_content, 

65 settings_snapshot=settings_snapshot, 

66 ) 

67 self.region = region 

68 self.time_period = time_period 

69 self.safe_search = safe_search 

70 self.search_language = search_language 

71 

72 # Get API key - check params, settings, or env vars 

73 serper_api_key = self._resolve_api_key( 

74 api_key, 

75 "search.engine.web.serper.api_key", 

76 engine_name="Serper", 

77 settings_snapshot=settings_snapshot, 

78 ) 

79 

80 self.api_key = serper_api_key 

81 self.base_url = self.BASE_URL 

82 # Note: self.engine_type is automatically set by parent BaseSearchEngine class 

83 

84 # Initialize per-query attributes (reset in _get_previews per search) 

85 self._knowledge_graph = None 

86 

87 # Full-content retrieval is the factory wrapper's job — the 

88 # wrapper fetches pages via ``batch_fetch_and_extract`` and 

89 # populates ``full_content`` on each result; the inner engine 

90 # only emits snippets from ``_get_previews``. 

91 

92 def _get_previews(self, query: str) -> List[Dict[str, Any]]: 

93 """ 

94 Get preview information from Serper API. 

95 

96 Args: 

97 query: The search query 

98 

99 Returns: 

100 List of preview dictionaries 

101 """ 

102 logger.info("Getting search results from Serper API") 

103 

104 # Reset per-query attributes to prevent leakage between searches 

105 self._knowledge_graph = None 

106 

107 try: 

108 # Build request payload 

109 payload = { 

110 "q": query, 

111 # Google's "num" tops out at 100 results per request; cap so a 

112 # large user-supplied max_results can't request an unbounded page. 

113 "num": min(self.max_results, 100), 

114 "gl": self.region, 

115 "hl": self.search_language, 

116 } 

117 

118 # Add optional parameters 

119 if self.time_period: 

120 # Map time periods to Serper's format 

121 time_mapping = { 

122 "day": "d", 

123 "week": "w", 

124 "month": "m", 

125 "year": "y", 

126 } 

127 if self.time_period in time_mapping: 

128 payload["tbs"] = f"qdr:{time_mapping[self.time_period]}" 

129 

130 # Apply rate limiting before request 

131 self._last_wait_time = self.rate_tracker.apply_rate_limit( 

132 self.engine_type 

133 ) 

134 

135 # Make API request 

136 headers = { 

137 "X-API-KEY": self.api_key, 

138 "Content-Type": "application/json", 

139 } 

140 

141 response = safe_post( 

142 self.base_url, 

143 headers=headers, 

144 json=payload, 

145 timeout=self.DEFAULT_TIMEOUT, 

146 ) 

147 

148 # Check for rate limits 

149 self._raise_if_rate_limit(requests.HTTPError(response=response)) 

150 

151 response.raise_for_status() 

152 

153 data = response.json() 

154 

155 # Extract organic results 

156 organic_results = data.get("organic", []) 

157 

158 # Format results as previews 

159 previews = [] 

160 for idx, result in enumerate(organic_results): 

161 # Extract display link 

162 link = result.get("link", "") 

163 display_link = self._extract_display_link(link) 

164 

165 preview = { 

166 "id": idx, 

167 "title": result.get("title", ""), 

168 "link": link, 

169 "snippet": result.get("snippet", ""), 

170 "displayed_link": display_link, 

171 "position": result.get("position", idx + 1), 

172 } 

173 

174 # Store full Serper result for later 

175 preview["_full_result"] = result 

176 

177 # Only include optional fields if present to avoid None values 

178 # This keeps the preview dict cleaner and saves memory 

179 if "sitelinks" in result: 

180 preview["sitelinks"] = result["sitelinks"] 

181 

182 if "date" in result: 

183 preview["date"] = result["date"] 

184 

185 if "attributes" in result: 

186 preview["attributes"] = result["attributes"] 

187 

188 previews.append(preview) 

189 

190 # Store the previews for potential full content retrieval 

191 self._search_results = previews 

192 

193 # Also store knowledge graph if available 

194 if "knowledgeGraph" in data: 

195 self._knowledge_graph = data["knowledgeGraph"] 

196 logger.info( 

197 f"Found knowledge graph for query: {data['knowledgeGraph'].get('title', 'Unknown')}" 

198 ) 

199 

200 return previews 

201 

202 except RateLimitError: 

203 raise # Re-raise rate limit errors 

204 except requests.exceptions.RequestException as e: 

205 self._record_search_failure(e) 

206 safe_msg = self._scrub_error(e) 

207 logger.warning(f"Error getting Serper API results: {safe_msg}") 

208 self._raise_if_rate_limit(e) 

209 return [] 

210 except Exception as e: 

211 self._record_search_failure(e) 

212 safe_msg = self._scrub_error(e) 

213 logger.warning( 

214 f"Unexpected error getting Serper API results: {safe_msg}" 

215 ) 

216 return [] 

217 

218 def _get_full_content( 

219 self, relevant_items: List[Dict[str, Any]] 

220 ) -> List[Dict[str, Any]]: 

221 """ 

222 Get full content for the relevant search results. 

223 Extends base implementation to include knowledge graph data. 

224 

225 Args: 

226 relevant_items: List of relevant preview dictionaries 

227 

228 Returns: 

229 List of result dictionaries with full content if requested 

230 """ 

231 results = super()._get_full_content(relevant_items) 

232 

233 # Include knowledge graph if available 

234 if results and hasattr(self, "_knowledge_graph"): 234 ↛ 237line 234 didn't jump to line 237 because the condition on line 234 was always true

235 results[0]["knowledge_graph"] = self._knowledge_graph 

236 

237 return results 

238 

239 def _temp_attributes(self): 

240 """Return list of temporary attribute names to clean up after run().""" 

241 return super()._temp_attributes() + [ 

242 "_knowledge_graph", 

243 ]