Coverage for src / local_deep_research / advanced_search_system / constraint_checking / threshold_checker.py: 13%
59 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-01-11 00:51 +0000
1"""
2Simple threshold-based constraint checker.
4This implementation uses simple yes/no threshold checking for constraints,
5making it faster but less nuanced than dual confidence checking.
6"""
8from typing import Dict, List, Tuple
10from loguru import logger
12from ..candidates.base_candidate import Candidate
13from ..constraints.base_constraint import Constraint
14from .base_constraint_checker import (
15 BaseConstraintChecker,
16 ConstraintCheckResult,
17)
20class ThresholdChecker(BaseConstraintChecker):
21 """
22 Simple threshold-based constraint checker.
24 This checker:
25 1. Uses simple LLM yes/no responses for constraint satisfaction
26 2. Makes rejection decisions based on simple thresholds
27 3. Faster than dual confidence but less detailed
28 """
30 def __init__(
31 self,
32 *args,
33 satisfaction_threshold: float = 0.7, # Minimum score to consider satisfied
34 required_satisfaction_rate: float = 0.8, # % of constraints that must be satisfied
35 **kwargs,
36 ):
37 """
38 Initialize threshold checker.
40 Args:
41 satisfaction_threshold: Minimum score for constraint satisfaction
42 required_satisfaction_rate: Percentage of constraints that must be satisfied
43 """
44 super().__init__(*args, **kwargs)
45 self.satisfaction_threshold = satisfaction_threshold
46 self.required_satisfaction_rate = required_satisfaction_rate
48 def check_candidate(
49 self, candidate: Candidate, constraints: List[Constraint]
50 ) -> ConstraintCheckResult:
51 """Check candidate using simple threshold analysis."""
52 logger.info(f"Checking candidate: {candidate.name} (threshold)")
54 constraint_scores = {}
55 detailed_results = []
56 satisfied_count = 0
57 total_constraints = len(constraints)
59 for constraint in constraints:
60 # Gather evidence
61 evidence_list = self._gather_evidence_for_constraint(
62 candidate, constraint
63 )
65 if evidence_list:
66 # Simple satisfaction check
67 satisfaction_score = self._check_constraint_satisfaction(
68 candidate, constraint, evidence_list
69 )
71 is_satisfied = satisfaction_score >= self.satisfaction_threshold
72 if is_satisfied:
73 satisfied_count += 1
75 # Store results
76 constraint_scores[constraint.value] = {
77 "total": satisfaction_score,
78 "satisfied": is_satisfied,
79 "weight": constraint.weight,
80 }
82 detailed_results.append(
83 {
84 "constraint": constraint.value,
85 "score": satisfaction_score,
86 "satisfied": is_satisfied,
87 "weight": constraint.weight,
88 "type": constraint.type.value,
89 }
90 )
92 self._log_constraint_result(
93 candidate, constraint, satisfaction_score, {}
94 )
96 else:
97 # No evidence - consider unsatisfied
98 constraint_scores[constraint.value] = {
99 "total": 0.0,
100 "satisfied": False,
101 "weight": constraint.weight,
102 }
104 detailed_results.append(
105 {
106 "constraint": constraint.value,
107 "score": 0.0,
108 "satisfied": False,
109 "weight": constraint.weight,
110 "type": constraint.type.value,
111 }
112 )
114 logger.info(
115 f"? {candidate.name} | {constraint.value}: No evidence found"
116 )
118 # Check rejection based on satisfaction rate
119 satisfaction_rate = (
120 satisfied_count / total_constraints if total_constraints > 0 else 0
121 )
122 should_reject = satisfaction_rate < self.required_satisfaction_rate
124 rejection_reason = None
125 if should_reject:
126 rejection_reason = f"Only {satisfied_count}/{total_constraints} constraints satisfied ({satisfaction_rate:.0%})"
128 # Calculate total score
129 if should_reject:
130 total_score = 0.0
131 else:
132 # Use satisfaction rate as score
133 total_score = satisfaction_rate
135 logger.info(
136 f"Final score for {candidate.name}: {total_score:.2%} ({satisfied_count}/{total_constraints} satisfied)"
137 )
139 return ConstraintCheckResult(
140 candidate=candidate,
141 total_score=total_score,
142 constraint_scores=constraint_scores,
143 should_reject=should_reject,
144 rejection_reason=rejection_reason,
145 detailed_results=detailed_results,
146 )
148 def should_reject_candidate(
149 self,
150 candidate: Candidate,
151 constraint: Constraint,
152 evidence_data: List[Dict],
153 ) -> Tuple[bool, str]:
154 """Simple rejection based on evidence availability and quality."""
155 if not evidence_data:
156 return (
157 True,
158 f"No evidence found for constraint '{constraint.value}'",
159 )
161 satisfaction_score = self._check_constraint_satisfaction(
162 candidate, constraint, evidence_data
163 )
165 if satisfaction_score < self.satisfaction_threshold:
166 return (
167 True,
168 f"Constraint '{constraint.value}' not satisfied (score: {satisfaction_score:.0%})",
169 )
171 return False, ""
173 def _check_constraint_satisfaction(
174 self,
175 candidate: Candidate,
176 constraint: Constraint,
177 evidence_list: List[Dict],
178 ) -> float:
179 """Check if constraint is satisfied using simple LLM prompt."""
180 # Combine evidence texts
181 combined_evidence = "\n".join(
182 [e.get("text", "")[:200] for e in evidence_list[:3]]
183 )
185 prompt = f"""
186Does the candidate "{candidate.name}" satisfy this constraint: "{constraint.value}"?
188Evidence:
189{combined_evidence}
191Consider the evidence and respond with a satisfaction score from 0.0 to 1.0 where:
192- 1.0 = Definitely satisfies the constraint
193- 0.5 = Partially satisfies or unclear
194- 0.0 = Definitely does not satisfy the constraint
196Score:
197"""
199 try:
200 response = self.model.invoke(prompt).content.strip()
202 # Extract score
203 import re
205 match = re.search(r"(\d*\.?\d+)", response)
206 if match:
207 score = float(match.group(1))
208 return max(0.0, min(score, 1.0))
210 except Exception:
211 logger.exception("Error checking constraint satisfaction")
213 return 0.5 # Default to neutral if parsing fails