| """Concept Bank vote blending shared by the playground and benchmark.""" |
|
|
| from __future__ import annotations |
|
|
| from typing import Any |
|
|
| CONCEPT_BANK_POLICY_VERSION = "agreement-gate-v2" |
|
|
|
|
| def _resolve_outcome(value: Any, outcomes: list[dict[str, str]]) -> int | None: |
| target = str(value).strip().casefold() |
| for index, outcome in enumerate(outcomes): |
| if target in {outcome["id"].casefold(), outcome["description"].casefold()}: |
| return index |
| aliases = {"true": "yes", "false": "no", "1": "yes", "0": "no"} |
| target = aliases.get(target, target) |
| for index, outcome in enumerate(outcomes): |
| if target in {outcome["id"].casefold(), outcome["description"].casefold()}: |
| return index |
| return None |
|
|
|
|
| def apply_concept_votes( |
| raw_probabilities: list[float], |
| signals: list[dict[str, Any]], |
| concepts: list[dict[str, Any]], |
| outcomes: list[dict[str, str]], |
| ) -> tuple[list[float], list[dict[str, Any]]]: |
| """Use linked bank votes only when they agree with the raw decision. |
| |
| Relevance is a relative ranking, not a calibrated measure of evidence. Until |
| bank reliability is measured, a conflicting vote must not override the model. |
| """ |
| votes = [0.0] * len(outcomes) |
| contributions = [] |
| for signal, concept in zip(signals, concepts): |
| relevance = float(signal["probability"]) |
| supports = concept.get("supports", []) or [] |
| contradicts = concept.get("contradicts", []) or [] |
| if isinstance(supports, str): |
| supports = [supports] |
| if isinstance(contradicts, str): |
| contradicts = [contradicts] |
| supported = sorted( |
| {i for value in supports if (i := _resolve_outcome(value, outcomes)) is not None} |
| ) |
| opposed = sorted( |
| {i for value in contradicts if (i := _resolve_outcome(value, outcomes)) is not None} |
| ) |
| concept_votes = [0.0] * len(outcomes) |
| if supported: |
| for i in supported: |
| concept_votes[i] += relevance / len(supported) |
| if opposed: |
| alternatives = [i for i in range(len(outcomes)) if i not in opposed] |
| if alternatives: |
| for i in alternatives: |
| concept_votes[i] += relevance / len(alternatives) |
| for i, amount in enumerate(concept_votes): |
| votes[i] += amount |
| contributions.append( |
| { |
| "description": signal["description"], |
| "probability": relevance, |
| "votes": [ |
| {"description": outcomes[i]["description"], "weight": amount} |
| for i, amount in enumerate(concept_votes) |
| if amount > 0 |
| ], |
| } |
| ) |
| total = sum(votes) |
| if total <= 0: |
| return raw_probabilities, [] |
| bank_probabilities = [value / total for value in votes] |
| raw_winner = max(range(len(raw_probabilities)), key=raw_probabilities.__getitem__) |
| bank_winner = max(range(len(bank_probabilities)), key=bank_probabilities.__getitem__) |
| if bank_winner != raw_winner: |
| return raw_probabilities, contributions |
| adjusted = [0.2 * raw + 0.8 * bank for raw, bank in zip(raw_probabilities, bank_probabilities)] |
| adjusted_total = sum(adjusted) |
| return [value / adjusted_total for value in adjusted], contributions |
|
|
