coflow version

LCToolFlow.py

@@ -6,6 +6,7 @@ import hydra
 from langchain.tools import BaseTool
 
 from aiflows.base_flows import AtomicFlow
+from aiflows.messages import FlowMessage
 
 
 class LCToolFlow(AtomicFlow):
@@ -81,7 +82,7 @@ class LCToolFlow(AtomicFlow):
         # ~~~ Instantiate flow ~~~
         return cls(**kwargs)
 
-    def run(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
+    def run(self, input_message: FlowMessage):
         """ This method runs the flow. It runs the backend on the input data.
         
         :param input_data: The input data of the flow.
@@ -89,7 +90,13 @@ class LCToolFlow(AtomicFlow):
         :return: The output data of the flow.
         :rtype: Dict[str, Any]
         """
+        input_data = input_message.data
         observation = self.backend.run(tool_input=input_data)
-
-        return {"observation": observation}
+        
+        reply = self._package_output_message(
+            input_message=input_message,
+            response = {"observation": observation}
+        )
+        
+        self.reply_to_message(reply=reply, to=input_message)
 

demo.yaml

@@ -1,4 +1,3 @@
-flow:
-  _target_: flow_modules.aiflows.LCToolFlowModule.LCToolFlow.instantiate_from_default_config
-  name: "demoLCToolFlow"
-  description: "An example flow that uses the LCToolFlowModule."
+_target_: flow_modules.aiflows.LCToolFlowModule.LCToolFlow.instantiate_from_default_config
+name: "demoLCToolFlow"
+description: "An example flow that uses the LCToolFlowModule."

run.py

@@ -6,12 +6,20 @@ 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, clear_cache
 
-CACHING_PARAMETERS.do_caching = False  # Set to True to enable caching
+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 in order to disable caching
 # clear_cache() # Uncomment this line to clear the cache
 
 logging.set_verbosity_debug()
@@ -23,38 +31,71 @@ 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, "query": "Michael Jordan age"}  # 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="LCToolFlowModule",
+        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="LCToolFlowModule",
+        config_overrides=None,
+        initial_state=None,
+        dispatch_point_override=None,
+    )   
+    
+
+    #6. ~~~ Get the data ~~~
+    data = {"id": 0, "query": "Obama's first name?"}  # Add your data here 
+    # 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")