Coverage for src / local_deep_research / advanced_search_system / constraint_checking / base_constraint_checker.py: 59%

1""" 

2Base constraint checker for inheritance-based constraint checking system. 

3 

4This module provides the base interface and common functionality for 

5constraint checking implementations. 

6""" 

7 

8from abc import ABC, abstractmethod 

9from dataclasses import dataclass 

10from typing import Dict, List, Optional, Tuple 

11 

12from langchain_core.language_models import BaseChatModel 

13from loguru import logger 

14 

15from ..candidates.base_candidate import Candidate 

16from ..constraints.base_constraint import Constraint 

17 

18 

19@dataclass 

20class ConstraintCheckResult: 

21 """Result of checking a candidate against all constraints.""" 

22 

23 candidate: Candidate 

24 total_score: float 

25 constraint_scores: Dict[str, Dict] 

26 should_reject: bool 

27 rejection_reason: Optional[str] 

28 detailed_results: List[Dict] 

29 

30 

31class BaseConstraintChecker(ABC): 

32 """ 

33 Base class for constraint checking implementations. 

34 

35 This provides the common interface and shared functionality that 

36 all constraint checkers should implement. 

37 """ 

38 

39 def __init__( 

40 self, 

41 model: BaseChatModel, 

42 evidence_gatherer=None, # Will be passed in from strategy 

43 **kwargs, 

44 ): 

45 """ 

46 Initialize the base constraint checker. 

47 

48 Args: 

49 model: Language model for evidence analysis 

50 evidence_gatherer: Function to gather evidence (from strategy) 

51 **kwargs: Additional parameters for specific implementations 

52 """ 

53 self.model = model 

54 self.evidence_gatherer = evidence_gatherer 

55 

56 @abstractmethod 

57 def check_candidate( 

58 self, candidate: Candidate, constraints: List[Constraint] 

59 ) -> ConstraintCheckResult: 

60 """ 

61 Check a candidate against all constraints. 

62 

63 Args: 

64 candidate: The candidate to check 

65 constraints: List of constraints to check against 

66 

67 Returns: 

68 ConstraintCheckResult: Complete evaluation result 

69 """ 

70 pass 

71 

72 @abstractmethod 

73 def should_reject_candidate( 

74 self, candidate: Candidate, constraint: Constraint, evidence_data: any 

75 ) -> Tuple[bool, str]: 

76 """ 

77 Determine if a candidate should be rejected for a specific constraint. 

78 

79 Args: 

80 candidate: The candidate being evaluated 

81 constraint: The constraint being checked 

82 evidence_data: Evidence data (format depends on implementation) 

83 

84 Returns: 

85 Tuple[bool, str]: (should_reject, reason) 

86 """ 

87 pass 

88 

89 def _gather_evidence_for_constraint( 

90 self, candidate: Candidate, constraint: Constraint 

91 ) -> List[Dict]: 

92 """Gather evidence for a constraint using the provided evidence gatherer.""" 

93 if self.evidence_gatherer: 

94 return self.evidence_gatherer(candidate, constraint) 

95 else: 

96 logger.warning( 

97 "No evidence gatherer provided - cannot gather evidence" 

98 ) 

99 return [] 

100 

101 def _log_constraint_result( 

102 self, 

103 candidate: Candidate, 

104 constraint: Constraint, 

105 score: float, 

106 details: Dict, 

107 ): 

108 """Log constraint evaluation result.""" 

109 symbol = "✓" if score >= 0.8 else "○" if score >= 0.5 else "✗" 

110 logger.info( 

111 f"{symbol} {candidate.name} | {constraint.value}: {int(score * 100)}%" 

112 ) 

113 

114 def _calculate_weighted_score( 

115 self, constraint_scores: List[float], weights: List[float] 

116 ) -> float: 

117 """Calculate weighted average score.""" 

118 if not constraint_scores or not weights: 

119 return 0.0 

120 return sum( 

121 s * w for s, w in zip(constraint_scores, weights, strict=False) 

122 ) / sum(weights)