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 formatted_sources = self._format_sources(documents) 

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

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

21 ) 

22 

23 output_prefix = self._get_output_instruction_prefix() 

24 

25 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. 

26 

27Question: {query} 

28 

29Sources: 

30{formatted_sources} 

31 

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

33 

34Provide 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], ..." 

35""" 

36 

37 response = self._invoke_with_streaming(prompt) 

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

39 

40 def analyze_followup( 

41 self, 

42 question: str, 

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

44 previous_knowledge: str, 

45 nr_of_links: int, 

46 ) -> Dict[str, Any]: 

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

48 documents = self._create_documents( 

49 search_results, nr_of_links=nr_of_links 

50 ) 

51 formatted_sources = self._format_sources(documents) 

52 # Add fact-checking step 

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

541. Cross-reference major claims between sources 

552. Identify and flag any contradictions 

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

574. Note when sources disagree 

58 

59Previous Knowledge: 

60{previous_knowledge} 

61 

62New Sources: 

63{formatted_sources} 

64 

65 Return any inconsistencies or conflicts found.""" 

66 if self.is_fact_checking_enabled(): 

67 fact_check_response = self._invoke_text(fact_check_prompt) 

68 else: 

69 fact_check_response = "" 

70 

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

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

73 ) 

74 

75 output_prefix = self._get_output_instruction_prefix() 

76 

77 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. 

78 

79Previous Knowledge: 

80{previous_knowledge} 

81 

82Question: {question} 

83 

84New Sources: 

85{formatted_sources} 

86 

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

88 

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

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

91 

92 response = self._invoke_with_streaming(prompt) 

93 

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