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) 

58 llm = llm_instance or get_llm(settings_snapshot=settings_snapshot) 

59 

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

61 if username and settings_snapshot is not None: 

62 # Create a copy to avoid modifying the original 

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

64 elif username: 

65 # Create new snapshot with username 

66 settings_snapshot = {"_username": username} 

67 

68 # Get search parameters 

69 params = { 

70 "search_tool": tool, 

71 "llm_instance": llm, 

72 "max_results": get_setting_from_snapshot( 

73 "search.max_results", 

74 10, 

75 username=username, 

76 settings_snapshot=settings_snapshot, 

77 ), 

78 "region": get_setting_from_snapshot( 

79 "search.region", 

80 "wt-wt", 

81 username=username, 

82 settings_snapshot=settings_snapshot, 

83 ), 

84 "time_period": get_setting_from_snapshot( 

85 "search.time_period", 

86 "all", 

87 username=username, 

88 settings_snapshot=settings_snapshot, 

89 ), 

90 "safe_search": get_setting_from_snapshot( 

91 "search.safe_search", 

92 True, 

93 username=username, 

94 settings_snapshot=settings_snapshot, 

95 ), 

96 "search_snippets_only": get_setting_from_snapshot( 

97 "search.snippets_only", 

98 True, 

99 username=username, 

100 settings_snapshot=settings_snapshot, 

101 ), 

102 "search_language": get_setting_from_snapshot( 

103 "search.search_language", 

104 "English", 

105 username=username, 

106 settings_snapshot=settings_snapshot, 

107 ), 

108 "max_filtered_results": get_setting_from_snapshot( 

109 "search.max_filtered_results", 

110 DEFAULT_MAX_FILTERED_RESULTS, 

111 username=username, 

112 settings_snapshot=settings_snapshot, 

113 ), 

114 } 

115 

116 # Log NULL parameters for debugging 

117 logger.info( 

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

119 ) 

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

121 

122 # Create search engine 

123 search_engine = factory_get_search( 

124 settings_snapshot=settings_snapshot, 

125 programmatic_mode=programmatic_mode, 

126 **params, 

127 ) 

128 

129 # Log the created engine type 

130 if search_engine: 

131 logger.info( 

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

133 ) 

134 else: 

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

136 

137 return search_engine