update template for coflows compatibility

demo.yaml

@@ -1,8 +1,3 @@
-input_interface:  # Connector between the "input data" and the Flow
-  #your input interface here
-output_interface:  # Connector between the Flow's output and the caller
-  #your output interface here
-
-flow:  # Overrides the for the flow config
-  #_target_: Your target here
-  #your flow overrides here
+# Overrides the for the flow config
+#_target_: Your target here
+#your flow overrides here

run.py

@@ -1,59 +1,114 @@
-"""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, ApiInfo
-from aiflows.utils.general_helpers import read_yaml_file
+from aiflows.flow_launchers import FlowLauncher
+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()
 
 
+dependencies = [
+    {"url": "aiflows/FlowModule", "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()
+
 
-    # ~~~ Instantiate the Flow ~~~
+    #2. ~~~~~Load flow config~~~~~~
     root_dir = "."
     cfg_path = os.path.join(root_dir, "demo.yaml")
     cfg = read_yaml_file(cfg_path)
+    
+    #2.1 ~~~ Set the API information ~~~
+    # OpenAI backend
+    api_information = [ApiInfo(backend_used="openai",
+                              api_key = os.getenv("OPENAI_API_KEY"))]
+    # # Azure backend
+    # 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") )
+    
+    
+    quick_load_api_keys(cfg, api_information, key="api_infos")
+
+    
+    #3. ~~~~ Serve The Flow ~~~~
+    serve_utils.recursive_serve_flow(
+        cl = cl,
+        flow_type="FlowModule",
+        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)
 
-    # ~~~ 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)
-        ),
-    }
-
-    # ~~~ 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,
+    #5. ~~~~~Mount the flow and get its proxy~~~~~~
+    proxy_flow = serve_utils.recursive_mount(
+        cl=cl,
+        client_id="local",
+        flow_type="FlowModule",
+        config_overrides=None,
+        initial_state=None,
+        dispatch_point_override=None,
+    )   
+    
+    #6. ~~~ Get the data ~~~
+    data = {"id": 0}  # 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, "ChatWithDemonstrationFlowModule")
+