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 .llm_config import get_llm 

8from .thread_settings import get_setting_from_snapshot 

9 

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

11QUALITY_CHECK_DDG_URLS = True 

12 

13 

14# get_setting_from_snapshot is now imported from thread_settings 

15 

16 

17def get_search_snippets_only_setting(username=None, settings_snapshot=None): 

18 """ 

19 Lazily retrieve the 'search.snippets_only' setting. 

20 

21 Args: 

22 username: Optional username for thread context 

23 settings_snapshot: Optional settings snapshot for thread safety 

24 """ 

25 return get_setting_from_snapshot( 

26 "search.snippets_only", 

27 True, 

28 username=username, 

29 settings_snapshot=settings_snapshot, 

30 ) 

31 

32 

33# Expose get_search function 

34def get_search( 

35 search_tool=None, 

36 llm_instance=None, 

37 username=None, 

38 settings_snapshot=None, 

39 programmatic_mode=False, 

40): 

41 """ 

42 Helper function to get search engine 

43 

44 Args: 

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

46 llm_instance: Override the LLM instance 

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

48 settings_snapshot: Settings snapshot from thread context 

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

50 """ 

51 

52 # Use specified tool or default from settings 

53 tool = search_tool or get_setting_from_snapshot( 

54 "search.tool", 

55 "searxng", 

56 username=username, 

57 settings_snapshot=settings_snapshot, 

58 ) 

59 

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

61 if isinstance(tool, dict): 

62 logger.warning( 

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

64 ) 

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

66 tool = tool["value"] 

67 

68 logger.info( 

69 f"Creating search engine with tool: {tool} (type: {type(tool)})" 

70 ) 

71 

72 # Get LLM instance (use provided or get fresh one) 

73 llm = llm_instance or get_llm(settings_snapshot=settings_snapshot) 

74 

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

76 if username and settings_snapshot is not None: 

77 # Create a copy to avoid modifying the original 

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

79 elif username: 

80 # Create new snapshot with username 

81 settings_snapshot = {"_username": username} 

82 

83 # Get search parameters 

84 params = { 

85 "search_tool": tool, 

86 "llm_instance": llm, 

87 "max_results": get_setting_from_snapshot( 

88 "search.max_results", 

89 10, 

90 username=username, 

91 settings_snapshot=settings_snapshot, 

92 ), 

93 "region": get_setting_from_snapshot( 

94 "search.region", 

95 "wt-wt", 

96 username=username, 

97 settings_snapshot=settings_snapshot, 

98 ), 

99 "time_period": get_setting_from_snapshot( 

100 "search.time_period", 

101 "all", 

102 username=username, 

103 settings_snapshot=settings_snapshot, 

104 ), 

105 "safe_search": get_setting_from_snapshot( 

106 "search.safe_search", 

107 True, 

108 username=username, 

109 settings_snapshot=settings_snapshot, 

110 ), 

111 "search_snippets_only": get_setting_from_snapshot( 

112 "search.snippets_only", 

113 True, 

114 username=username, 

115 settings_snapshot=settings_snapshot, 

116 ), 

117 "search_language": get_setting_from_snapshot( 

118 "search.search_language", 

119 "English", 

120 username=username, 

121 settings_snapshot=settings_snapshot, 

122 ), 

123 "max_filtered_results": get_setting_from_snapshot( 

124 "search.max_filtered_results", 

125 5, 

126 username=username, 

127 settings_snapshot=settings_snapshot, 

128 ), 

129 } 

130 

131 # Log NULL parameters for debugging 

132 logger.info( 

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

134 ) 

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

136 

137 # Create search engine 

138 search_engine = factory_get_search( 

139 settings_snapshot=settings_snapshot, 

140 programmatic_mode=programmatic_mode, 

141 **params, 

142 ) 

143 

144 # Log the created engine type 

145 if search_engine: 

146 logger.info( 

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

148 ) 

149 else: 

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

151 

152 return search_engine