Merge pull request #665 from praekeltfoundation/nlu-intent-labelling

Nluclassifier for feedback labelling

Author: DevChima

ndoh_hub/settings.py

@@ -61,6 +61,7 @@
     "changes",
     "eventstore",
     "aaq",
+    "nluclassifier",
 )
 
 
@@ -110,6 +111,11 @@
             "handlers": ["console"],
             "propagate": False,
         },
+        "nluclassifier": {
+            "handlers": ["console"],
+            "level": "INFO",
+            "propagate": False,
+        },
     },
 }
 

ndoh_hub/urls.py

@@ -94,6 +94,7 @@
     path("api/v3/", include(v3router.urls)),
     path("api/v4/", include(v4router.urls)),
     path("api/v5/", include(v5router.urls)),
+    path("nluclassifier/", include("nluclassifier.urls")),
     path(
         "metrics", internal_only(django_prometheus.ExportToDjangoView), name="metrics"
     ),

nluclassifier/tasks.py

@@ -0,0 +1,119 @@
+import logging
+from urllib.parse import urljoin
+
+import requests
+from celery.exceptions import SoftTimeLimitExceeded
+from django.conf import settings
+from requests.auth import HTTPBasicAuth
+from requests.exceptions import RequestException
+
+from ndoh_hub.celery import app
+
+logger = logging.getLogger(__name__)
+
+CELERY_TASK_OPTIONS = {
+    "autoretry_for": (RequestException, SoftTimeLimitExceeded),
+    "retry_backoff": True,
+    "max_retries": 15,
+    "acks_late": True,
+    "soft_time_limit": 10,
+    "time_limit": 15,
+}
+
+
+@app.task(**CELERY_TASK_OPTIONS)
+def process_feedback_for_labeling(message_id: str, inbound_message: str):
+    """
+    Call the NLU endpoint for feedback classification.
+    LLabel the message in Turn using the detected intent.
+    """
+
+    nlu_label = None
+
+    NLU_URL = settings.INTENT_CLASSIFIER_URL
+    NLU_USER = settings.INTENT_CLASSIFIER_USER
+    NLU_PASS = settings.INTENT_CLASSIFIER_PASS
+    # fmt: off
+    try:
+        nlu_endpoint = urljoin(NLU_URL, "/nlu/feedback/")
+        params = {"question": inbound_message}
+
+        logger.info(f"Calling NLU for message {message_id} at {nlu_endpoint}")
+
+        response = requests.get(
+            nlu_endpoint,
+            params=params,
+            auth=HTTPBasicAuth(NLU_USER, NLU_PASS),
+            timeout=CELERY_TASK_OPTIONS["soft_time_limit"],
+        )
+        response.raise_for_status()
+
+        nlu_label = response.json().get("intent")
+
+        logger.info(f"NLU Intent for {message_id}: {nlu_label}")
+
+    except RequestException as e:
+        logger.warning(
+            f"NLU service failed for message {message_id}. "
+            f"Retrying. Error: {e}"
+            )
+        raise
+    except Exception as e:
+        logger.error(
+            f"NLU non-retriable failure for message {message_id}: "
+            f"{e}"
+            )
+        return
+
+    if nlu_label and nlu_label.lower() in ("compliment", "complaint"):
+
+        TURN_TOKEN = settings.TURN_TOKEN
+        TURN_URL = settings.TURN_URL
+
+        turn_endpoint = urljoin(TURN_URL, f"v1/messages/{message_id}/labels")
+
+        label_payload = {"labels": [nlu_label.lower()]}
+
+        headers = {
+            "Authorization": f"Bearer {TURN_TOKEN}",
+            "Accept": "application/vnd.v1+json",
+            "Content-Type": "application/json",
+        }
+
+        try:
+            logger.info(
+                f"Labeling message {message_id} in Turn at {turn_endpoint} "
+                f"with label: {nlu_label}"
+                )
+
+            turn_response = requests.post(
+                turn_endpoint,
+                json=label_payload,
+                headers=headers,
+                timeout=CELERY_TASK_OPTIONS["soft_time_limit"],
+            )
+
+            turn_response.raise_for_status()
+
+            logger.info(
+                f"Successfully labeled message {message_id} "
+                f"as: {nlu_label}"
+                )
+
+        except RequestException as e:
+
+            logger.warning(
+                f"Turn API failed to label message {message_id}. "
+                f"Retrying. Error: {e}"
+                )
+            raise
+        except Exception as e:
+            logger.error(
+                f"Turn API unrecoverable error for message {message_id}: {e}"
+            )
+            return
+    else:
+        logger.info(
+            f"NLU result was '{nlu_label}' (not 'Compliment' or 'Complaint'). "
+            f"No Turn label applied for message {message_id}."
+            )

nluclassifier/tests/test_tasks.py

@@ -0,0 +1,202 @@
+import logging
+from unittest import mock
+from urllib.parse import urljoin
+
+import requests.exceptions
+from django.conf import settings
+from django.test import TestCase
+
+from nluclassifier.tasks import process_feedback_for_labeling
+
+logger = logging.getLogger(__name__)
+
+
+# fmt: off
+class NLUClassifierTaskTests(TestCase):
+    """
+    Tests for the Celery task that handles NLU classification
+    and Turn message labeling. The tests verify API calls and
+    ensure RequestExceptions are re-raised for Celery's autoretry
+    """
+
+    def setUp(self):
+        self.settings = settings
+
+        self.message_id = "12345"
+        self.inbound_message = "I am not happy with your service."
+
+        self.expected_intent = "COMPLAINT"
+
+    def test_successful_classification_and_labeling(self):
+        """
+        Tests the end-to-end flow where NLU classifies the message and
+        Turn successfully applies the label
+        """
+        nlu_success_mock = mock.Mock(status_code=200)
+        nlu_success_mock.json.return_value = {
+            "intent": self.expected_intent,
+            "model_version": "2025-09-29-v1",
+            "parent_label": "FEEDBACK",
+            "probability": 0.3805,
+            "review_status": "NEEDS_REVIEW",
+        }
+        nlu_success_mock.raise_for_status.return_value = None
+
+        turn_success_mock = mock.Mock(status_code=202)
+        turn_success_mock.raise_for_status.return_value = None
+
+        with (
+            mock.patch("requests.get") as mock_get,
+            mock.patch("requests.post") as mock_post,
+        ):
+
+            mock_get.return_value = nlu_success_mock
+            mock_post.return_value = turn_success_mock
+
+            result = process_feedback_for_labeling(
+                self.message_id, self.inbound_message
+            )
+
+            expected_nlu_endpoint = urljoin(
+                self.settings.INTENT_CLASSIFIER_URL, "/nlu/feedback/"
+            )
+
+            self.assertEqual(mock_get.call_count, 1)
+            nlu_call_args = mock_get.call_args
+            self.assertEqual(nlu_call_args[0][0], expected_nlu_endpoint)
+            self.assertEqual(
+                nlu_call_args[1]["params"]["question"], self.inbound_message
+            )
+
+            self.assertEqual(mock_post.call_count, 1)
+            turn_call_args = mock_post.call_args
+            self.assertIn(
+                f"messages/{self.message_id}/labels",
+                turn_call_args[0][0]
+                )
+            self.assertEqual(
+                turn_call_args[1]["headers"]["Authorization"],
+                f"Bearer {self.settings.TURN_TOKEN}",
+            )
+
+            turn_payload = turn_call_args[1]["json"]
+            self.assertEqual(
+                turn_payload["labels"],
+                [self.expected_intent.lower()]
+                )
+
+            self.assertTrue(result is None)
+
+    def test_nlu_api_failure_raises_exception_for_celery_retry(self):
+        """
+        Tests that if the NLU API returns a failure,
+        a RequestException is raised
+        """
+        nlu_fail_mock = mock.Mock(status_code=404)
+        error = requests.exceptions.HTTPError("404 Not Found")
+        nlu_fail_mock.raise_for_status.side_effect = error
+
+        with (
+            mock.patch("requests.get") as mock_get,
+            mock.patch("requests.post") as mock_post,
+            self.assertLogs(
+                "nluclassifier.tasks", level="WARNING") as log_context,
+        ):
+
+            mock_get.return_value = nlu_fail_mock
+
+            with self.assertRaises(requests.exceptions.HTTPError):
+                process_feedback_for_labeling(
+                    self.message_id, self.inbound_message)
+
+            self.assertEqual(mock_get.call_count, 1)
+
+            self.assertEqual(mock_post.call_count, 0)
+
+            self.assertTrue(
+                any(
+                    "NLU service failed for message" in output
+                    for output in log_context.output
+                )
+            )
+
+    def test_turn_api_failure_raises_exception_for_retry(self):
+        """
+        Tests that if NLU succeeds but the Turn API fails,
+        a RequestException is rraised to trigger a retry.
+        """
+        nlu_success_mock = mock.Mock(status_code=200)
+        nlu_success_mock.json.return_value = {
+            "intent": self.expected_intent,
+            "model_version": "2025-09-29-v1",
+            "parent_label": "FEEDBACK",
+            "probability": 0.3805,
+            "review_status": "NEEDS_REVIEW",
+        }
+        nlu_success_mock.raise_for_status.return_value = None
+
+        turn_fail_mock = mock.Mock(status_code=500)
+        error = requests.exceptions.HTTPError("500 Server Error")
+        turn_fail_mock.raise_for_status.side_effect = error
+
+        with (
+            mock.patch("requests.get") as mock_get,
+            mock.patch("requests.post") as mock_post,
+            self.assertLogs(
+                "nluclassifier.tasks", level="WARNING") as log_context,
+        ):
+            mock_get.return_value = nlu_success_mock
+            mock_post.return_value = turn_fail_mock
+
+            with self.assertRaises(requests.exceptions.HTTPError):
+                process_feedback_for_labeling(
+                    self.message_id, self.inbound_message)
+
+            self.assertEqual(mock_get.call_count, 1)
+
+            self.assertEqual(mock_post.call_count, 1)
+
+            self.assertTrue(
+                any("Turn API failed to label message" in output
+                    for output in log_context.output)
+            )
+
+    def test_no_label_applied_on_unhandled_intent(self):
+        """
+        Tests that if NLU returns an intent that is not
+        'compliment' or 'complaint'.
+        No Turn API call is made.
+        """
+        nlu_success_mock = mock.Mock(status_code=200)
+
+        nlu_success_mock.json.return_value = {
+            "intent": "None",
+            "model_version": "2025-09-29-v1",
+            "parent_label": "SENSITIVE_EXIT",
+            "probability": 0.3357,
+            "review_status": "NEEDS_REVIEW",
+        }
+        nlu_success_mock.raise_for_status.return_value = None
+
+        with (
+            mock.patch("requests.get") as mock_get,
+            mock.patch("requests.post") as mock_post,
+            self.assertLogs(
+                "nluclassifier.tasks", level="INFO") as log_context,
+        ):
+
+            mock_get.return_value = nlu_success_mock
+
+            result = process_feedback_for_labeling(
+                self.message_id, self.inbound_message
+            )
+
+            self.assertEqual(mock_get.call_count, 1)
+
+            self.assertEqual(mock_post.call_count, 0)
+
+            self.assertTrue(
+                any("No Turn label applied" in output
+                    for output in log_context.output)
+            )
+            self.assertTrue(result is None)