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: Whether to include full webpage content in results 

54 max_filtered_results: Maximum number of results to keep after filtering 

55 settings_snapshot: Settings snapshot for thread context 

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

57 """ 

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

59 super().__init__( 

60 llm=llm, 

61 max_filtered_results=max_filtered_results, 

62 max_results=max_results, 

63 include_full_content=include_full_content, 

64 settings_snapshot=settings_snapshot, 

65 ) 

66 self.region = region 

67 self.time_period = time_period 

68 self.safe_search = safe_search 

69 self.search_language = search_language 

70 

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

72 serper_api_key = self._resolve_api_key( 

73 api_key, 

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

75 engine_name="Serper", 

76 settings_snapshot=settings_snapshot, 

77 ) 

78 

79 self.api_key = serper_api_key 

80 self.base_url = self.BASE_URL 

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

82 

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

84 self._knowledge_graph = None 

85 

86 # If full content is requested, initialize FullSearchResults 

87 self._init_full_search( 

88 web_search=None, # We'll handle the search ourselves 

89 language=search_language, 

90 max_results=max_results, 

91 region=region, 

92 time_period=time_period, 

93 safe_search="Moderate" if safe_search else "Off", 

94 ) 

95 

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

97 """ 

98 Get preview information from Serper API. 

99 

100 Args: 

101 query: The search query 

102 

103 Returns: 

104 List of preview dictionaries 

105 """ 

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

107 

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

109 self._knowledge_graph = None 

110 

111 try: 

112 # Build request payload 

113 payload = { 

114 "q": query, 

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

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

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

118 "gl": self.region, 

119 "hl": self.search_language, 

120 } 

121 

122 # Add optional parameters 

123 if self.time_period: 

124 # Map time periods to Serper's format 

125 time_mapping = { 

126 "day": "d", 

127 "week": "w", 

128 "month": "m", 

129 "year": "y", 

130 } 

131 if self.time_period in time_mapping: 

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

133 

134 # Apply rate limiting before request 

135 self._last_wait_time = self.rate_tracker.apply_rate_limit( 

136 self.engine_type 

137 ) 

138 

139 # Make API request 

140 headers = { 

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

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

143 } 

144 

145 response = safe_post( 

146 self.base_url, 

147 headers=headers, 

148 json=payload, 

149 timeout=self.DEFAULT_TIMEOUT, 

150 ) 

151 

152 # Check for rate limits 

153 self._raise_if_rate_limit(response.status_code) 

154 

155 response.raise_for_status() 

156 

157 data = response.json() 

158 

159 # Extract organic results 

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

161 

162 # Format results as previews 

163 previews = [] 

164 for idx, result in enumerate(organic_results): 

165 # Extract display link 

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

167 display_link = self._extract_display_link(link) 

168 

169 preview = { 

170 "id": idx, 

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

172 "link": link, 

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

174 "displayed_link": display_link, 

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

176 } 

177 

178 # Store full Serper result for later 

179 preview["_full_result"] = result 

180 

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

182 # This keeps the preview dict cleaner and saves memory 

183 if "sitelinks" in result: 

184 preview["sitelinks"] = result["sitelinks"] 

185 

186 if "date" in result: 

187 preview["date"] = result["date"] 

188 

189 if "attributes" in result: 

190 preview["attributes"] = result["attributes"] 

191 

192 previews.append(preview) 

193 

194 # Store the previews for potential full content retrieval 

195 self._search_results = previews 

196 

197 # Also store knowledge graph if available 

198 if "knowledgeGraph" in data: 

199 self._knowledge_graph = data["knowledgeGraph"] 

200 logger.info( 

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

202 ) 

203 

204 return previews 

205 

206 except RateLimitError: 

207 raise # Re-raise rate limit errors 

208 except requests.exceptions.RequestException as e: 

209 safe_msg = self._scrub_error(e) 

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

211 self._raise_if_rate_limit(e) 

212 return [] 

213 except Exception as e: 

214 safe_msg = self._scrub_error(e) 

215 logger.warning( 

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

217 ) 

218 return [] 

219 

220 def _get_full_content( 

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

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

223 """ 

224 Get full content for the relevant search results. 

225 Extends base implementation to include knowledge graph data. 

226 

227 Args: 

228 relevant_items: List of relevant preview dictionaries 

229 

230 Returns: 

231 List of result dictionaries with full content if requested 

232 """ 

233 results = super()._get_full_content(relevant_items) 

234 

235 # Include knowledge graph if available 

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

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

238 

239 return results 

240 

241 def _temp_attributes(self): 

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

243 return super()._temp_attributes() + [ 

244 "_knowledge_graph", 

245 ]