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 ..constraints.base_constraint import Constraint 

11from .base_evidence import Evidence, EvidenceType 

12 

13 

14class EvidenceEvaluator: 

15 """Evaluates evidence quality and relevance.""" 

16 

17 def __init__(self, model: BaseChatModel): 

18 """Initialize the evidence evaluator.""" 

19 self.model = model 

20 self.source_reliability = { 

21 "official": 1.0, 

22 "research": 0.95, 

23 "news": 0.8, 

24 "community": 0.6, 

25 "inference": 0.5, 

26 "speculation": 0.3, 

27 } 

28 

29 def extract_evidence( 

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

31 ) -> Evidence: 

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

33 prompt = f""" 

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

35 

36Constraint: {constraint.description} 

37Constraint Type: {constraint.type.value} 

38Required Value: {constraint.value} 

39 

40Search Results: 

41{search_result[:3000]} 

42 

43Provide: 

441. CLAIM: What the evidence claims about the constraint 

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

463. SOURCE: Where this evidence comes from 

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

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

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

50 

51Format: 

52CLAIM: [specific claim] 

53TYPE: [evidence type] 

54SOURCE: [source description] 

55CONFIDENCE: [0.0-1.0] 

56REASONING: [explanation] 

57QUOTE: [relevant text] 

58""" 

59 

60 response = self.model.invoke(prompt) 

61 content = response.content 

62 

63 # Parse response 

64 parsed = self._parse_evidence_response(content) 

65 

66 # Create evidence object 

67 # Safely parse confidence value, handling potential errors 

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

69 try: 

70 confidence = float(confidence_str) 

71 # Ensure confidence is between 0 and 1 

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

73 except ValueError: 

74 logger.warning( 

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

76 ) 

77 confidence = 0.5 

78 

79 evidence = Evidence( 

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

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

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

83 confidence=confidence, 

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

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

86 metadata={ 

87 "candidate": candidate, 

88 "constraint_id": constraint.id, 

89 "constraint_type": constraint.type.value, 

90 }, 

91 ) 

92 

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

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

95 

96 return evidence 

97 

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

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

100 import re 

101 

102 parsed = {} 

103 

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

105 if ":" in line: 

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

107 key = key.strip().lower() 

108 value = value.strip() 

109 

110 if key in [ 

111 "claim", 

112 "type", 

113 "source", 

114 "confidence", 

115 "reasoning", 

116 "quote", 

117 ]: 

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

119 if key == "confidence": 

120 # Extract the first float from the value string 

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

122 if match: 

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

124 else: 

125 parsed[key] = value 

126 else: 

127 parsed[key] = value 

128 

129 return parsed 

130 

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

132 """Parse evidence type from string.""" 

133 type_map = { 

134 "direct_statement": EvidenceType.DIRECT_STATEMENT, 

135 "official_record": EvidenceType.OFFICIAL_RECORD, 

136 "research_finding": EvidenceType.RESEARCH_FINDING, 

137 "news_report": EvidenceType.NEWS_REPORT, 

138 "statistical_data": EvidenceType.STATISTICAL_DATA, 

139 "inference": EvidenceType.INFERENCE, 

140 "correlation": EvidenceType.CORRELATION, 

141 "speculation": EvidenceType.SPECULATION, 

142 } 

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

144 

145 def _assess_match_quality( 

146 self, evidence: Evidence, constraint: Constraint 

147 ) -> float: 

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

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

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

151 return 1.0 

152 if any( 

153 word in evidence.claim.lower() 

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

155 ): 

156 return 0.8 

157 return 0.6 # Partial match at best