Coverage for src/local_deep_research/config/search_config.py: 97%

1# local_deep_research/config.py 

2from loguru import logger 

3 

4from ..web_search_engines.search_engine_factory import ( 

5 get_search as factory_get_search, 

6) 

7from .constants import DEFAULT_MAX_FILTERED_RESULTS 

8from .llm_config import get_llm 

9from .thread_settings import get_setting_from_snapshot 

10from ..security.log_sanitizer import redact_and_bound_for_log 

11 

12# Whether to check the quality search results using the LLM. 

13QUALITY_CHECK_DDG_URLS = True 

14 

15 

16# get_setting_from_snapshot is now imported from thread_settings 

17 

18 

19# Expose get_search function 

20def get_search( 

21 search_tool=None, 

22 llm_instance=None, 

23 username=None, 

24 settings_snapshot=None, 

25 programmatic_mode=False, 

26): 

27 """ 

28 Helper function to get search engine 

29 

30 Args: 

31 search_tool: Override the search tool setting (e.g. searxng, wikipedia) 

32 llm_instance: Override the LLM instance 

33 username: Optional username for thread context (e.g., background research threads) 

34 settings_snapshot: Settings snapshot from thread context 

35 programmatic_mode: If True, disables database operations and metrics tracking 

36 """ 

37 

38 # Use specified tool or default from settings 

39 tool = search_tool or get_setting_from_snapshot( 

40 "search.tool", 

41 "searxng", 

42 username=username, 

43 settings_snapshot=settings_snapshot, 

44 ) 

45 

46 # Debug: Check if we got a dict instead of a string 

47 if isinstance(tool, dict): 

48 logger.warning( 

49 f"Got dict for search.tool, extracting value: {tool.get('value')}" 

50 ) 

51 if "value" in tool: 51 ↛ 55line 51 didn't jump to line 55 because the condition on line 51 was always true

52 tool = tool["value"] 

53 

54 # (#6938: the tool name can be a request-supplied string.) 

55 logger.info( 

56 "Creating search engine with tool: {} (type: {})", 

57 redact_and_bound_for_log(str(tool), 200), 

58 type(tool), 

59 ) 

60 

61 # Get LLM instance (use provided or get fresh one). Pass username 

62 # explicitly: this runs BEFORE ``_username`` is injected into the 

63 # snapshot below, so without it a per-user registered LLM (and per-user 

64 # egress classification) would be invisible to this fresh get_llm. 

65 llm = llm_instance or get_llm( 

66 settings_snapshot=settings_snapshot, username=username 

67 ) 

68 

69 # Ensure username is in settings_snapshot for search engines that need it (e.g., library) 

70 if username and settings_snapshot is not None: 

71 # Create a copy to avoid modifying the original 

72 settings_snapshot = {**settings_snapshot, "_username": username} 

73 elif username: 

74 # Create new snapshot with username 

75 settings_snapshot = {"_username": username} 

76 

77 # Get search parameters 

78 params = { 

79 "search_tool": tool, 

80 "llm_instance": llm, 

81 "max_results": get_setting_from_snapshot( 

82 "search.max_results", 

83 10, 

84 username=username, 

85 settings_snapshot=settings_snapshot, 

86 ), 

87 "region": get_setting_from_snapshot( 

88 "search.region", 

89 "wt-wt", 

90 username=username, 

91 settings_snapshot=settings_snapshot, 

92 ), 

93 "time_period": get_setting_from_snapshot( 

94 "search.time_period", 

95 "all", 

96 username=username, 

97 settings_snapshot=settings_snapshot, 

98 ), 

99 "safe_search": get_setting_from_snapshot( 

100 "search.safe_search", 

101 True, 

102 username=username, 

103 settings_snapshot=settings_snapshot, 

104 ), 

105 "search_snippets_only": get_setting_from_snapshot( 

106 "search.snippets_only", 

107 True, 

108 username=username, 

109 settings_snapshot=settings_snapshot, 

110 ), 

111 "search_language": get_setting_from_snapshot( 

112 "search.search_language", 

113 "English", 

114 username=username, 

115 settings_snapshot=settings_snapshot, 

116 ), 

117 "max_filtered_results": get_setting_from_snapshot( 

118 "search.max_filtered_results", 

119 DEFAULT_MAX_FILTERED_RESULTS, 

120 username=username, 

121 settings_snapshot=settings_snapshot, 

122 ), 

123 } 

124 

125 # Log NULL parameters for debugging 

126 logger.info( 

127 f"Search config: tool={tool}, max_results={params['max_results']}, time_period={params['time_period']}" 

128 ) 

129 logger.info(f"Full params dict: {params}") 

130 

131 # Create search engine 

132 search_engine = factory_get_search( 

133 settings_snapshot=settings_snapshot, 

134 programmatic_mode=programmatic_mode, 

135 **params, 

136 ) 

137 

138 # Log the created engine type 

139 if search_engine: 

140 logger.info( 

141 f"Successfully created search engine of type: {type(search_engine).__name__}" 

142 ) 

143 else: 

144 logger.error(f"Failed to create search engine for tool: {tool}") 

145 

146 return search_engine