Clean Versions

ControllerExecutorFlow.py

@@ -61,7 +61,7 @@ class ControllerExecutorFlow(CompositeFlow):
     
     - `answer` (str): The answer of the flow to the query (e.g. "Michael Jordan is a basketball player and business man. He was born on February 17, 1963.")
     - `status` (str): The status of the flow. It can be "finished" or "unfinished". If the status is "unfinished", it's usually because the maximum amount of rounds was reached before the model found an answer.
-    
+
     :param flow_config: The configuration of the flow (see Configuration Parameters).
     :param subflows: A list of subflows. Required when instantiating the subflow programmatically (it replaces subflows_config from flow_config).                                  
     """
@@ -77,10 +77,10 @@ class ControllerExecutorFlow(CompositeFlow):
         )
         
         self.reply_interface = KeyInterface(
-            keys_to_select = ["answer","status", "EARLY_EXIT"],
+            keys_to_select = ["answer","status"],
         )
         
-        self.next_flow_to_call = {
+        self.next_state = {
             None: "Controller",
             "Controller": "Executor",
             "Executor": "Controller" 
@@ -95,12 +95,12 @@ class ControllerExecutorFlow(CompositeFlow):
         )
         self.send_message(reply)
     
-    def get_next_flow_to_call(self):
-        
+    def get_next_state(self):
+        """ """
         if self.flow_config["max_rounds"] is not None and self.flow_state["current_round"] >= self.flow_config["max_rounds"]:
             return None
         
-        return self.next_flow_to_call[self.flow_state["last_called"]]
+        return self.next_state[self.flow_state["last_state"]]
     
     def _on_reach_max_round(self):
         """ This method is called when the flow reaches the maximum amount of rounds. It updates the state of the flow and starts the process of terminating the flow."""
@@ -111,12 +111,14 @@ class ControllerExecutorFlow(CompositeFlow):
         })
         
     def set_up_flow_state(self):
+        """ Sets up the flow state."""
         super().set_up_flow_state()
-        self.flow_state["last_called"] = None
+        self.flow_state["last_state"] = None
         self.flow_state["current_round"] = 0
         
     
     def call_controller(self):
+        """ Calls the controller: the flow that decides which command to call next."""
         #first_round  
         if self.flow_state["current_round"] == 0:
             input_interface = self.input_interface_first_round_controller
@@ -134,6 +136,7 @@ class ControllerExecutorFlow(CompositeFlow):
             )
         
     def call_executor(self):
+        """ Calls the flow that executes the command instructed by the ControllerAtomicFlow."""
     
         #call executor
         executor_branch_to_call = self.flow_state["command"]
@@ -149,13 +152,18 @@ class ControllerExecutorFlow(CompositeFlow):
             
             
     def register_data_to_state(self, input_message):
-        last_called = self.flow_state["last_called"]
+        """This method registers the input message data to the flow state. It's everytime a new input message is received.
+        
+        :param input_message: The input message
+        :type input_message: FlowMessage
+        """
+        last_state = self.flow_state["last_state"]
         
-        if last_called is None:
+        if last_state is None:
             self.flow_state["input_message"] = input_message
             self.flow_state["goal"] = input_message.data["goal"]
         
-        elif last_called == "Executor":
+        elif last_state == "Executor":
             self.flow_state["observation"] = input_message.data
         
         else:
@@ -190,15 +198,15 @@ class ControllerExecutorFlow(CompositeFlow):
      
         self.register_data_to_state(input_message)       
         
-        flow_to_call = self.get_next_flow_to_call()
+        current_state = self.get_next_state()
         
         if self.flow_state.get("early_exit_flag",False):
             self.generate_reply()
         
-        elif flow_to_call == "Controller":
+        elif current_state == "Controller":
             self.call_controller()
             
-        elif flow_to_call == "Executor":
+        elif current_state == "Executor":
             self.call_executor()
             self.flow_state["current_round"] += 1
             
@@ -206,5 +214,5 @@ class ControllerExecutorFlow(CompositeFlow):
             self._on_reach_max_round()
             self.generate_reply()
         
-        self.flow_state["last_called"] = flow_to_call
+        self.flow_state["last_state"] = current_state
         

ControllerExecutorFlow.yaml

@@ -14,6 +14,9 @@ subflows_config:
   Controller:
     name: "ControllerAtomicFlow"
     description: "A flow that calls other flows to solve a problem."
+    flow_class_name: flow_modules.aiflows.ControllerExecutorFlowModule.ControllerAtomicFlow
+    flow_endpoint: ControllerAtomicFlow
+    user_id: local
     _target_: flow_modules.aiflows.ControllerExecutorFlowModule.ControllerAtomicFlow.instantiate_from_default_config
     commands:
       finish:
@@ -25,7 +28,13 @@ subflows_config:
 #        description: "Performs a search on Wikipedia."
 #        input_args: ["search_term"]
 
-
+  wiki_search:
+    _target_: flow_modules.aiflows.ControllerExecutorFlowModule.WikiSearchAtomicFlow.instantiate_from_default_config
+    flow_class_name: flow_modules.aiflows.ControllerExecutorFlowModule.WikiSearchAtomicFlow
+    flow_endpoint: WikiSearchAtomicFlow
+    user_id: local
+    name: "WikiSearchAtomicFlow"
+    description: "A flow that searches Wikipedia for information."
 # E.g.,
   # wiki_search:
   #   _target_: .WikiSearchAtomicFlow.instantiate_from_default_config

demo.yaml

@@ -4,11 +4,12 @@ max_rounds: 30
 ### Subflows specification
 subflows_config:
   Controller:
-    _target_: aiflows.base_flows.AtomicFlow.instantiate_from_default_config
+    _target_: flow_modules.aiflows.ControllerExecutorFlowModule.ControllerAtomicFlow.instantiate_from_default_config
     name: "Proxy Controller"
-    description: "A flow that acts as a proxy for the controller."
+    flow_class_name: flow_modules.aiflows.ControllerExecutorFlowModule.ControllerAtomicFlow
     flow_endpoint: ControllerAtomicFlow
     user_id: local
+    description: "A flow that acts as a proxy for the controller."
     commands:
       wiki_search:
         description: "Performs a search on Wikipedia."
@@ -26,8 +27,9 @@ subflows_config:
 
 
   wiki_search:
-    _target_: aiflows.base_flows.AtomicFlow.instantiate_from_default_config
-    name: "WikiSearchAtomicFlow"
-    description: "A flow that searches Wikipedia for information."
+    _target_: flow_modules.aiflows.ControllerExecutorFlowModule.WikiSearchAtomicFlow.instantiate_from_default_config
+    flow_class_name: flow_modules.aiflows.ControllerExecutorFlowModule.WikiSearchAtomicFlow
     flow_endpoint: WikiSearchAtomicFlow
     user_id: local
+    name: "WikiSearchAtomicFlow"
+    description: "A flow that searches Wikipedia for information."

run.py

@@ -60,19 +60,19 @@ if __name__ == "__main__":
     #3. ~~~~ Serve The Flow ~~~~
     
     
-    serve_utils.serve_flow(
-        cl = cl,
-        flow_class_name="flow_modules.aiflows.ControllerExecutorFlowModule.WikiSearchAtomicFlow",
-        flow_endpoint="WikiSearchAtomicFlow",
-    )
+    # serve_utils.recursive_serve_flow(
+    #     cl = cl,
+    #     flow_class_name="flow_modules.aiflows.ControllerExecutorFlowModule.WikiSearchAtomicFlow",
+    #     flow_endpoint="WikiSearchAtomicFlow",
+    # )
     
-    serve_utils.serve_flow(
-        cl = cl,
-        flow_class_name="flow_modules.aiflows.ControllerExecutorFlowModule.ControllerAtomicFlow",
-        flow_endpoint="ControllerAtomicFlow",
-    )
+    # serve_utils.serve_flow(
+    #     cl = cl,
+    #     flow_class_name="flow_modules.aiflows.ControllerExecutorFlowModule.ControllerAtomicFlow",
+    #     flow_endpoint="ControllerAtomicFlow",
+    # )
     
-    serve_utils.serve_flow(
+    serve_utils.recursive_serve_flow(
         cl = cl,
         flow_class_name="flow_modules.aiflows.ControllerExecutorFlowModule.ControllerExecutorFlow",
         flow_endpoint="ControllerExecutorFlow",