coflows compatible

demo.yaml

@@ -1,6 +1,5 @@
-flow:
-  _target_: flow_modules.aiflows.FixedReplyFlowModule.FixedReplyFlow.instantiate_from_default_config
-  name: "FixedReplyFlow"
-  description: "A demo of the FixedReplyFlow."
-  fixed_reply: "This is a fixed reply."
+_target_: flow_modules.aiflows.FixedReplyFlowModule.FixedReplyFlow.instantiate_from_default_config
+name: "FixedReplyFlow"
+description: "A demo of the FixedReplyFlow."
+fixed_reply: "This is a fixed reply."
   

fixed_reply.py

@@ -1,7 +1,7 @@
 from typing import Dict, Any
 
 from aiflows.base_flows import AtomicFlow
-
+from aiflows.messages import FlowMessage
 
 class FixedReplyFlow(AtomicFlow):
     """ This class implements a FixedReplyFlow. It's used to reply with a fixed reply.
@@ -38,14 +38,15 @@ class FixedReplyFlow(AtomicFlow):
         super().__init__(**kwargs)
 
     def run(self,
-            input_data: Dict[str, Any]) -> Dict[str, Any]:
+            input_message: FlowMessage):
         """ Runs the FixedReplyFlow. It's used to reply with a fixed reply.
         
-        :param input_data: The input data dictionary
-        :type input_data: Dict[str, Any]
-        :return: The fixed reply
-        :rtype: Dict[str, Any]
+        :param input_message: The input message
+        :type input_message: FlowMessage
         """
-
-        return {"fixed_reply": self.flow_config["fixed_reply"]}
+        reply = self._package_output_message(
+            input_message=input_message,
+            response={"fixed_reply": self.flow_config["fixed_reply"]}
+        )
+        self.reply_to_message(reply=reply, to=input_message)
 

run.py

@@ -1,58 +1,103 @@
+"""A simple script to run a Flow that can be used for development and debugging."""
+
 import os
 
 import hydra
 
+import aiflows
 from aiflows.flow_launchers import FlowLauncher
-from aiflows.utils.general_helpers import read_yaml_file
+from aiflows.backends.api_info import ApiInfo
+from aiflows.utils.general_helpers import read_yaml_file, quick_load_api_keys
 
 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()
-logging.auto_set_dir()
+
 
 dependencies = [
-    {"url": "aiflows/FixedReplyFlowModule", "revision": os.getcwd()},
+    {"url": "aiflows/FixedReplyFlowModule", "revision": os.getcwd()}
 ]
+
 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.recursive_serve_flow(
+        cl = cl,
+        flow_type="FixedReplyFlowModule",
+        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="FixedReplyFlowModule",
+        config_overrides=None,
+        initial_state=None,
+        dispatch_point_override=None,
+    )   
+    
+    #6. ~~~ Get the data ~~~
+    data = {"id": 0}
+   
+    
+    #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, "FlowModule")
+