Coverage for src/local_deep_research/citation_handlers/forced_answer_citation_handler.py: 100%

1""" 

2Forced answer citation handler - optimized for BrowseComp-style questions. 

3Always provides a specific answer, never returns "cannot determine". 

4""" 

5 

6from datetime import datetime, timezone 

7from typing import Any, Dict, List, Union 

8 

9from loguru import logger 

10 

11from .base_citation_handler import BaseCitationHandler 

12 

13 

14class ForcedAnswerCitationHandler(BaseCitationHandler): 

15 """Citation handler that forces direct answers for benchmark questions.""" 

16 

17 # Keep the base's terse source format: answer extraction slices sources 

18 # to 1500 characters, so identity lines would crowd out evidence. 

19 

20 def analyze_initial( 

21 self, query: str, search_results: Union[str, List[Dict]] 

22 ) -> Dict[str, Any]: 

23 """Initial analysis with forced answer generation.""" 

24 documents = self._create_documents(search_results) 

25 formatted_sources = self._format_sources(documents) 

26 

27 current_timestamp = datetime.now(timezone.utc).strftime( 

28 "%Y-%m-%d %H:%M" 

29 ) 

30 

31 output_prefix = self._get_output_instruction_prefix() 

32 

33 prompt = f"""{output_prefix}Analyze the following information and provide a DIRECT answer to the question. Include citations using numbers in square brackets [1], [2], etc. 

34 

35Question: {query} 

36 

37Sources: 

38{formatted_sources} 

39 

40Current time is {current_timestamp} UTC for verifying temporal references in sources. 

41 

42CRITICAL INSTRUCTIONS: 

431. Start your response with a direct answer to the question 

442. NEVER say "I cannot determine" or "insufficient information" 

453. If unsure between options, choose the MOST LIKELY based on evidence 

464. After the direct answer, provide supporting analysis with citations 

475. Do not create the bibliography, it will be provided automatically. 

48 

49Example response format: 

50"[Direct Answer]. According to [1], this is supported by..." 

51""" 

52 

53 response = self._invoke_with_streaming(prompt) 

54 

55 # If the response still doesn't have a direct answer, extract one 

56 if self._needs_answer_extraction(response, query): 

57 response = self._extract_direct_answer( 

58 query, response, formatted_sources 

59 ) 

60 

61 return {"content": response, "documents": documents} 

62 

63 def analyze_followup( 

64 self, 

65 question: str, 

66 search_results: Union[str, List[Dict]], 

67 previous_knowledge: str, 

68 nr_of_links: int, 

69 ) -> Dict[str, Any]: 

70 """Follow-up analysis with forced answer generation.""" 

71 documents = self._create_documents( 

72 search_results, nr_of_links=nr_of_links 

73 ) 

74 formatted_sources = self._format_sources(documents) 

75 

76 # Fact-checking step (if enabled) 

77 fact_check_response = "" 

78 if self.is_fact_checking_enabled(): 

79 fact_check_prompt = f"""Analyze these sources for factual consistency: 

801. Cross-reference major claims between sources 

812. Identify the most frequently mentioned answer 

823. Note any conflicts but identify the most likely correct answer 

83 

84Previous Knowledge: 

85{previous_knowledge} 

86 

87New Sources: 

88{formatted_sources} 

89 

90Return the most likely answer based on evidence consistency.""" 

91 fact_check_response = self._invoke_text(fact_check_prompt) 

92 

93 current_timestamp = datetime.now(timezone.utc).strftime( 

94 "%Y-%m-%d %H:%M" 

95 ) 

96 

97 output_prefix = self._get_output_instruction_prefix() 

98 

99 prompt = f"""{output_prefix}Using the previous knowledge and new sources, provide a DIRECT answer to the question. Include citations using numbers in square brackets. 

100 

101Previous Knowledge: 

102{previous_knowledge} 

103 

104Question: {question} 

105 

106New Sources: 

107{formatted_sources} 

108 

109Current time is {current_timestamp} UTC for verifying temporal references in sources. 

110 

111Fact Analysis: {fact_check_response} 

112 

113CRITICAL INSTRUCTIONS: 

1141. You MUST start with a direct, specific answer 

1152. NEVER say "I cannot determine" or similar phrases 

1163. If the question asks for a name, provide a specific name 

1174. If the question asks for a place, provide a specific place 

1185. If unsure, choose the answer with the most supporting evidence 

1196. Format: "[Direct Answer]. Supporting evidence from [1], [2]..." 

1207. Do not create the bibliography, it will be provided automatically. 

121 

122Remember: A wrong answer is better than no answer for this task.""" 

123 

124 content = self._invoke_with_streaming(prompt) 

125 

126 # Final check - if still no direct answer, force extraction 

127 if self._needs_answer_extraction(content, question): 

128 content = self._extract_direct_answer( 

129 question, content, formatted_sources 

130 ) 

131 logger.info(f"Forced answer extraction applied: {content[:100]}...") 

132 

133 return {"content": content, "documents": documents} 

134 

135 def _needs_answer_extraction(self, content: str, query: str) -> bool: 

136 """Check if the response needs forced answer extraction.""" 

137 no_answer_indicators = [ 

138 "cannot determine", 

139 "unable to find", 

140 "insufficient", 

141 "unclear", 

142 "not enough", 

143 "cannot provide", 

144 "no specific answer", 

145 "cannot definitively", 

146 ] 

147 

148 content_lower = content.lower() 

149 

150 # Check for no-answer indicators 

151 for indicator in no_answer_indicators: 

152 if indicator in content_lower: 

153 return True 

154 

155 # Check if it's a direct question but no direct answer given 

156 if query.lower().startswith( 

157 ("what", "who", "which", "where", "when", "name") 

158 ): 

159 # Look for a direct answer pattern in first 100 chars 

160 first_part = content[:100].lower() 

161 if not any( 

162 word in first_part for word in ["is", "was", "are", "were", ":"] 

163 ): 

164 return True 

165 

166 return False 

167 

168 def _extract_direct_answer( 

169 self, query: str, content: str, sources: str 

170 ) -> str: 

171 """Force extraction of a direct answer using LLM.""" 

172 extraction_prompt = f"""Based on the content below, extract a SINGLE, DIRECT answer to the question. 

173 

174Question: {query} 

175 

176Content: {content[:1500]} 

177 

178Sources: {sources[:1500]} 

179 

180RULES: 

1811. Respond with ONLY the answer itself (name, place, number, etc.) 

1822. No explanations, just the answer 

1833. If multiple candidates exist, pick the one mentioned most 

1844. If truly no information exists, make an educated guess 

185 

186Answer:""" 

187 

188 try: 

189 answer = self._invoke_text(extraction_prompt) 

190 if not answer: 

191 return content 

192 

193 # Format as a proper response 

194 return f"{answer}. Based on the available sources, this appears to be the most likely answer. {content}" 

195 

196 except Exception: 

197 logger.exception("Error in forced answer extraction") 

198 # Fallback - just prepend a guess 

199 return f"Based on the available evidence, the most likely answer appears to be related to the search results. {content}"