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 

10 

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

12QUALITY_CHECK_DDG_URLS = True 

13 

14 

15# get_setting_from_snapshot is now imported from thread_settings 

16 

17 

18# Expose get_search function 

19def get_search( 

20 search_tool=None, 

21 llm_instance=None, 

22 username=None, 

23 settings_snapshot=None, 

24 programmatic_mode=False, 

25): 

26 """ 

27 Helper function to get search engine 

28 

29 Args: 

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

31 llm_instance: Override the LLM instance 

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

33 settings_snapshot: Settings snapshot from thread context 

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

35 """ 

36 

37 # Use specified tool or default from settings 

38 tool = search_tool or get_setting_from_snapshot( 

39 "search.tool", 

40 "searxng", 

41 username=username, 

42 settings_snapshot=settings_snapshot, 

43 ) 

44 

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

46 if isinstance(tool, dict): 

47 logger.warning( 

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

49 ) 

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

51 tool = tool["value"] 

52 

53 logger.info( 

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

55 ) 

56 

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

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

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

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

61 llm = llm_instance or get_llm( 

62 settings_snapshot=settings_snapshot, username=username 

63 ) 

64 

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

66 if username and settings_snapshot is not None: 

67 # Create a copy to avoid modifying the original 

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

69 elif username: 

70 # Create new snapshot with username 

71 settings_snapshot = {"_username": username} 

72 

73 # Get search parameters 

74 params = { 

75 "search_tool": tool, 

76 "llm_instance": llm, 

77 "max_results": get_setting_from_snapshot( 

78 "search.max_results", 

79 10, 

80 username=username, 

81 settings_snapshot=settings_snapshot, 

82 ), 

83 "region": get_setting_from_snapshot( 

84 "search.region", 

85 "wt-wt", 

86 username=username, 

87 settings_snapshot=settings_snapshot, 

88 ), 

89 "time_period": get_setting_from_snapshot( 

90 "search.time_period", 

91 "all", 

92 username=username, 

93 settings_snapshot=settings_snapshot, 

94 ), 

95 "safe_search": get_setting_from_snapshot( 

96 "search.safe_search", 

97 True, 

98 username=username, 

99 settings_snapshot=settings_snapshot, 

100 ), 

101 "search_snippets_only": get_setting_from_snapshot( 

102 "search.snippets_only", 

103 True, 

104 username=username, 

105 settings_snapshot=settings_snapshot, 

106 ), 

107 "search_language": get_setting_from_snapshot( 

108 "search.search_language", 

109 "English", 

110 username=username, 

111 settings_snapshot=settings_snapshot, 

112 ), 

113 "max_filtered_results": get_setting_from_snapshot( 

114 "search.max_filtered_results", 

115 DEFAULT_MAX_FILTERED_RESULTS, 

116 username=username, 

117 settings_snapshot=settings_snapshot, 

118 ), 

119 } 

120 

121 # Log NULL parameters for debugging 

122 logger.info( 

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

124 ) 

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

126 

127 # Create search engine 

128 search_engine = factory_get_search( 

129 settings_snapshot=settings_snapshot, 

130 programmatic_mode=programmatic_mode, 

131 **params, 

132 ) 

133 

134 # Log the created engine type 

135 if search_engine: 

136 logger.info( 

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

138 ) 

139 else: 

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

141 

142 return search_engine