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

1""" 

2Standard citation handler - the original implementation. 

3""" 

4 

5from datetime import datetime, timezone 

6from typing import Any, Dict, List, Union 

7 

8from .base_citation_handler import BaseCitationHandler 

9 

10 

11class StandardCitationHandler(BaseCitationHandler): 

12 """Standard citation handler with detailed analysis.""" 

13 

14 def analyze_initial( 

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

16 ) -> Dict[str, Any]: 

17 documents = self._create_documents(search_results) 

18 if not documents: 

19 return self._no_sources_response(query) 

20 formatted_sources = self._format_sources(documents) 

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

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

23 ) 

24 

25 output_prefix = self._get_output_instruction_prefix() 

26 

27 prompt = f"""{output_prefix}Analyze the following information concerning the question and include citations using numbers in square brackets [1], [2], etc. When citing, use the source number provided at the start of each source. 

28 

29Question: {query} 

30 

31Sources: 

32{formatted_sources} 

33 

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

35 

36Provide a detailed analysis with citations. Do not create the bibliography, it will be provided automatically. Never make up sources. Never write or create urls. Only write text relevant to the question. Example format: "According to the research [1], ..." 

37""" 

38 

39 response = self._invoke_with_streaming(prompt) 

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

41 

42 def analyze_followup( 

43 self, 

44 question: str, 

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

46 previous_knowledge: str, 

47 nr_of_links: int, 

48 ) -> Dict[str, Any]: 

49 """Process follow-up analysis with citations.""" 

50 documents = self._create_documents( 

51 search_results, nr_of_links=nr_of_links 

52 ) 

53 # With previous knowledge present the LLM can still cite prior 

54 # sources legitimately; with neither, refuse to synthesize. 

55 if not documents and not (previous_knowledge or "").strip(): 

56 return self._no_sources_response(question) 

57 formatted_sources = self._format_sources(documents) 

58 # Add fact-checking step 

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

601. Cross-reference major claims between sources 

612. Identify and flag any contradictions 

623. Verify basic facts (dates, company names, ownership) 

634. Note when sources disagree 

64 

65Previous Knowledge: 

66{previous_knowledge} 

67 

68New Sources: 

69{formatted_sources} 

70 

71 Return any inconsistencies or conflicts found.""" 

72 if self.is_fact_checking_enabled(): 

73 fact_check_response = self._invoke_text(fact_check_prompt) 

74 else: 

75 fact_check_response = "" 

76 

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

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

79 ) 

80 

81 output_prefix = self._get_output_instruction_prefix() 

82 

83 prompt = f"""{output_prefix}Using the previous knowledge and new sources, answer the question. Include citations using numbers in square brackets [1], [2], etc. When citing, use the source number provided at the start of each source. Reflect information from sources critically. 

84 

85Previous Knowledge: 

86{previous_knowledge} 

87 

88Question: {question} 

89 

90New Sources: 

91{formatted_sources} 

92 

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

94 

95Reflect information from sources critically based on: {fact_check_response}. Never invent sources. 

96Provide a detailed answer with citations. Do not create the bibliography, it will be provided automatically. Example format: "According to [1], ..." """ 

97 

98 response = self._invoke_with_streaming(prompt) 

99 

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