working version coflows

ControllerExecutorFlow.py

@@ -1,15 +1,15 @@
 from typing import Dict, Any
 
-from aiflows.base_flows import CircularFlow
+from aiflows.base_flows import CompositeFlow
 from aiflows.utils import logging
 
 from .ControllerAtomicFlow import ControllerAtomicFlow
-
+from aiflows.interfaces import KeyInterface
 logging.set_verbosity_debug()
 log = logging.get_logger(__name__)
 
 
-class ControllerExecutorFlow(CircularFlow):
+class ControllerExecutorFlow(CompositeFlow):
     """ This class implements a ControllerExecutorFlow. It's composed of a ControllerAtomicFlow and an ExecutorFlow.
     Where typically the ControllerAtomicFlow is uses a LLM to decide which command to call and the ExecutorFlow (branching flow) is used to execute the command.
     
@@ -64,29 +64,33 @@ class ControllerExecutorFlow(CircularFlow):
     :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).                                  
     """
-    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."""
-        self._state_update_dict({
-            "answer": "The maximum amount of rounds was reached before the model found an answer.",
-            "status": "unfinished"
-        })
+    def __init__(self, flow_config: Dict[str, Any], subflows: Dict[str, Any] = None):
+        super().__init__(flow_config, subflows)
+        
+    def set_up_flow_state(self):
+        super().set_up_flow_state()
+        
+        
+    def run(self,input_data):
 
-    @CircularFlow.output_msg_payload_processor
-    def detect_finish_or_continue(self, output_payload: Dict[str, Any], src_flow: ControllerAtomicFlow) -> Dict[str, Any]:
-        """ This method is called when the ExecutorAtomicFlow receives a message from the ControllerAtomicFlow. It checks if the flow should finish or continue.
+        executor_reply = input_data
         
-        :param output_payload: The output payload of the ControllerAtomicFlow.
-        :type output_payload: Dict[str, Any]
-        :param src_flow: The ControllerAtomicFlow.
-        :type src_flow: ControllerAtomicFlow
-        :return: The output payload of the ControllerAtomicFlow.
-        """
-        command = output_payload["command"]
-        if command == "finish":
-            return {
-                "EARLY_EXIT": True,
-                "answer": output_payload["command_args"]["answer"],
-                "status": "finished"
+        for round in range(self.flow_config["max_rounds"]):
+            controller_reply = self.ask_subflow("Controller", executor_reply).get_data()
+            if controller_reply["command"] == "finish":
+                return {
+                    "EARLY_EXIT": True,
+                    "answer": controller_reply["command_args"]["answer"],
+                    "status": "finished"
+                }
+            
+            
+            
+            executor_reply = {
+                    "observation":  self.ask_subflow(controller_reply["command"], controller_reply["command_args"]).get_data()
             }
-        else:
-            return output_payload
+           
+        return  {
+            "answer": "The maximum amount of rounds was reached before the model found an answer.",
+            "status": "unfinished"
+        }  

ControllerExecutorFlow.yaml

@@ -12,6 +12,8 @@ output_interface:
 ### Subflows specification
 subflows_config:
   Controller:
+    name: "ControllerAtomicFlow"
+    description: "A flow that calls other flows to solve a problem."
     _target_: flow_modules.aiflows.ControllerAtomicFlow.instantiate_from_default_config
     finish:
       description: "Signal that the objective has been satisfied, and returns the answer to the user."
@@ -23,38 +25,10 @@ subflows_config:
 #        input_args: ["search_term"]
 
 
-  Executor:
-    _target_: aiflows.base_flows.BranchingFlow.instantiate_from_default_config
 # E.g.,
-#    subflows_config:
-#      wiki_search:
-#        _target_: .WikiSearchAtomicFlow.instantiate_from_default_config
+  # wiki_search:
+  #   _target_: .WikiSearchAtomicFlow.instantiate_from_default_config
 
 
 early_exit_key: "EARLY_EXIT"
 
-topology:
-  - goal: "Select the next action and prepare the input for the executor."
-    input_interface:
-      _target_: aiflows.interfaces.KeyInterface
-      additional_transformations:
-        - _target_: aiflows.data_transformations.KeyMatchInput
-    flow: Controller
-    output_interface:
-      _target_: ControllerExecutorFlow.detect_finish_or_continue
-    reset: false
-
-  - goal: "Execute the action specified by the Controller."
-    input_interface:
-      _target_: aiflows.interfaces.KeyInterface
-      keys_to_rename:
-        command: branch
-        command_args: branch_input_data
-      keys_to_select: ["branch", "branch_input_data"]
-    flow: Executor
-    output_interface:
-      _target_: aiflows.interfaces.KeyInterface
-      keys_to_rename:
-        branch_output_data: observation
-      keys_to_select: ["observation"]
-    reset: false

demo.yaml

@@ -1,27 +1,29 @@
-flow:
-  _target_: flow_modules.aiflows.ControllerExecutorFlowModule.ControllerExecutorFlow.instantiate_from_default_config
-  max_rounds: 30
+_target_: flow_modules.aiflows.ControllerExecutorFlowModule.ControllerExecutorFlow.instantiate_from_default_config
+max_rounds: 30
 
-  ### Subflows specification
-  subflows_config:
-    Controller:
-      _target_: flow_modules.aiflows.ControllerExecutorFlowModule.ControllerAtomicFlow.instantiate_from_default_config
-      commands:
-        wiki_search:
-          description: "Performs a search on Wikipedia."
-          input_args: [ "search_term" ]
-        finish:
-          description: "Signal that the objective has been satisfied, and returns the answer to the user."
-          input_args: [ "answer" ]
-      backend:
-        _target_: aiflows.backends.llm_lite.LiteLLMBackend
-        api_infos: ???
-        model_name: 
-          openai: "gpt-3.5-turbo"
-          azure: "azure/gpt-4"
+### Subflows specification
+subflows_config:
+  Controller:
+    _target_: flow_modules.aiflows.ControllerExecutorFlowModule.ControllerAtomicFlow.instantiate_from_default_config
+    commands:
+      wiki_search:
+        description: "Performs a search on Wikipedia."
+        input_args: ["search_term"]
+      ddg_search:
+        description: "Query the search engine DuckDuckGo."
+        input_args: ["query"]
+      finish:
+        description: "Signal that the objective has been satisfied, and returns the answer to the user."
+        input_args: ["answer"]
+    backend:
+      _target_: aiflows.backends.llm_lite.LiteLLMBackend
+      api_infos: ???
+      model_name:
+        openai: "gpt-3.5-turbo"
+        azure: "azure/gpt-4"
 
-    Executor:
-      _target_: aiflows.base_flows.BranchingFlow.instantiate_from_default_config
-      subflows_config:
-        wiki_search:
-          _target_: flow_modules.aiflows.ControllerExecutorFlowModule.WikiSearchAtomicFlow.instantiate_from_default_config
+
+  wiki_search:
+    _target_: flow_modules.aiflows.ControllerExecutorFlowModule.WikiSearchAtomicFlow.instantiate_from_default_config
+    name: "WikiSearchAtomicFlow"
+    description: "A flow that searches Wikipedia for information."

run.py

@@ -5,73 +5,91 @@ import hydra
 import aiflows
 from aiflows.flow_launchers import FlowLauncher
 from aiflows.backends.api_info import ApiInfo
-from aiflows.utils.general_helpers import read_yaml_file
+from aiflows.utils.general_helpers import read_yaml_file, quick_load_api_keys
 
 from aiflows import logging
 from aiflows.flow_cache import CACHING_PARAMETERS, clear_cache
+from aiflows.utils import serve_utils
+from aiflows.workers import run_dispatch_worker_thread
+from aiflows.messages import FlowMessage
+from aiflows.interfaces import KeyInterface
 
 CACHING_PARAMETERS.do_caching = False  # Set to True in order to disable caching
 # clear_cache() # Uncomment this line to clear the cache
 
 logging.set_verbosity_debug()
 
+from aiflows import flow_verse
+# ~~~ Load Flow dependecies from FlowVerse ~~~
 dependencies = [
-    {"url": "aiflows/ControllerExecutorFlowModule", "revision": os.getcwd()},
+    {"url": "aiflows/ControllerExecutorFlowModule", "revision":  os.getcwd()},
 ]
-from aiflows import flow_verse
 
 flow_verse.sync_dependencies(dependencies)
 
 if __name__ == "__main__":
     # ~~~ Set the API information ~~~
     # OpenAI backend
-    api_information = [ApiInfo(backend_used="openai",
-                              api_key = os.getenv("OPENAI_API_KEY"))]
+    api_information = [ApiInfo(backend_used="openai", api_key=os.getenv("OPENAI_API_KEY"))]
     # Azure backend
-    # api_information = ApiInfo(backend_used = "azure",
+    # api_information = [ApiInfo(backend_used = "azure",
     #                           api_base = os.getenv("AZURE_API_BASE"),
     #                           api_key = os.getenv("AZURE_OPENAI_KEY"),
-    #                           api_version =  os.getenv("AZURE_API_VERSION") )
+    #                           api_version =  os.getenv("AZURE_API_VERSION") )]
 
-    path_to_output_file = None
-    # path_to_output_file = "output.jsonl"  # Uncomment this line to save the output to disk
+    FLOW_MODULES_PATH = "./"
     
+    jwt = os.getenv("COLINK_JWT")
+    addr = os.getenv("LOCAL_COLINK_ADDRESS")
+    
+    cl = serve_utils.start_colink_component(
+        "Reverse Number Demo",
+        {"jwt": jwt, "addr": addr}
+    )
+    # path_to_output_file = "output.jsonl"  # Uncomment this line to save the output to disk
+
     root_dir = "."
     cfg_path = os.path.join(root_dir, "demo.yaml")
     cfg = read_yaml_file(cfg_path)
-    # print(cfg["flow"].keys())
-    cfg["flow"]["subflows_config"]["Controller"]["backend"]["api_infos"] = api_information
+    # put the API information in the config
+    serve_utils.recursive_serve_flow(
+        cl = cl,
+        flow_type="ReAct_served",
+        default_config=cfg,
+        default_state=None,
+        default_dispatch_point="coflows_dispatch",
+    )
     
-    # ~~~ Instantiate the Flow ~~~
-    flow_with_interfaces = {
-        "flow": hydra.utils.instantiate(cfg['flow'], _recursive_=False, _convert_="partial"),
-        "input_interface": (
-            None
-            if cfg.get( "input_interface", None) is None
-            else hydra.utils.instantiate(cfg['input_interface'], _recursive_=False)
-        ),
-        "output_interface": (
-            None
-            if cfg.get( "output_interface", None) is None
-            else hydra.utils.instantiate(cfg['output_interface'], _recursive_=False)
-        ),
-    }
+    #in case you haven't started the dispatch worker thread, uncomment
+    #run_dispatch_worker_thread(cl, dispatch_point="coflows_dispatch", flow_modules_base_path=FLOW_MODULES_PATH)
+
+    quick_load_api_keys(cfg, api_information, key="api_infos")
 
     # ~~~ Get the data ~~~
     # This can be a list of samples
     # data = {"id": 0, "goal": "Answer the following question: What is the population of Canada?"}  # Uses wikipedia
-    data = {"id": 0, "goal": "Answer the following question: Who was the NBA champion in 2023?"}
-
+    # data = {"id": 0, "goal": "Answer the following question: Who was the NBA champion in 2023?"}
+    data = {
+        "id": 0,
+        "goal": "Answer the following question: What is the profession and date of birth of Michael Jordan?",
+    }
     # ~~~ Run inference ~~~
-    path_to_output_file = None
-    # path_to_output_file = "output.jsonl"  # Uncomment this line to save the output to disk
-
-    _, outputs = FlowLauncher.launch(
-        flow_with_interfaces=flow_with_interfaces,
-        data=data,
-        path_to_output_file=path_to_output_file
-    )
-
+    proxy_flow = serve_utils.recursive_mount(
+        cl=cl,
+        client_id="local",
+        flow_type="ReAct_served",
+        config_overrides=cfg,
+        initial_state=None,
+        dispatch_point_override=None,
+    )   
     # ~~~ Print the output ~~~
-    flow_output_data = outputs[0]
-    print(flow_output_data)
+    input_message = FlowMessage(
+        data= data,
+        src_flow="Coflows team",
+        dst_flow=proxy_flow,
+        is_input_msg=True
+    )
+    
+    future = proxy_flow.ask(input_message)
+    
+    print(future.get_data())