ai-device/intelligent_cabin/archive/tests/test_router_decisions.py

from __future__ import annotations

import unittest

from app.schemas.debug import IntentCandidate, MatcherStageDebug
from app.schemas.intent import IntentDefinition
from app.services.intent_registry import IntentRegistry
from app.services.router import IntentMatchResult, MultiStageIntentMatcher


class _FakeMatcher:
    def __init__(self, stage_debug: MatcherStageDebug) -> None:
        self._stage_debug = stage_debug

    def match(self, text: str) -> IntentMatchResult:
        _ = text
        return IntentMatchResult(intent=None, stage_debug=self._stage_debug)


def _intent(intent_id: str) -> IntentDefinition:
    return IntentDefinition(
        intent_id=intent_id,
        plugin_id=f"mock.{intent_id}",
        domain="test",
        keywords=[],
        examples=[],
    )


class RouterDecisionTests(unittest.TestCase):
    def setUp(self) -> None:
        self.registry = IntentRegistry([_intent("alpha"), _intent("beta"), _intent("gamma")])

    def test_execute_when_bert_classifier_is_clear(self) -> None:
        matcher = MultiStageIntentMatcher(
            registry=self.registry,
            matchers=[
                _FakeMatcher(
                    MatcherStageDebug(
                        stage="classifier",
                        accepted=True,
                        selected_intent="alpha",
                        score=0.92,
                        reason="classifier selected best candidate",
                        backend="joint-bert-local",
                        candidates=[
                            IntentCandidate(intent_id="alpha", score=0.92, reason="classifier", model_name="joint-bert-local"),
                            IntentCandidate(intent_id="beta", score=0.21, reason="classifier", model_name="joint-bert-local"),
                        ],
                    )
                ),
            ],
        )

        result = matcher.match("alpha")

        self.assertEqual(result.debug.decision, "execute")
        self.assertEqual(result.intent.intent_id if result.intent else None, "alpha")

    def test_clarify_when_bert_top_candidates_are_too_close(self) -> None:
        matcher = MultiStageIntentMatcher(
            registry=self.registry,
            matchers=[
                _FakeMatcher(
                    MatcherStageDebug(
                        stage="classifier",
                        accepted=True,
                        selected_intent="alpha",
                        score=0.22,
                        reason="classifier selected best candidate",
                        backend="bert-local",
                        metadata={"threshold": 0.2},
                        candidates=[
                            IntentCandidate(intent_id="alpha", score=0.31, reason="classifier", model_name="bert-local"),
                            IntentCandidate(intent_id="beta", score=0.28, reason="classifier", model_name="bert-local"),
                        ],
                    )
                ),
            ],
            route_to_cloud_threshold=0.2,
        )

        result = matcher.match("ambiguous request")

        self.assertEqual(result.debug.decision, "clarify")
        self.assertIsNone(result.intent)
        self.assertEqual(result.debug.confidence_grade, "medium")

    def test_route_to_cloud_when_bert_signal_is_weak_but_known(self) -> None:
        matcher = MultiStageIntentMatcher(
            registry=self.registry,
            matchers=[
                _FakeMatcher(
                    MatcherStageDebug(
                        stage="classifier",
                        accepted=False,
                        selected_intent="alpha",
                        score=0.29,
                        reason="classifier below execute threshold",
                        backend="joint-bert-local",
                        candidates=[
                            IntentCandidate(intent_id="alpha", score=0.29, reason="classifier", model_name="joint-bert-local"),
                            IntentCandidate(intent_id="beta", score=0.14, reason="classifier", model_name="joint-bert-local"),
                        ],
                    )
                ),
            ],
        )

        result = matcher.match("weak symbolic request")

        self.assertEqual(result.debug.decision, "route_to_cloud")
        self.assertIsNone(result.intent)

    def test_reject_when_no_branch_has_usable_signal(self) -> None:
        matcher = MultiStageIntentMatcher(
            registry=self.registry,
            matchers=[
                _FakeMatcher(
                    MatcherStageDebug(
                        stage="classifier",
                        accepted=False,
                        score=0.12,
                        reason="classifier below threshold",
                        backend="bert-local",
                        metadata={"threshold": 0.2},
                        candidates=[],
                    )
                ),
            ],
        )

        result = matcher.match("unknown request")

        self.assertEqual(result.debug.decision, "reject")
        self.assertTrue(result.debug.unknown_detected)
        self.assertIsNone(result.intent)

    def test_route_to_cloud_for_low_confidence_classifier_only_bert_signal(self) -> None:
        matcher = MultiStageIntentMatcher(
            registry=self.registry,
            matchers=[
                _FakeMatcher(
                    MatcherStageDebug(
                        stage="classifier",
                        accepted=True,
                        selected_intent="alpha",
                        score=0.31,
                        reason="classifier selected best candidate",
                        backend="bert-local",
                        metadata={"threshold": 0.0, "top_margin": 0.04},
                        candidates=[
                            IntentCandidate(intent_id="alpha", score=0.31, reason="classifier", model_name="bert-local"),
                            IntentCandidate(intent_id="beta", score=0.27, reason="classifier", model_name="bert-local"),
                        ],
                    )
                ),
            ],
        )

        result = matcher.match("bert only weak request")

        self.assertEqual(result.debug.decision, "route_to_cloud")
        self.assertIsNone(result.intent)

    def test_execute_for_high_confidence_classifier_only_bert_signal(self) -> None:
        matcher = MultiStageIntentMatcher(
            registry=self.registry,
            matchers=[
                _FakeMatcher(
                    MatcherStageDebug(
                        stage="classifier",
                        accepted=True,
                        selected_intent="alpha",
                        score=0.92,
                        reason="classifier selected best candidate",
                        backend="bert-local",
                        metadata={"threshold": 0.0, "top_margin": 0.63},
                        candidates=[
                            IntentCandidate(intent_id="alpha", score=0.92, reason="classifier", model_name="bert-local"),
                            IntentCandidate(intent_id="beta", score=0.29, reason="classifier", model_name="bert-local"),
                        ],
                    )
                ),
            ],
        )

        result = matcher.match("bert only strong request")

        self.assertEqual(result.debug.decision, "execute")
        self.assertEqual(result.intent.intent_id if result.intent else None, "alpha")


if __name__ == "__main__":
    unittest.main()