Merge pull request #8929 from RasaHQ/fix-intent-id-formatting

Convert Intent ID Hashes from Integer to String

changelog/8929.improvement.md

@@ -0,0 +1 @@
+Added optional flag to convert intent ID hashes from integer to string in the `KafkaEventBroker`.

data/test_endpoints/event_brokers/kafka_sasl_plaintext_endpoint.yml

@@ -7,3 +7,4 @@ event_broker:
   sasl_username: username
   sasl_password: password
   sasl_mechanism: PLAIN
+  convert_intent_id_to_string: True

rasa/core/brokers/kafka.py

@@ -34,6 +34,7 @@ def __init__(
         ssl_check_hostname: bool = False,
         security_protocol: Text = "SASL_PLAINTEXT",
         loglevel: Union[int, Text] = logging.ERROR,
+        convert_intent_id_to_string: bool = False,
         **kwargs: Any,
     ) -> None:
         """Kafka event broker.
@@ -68,7 +69,8 @@ def __init__(
             security_protocol: Protocol used to communicate with brokers.
                 Valid values are: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL.
             loglevel: Logging level of the kafka logger.
-
+            convert_intent_id_to_string: Optional flag to configure whether intent ID's
+                are converted from an integer to a string.
         """
         import kafka
 
@@ -85,6 +87,7 @@ def __init__(
         self.ssl_certfile = ssl_certfile
         self.ssl_keyfile = ssl_keyfile
         self.ssl_check_hostname = ssl_check_hostname
+        self.convert_intent_id_to_string = convert_intent_id_to_string
 
         logging.getLogger("kafka").setLevel(loglevel)
 
@@ -107,6 +110,8 @@ def publish(
         retry_delay_in_seconds: float = 5,
     ) -> None:
         """Publishes events."""
+        if self.convert_intent_id_to_string:
+            event = self._convert_intent_id_to_string(event)
         if self.producer is None:
             self._create_producer()
             connected = self.producer.bootstrap_connected()
@@ -200,5 +205,17 @@ def _publish(self, event: Dict[Text, Any]) -> None:
         )
         self.producer.send(self.topic, value=event, key=partition_key)
 
+    def _convert_intent_id_to_string(self, event: Dict[Text, Any]) -> Dict[Text, Any]:
+        if event.get("event", "") == "user" and "id" in event.get("parse_data", {}).get(
+            "intent", {}
+        ):
+            event["parse_data"]["intent"]["id"] = str(
+                event["parse_data"]["intent"]["id"]
+            )
+            for idx, parse_data in enumerate(event["parse_data"]["intent_ranking"]):
+                parse_data["id"] = str(parse_data["id"])
+                event["parse_data"]["intent_ranking"][idx] = parse_data
+        return event
+
     def _close(self) -> None:
         self.producer.close()

tests/core/test_broker.py

@@ -262,6 +262,7 @@ async def test_kafka_broker_from_config():
         topic="topic",
         partition_by_sender=True,
         security_protocol="SASL_PLAINTEXT",
+        convert_intent_id_to_string=True,
     )
 
     assert actual.url == expected.url
@@ -270,6 +271,54 @@ async def test_kafka_broker_from_config():
     assert actual.sasl_mechanism == expected.sasl_mechanism
     assert actual.topic == expected.topic
     assert actual.partition_by_sender == expected.partition_by_sender
+    assert actual.convert_intent_id_to_string == expected.convert_intent_id_to_string
+
+
+async def test_kafka_broker_convert_intent_id_to_string():
+    user_event = {
+        "timestamp": 1517821726.200036,
+        "metadata": {},
+        "parse_data": {
+            "entities": [],
+            "intent": {"confidence": 0.54, "name": "greet", "id": 7703045398849936579},
+            "message_id": "987654321",
+            "metadata": {},
+            "text": "/greet",
+            "intent_ranking": [
+                {"confidence": 0.54, "name": "greet", "id": 7703045398849936579},
+                {"confidence": 0.31, "name": "goodbye", "id": -5127945386715371244},
+                {"confidence": 0.15, "name": "default", "id": 1699173715362944540},
+            ],
+        },
+        "event": "user",
+        "text": "/greet",
+        "input_channel": "rest",
+        "message_id": "987654321",
+    }
+    actual = KafkaEventBroker(
+        "localhost",
+        sasl_username="username",
+        sasl_password="password",
+        sasl_mechanism="PLAIN",
+        topic="topic",
+        partition_by_sender=True,
+        security_protocol="SASL_PLAINTEXT",
+        convert_intent_id_to_string=True,
+    )
+
+    converted_user_event = actual._convert_intent_id_to_string(user_event)
+    intent_ranking = user_event["parse_data"]["intent_ranking"]
+    converted_intent_ranking = converted_user_event["parse_data"]["intent_ranking"]
+
+    assert converted_user_event["parse_data"]["intent"]["id"] == str(
+        user_event["parse_data"]["intent"]["id"]
+    )
+    assert all(
+        converted_parse_data["id"] == str(parse_data["id"])
+        for parse_data, converted_parse_data in zip(
+            intent_ranking, converted_intent_ranking
+        )
+    )
 
 
 @pytest.mark.parametrize(