coflows compatible

__init__.py

@@ -1,6 +1,6 @@
 # ~~~ Specify the dependencies ~~
 dependencies = [
-    {"url": "aiflows/ChatFlowModule", "revision": "main"}
+    {"url": "aiflows/ChatFlowModule", "revision": "coflows"}
 ]
 from aiflows import flow_verse
 flow_verse.sync_dependencies(dependencies)

demo.yaml

@@ -1,20 +1,19 @@
-flow:
-  _target_: flow_modules.aiflows.VisionFlowModule.VisionAtomicFlow.instantiate_from_default_config
-  name: "Demo Vision Flow"
-  description: "A flow that, given a textual input, and a set of images and/or videos, generates a textual output."
-  backend:
-    api_infos: ???
-  
-  system_message_prompt_template:
-    template: |2-
-      You are a helpful chatbot that truthfully answers questions.
-    input_variables: []
-    partial_variables: {}
-  
-  init_human_message_prompt_template:
-    template: |2-
-      {{query}}
-    input_variables: ["query"]
-    partial_variables: {}
+_target_: flow_modules.aiflows.VisionFlowModule.VisionAtomicFlow.instantiate_from_default_config
+name: "Demo Vision Flow"
+description: "A flow that, given a textual input, and a set of images and/or videos, generates a textual output."
+backend:
+  api_infos: ???
+
+system_message_prompt_template:
+  template: |2-
+    You are a helpful chatbot that truthfully answers questions.
+  input_variables: []
+  partial_variables: {}
+
+init_human_message_prompt_template:
+  template: |2-
+    {{query}}
+  input_variables: ["query"]
+  partial_variables: {}
   
   

run.py

@@ -2,59 +2,83 @@ 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.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
+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()  # Uncomment this line to see verbose logs
+logging.set_verbosity_debug()
 
-from aiflows import flow_verse
 
 dependencies = [
-    {"url": "aiflows/VisionFlowModule", "revision": os.getcwd()},
+    {"url": "aiflows/VisionFlowModule", "revision": os.getcwd()}
 ]
-flow_verse.sync_dependencies(dependencies)
 
+from aiflows import flow_verse
+flow_verse.sync_dependencies(dependencies)
 if __name__ == "__main__":
-    # ~~~ Set the API information ~~~
-    # OpenAI backend
+    
+    #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)
+    
+    #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")
 
-    root_dir = "."
-    cfg_path = os.path.join(root_dir, "demo.yaml")
-    cfg = read_yaml_file(cfg_path)
     
-    cfg["flow"]["backend"]["api_infos"] = api_information
-    
-    # ~~~ 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)
-        ),
-    }
+    #3. ~~~~ Serve The Flow ~~~~
+    serve_utils.recursive_serve_flow(
+        cl = cl,
+        flow_type="VisionFlowModule",
+        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="VisionFlowModule",
+        config_overrides=None,
+        initial_state=None,
+        dispatch_point_override=None,
+    )   
+    
+    #6. ~~~ Get the data ~~~
     url_image = {"type": "url",
                  "image": "https://upload.wikimedia.org/wikipedia/commons/thumb/d/dd/Gfp-wisconsin-madison-the-nature-boardwalk.jpg/2560px-Gfp-wisconsin-madison-the-nature-boardwalk.jpg"}
     
@@ -75,17 +99,33 @@ if __name__ == "__main__":
     #         "question": "These are frames from a video that I want to upload. Generate a compelling description that I can upload along with the video.",
     #         "data": {"video": video}}  # This can be a list of samples
     
-
-    # ~~~ 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,
+    #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, "VisionFlowModule")    o_caching = False  # Set to True to enable caching
+