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

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 Any, Optional 

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[str, Any]] 

26 should_reject: bool 

27 rejection_reason: Optional[str] 

28 detailed_results: list[dict[str, Any]] 

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[str, Any]]: 

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

93 if self.evidence_gatherer: 

94 return self.evidence_gatherer(candidate, constraint) 

95 logger.warning("No evidence gatherer provided - cannot gather evidence") 

96 return [] 

97 

98 def _log_constraint_result( 

99 self, 

100 candidate: Candidate, 

101 constraint: Constraint, 

102 score: float, 

103 details: dict[str, Any], 

104 ): 

105 """Log constraint evaluation result.""" 

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

107 logger.info( 

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

109 ) 

110 

111 def _calculate_weighted_score( 

112 self, constraint_scores: list[float], weights: list[float] 

113 ) -> float: 

114 """Calculate weighted average score.""" 

115 if not constraint_scores or not weights: 

116 return 0.0 

117 return sum( 

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

119 ) / sum(weights)