coflows-compatible

AutoGPTFlow.py

@@ -1,17 +1,15 @@
 import sys
 from typing import Dict, Any
 
-from aiflows.base_flows import CircularFlow
 from aiflows.utils import logging
 
 logging.set_verbosity_debug()
 
 log = logging.get_logger(__name__)
+from aiflows.interfaces import KeyInterface
+from flow_modules.aiflows.ControllerExecutorFlowModule import ControllerExecutorFlow
 
-from flow_modules.aiflows.ControllerExecutorFlowModule import ControllerAtomicFlow
-from flow_modules.aiflows.VectorStoreFlowModule import ChromaDBFlow
-
-class AutoGPTFlow(CircularFlow):
+class AutoGPTFlow(ControllerExecutorFlow):
     """ This class implements a (very basic) AutoGPT flow. It is a flow that consists of multiple sub-flows that are executed circularly. It Contains the following subflows:
     
     - A Controller Flow: A Flow that controls which subflow of the Executor Flow to execute next.
@@ -71,15 +69,136 @@ class AutoGPTFlow(CircularFlow):
     :param subflows: A list of subflows constituating the circular flow. Required when instantiating the subflow programmatically (it replaces subflows_config from flow_config).
     :type subflows: List[Flow]
     """
-    def _on_reach_max_round(self):
-        """ This method is called when the flow reaches the max_rounds."""
-        self._state_update_dict({
-            "answer": "The maximum amount of rounds was reached before the model found an answer.",
+    
+    
+    def __init__(self, **kwargs):
+        super().__init__( **kwargs)
+        self.rename_human_output_interface = KeyInterface(
+            keys_to_rename={"human_input": "human_feedback"}
+        )
+    
+    def set_up_flow_state(self):
+        super().set_up_flow_state()
+        self.flow_state["early_exit_flag"] = True
+        self.flow_state["is_first_round"] = True
+        
+    def memory_read_step(self):
+        memory_read_input = self.prepare_memory_read_input()
+        output = self.ask_subflow("Memory", memory_read_input).get_data()
+        memory_read_output = self.prepare_memory_read_output(output)
+        
+        return memory_read_output
+    
+    def memory_write_step(self):
+        memory_write_input = self.prepare_memory_write_input()
+        self.tell_subflow("Memory", memory_write_input)
+    
+    def controller_executor_step(self, output_memory_retrieval):
+        
+        if self.flow_state["is_first_round"]:
+            additional_input_ctrl_ex = {
+                "goal": self.flow_state["goal"],
+            }
+        else:
+            additional_input_ctrl_ex = {
+                "observation": self.flow_state["observation"],
+                "human_feedback": self.flow_state["human_feedback"],
+            }
+        
+        input_ctrl_ex =  {"executor_reply": {**output_memory_retrieval,**additional_input_ctrl_ex}}
+        
+        output_ctrl_ex = self._single_round_controller_executor(input_ctrl_ex)
+        
+        self.flow_state["early_exit_flag"] = output_ctrl_ex.get("EARLY_EXIT",False)
+        
+        if self.flow_state["early_exit_flag"]:
+            return output_ctrl_ex
+        
+        controller_reply = output_ctrl_ex["controller_reply"]
+        executor_reply = output_ctrl_ex["executor_reply"]
+        
+        self._state_update_dict(
+            {
+                "command": controller_reply["command"],
+                "command_args": controller_reply["command_args"],
+                "observation": executor_reply["observation"],
+            }
+        )
+        return output_ctrl_ex
+    
+    
+    def human_feedback_step(self):
+            
+        human_feedback_input_variables = {
+            "goal": self.flow_state["goal"],
+            "command": self.flow_state["command"],
+            "command_args": self.flow_state["command_args"],
+            "observation": self.flow_state["observation"],
+        }
+        
+        human_feedback = self.rename_human_output_interface(
+            self.ask_subflow("HumanFeedback", human_feedback_input_variables).get_data()
+        )
+        
+        self.flow_state["human_feedback"] = human_feedback["human_feedback"]
+        
+        if human_feedback["human_feedback"].strip().lower() == "q":
+            self.flow_state["early_exit_flag"] = True
+            return {
+                "EARLY_EXIT": True,
+                "answer": "The user has chosen to exit before a final answer was generated.",
+                "status": "unfinished",
+            }
+        
+        return human_feedback
+        
+    
+    def _single_round_autogpt(self):
+        
+        
+        #1. Memory Retrieval
+        output_memory_retrieval = self.memory_read_step()
+        
+        
+        #2. ControllerExecutor
+        output_ctrl_ex = self.controller_executor_step(output_memory_retrieval)
+        
+        if self.flow_state["early_exit_flag"]:
+            return output_ctrl_ex
+        
+        #3. HumanFeedback
+        output_human_feedback = self.human_feedback_step()
+        
+        if self.flow_state["early_exit_flag"]:
+            return output_human_feedback
+        
+        #4. Memory Write
+        self.memory_write_step()
+        
+        return {** output_ctrl_ex, **output_human_feedback}
+        
+        
+        
+        
+    
+    def run(self,input_data):
+        self._state_update_dict({"goal": input_data["goal"]})
+        
+        for round in range(self.flow_config["max_rounds"]):
+            output = self._single_round_autogpt()
+            
+            self.flow_state["is_first_round"] = False
+            
+            if self.flow_state["early_exit_flag"]:
+                return output
+        
+        return {
+            "EARLY_EXIT": False,
+            "answer": output,
             "status": "unfinished"
-        })
-
-    @staticmethod
-    def _get_memory_key(flow_state):
+        }
+        
+    def _get_memory_key(self):
         """ This method returns the memory key that is used to retrieve memories from the ChromaDB model.
         
         :param flow_state: The state of the flow
@@ -87,11 +206,11 @@ class AutoGPTFlow(CircularFlow):
         :return: The current context
         :rtype: str
         """
-        goal = flow_state.get("goal")
-        last_command = flow_state.get("command")
-        last_command_args = flow_state.get("command_args")
-        last_observation = flow_state.get("observation")
-        last_human_feedback = flow_state.get("human_feedback")
+        goal = self.flow_state.get("goal")
+        last_command = self.flow_state.get("command")
+        last_command_args = self.flow_state.get("command_args")
+        last_observation = self.flow_state.get("observation")
+        last_human_feedback = self.flow_state.get("human_feedback")
 
         if last_command is None:
             return ""
@@ -118,99 +237,25 @@ class AutoGPTFlow(CircularFlow):
 
         return current_context
 
-    @CircularFlow.input_msg_payload_builder
-    def prepare_memory_read_input(self, flow_state: Dict[str, Any], dst_flow: ChromaDBFlow) -> Dict[str, Any]:
-        """ This method prepares the input for the Memory Flow. It is called before the Memory Flow is called.
-            A (very) basic example implementation of how the memory retrieval could be constructed.
-            
-        :param flow_state: The state of the flow
-        :type flow_state: Dict[str, Any]
-        :param dst_flow: The destination flow
-        :type dst_flow: Flow
-        :return: The input message for the Memory Flow
-        :rtype: Dict[str, Any]
-        """
-        query = self._get_memory_key(flow_state)
+    def prepare_memory_read_input(self) -> Dict[str, Any]:
+    
+        query = self._get_memory_key()
 
         return {
             "operation": "read",
             "content": query
         }
 
-    @CircularFlow.output_msg_payload_processor
-    def prepare_memory_read_output(self, output_payload: Dict[str, Any], src_flow: ControllerAtomicFlow):
-        """ This method processes the output of the Memory Flow. It is called after the Memory Flow is called.
-        
-        :param output_payload: The output payload of the Memory Flow
-        :type output_payload: Dict[str, Any]
-        :param src_flow: The source flow
-        :type src_flow: Flow
-        :return: The processed output payload
-        :rtype: Dict[str, Any]
-        """
-        retrieved_memories = output_payload["retrieved"][0][1:]
+    def prepare_memory_read_output(self, data: Dict[str, Any]):
+    
+        retrieved_memories = data["retrieved"][0][1:]
         return {"memory": "\n".join(retrieved_memories)}
 
-    @CircularFlow.input_msg_payload_builder
-    def prepare_memory_write_input(self, flow_state: Dict[str, Any], dst_flow: ChromaDBFlow) -> Dict[str, Any]:
-        """ This method prepares the input for the Memory Flow. It is called before the Memory Flow is called.
-            A (very) basic example implementation of how the memory population could be constructed.
-            
-        :param flow_state: The state of the flow
-        :type flow_state: Dict[str, Any]
-        :param dst_flow: The destination flow
-        :type dst_flow: Flow
-        :return: The input message to write the Memory Flow
-        :rtype: Dict[str, Any]
-        """""
-        query = self._get_memory_key(flow_state)
+    def prepare_memory_write_input(self) -> Dict[str, Any]:
+
+        query = self._get_memory_key()
 
         return {
             "operation": "write",
             "content": str(query)
         }
-
-    @CircularFlow.output_msg_payload_processor
-    def detect_finish_or_continue(self, output_payload: Dict[str, Any], src_flow: ControllerAtomicFlow) -> Dict[
-        str, Any]:
-        """ This method detects whether the Controller flow has generated a "finish" command or not to terminate the flow. . It is called after the Controller Flow is called.
-        
-        :param output_payload: The output payload of the Controller Flow
-        :type output_payload: Dict[str, Any]
-        :param src_flow: The source flow
-        :type src_flow: Flow
-        :return: The processed output payload
-        :rtype: Dict[str, Any]
-        """
-        command = output_payload["command"]
-        if command == "finish":
-            return {
-                "EARLY_EXIT": True,
-                "answer": output_payload["command_args"]["answer"],
-                "status": "finished"
-            }
-        else:
-            return output_payload
-
-    @CircularFlow.output_msg_payload_processor
-    def detect_finish_in_human_input(self, output_payload: Dict[str, Any], src_flow: ControllerAtomicFlow) -> Dict[
-        str, Any]:
-        """ This method detects whether the HumanFeedback (the human/user) flow has generated a "finish" command or not to terminate the flow. It is called after the HumanFeedback Flow is called.
-        
-        :param output_payload: The output payload of the HumanFeedback Flow
-        :type output_payload: Dict[str, Any]
-        :param src_flow: The source flow
-        :type src_flow: Flow
-        :return: The processed output payload
-        :rtype: Dict[str, Any]
-        """
-        
-        human_feedback = output_payload["human_input"]
-        if human_feedback.strip().lower() == "q":
-            return {
-                "EARLY_EXIT": True,
-                "answer": "The user has chosen to exit before a final answer was generated.",
-                "status": "unfinished"
-            }
-
-        return {"human_feedback": human_feedback}

AutoGPTFlow.yaml

@@ -43,18 +43,18 @@ subflows_config:
       - "human_feedback"
       - "memory"
 
-  Executor:
-    _target_: aiflows.base_flows.BranchingFlow.instantiate_from_default_config
-# E.g.,
-#    subflows_config:
-#      wiki_search:
-#        _target_: .WikiSearchAtomicFlow.instantiate_from_default_config
-#      ddg_search:
-#        _target_: flows.application_flows.LCToolFlowModule.LCToolFlow.instantiate_from_default_config
-#        backend:
-#          _target_: langchain.tools.DuckDuckGoSearchRun
+  
+  
+  # wiki_search:
+  #   _target_: .WikiSearchAtomicFlow.instantiate_from_default_config
+  # ddg_search:
+  #   _target_: flows.application_flows.LCToolFlowModule.LCToolFlow.instantiate_from_default_config
+  #   backend:
+  #     _target_: langchain.tools.DuckDuckGoSearchRun
 
   HumanFeedback:
+    name: HumanFeedbackFlow
+    description: "A flow that requests feedback from a human."
     _target_: flow_modules.aiflows.HumanStandardInputFlowModule.HumanStandardInputFlow.instantiate_from_default_config
     request_multi_line_input_flag: False
     query_message_prompt_template:
@@ -88,50 +88,3 @@ subflows_config:
     _target_: flow_modules.aiflows.VectorStoreFlowModule.ChromaDBFlow.instantiate_from_default_config
     n_results: 2
 
-topology:
-  - goal: "Retrieve relevant information from memory."
-    input_interface:
-      _target_: AutoGPTFlow.prepare_memory_read_input # An interface defined as a function in the Flow class
-    flow: Memory
-    output_interface:
-      _target_: AutoGPTFlow.prepare_memory_read_output
-    reset: false
-
-  - goal: "Select the next action and prepare the input for the executor."
-    input_interface:
-      _target_: aiflows.interfaces.KeyInterface
-      additional_transformations:
-        - _target_: aiflows.data_transformations.KeyMatchInput
-    flow: Controller
-    output_interface:
-      _target_: AutoGPTFlow.detect_finish_or_continue # An interface defined as a function in the Flow class
-    reset: false
-
-  - goal: "Execute the action specified by the Controller."
-    input_interface:
-      _target_: aiflows.interfaces.KeyInterface
-      keys_to_rename:
-        command: branch
-        command_args: branch_input_data
-      keys_to_select: ["branch", "branch_input_data"]
-    flow: Executor
-    output_interface:
-      _target_: aiflows.interfaces.KeyInterface
-      keys_to_rename:
-        branch_output_data: observation
-      keys_to_select: ["observation"]
-    reset: false
-
-  - goal: "Ask the user for feedback."
-    input_interface:
-      _target_: aiflows.interfaces.KeyInterface
-    flow: HumanFeedback
-    output_interface:
-      _target_: AutoGPTFlow.detect_finish_in_human_input
-    reset: false
-
-  - goal: "Write relevant information to memory"
-    input_interface:
-      _target_: AutoGPTFlow.prepare_memory_write_input  # An interface defined as a function in the Flow class
-    flow: Memory
-    reset: false

__init__.py

@@ -1,7 +1,7 @@
 # ~~~ Specify the dependencies ~~~
 dependencies = [
     {"url": "aiflows/ControllerExecutorFlowModule",
-     "revision": "main"},
+     "revision": "coflows"},
     {"url": "aiflows/HumanStandardInputFlowModule",
      "revision": "main"},
     {"url": "aiflows/VectorStoreFlowModule",

demo.yaml

@@ -1,41 +1,45 @@
-flow:
-  _target_: flow_modules.aiflows.AutoGPTFlowModule.AutoGPTFlow.instantiate_from_default_config
-  max_rounds: 30
+_target_: flow_modules.aiflows.AutoGPTFlowModule.AutoGPTFlow.instantiate_from_default_config
+max_rounds: 30
 
-  ### Subflows specification
-  subflows_config:
-    Controller:
-      _target_: flow_modules.aiflows.ControllerExecutorFlowModule.ControllerAtomicFlow.instantiate_from_default_config
-      commands:
-        wiki_search:
-          description: "Performs a search on Wikipedia."
-          input_args: [ "search_term" ]
-        finish:
-          description: "Signal that the objective has been satisfied, and returns the answer to the user."
-          input_args: [ "answer" ]
-      backend:
-        api_infos: ???
-      human_message_prompt_template:
-        template: |2-
-          Here is the response to your last action:
-          {{observation}}
-          Here is the feedback from the user:
-          {{human_feedback}}
-        input_variables:
-          - "observation"
-          - "human_feedback"
-      input_interface_initialized:
+### Subflows specification
+subflows_config:
+  #ControllerFlow Configuration
+  Controller:
+    _target_: flow_modules.aiflows.ControllerExecutorFlowModule.ControllerAtomicFlow.instantiate_from_default_config
+    commands:
+      wiki_search:
+        description: "Performs a search on Wikipedia."
+        input_args: ["search_term"]
+      finish:
+        description: "Signal that the objective has been satisfied, and returns the answer to the user."
+        input_args: ["answer"]
+    backend:
+      api_infos: ???
+    human_message_prompt_template:
+      template: |2-
+        Here is the response to your last action:
+        {{observation}}
+        Here is the feedback from the user:
+        {{human_feedback}}
+      input_variables:
         - "observation"
         - "human_feedback"
+    input_interface_initialized:
+      - "observation"
+      - "human_feedback"
 
-    Executor:
-      _target_: aiflows.base_flows.BranchingFlow.instantiate_from_default_config
-      subflows_config:
-        wiki_search:
-          _target_: flow_modules.aiflows.ControllerExecutorFlowModule.WikiSearchAtomicFlow.instantiate_from_default_config
+    previous_messages:
+      last_k: 1
+      first_k: 2
 
 
-    Memory:
-      backend:
-        api_infos: ???
+  wiki_search:
+    _target_: flow_modules.aiflows.ControllerExecutorFlowModule.WikiSearchAtomicFlow.instantiate_from_default_config
+
+
+  #MemoryFlow Configuration
+  Memory:
+    backend:
+      model_name: none
+      api_infos: ???
 

run.py

@@ -5,72 +5,95 @@ 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
 
 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()
 
+from aiflows import flow_verse
+# ~~~ Load Flow dependecies from FlowVerse ~~~
 dependencies = [
     {"url": "aiflows/AutoGPTFlowModule", "revision": os.getcwd()},
-    {"url": "aiflows/LCToolFlowModule", "revision": "main"},
+    {"url": "aiflows/LCToolFlowModule", "revision": "80c0c76181d90846ebff1057b8951d9689f93b62"},
 ]
-from aiflows import flow_verse
 
 flow_verse.sync_dependencies(dependencies)
 if __name__ == "__main__":
     # ~~~ Set the API information ~~~
     # OpenAI backend
-    api_information = [ApiInfo(backend_used="openai",
-                              api_key = os.getenv("OPENAI_API_KEY"))]
+    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") )
+    
+    FLOW_MODULES_PATH = "./"
+    
+    jwt = os.getenv("COLINK_JWT")
+    addr = os.getenv("LOCAL_COLINK_ADDRESS")
+    
+    cl = serve_utils.start_colink_component(
+        "Reverse Number Demo",
+        {"jwt": jwt, "addr": addr}
+    )
 
     root_dir = "."
     cfg_path = os.path.join(root_dir, "demo.yaml")
     cfg = read_yaml_file(cfg_path)
-    cfg["flow"]["subflows_config"]["Controller"]["backend"]["api_infos"] = api_information
-    cfg["flow"]["subflows_config"]["Memory"]["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)
-        ),
-    }
+    
+    serve_utils.recursive_serve_flow(
+        cl = cl,
+        flow_type="demo_served",
+        default_config=cfg,
+        default_state=None,
+        default_dispatch_point="coflows_dispatch",
+    )
+    
+    #in case you haven't started the dispatch worker thread, uncomment the 2 lines
+    #run_dispatch_worker_thread(cl, dispatch_point="coflows_dispatch", flow_modules_base_path=FLOW_MODULES_PATH)
+    #run_dispatch_worker_thread(cl, dispatch_point="coflows_dispatch", flow_modules_base_path=FLOW_MODULES_PATH)
+
+    quick_load_api_keys(cfg, api_information, key="api_infos")
+
 
     # ~~~ Get the data ~~~
     # data = {"id": 0, "goal": "Answer the following question: What is the population of Canada?"}  # Uses wikipedia
     # data = {"id": 0, "goal": "Answer the following question: Who was the NBA champion in 2023?"}  # Uses duckduckgo
-    data = {"id": 0, "goal": "Answer the following question: What is the profession and date of birth of Michael Jordan?"}
+    data = {
+        "id": 0,
+        "goal": "Answer the following question: What is the profession and date of birth of Michael Jordan?",
+    }
     # At first, we retrieve information about Michael Jordan the basketball player
     # If we provide feedback, only in the first round, that we are not interested in the basketball player,
     #   but the statistician, and skip the feedback in the next rounds, we get the correct answer
 
     # ~~~ 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,
-        data=data,
-        path_to_output_file=path_to_output_file,
-    )
-
+    proxy_flow = serve_utils.recursive_mount(
+        cl=cl,
+        client_id="local",
+        flow_type="demo_served",
+        config_overrides=cfg,
+        initial_state=None,
+        dispatch_point_override=None,
+    )   
     # ~~~ Print the output ~~~
-    flow_output_data = outputs[0]
-    print(flow_output_data)
+    input_message = FlowMessage(
+        data= data,
+        src_flow="Coflows team",
+        dst_flow=proxy_flow,
+        is_input_msg=True
+    )
+    
+    future = proxy_flow.ask(input_message)
+    
+    print(future.get_data())