Coverage for src/local_deep_research/advanced_search_system/evidence/evaluator.py: 100%

1""" 

2Evidence evaluator for assessing evidence quality and relevance. 

3""" 

4 

5from typing import Dict 

6 

7from langchain_core.language_models import BaseChatModel 

8from loguru import logger 

9 

10from ...utilities.json_utils import get_llm_response_text 

11from ..constraints.base_constraint import Constraint 

12from .base_evidence import Evidence, EvidenceType 

13from ...utilities.llm_utils import invoke_llm_sync 

14 

15 

16class EvidenceEvaluator: 

17 """Evaluates evidence quality and relevance.""" 

18 

19 def __init__(self, model: BaseChatModel): 

20 """Initialize the evidence evaluator.""" 

21 self.model = model 

22 self.source_reliability = { 

23 "official": 1.0, 

24 "research": 0.95, 

25 "news": 0.8, 

26 "community": 0.6, 

27 "inference": 0.5, 

28 "speculation": 0.3, 

29 } 

30 

31 def extract_evidence( 

32 self, search_result: str, candidate: str, constraint: Constraint 

33 ) -> Evidence: 

34 """Extract evidence from search results for a specific constraint.""" 

35 prompt = f""" 

36Extract evidence regarding whether "{candidate}" satisfies this constraint: 

37 

38Constraint: {constraint.description} 

39Constraint Type: {constraint.type.value} 

40Required Value: {constraint.value} 

41 

42Search Results: 

43{search_result[:3000]} 

44 

45Provide: 

461. CLAIM: What the evidence claims about the constraint 

472. TYPE: direct_statement, official_record, research_finding, news_report, statistical_data, inference, correlation, or speculation 

483. SOURCE: Where this evidence comes from 

494. CONFIDENCE: How confident you are this evidence is accurate (0.0-1.0) 

505. REASONING: Why this evidence supports or refutes the constraint 

516. QUOTE: Relevant quote from the search results (if any) 

52 

53Format: 

54CLAIM: [specific claim] 

55TYPE: [evidence type] 

56SOURCE: [source description] 

57CONFIDENCE: [0.0-1.0] 

58REASONING: [explanation] 

59QUOTE: [relevant text] 

60""" 

61 

62 response = invoke_llm_sync(self.model, prompt) 

63 content = get_llm_response_text(response) 

64 

65 # Parse response 

66 parsed = self._parse_evidence_response(content) 

67 

68 # Create evidence object 

69 # Safely parse confidence value, handling potential errors 

70 confidence_str = parsed.get("confidence", "0.5") 

71 try: 

72 confidence = float(confidence_str) 

73 # Ensure confidence is between 0 and 1 

74 confidence = max(0.0, min(1.0, confidence)) 

75 except ValueError: 

76 logger.warning( 

77 f"Failed to parse confidence value: {confidence_str}" 

78 ) 

79 confidence = 0.5 

80 

81 evidence = Evidence( 

82 claim=parsed.get("claim", "No clear claim"), 

83 type=self._parse_evidence_type(parsed.get("type", "speculation")), 

84 source=parsed.get("source", "Unknown"), 

85 confidence=confidence, 

86 reasoning=parsed.get("reasoning", ""), 

87 raw_text=parsed.get("quote", ""), 

88 metadata={ 

89 "candidate": candidate, 

90 "constraint_id": constraint.id, 

91 "constraint_type": constraint.type.value, 

92 }, 

93 ) 

94 

95 # Adjust confidence based on how well it matches the constraint 

96 evidence.confidence *= self._assess_match_quality(evidence, constraint) 

97 

98 return evidence 

99 

100 def _parse_evidence_response(self, content: str) -> Dict[str, str]: 

101 """Parse the LLM response into evidence components.""" 

102 import re 

103 

104 parsed = {} 

105 

106 for line in content.strip().split("\n"): 

107 if ":" in line: 

108 key, value = line.split(":", 1) 

109 key = key.strip().lower() 

110 value = value.strip() 

111 

112 if key in [ 

113 "claim", 

114 "type", 

115 "source", 

116 "confidence", 

117 "reasoning", 

118 "quote", 

119 ]: 

120 # Special handling for confidence to extract just the float value 

121 if key == "confidence": 

122 # Extract the first float from the value string 

123 match = re.search(r"(\d*\.?\d+)", value) 

124 if match: 

125 parsed[key] = match.group(1) 

126 else: 

127 parsed[key] = value 

128 else: 

129 parsed[key] = value 

130 

131 return parsed 

132 

133 def _parse_evidence_type(self, type_str: str) -> EvidenceType: 

134 """Parse evidence type from string.""" 

135 type_map = { 

136 "direct_statement": EvidenceType.DIRECT_STATEMENT, 

137 "official_record": EvidenceType.OFFICIAL_RECORD, 

138 "research_finding": EvidenceType.RESEARCH_FINDING, 

139 "news_report": EvidenceType.NEWS_REPORT, 

140 "statistical_data": EvidenceType.STATISTICAL_DATA, 

141 "inference": EvidenceType.INFERENCE, 

142 "correlation": EvidenceType.CORRELATION, 

143 "speculation": EvidenceType.SPECULATION, 

144 } 

145 return type_map.get(type_str.lower(), EvidenceType.SPECULATION) 

146 

147 def _assess_match_quality( 

148 self, evidence: Evidence, constraint: Constraint 

149 ) -> float: 

150 """Assess how well the evidence matches the constraint.""" 

151 # This is a simplified version - could be made more sophisticated 

152 if constraint.value.lower() in evidence.claim.lower(): 

153 return 1.0 

154 if any( 

155 word in evidence.claim.lower() 

156 for word in constraint.value.lower().split() 

157 ): 

158 return 0.8 

159 return 0.6 # Partial match at best