coflows compatible

HumanStandardInputFlow.py

@@ -8,7 +8,7 @@ from aiflows.base_flows import AtomicFlow
 from aiflows.messages import UpdateMessage_Generic
 
 from aiflows.utils import logging
-
+from aiflows.messages import FlowMessage
 # logging.set_verbosity_debug()  # ToDo: Has no effect on the logger for __name__. Level is warn, and info is not printed
 log = logging.get_logger(f"aiflows.{__name__}")  # ToDo: Is there a better fix?
 
@@ -142,7 +142,7 @@ class HumanStandardInputFlow(AtomicFlow):
         return human_input
 
     def run(self,
-            input_data: Dict[str, Any]) -> Dict[str, Any]:
+            input_message: FlowMessage):
         """ Runs the HumanStandardInputFlow. It's used to read input from the user/human's standard input.
         
         :param input_data: The input data dictionary
@@ -150,8 +150,10 @@ class HumanStandardInputFlow(AtomicFlow):
         :return: The output data dictionary
         :rtype: Dict[str, Any]
         """
-
+        input_data = input_message.data
+        
         query_message = self._get_message(self.query_message_prompt_template, input_data)
+        
         state_update_message = UpdateMessage_Generic(
             created_by=self.flow_config['name'],
             updated_flow=self.flow_config["name"],
@@ -162,4 +164,6 @@ class HumanStandardInputFlow(AtomicFlow):
         log.info(query_message)
         human_input = self._read_input()
 
-        return {"human_input": human_input}
+        reply_message = self._package_output_message(input_message = input_message, response = {"human_input": human_input})
+        
+        self.reply_to_message(reply = reply_message, to = input_message)

demo.yaml

@@ -1,17 +1,15 @@
-input_interface:
-  _target_: aiflows.interfaces.KeyInterface
-flow:
-  _target_: flow_modules.aiflows.HumanStandardInputFlowModule.HumanStandardInputFlow.instantiate_from_default_config
-  name: "HumanStandardInputFlow"
-  description: "A demo of the HumanStandardInputFlow."
-  request_multi_line_input_flag: True
-  end_of_input_string: EOI
-  query_message_prompt_template: 
-    _target_: aiflows.prompt_template.JinjaPrompt
-    template: |2-
-      Please enter your input.
 
-      What's opinion on this statement?
-      {{statement}}
-    input_variables: []
-    partial_variables: {"statement": "This Flow works well."}
+_target_: flow_modules.aiflows.HumanStandardInputFlowModule.HumanStandardInputFlow.instantiate_from_default_config
+name: "HumanStandardInputFlow"
+description: "A demo of the HumanStandardInputFlow."
+request_multi_line_input_flag: True
+end_of_input_string: EOI
+query_message_prompt_template: 
+  _target_: aiflows.prompt_template.JinjaPrompt
+  template: |2-
+    Please enter your input.
+
+    What's opinion on this statement?
+    {{statement}}
+  input_variables: [statement]
+  partial_variables: {}

run.py

@@ -2,14 +2,22 @@ import os
 
 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 import logging
-from aiflows.flow_cache import CACHING_PARAMETERS
+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
+from aiflows.utils.colink_utils import start_colink_server
+from aiflows.workers import run_dispatch_worker_thread
 
-CACHING_PARAMETERS.do_caching = False  # Set to True to enable caching
+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()
@@ -21,38 +29,70 @@ from aiflows import flow_verse
 flow_verse.sync_dependencies(dependencies)
 
 if __name__ == "__main__":
-
+   
+    #1. ~~~~~ Set up a colink server ~~~~
+    FLOW_MODULES_PATH = "./"
+    
+    cl = start_colink_server()
+    
+    #2. ~~~~~Load flow config~~~~~~
     root_dir = "."
     cfg_path = os.path.join(root_dir, "demo.yaml")
     cfg = read_yaml_file(cfg_path)
-    flow_with_interfaces = {
-        "flow": hydra.utils.instantiate(cfg['flow'], _recursive_=False, _convert_="partial"),
-        "input_interface": (
-            None
-            if getattr(cfg, "input_interface", None) is None
-            else hydra.utils.instantiate(cfg['input_interface'], _recursive_=False)
-        ),
-        "output_interface": (
-            None
-            if getattr(cfg, "output_interface", None) is None
-            else hydra.utils.instantiate(cfg['output_interface'], _recursive_=False)
-        ),
-    }
-
-    # ~~~ Get the data ~~~
-    # This can be a list of samples
-    data = {"id": 0}  # Add your data here
-
-    # ~~~ 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,
+
+    #3. ~~~~ Serve The Flow ~~~~
+    serve_utils.serve_flow(
+        cl = cl,
+        flow_type="HumanStandardInputFlowModule",
+        default_config=cfg,
+        default_state=None,
+        default_dispatch_point="coflows_dispatch"
+    )
+    
+    #4. ~~~~~Start A Worker Thread~~~~~
+    run_dispatch_worker_thread(cl, dispatch_point="coflows_dispatch", flow_modules_base_path=FLOW_MODULES_PATH)
+
+    #5. ~~~~~Mount the flow and get its proxy~~~~~~
+    proxy_flow = serve_utils.recursive_mount(
+        cl=cl,
+        client_id="local",
+        flow_type="HumanStandardInputFlowModule",
+        config_overrides=None,
+        initial_state=None,
+        dispatch_point_override=None,
+    )   
+    
+
+    #6. ~~~ Get the data ~~~
+    data = {"id": 0, "statement": "This flow works well"}  # This can be a list of samples
+    # data = {"id": 0, "question": "Who was the NBA champion in 2023?"}  # This can be a list of samples
+    
+    #option1: use the FlowMessage class
+    input_message = FlowMessage(
         data=data,
-        path_to_output_file=path_to_output_file,
     )
 
+    #option2: use the proxy_flow
+    #input_message = proxy_flow._package_input_message(data = data)
+    
+    #7. ~~~ Run inference ~~~
+    future = proxy_flow.send_message_blocking(input_message)
+    
+    #uncomment this line if you would like to get the full message back
+    #reply_message = future.get_message()
+    reply_data = future.get_data()
+    
     # ~~~ Print the output ~~~
-    flow_output_data = outputs[0]
-    print(flow_output_data)
+    print("~~~~~~Reply~~~~~~")
+    print(reply_data)
+    
+    
+    #8. ~~~~ (Optional) apply output interface on reply ~~~~
+    # output_interface = KeyInterface(
+    #     keys_to_rename={"api_output": "answer"},
+    # )
+    # print("Output: ", output_interface(reply_data))
+    
+    
+    #9. ~~~~~Optional: Unserve Flow~~~~~~
+    # serve_utils.delete_served_flow(cl, "ReverseNumberAtomicFlow_served")