Coverage for src/local_deep_research/web_search_engines/engines/search_engine_serpapi.py: 100%

1from ...security.secure_logging import logger 

2from typing import Any, Dict, List, Optional 

3 

4from langchain_community.utilities import SerpAPIWrapper 

5from langchain_core.language_models import BaseLLM 

6 

7from ..rate_limiting import RateLimitError 

8from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity 

9 

10 

11class SerpAPISearchEngine(BaseSearchEngine): 

12 """Google search engine implementation using SerpAPI with two-phase approach""" 

13 

14 # Mark as public search engine 

15 is_public = True 

16 egress_sensitivity = Sensitivity.NON_SENSITIVE 

17 egress_exposure = Exposure.EXPOSING 

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

19 is_generic = True 

20 

21 def __init__( 

22 self, 

23 max_results: int = 10, 

24 region: str = "us", 

25 time_period: str = "y", 

26 safe_search: bool = True, 

27 search_language: str = "English", 

28 api_key: Optional[str] = None, 

29 language_code_mapping: Optional[Dict[str, str]] = None, 

30 llm: Optional[BaseLLM] = None, 

31 include_full_content: bool = False, 

32 max_filtered_results: Optional[int] = None, 

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

34 **kwargs, 

35 ): 

36 """ 

37 Initialize the SerpAPI search engine. 

38 

39 Args: 

40 max_results: Maximum number of search results 

41 region: Region code for search results 

42 time_period: Time period for search results 

43 safe_search: Whether to enable safe search 

44 search_language: Language for search results 

45 api_key: SerpAPI API key (can also be set via LDR_SEARCH_ENGINE_WEB_SERPAPI_API_KEY env var or in UI settings) 

46 language_code_mapping: Mapping from language names to codes 

47 llm: Language model for relevance filtering 

48 include_full_content: Whether to include full webpage content in results 

49 max_filtered_results: Maximum number of results to keep after filtering 

50 settings_snapshot: Settings snapshot for thread context 

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

52 """ 

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

54 super().__init__( 

55 llm=llm, 

56 max_filtered_results=max_filtered_results, 

57 max_results=max_results, 

58 include_full_content=include_full_content, 

59 settings_snapshot=settings_snapshot, 

60 ) 

61 

62 # Set up language code mapping 

63 if language_code_mapping is None: 

64 from ...utilities.search_utilities import LANGUAGE_CODE_MAP 

65 

66 language_code_mapping = LANGUAGE_CODE_MAP 

67 

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

69 serpapi_api_key = self._resolve_api_key( 

70 api_key, 

71 "search.engine.web.serpapi.api_key", 

72 engine_name="SerpAPI", 

73 settings_snapshot=settings_snapshot, 

74 ) 

75 # Store for error-message redaction (BaseSearchEngine._scrub_error 

76 # reads self.api_key via _secret_attrs). SerpAPIWrapper sends the key 

77 # as a URL param the regex pass already catches, but storing it gives 

78 # the literal-redaction pass a belt-and-suspenders for any other shape. 

79 self.api_key = serpapi_api_key 

80 

81 # Get language code 

82 language_code = language_code_mapping.get(search_language.lower(), "en") 

83 

84 # Initialize SerpAPI wrapper 

85 self.engine = SerpAPIWrapper( 

86 serpapi_api_key=serpapi_api_key, 

87 params={ 

88 "engine": "google", 

89 "hl": language_code, 

90 "gl": region, 

91 "safe": "active" if safe_search else "off", 

92 "tbs": f"qdr:{time_period}", 

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

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

95 # Use the base-class-normalized self.max_results (positive int) 

96 # rather than the raw arg, which could be None/str. 

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

98 }, 

99 ) 

100 

101 # If full content is requested, initialize FullSearchResults 

102 self._init_full_search( 

103 web_search=self.engine, 

104 language=search_language, 

105 max_results=max_results, 

106 region=region, 

107 time_period=time_period, 

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

109 ) 

110 

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

112 """ 

113 Get preview information from SerpAPI. 

114 

115 Args: 

116 query: The search query 

117 

118 Returns: 

119 List of preview dictionaries 

120 """ 

121 logger.info("Getting search results from SerpAPI") 

122 

123 try: 

124 # Get search results from SerpAPI 

125 organic_results = self.engine.results(query).get( 

126 "organic_results", [] 

127 ) 

128 

129 # Format results as previews 

130 previews: list[dict[str, Any]] = [] 

131 for result in organic_results: 

132 preview = { 

133 "id": result.get( 

134 "position", len(previews) 

135 ), # Use position as ID 

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

137 "link": result.get("link", ""), 

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

139 "displayed_link": result.get("displayed_link", ""), 

140 "position": result.get("position"), 

141 } 

142 

143 # Store full SerpAPI result for later 

144 preview["_full_result"] = result 

145 

146 previews.append(preview) 

147 

148 # Store the previews for potential full content retrieval 

149 self._search_results = previews 

150 

151 return previews 

152 

153 except RateLimitError: 

154 raise 

155 except Exception as e: 

156 safe_msg = self._scrub_error(e) 

157 logger.warning(f"Error getting SerpAPI results: {safe_msg}") 

158 self._raise_if_rate_limit(e) 

159 return []