diff --git a/python/src/agent_squad/neutrosophic/__init__.py b/python/src/agent_squad/neutrosophic/__init__.py new file mode 100644 index 00000000..1cb524a1 --- /dev/null +++ b/python/src/agent_squad/neutrosophic/__init__.py @@ -0,0 +1,19 @@ +from .consensus import neutrosophic_consensus, neutrosophic_evidence_consensus +from .decision import DecisionAction, DecisionThresholds, decide +from .operators import n_conorm, n_norm, negate +from .scorer import score_classifier_confidence, score_text_response +from .triplet import Triplet + +__all__ = [ + "DecisionAction", + "DecisionThresholds", + "Triplet", + "decide", + "n_conorm", + "n_norm", + "negate", + "neutrosophic_consensus", + "neutrosophic_evidence_consensus", + "score_classifier_confidence", + "score_text_response", +] diff --git a/python/src/agent_squad/neutrosophic/consensus.py b/python/src/agent_squad/neutrosophic/consensus.py new file mode 100644 index 00000000..3519a5a8 --- /dev/null +++ b/python/src/agent_squad/neutrosophic/consensus.py @@ -0,0 +1,46 @@ +from collections.abc import Iterable + +from agent_squad.neutrosophic.operators import n_conorm +from agent_squad.neutrosophic.triplet import Triplet + + +def neutrosophic_consensus(responses: Iterable[Triplet]) -> Triplet: + """Fuse agent response scores using repeated N-conorm aggregation.""" + iterator = iter(responses) + try: + consensus = next(iterator) + except StopIteration as exc: + raise ValueError("responses must contain at least one triplet") from exc + + for response in iterator: + consensus = n_conorm(consensus, response) + + return consensus + + +def neutrosophic_evidence_consensus(responses: Iterable[Triplet]) -> Triplet: + """Fuse agent evidence while preserving contradiction as indeterminacy. + + N-conorm is still the formal OR operator. For multi-agent response fusion, + however, strong counter-evidence should not disappear just because another + response has low falsity. This consensus keeps the strongest truth, keeps + the strongest falsity, and raises indeterminacy when the evidence disagrees. + """ + triplets = list(responses) + if not triplets: + raise ValueError("responses must contain at least one triplet") + + formal_union = neutrosophic_consensus(triplets) + truth_values = [response.T for response in triplets] + falsity_values = [response.F for response in triplets] + + truth_spread = max(truth_values) - min(truth_values) + falsity_spread = max(falsity_values) - min(falsity_values) + contradiction = min(max(truth_values), max(falsity_values)) + conflict = max(truth_spread, falsity_spread, contradiction) + + return Triplet( + T=formal_union.T, + I=max(formal_union.I, conflict), + F=max(falsity_values), + ) diff --git a/python/src/agent_squad/neutrosophic/decision.py b/python/src/agent_squad/neutrosophic/decision.py new file mode 100644 index 00000000..094a6297 --- /dev/null +++ b/python/src/agent_squad/neutrosophic/decision.py @@ -0,0 +1,38 @@ +from dataclasses import dataclass +from enum import Enum + +from agent_squad.neutrosophic.triplet import Triplet + + +class DecisionAction(str, Enum): + CLARIFY = "CLARIFY" + CONFIDENCE = "CONFIDENCE" + CAVEAT = "CAVEAT" + REJECT = "REJECT" + + +@dataclass(frozen=True) +class DecisionThresholds: + indeterminacy: float = 0.6 + falsity: float = 0.5 + truth: float = 0.7 + + def __post_init__(self) -> None: + Triplet( + T=self.truth, + I=self.indeterminacy, + F=self.falsity, + ) + + +def decide(value: Triplet, thresholds: DecisionThresholds | None = None) -> DecisionAction: + """Choose a response action from a neutrosophic triplet.""" + thresholds = thresholds or DecisionThresholds() + + if value.I > thresholds.indeterminacy: + return DecisionAction.CLARIFY + if value.F > thresholds.falsity: + return DecisionAction.REJECT + if value.T > thresholds.truth: + return DecisionAction.CONFIDENCE + return DecisionAction.CAVEAT diff --git a/python/src/agent_squad/neutrosophic/operators.py b/python/src/agent_squad/neutrosophic/operators.py new file mode 100644 index 00000000..274a6ede --- /dev/null +++ b/python/src/agent_squad/neutrosophic/operators.py @@ -0,0 +1,24 @@ +from agent_squad.neutrosophic.triplet import Triplet + + +def n_norm(left: Triplet, right: Triplet) -> Triplet: + """Smarandache N-norm: AND over two neutrosophic triplets.""" + return Triplet( + T=min(left.T, right.T), + I=max(left.I, right.I), + F=max(left.F, right.F), + ) + + +def n_conorm(left: Triplet, right: Triplet) -> Triplet: + """Smarandache N-conorm: OR over two neutrosophic triplets.""" + return Triplet( + T=max(left.T, right.T), + I=min(left.I, right.I), + F=min(left.F, right.F), + ) + + +def negate(value: Triplet) -> Triplet: + """Neutrosophic negation: swap truth and falsity, keep indeterminacy.""" + return Triplet(T=value.F, I=value.I, F=value.T) diff --git a/python/src/agent_squad/neutrosophic/scorer.py b/python/src/agent_squad/neutrosophic/scorer.py new file mode 100644 index 00000000..2fed81c7 --- /dev/null +++ b/python/src/agent_squad/neutrosophic/scorer.py @@ -0,0 +1,94 @@ +import math +import re + +from agent_squad.neutrosophic.triplet import Triplet + +HEDGING_PATTERNS = ( + "maybe", + "might", + "possibly", + "probably", + "not sure", + "unclear", + "unknown", + "ambiguous", + "insufficient", + "could be", + "it depends", +) + +ERROR_PATTERNS = ( + "error", + "failed", + "failure", + "cannot", + "can't", + "unable", + "invalid", + "contradiction", + "conflict", + "not possible", + "i do not know", + "i don't know", +) + +REFUSAL_PATTERNS = ( + "i can't help", + "i cannot help", + "i cannot comply", + "i can't comply", + "sorry, but i can't", + "sorry, but i cannot", +) + + +def score_text_response(text: str) -> Triplet: + """Score plain text with deterministic baseline heuristics. + + This scorer is intentionally lightweight. It provides a dependency-free + baseline until framework integrations can inject model-specific scorers. + """ + if not isinstance(text, str): + raise TypeError("text must be a string") + + normalized = " ".join(text.lower().split()) + if not normalized: + return Triplet(T=0.0, I=1.0, F=0.0) + + hedge_count = _count_matches(normalized, HEDGING_PATTERNS) + error_count = _count_matches(normalized, ERROR_PATTERNS) + refusal_count = _count_matches(normalized, REFUSAL_PATTERNS) + word_count = len(re.findall(r"\b\w+\b", normalized)) + + substance = min(word_count / 80, 1.0) + directness = 1.0 if hedge_count == 0 else max(0.0, 1.0 - (hedge_count * 0.2)) + + truth = _clamp((0.35 + (0.45 * substance) + (0.20 * directness)) - (0.20 * error_count)) + indeterminacy = _clamp((0.15 if word_count >= 8 else 0.45) + (0.18 * hedge_count)) + falsity = _clamp((0.22 * error_count) + (0.35 * refusal_count)) + + return Triplet(T=truth, I=indeterminacy, F=falsity) + + +def score_classifier_confidence(confidence: float, selected: bool) -> Triplet: + """Map legacy classifier confidence into a neutrosophic triplet.""" + if isinstance(confidence, bool): + raise TypeError("confidence must be a number") + + confidence_value = float(confidence) + if not math.isfinite(confidence_value): + raise ValueError("confidence must be finite") + + confidence_value = _clamp(confidence_value) + if selected: + return Triplet(T=confidence_value, I=1 - confidence_value, F=0) + + return Triplet(T=0, I=max(1 - confidence_value, 0.7), F=0) + + +def _count_matches(text: str, patterns: tuple[str, ...]) -> int: + return sum(len(re.findall(rf"(? float: + return max(0.0, min(1.0, value)) diff --git a/python/src/agent_squad/neutrosophic/triplet.py b/python/src/agent_squad/neutrosophic/triplet.py new file mode 100644 index 00000000..33d54f74 --- /dev/null +++ b/python/src/agent_squad/neutrosophic/triplet.py @@ -0,0 +1,30 @@ +from dataclasses import dataclass +from math import isfinite +from numbers import Real + + +@dataclass(frozen=True) +class Triplet: + """Neutrosophic truth, indeterminacy, and falsity values.""" + + T: float + I: float + F: float + + def __post_init__(self) -> None: + object.__setattr__(self, "T", self._validate_component("T", self.T)) + object.__setattr__(self, "I", self._validate_component("I", self.I)) + object.__setattr__(self, "F", self._validate_component("F", self.F)) + + @staticmethod + def _validate_component(name: str, value: Real) -> float: + if isinstance(value, bool) or not isinstance(value, Real): + raise TypeError(f"{name} must be a number") + + numeric_value = float(value) + if not isfinite(numeric_value): + raise ValueError(f"{name} must be finite") + if not 0 <= numeric_value <= 1: + raise ValueError(f"{name} must be between 0 and 1") + + return numeric_value diff --git a/python/src/tests/neutrosophic/__init__.py b/python/src/tests/neutrosophic/__init__.py new file mode 100644 index 00000000..8b137891 --- /dev/null +++ b/python/src/tests/neutrosophic/__init__.py @@ -0,0 +1 @@ + diff --git a/python/src/tests/neutrosophic/test_consensus.py b/python/src/tests/neutrosophic/test_consensus.py new file mode 100644 index 00000000..010232fc --- /dev/null +++ b/python/src/tests/neutrosophic/test_consensus.py @@ -0,0 +1,36 @@ +import pytest + +from agent_squad.neutrosophic import Triplet, neutrosophic_consensus, neutrosophic_evidence_consensus + + +def test_neutrosophic_consensus_fuses_with_n_conorm(): + responses = [ + Triplet(T=0.3, I=0.8, F=0.6), + Triplet(T=0.9, I=0.5, F=0.4), + Triplet(T=0.7, I=0.2, F=0.1), + ] + + assert neutrosophic_consensus(responses) == Triplet(T=0.9, I=0.2, F=0.1) + + +def test_neutrosophic_consensus_rejects_empty_input(): + with pytest.raises(ValueError, match="at least one triplet"): + neutrosophic_consensus([]) + + +def test_neutrosophic_evidence_consensus_preserves_conflicting_falsity(): + responses = [ + Triplet(T=0.9, I=0.1, F=0.0), + Triplet(T=0.2, I=0.2, F=0.8), + ] + + consensus = neutrosophic_evidence_consensus(responses) + + assert consensus.T == 0.9 + assert consensus.I == 0.8 + assert consensus.F == 0.8 + + +def test_neutrosophic_evidence_consensus_rejects_empty_input(): + with pytest.raises(ValueError, match="at least one triplet"): + neutrosophic_evidence_consensus([]) diff --git a/python/src/tests/neutrosophic/test_decision.py b/python/src/tests/neutrosophic/test_decision.py new file mode 100644 index 00000000..83ca714a --- /dev/null +++ b/python/src/tests/neutrosophic/test_decision.py @@ -0,0 +1,39 @@ +import pytest + +from agent_squad.neutrosophic import DecisionAction, DecisionThresholds, Triplet, decide + + +def test_decide_clarify_has_priority_over_reject(): + value = Triplet(T=0.9, I=0.7, F=0.9) + + assert decide(value) == DecisionAction.CLARIFY + + +def test_decide_reject_when_falsity_is_high(): + value = Triplet(T=0.9, I=0.2, F=0.6) + + assert decide(value) == DecisionAction.REJECT + + +def test_decide_confidence_when_truth_is_high(): + value = Triplet(T=0.8, I=0.2, F=0.1) + + assert decide(value) == DecisionAction.CONFIDENCE + + +def test_decide_caveat_for_low_confidence_middle_state(): + value = Triplet(T=0.5, I=0.2, F=0.1) + + assert decide(value) == DecisionAction.CAVEAT + + +def test_decide_accepts_custom_thresholds(): + value = Triplet(T=0.65, I=0.2, F=0.1) + thresholds = DecisionThresholds(truth=0.6) + + assert decide(value, thresholds) == DecisionAction.CONFIDENCE + + +def test_decision_thresholds_validate_component_bounds(): + with pytest.raises(ValueError, match="between 0 and 1"): + DecisionThresholds(indeterminacy=1.2) diff --git a/python/src/tests/neutrosophic/test_operators.py b/python/src/tests/neutrosophic/test_operators.py new file mode 100644 index 00000000..217b58ab --- /dev/null +++ b/python/src/tests/neutrosophic/test_operators.py @@ -0,0 +1,19 @@ +from agent_squad.neutrosophic import Triplet, n_conorm, n_norm, negate + + +def test_n_norm_uses_min_truth_max_indeterminacy_max_falsity(): + left = Triplet(T=0.8, I=0.2, F=0.1) + right = Triplet(T=0.4, I=0.7, F=0.5) + + assert n_norm(left, right) == Triplet(T=0.4, I=0.7, F=0.5) + + +def test_n_conorm_uses_max_truth_min_indeterminacy_min_falsity(): + left = Triplet(T=0.8, I=0.2, F=0.1) + right = Triplet(T=0.4, I=0.7, F=0.5) + + assert n_conorm(left, right) == Triplet(T=0.8, I=0.2, F=0.1) + + +def test_negate_swaps_truth_and_falsity(): + assert negate(Triplet(T=0.8, I=0.2, F=0.1)) == Triplet(T=0.1, I=0.2, F=0.8) diff --git a/python/src/tests/neutrosophic/test_scorer.py b/python/src/tests/neutrosophic/test_scorer.py new file mode 100644 index 00000000..73089b8c --- /dev/null +++ b/python/src/tests/neutrosophic/test_scorer.py @@ -0,0 +1,74 @@ +import pytest + +from agent_squad.neutrosophic import Triplet, score_classifier_confidence, score_text_response + + +def test_score_text_response_rejects_non_string_input(): + with pytest.raises(TypeError, match="text must be a string"): + score_text_response(None) + + +def test_score_text_response_marks_empty_text_as_indeterminate(): + assert score_text_response("") == Triplet(T=0.0, I=1.0, F=0.0) + + +def test_score_text_response_returns_bounded_triplet(): + score = score_text_response("Maybe this could be unclear and failed because of an error.") + + assert 0 <= score.T <= 1 + assert 0 <= score.I <= 1 + assert 0 <= score.F <= 1 + + +def test_score_text_response_increases_indeterminacy_for_hedging(): + direct = score_text_response("The request should be routed to the billing agent for invoice help.") + hedged = score_text_response("Maybe it could be routed to billing, but it depends and is unclear.") + + assert hedged.I > direct.I + + +def test_score_text_response_increases_falsity_for_errors(): + direct = score_text_response("The task completed successfully with a clear answer.") + failed = score_text_response("The task failed with an invalid response and an error.") + + assert failed.F > direct.F + + +def test_score_text_response_does_not_match_substrings_as_patterns(): + direct = score_text_response("The terror radius changed after the update.") + failed = score_text_response("The update returned an error.") + + assert failed.F > direct.F + + +def test_score_text_response_scores_direct_substantive_answer_with_more_truth(): + short = score_text_response("Maybe.") + substantive = score_text_response( + "The billing agent is the correct destination because the user asks about invoices, refunds, " + "and account balance details that match the billing agent description." + ) + + assert substantive.T > short.T + + +def test_score_classifier_confidence_maps_selected_agent_to_truth_and_indeterminacy(): + score = score_classifier_confidence(0.8, selected=True) + + assert score.T == 0.8 + assert score.I == pytest.approx(0.2) + assert score.F == 0.0 + + +def test_score_classifier_confidence_maps_no_agent_to_clarification_indeterminacy(): + assert score_classifier_confidence(0.95, selected=False) == Triplet(T=0.0, I=0.7, F=0.0) + + +def test_score_classifier_confidence_rejects_boolean_confidence(): + with pytest.raises(TypeError, match="confidence must be a number"): + score_classifier_confidence(True, selected=True) + + +@pytest.mark.parametrize("confidence", [float("nan"), float("inf"), float("-inf")]) +def test_score_classifier_confidence_rejects_non_finite_confidence(confidence): + with pytest.raises(ValueError, match="confidence must be finite"): + score_classifier_confidence(confidence, selected=True) diff --git a/python/src/tests/neutrosophic/test_triplet.py b/python/src/tests/neutrosophic/test_triplet.py new file mode 100644 index 00000000..c6fe8cb7 --- /dev/null +++ b/python/src/tests/neutrosophic/test_triplet.py @@ -0,0 +1,26 @@ +import pytest + +from agent_squad.neutrosophic import Triplet + + +def test_triplet_accepts_valid_values(): + triplet = Triplet(T=1, I=0.5, F=0) + + assert triplet.T == 1.0 + assert triplet.I == 0.5 + assert triplet.F == 0.0 + + +@pytest.mark.parametrize("field", ["T", "I", "F"]) +def test_triplet_rejects_out_of_range_values(field): + values = {"T": 0.2, "I": 0.3, "F": 0.4} + values[field] = 1.1 + + with pytest.raises(ValueError, match="between 0 and 1"): + Triplet(**values) + + +@pytest.mark.parametrize("bad_value", [True, "0.5", None]) +def test_triplet_rejects_non_numeric_values(bad_value): + with pytest.raises(TypeError, match="must be a number"): + Triplet(T=bad_value, I=0.2, F=0.3)