| 26 | |
| 27 | |
| 28 | class Ensemble: |
| 29 | def __init__( |
| 30 | self, |
| 31 | weights: list[float], |
| 32 | risk_tolerance: float = 0.5, |
| 33 | direct_threshold: float = 0.55, |
| 34 | calibrator: "PlattCalibrator | None" = None, |
| 35 | ): |
| 36 | self._weights = weights |
| 37 | self._risk_tolerance = risk_tolerance |
| 38 | self._threshold = direct_threshold + (0.5 - risk_tolerance) * 0.3 |
| 39 | self._calibrator = calibrator |
| 40 | |
| 41 | def decide(self, votes: list[TierVote]) -> EnsembleResult: |
| 42 | if len(votes) != len(self._weights): |
| 43 | raise ValueError(f"Expected {len(self._weights)} votes, got {len(votes)}") |
| 44 | tier_scores = [0.0, 0.0, 0.0, 0.0] |
| 45 | total_weight = 0.0 |
| 46 | confidence_weighted_sum = 0.0 |
| 47 | confidence_nominal_weight = 0.0 |
| 48 | |
| 49 | for vote, weight in zip(votes, self._weights): |
| 50 | if vote.tier_id is None: |
| 51 | continue |
| 52 | # Adaptive weighting: confidence² amplifies high-confidence signals |
| 53 | # A signal at 0.9 confidence gets 0.81 weight vs 0.49 at 0.7 (LENS 2025) |
| 54 | w = vote.confidence * vote.confidence * weight |
| 55 | tier_scores[vote.tier_id] += w |
| 56 | total_weight += w |
| 57 | confidence_weighted_sum += vote.confidence * weight |
| 58 | confidence_nominal_weight += weight |
| 59 | |
| 60 | if total_weight == 0: |
| 61 | return EnsembleResult( |
| 62 | tier_id=None, confidence=0.0, raw_confidence=0.0, |
| 63 | method="abstain", tier_scores=tier_scores, |
| 64 | ) |
| 65 | |
| 66 | normalized = [s / total_weight for s in tier_scores] |
| 67 | best_tier = max(range(4), key=lambda i: normalized[i]) |
| 68 | signal_confidence = ( |
| 69 | confidence_weighted_sum / confidence_nominal_weight |
| 70 | if confidence_nominal_weight > 0 |
| 71 | else 0.0 |
| 72 | ) |
| 73 | raw_confidence = normalized[best_tier] * signal_confidence |
| 74 | |
| 75 | # Apply calibration if available |
| 76 | confidence = raw_confidence |
| 77 | if self._calibrator is not None: |
| 78 | confidence = self._calibrator.calibrate(raw_confidence) |
| 79 | |
| 80 | if confidence >= self._threshold: |
| 81 | # Low-confidence escalation: weaker LOW predictions are unreliable |
| 82 | # on ambiguous prompts. Bump to MID where a more capable model |
| 83 | # reduces hallucination risk. |
| 84 | if best_tier == 0 and confidence < _LOW_ESCALATION_THRESHOLD: |
| 85 | return EnsembleResult( |
no outgoing calls