Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 19 additions & 0 deletions python/src/agent_squad/neutrosophic/__init__.py
Original file line number Diff line number Diff line change
@@ -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",
]
46 changes: 46 additions & 0 deletions python/src/agent_squad/neutrosophic/consensus.py
Original file line number Diff line number Diff line change
@@ -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),
)
38 changes: 38 additions & 0 deletions python/src/agent_squad/neutrosophic/decision.py
Original file line number Diff line number Diff line change
@@ -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
24 changes: 24 additions & 0 deletions python/src/agent_squad/neutrosophic/operators.py
Original file line number Diff line number Diff line change
@@ -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)
94 changes: 94 additions & 0 deletions python/src/agent_squad/neutrosophic/scorer.py
Original file line number Diff line number Diff line change
@@ -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)
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Reject out-of-range classifier confidence

When a classifier returns a malformed confidence outside the documented 0–1 range, this clamp silently turns it into an extreme certainty instead of surfacing the bad value; for example, score_classifier_confidence(2, selected=True) becomes Triplet(T=1, I=0, F=0). Since the downstream decision helper treats high truth/low indeterminacy as confident, an invalid model output can be promoted to the strongest possible route rather than being handled as bad/uncertain input.

Useful? React with 👍 / 👎.

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"(?<!\w){re.escape(pattern)}(?!\w)", text)) for pattern in patterns)


def _clamp(value: float) -> float:
return max(0.0, min(1.0, value))
30 changes: 30 additions & 0 deletions python/src/agent_squad/neutrosophic/triplet.py
Original file line number Diff line number Diff line change
@@ -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
1 change: 1 addition & 0 deletions python/src/tests/neutrosophic/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@

36 changes: 36 additions & 0 deletions python/src/tests/neutrosophic/test_consensus.py
Original file line number Diff line number Diff line change
@@ -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([])
39 changes: 39 additions & 0 deletions python/src/tests/neutrosophic/test_decision.py
Original file line number Diff line number Diff line change
@@ -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)
19 changes: 19 additions & 0 deletions python/src/tests/neutrosophic/test_operators.py
Original file line number Diff line number Diff line change
@@ -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)
Loading