feat: added JeevesListener for unified DCT on both Jeeves api and Jeeves agents

Author: cristibleotiu

constants.py

@@ -157,6 +157,19 @@ class JeevesCt:
     "KEYSOFT_JEEVES",
   ]
 
+  JEEVES_AGENT_SIGNATURES = [
+    "DOC_EMBEDDING_AGENT",
+    "LLM_AGENT",
+  ]
+
+  JEEVES_PLUGIN_SIGNATURES = JEEVES_API_SIGNATURES + JEEVES_AGENT_SIGNATURES
+
+  UNIFIED_PATH_FILTER = [
+    None,
+    None,
+    JEEVES_PLUGIN_SIGNATURES,
+    None
+  ]
   AGENT_PATH_FILTER = [
     None,
     None,
@@ -166,10 +179,7 @@ class JeevesCt:
   API_PATH_FILTER = [
     None,
     None,
-    [
-      "DOC_EMBEDDING_AGENT",
-      "LLM_AGENT"
-    ],
+    JEEVES_AGENT_SIGNATURES,
     None
   ]
 

docker-compose/start_n_containers.bat

@@ -6,18 +6,21 @@ set DEPLOY_DEBUG=false
 set USE_LOCAL_IMAGE=true
 set USE_GPU=true
 set USE_IPC_HOST=true
-set EXPOSED_PORT=15033
+set EXPOSED_PORT=15034
+set USE_CACHED_DOCKER_COMPOSE=true
 
 
 REM Hardcoded number of containers
-set NUM_CONTAINERS=1
+set NUM_CONTAINERS=2
 set NUM_SUPERVISORS=0
+set START_CONTAINER_INDEX=1
+set END_CONTAINER_INDEX=%NUM_CONTAINERS% + %START_CONTAINER_INDEX% - 1
 
 
 if !USE_LOCAL_IMAGE! == true (
     set CONTAINER_IMAGE=local_edge_node
 ) else (
-    set CONTAINER_IMAGE=ratio1/edge_node:develop
+    set CONTAINER_IMAGE=ratio1/edge_node:devnet
 )
 
 if !DEPLOY_DEBUG! == true (
@@ -63,9 +66,11 @@ set WATCHTOWER_LABEL=com.centurylinklabs.watchtower.enable=true
 REM Generate the list of container IDs dynamically
 for /l %%i in (1,1,%NUM_CONTAINERS%) do (
     REM Generic container ID and edge node ID
-    set CONTAINER_IDS[%%i]=!GENERIC_CONTAINER_ID!%%i
-    set EDGE_NODE_IDS[%%i]=!GENERIC_EDGE_NODE_ID!%%i
-    set CONTAINER_VOLUMES[%%i]=!GENERIC_CONTAINER_VOLUME!%%i
+    set /a CURRENT_CONTAINER_IDX = %%i + !START_CONTAINER_INDEX! - 1
+    echo Current container index: !CURRENT_CONTAINER_IDX!
+    set CONTAINER_IDS[%%i]=!GENERIC_CONTAINER_ID!!CURRENT_CONTAINER_IDX!
+    set EDGE_NODE_IDS[%%i]=!GENERIC_EDGE_NODE_ID!!CURRENT_CONTAINER_IDX!
+    set CONTAINER_VOLUMES[%%i]=!GENERIC_CONTAINER_VOLUME!!CURRENT_CONTAINER_IDX!
 
     REM Check if the container is a supervisor
     if %%i leq %NUM_SUPERVISORS% (
@@ -76,6 +81,14 @@ for /l %%i in (1,1,%NUM_CONTAINERS%) do (
 )
 
 
+REM Check if we should use cached docker-compose.yaml
+if !USE_CACHED_DOCKER_COMPOSE! == true (
+    if exist docker-compose.yaml (
+        echo Using cached docker-compose.yaml
+        goto :START_CONTAINERS
+    )
+)
+
 REM Generate docker-compose.yaml dynamically
 echo services: > docker-compose.yaml
 for /l %%i in (1,1,%NUM_CONTAINERS%) do (
@@ -153,6 +166,7 @@ REM Maybe unnecessary
 @REM     echo   !CONTAINER_VOLUMES[%%i]!: >> docker-compose.yaml
 @REM )
 
+:START_CONTAINERS
 REM Pull the containers
 docker-compose pull
 

extensions/business/mixins/nlp_agent_mixin.py

@@ -1,27 +1,60 @@
 NLP_AGENT_MIXIN_CONFIG = {
   'OBJECT_TYPE': [],
   "ALLOW_EMPTY_INPUTS": False,  # if this is set to true the on-idle will be triggered continuously the process
+  "DEBUG_MODE": True,
 
   "VALIDATION_RULES": {
   },
 }
 
 
 class _NlpAgentMixin(object):
-  def compute_and_send_responses(self, inferences):
+  def Pd(self, msg, **kwargs):
+    if self.cfg_debug_mode:
+      self.P(msg, **kwargs)
+    return
+
+  def filter_valid_inferences(self, inferences, return_idxs=False):
+    res = []
+    idxs = []
+    for idx, inf in enumerate(inferences):
+      if isinstance(inf, dict) and inf.get("IS_VALID", True):
+        res.append(inf)
+        idxs.append(idx)
+    # endfor inferences
+    return res if not return_idxs else (res, idxs)
+
+  def inference_to_response(self, inference, model_name):
+    return {
+      'REQUEST_ID': inference.get('REQUEST_ID'),
+      'MODEL_NAME': model_name,
+      'TEXT_RESPONSE': inference.get('text'),
+    }
+
+  def handle_inferences(self, inferences, data=None):
+    if not isinstance(inferences, list):
+      return
     model_name = inferences[0].get('MODEL_NAME', None) if len(inferences) > 0 else None
+    cnt_initial_inferences = len(inferences)
+    inferences, valid_idxs = self.filter_valid_inferences(inferences, return_idxs=True)
+    self.Pd(f"Filtered {cnt_initial_inferences} inferences to {len(inferences)} valid inferences.")
+    if data is not None:
+      filtered_data = [
+        data[idx] for idx in valid_idxs
+      ]
+      if len(filtered_data) > 0:
+        self.Pd(f"Received requests: {self.json_dumps(self.shorten_str(filtered_data), indent=2)}")
+    # endif data is not None
+
     for inf in inferences:
-      request_result = {
-        'REQUEST_ID': inf.get('REQUEST_ID'),
-        'MODEL_NAME': model_name,
-        'TEXT_RESPONSE': inf.get('text'),
-      }
+      request_id = inf.get('REQUEST_ID', None)
+      self.Pd(f"Processing inference for request ID: {request_id}, model: {model_name}")
+      request_result = self.inference_to_response(inf, model_name)
       current_payload_kwargs = {
         'result': request_result,
-        'request_id': inf.get('REQUEST_ID'),
+        'request_id': request_id,
       }
       self.add_payload_by_fields(**current_payload_kwargs)
     # endfor inferences
     return
 
-

extensions/business/nlp/doc_embedding_agent.py

@@ -1,10 +1,12 @@
 from naeural_core.business.base import BasePluginExecutor as BasePlugin
+from extensions.business.mixins.nlp_agent_mixin import _NlpAgentMixin, NLP_AGENT_MIXIN_CONFIG
 
 __VER__ = '0.1.0.0'
 
 _CONFIG = {
   # mandatory area
   **BasePlugin.CONFIG,
+  **NLP_AGENT_MIXIN_CONFIG,
 
   'MAX_INPUTS_QUEUE_SIZE': 64,
 
@@ -21,14 +23,9 @@
 }
 
 
-class DocEmbeddingAgentPlugin(BasePlugin):
+class DocEmbeddingAgentPlugin(BasePlugin, _NlpAgentMixin):
   CONFIG = _CONFIG
 
-  def D(self, msg, **kwargs):
-    if self.cfg_debug_mode:
-      self.P(msg, **kwargs)
-    return
-
   def on_init(self):
     self.__last_status_time = None
     self.__last_inference_meta = None
@@ -80,22 +77,13 @@ def maybe_send_status(self, inf_meta):
     # endif time to send status
     return
 
+  def inference_to_response(self, inference, model_name):
+    return inference
+
   def _process(self):
     inf_meta = self.dataapi_inferences_meta().get(self.cfg_ai_engine)
     self.maybe_send_status(inf_meta)
+    data = self.dataapi_struct_datas()
     inferences = self.dataapi_struct_data_inferences()
-    if inferences is None or len(inferences) == 0 or isinstance(inferences[0], list):
-      return
-    self.D(f'[Agent]Processing inferences: {str(inferences)[:50]}')
-    for inf in inferences:
-      self.D(f'[Agent]Processing inference: {inf["REQUEST_ID"]}')
-      # For each inference a response payload will be created
-      request_id = inf.get('REQUEST_ID')
-      request_result = inf
-      self.D(f'[Agent]Processing inference: {request_result}')
-      self.add_payload_by_fields(
-        result=request_result,
-        request_id=request_id,
-      )
-    # endfor inferences
+    self.handle_inferences(inferences, data=data)
     return

extensions/business/nlp/llm_agent.py

@@ -24,13 +24,7 @@ class LlmAgentPlugin(BasePlugin, _NlpAgentMixin):
   def _process(self):
     # we always receive input from the upstream due to the fact that _process
     # is called only when we have input based on ALLOW_EMPTY_INPUTS=False (from NLP_AGENT_MIXIN_CONFIG)
-    data = self.dataapi_struct_data()
+    data = self.dataapi_struct_datas()
     inferences = self.dataapi_struct_data_inferences()
-    if len(inferences) == 0:
-      return
-    if isinstance(inferences[0], list):
-      return
-    self.P(f"Received request:\n{self.json_dumps(self.shorten_str(data), indent=2)}")
-    self.P(f"Received inferences:\n{inferences}")
-    self.compute_and_send_responses(inferences)
+    self.handle_inferences(inferences=inferences, data=data)
     return

extensions/data/default/jeeves/jeeves_agent_listener.py

@@ -0,0 +1,56 @@
+from extensions.data.default.jeeves.jeeves_listener import JeevesListenerDataCapture as BaseClass
+from constants import JeevesCt
+
+
+_CONFIG = {
+  **BaseClass.CONFIG,
+
+  "PATH_FILTER": JeevesCt.AGENT_PATH_FILTER,
+  "PING_PERIOD": 0.5,  # seconds between pings
+
+  'PING_ENABLED': False,  # whether to send ping inputs
+
+  'VALIDATION_RULES': {
+    **BaseClass.CONFIG['VALIDATION_RULES'],
+  },
+}
+
+
+class JeevesAgentListenerDataCapture(BaseClass):
+  CONFIG = _CONFIG
+
+  def __init__(self, **kwargs):
+    super(JeevesAgentListenerDataCapture, self).__init__(**kwargs)
+    self.last_ping_time = 0
+    return
+
+  def add_ping_input(self):
+    self._add_inputs(
+      [
+        self._new_input(struct_data={
+          'ping': True
+        })
+      ]
+    )
+
+  def needs_ping(self):
+    """
+    Check if it's time to send a ping input.
+    """
+    if not self.cfg_ping_enabled:
+      return False
+    if self._deque is None or len(self._deque) > 0:
+      return False
+    return self.time() - self.last_ping_time >= self.cfg_ping_period
+
+  def _run_data_aquisition_step(self):
+    """
+    Override to of _run_data_aquisition_step to add ping inputs for Jeeves agents.
+    """
+    super()._run_data_aquisition_step()
+    if self.needs_ping():
+      # Add a ping input for Jeeves agents
+      self.add_ping_input()
+      self.last_ping_time = self.time()
+    # endif needs_ping
+    return

…/data/default/iot/jeeves_api_listener.py …ta/default/jeeves/jeeves_api_listener.pyextensions/data/default/iot/jeeves_api_listener.py renamed to extensions/data/default/jeeves/jeeves_api_listener.py extensions/data/default/iot/jeeves_api_listener.py renamed to extensions/data/default/jeeves/jeeves_api_listener.py

@@ -1,4 +1,4 @@
-from naeural_core.data.default.iot.network_listener import NetworkListenerDataCapture as BaseClass
+from extensions.data.default.jeeves.jeeves_listener import JeevesListenerDataCapture as BaseClass
 from constants import JeevesCt
 
 

…fault/iot/jeeves_embed_agent_listener.py …lt/jeeves/jeeves_embed_agent_listener.pyextensions/data/default/iot/jeeves_embed_agent_listener.py renamed to extensions/data/default/jeeves/jeeves_embed_agent_listener.py extensions/data/default/iot/jeeves_embed_agent_listener.py renamed to extensions/data/default/jeeves/jeeves_embed_agent_listener.py

@@ -1,4 +1,4 @@
-from extensions.data.default.iot.jeeves_agent_listener import JeevesAgentListenerDataCapture as BaseClass
+from extensions.data.default.jeeves.jeeves_agent_listener import JeevesAgentListenerDataCapture as BaseClass
 from constants import JeevesCt