CtrlExMemFlow.py

from typing import Dict, Any

from aiflows.base_flows import CircularFlow
from aiflows.utils import logging
from abc import ABC, abstractmethod


logging.set_verbosity_debug()
log = logging.get_logger(__name__)


class CtrlExMemFlow(CircularFlow, ABC):
    """This class is the controller-executor agent with memory reading and memory writing, it will execute the plan
    generated by the planner. This flow is, at a higher level, a circular flow, it runs until either max_round is
    reached, or the controller decides to early exit (see: detect_finish_or_continue)

    The brain of the flow is the controller, the controller decides what action (which branch of the branching flow)
    to take next. The controller can also decide to early exit the flow, in which case the flow will stop. After the
    controller decides what action to take, the controller will pass the action to the executor, the executor will
    execute the action, yielding `result` and `summary`, which respectively will be passed to the controller and the
    memory writer (into logs of the upper layer of flow). Depending on the `result`, the controller will decide what
    action to take next.

    *Configuration Parameters*:
    - `name` (str): Name of the flow.
    - `description` (str): Description of the flow.
    - `max_round` (int): The maximum number of rounds the flow will run. Default: 30.
    - `subflows_config` (dict): The configuration of the subflows.
        - `Controller` (dict): The configuration of the controller. It is important that the target of the controller (instance customizable by the user) and the api information should be specified.
        - `Executor` (dict): The executor of the flow, it is supposed to be a branching flow. To instantiate the executor, the user needs to specify the subflows of the executor (i.e. the actual tools that can be used by the agent, e.g. a flow for interpreting code)
        - `MemoryWriting` (dict): The configuration of the memory writer. There is an existing memory writing flow implemented.
        - 'MemoryReading' (dict): The configuration of the memory reader. There is an existing memory reading flow implemented.
    - `topology` (List): The topology of the subflows, notice that the output interface of the Controller must be implemented and specified.

    *Input Interface*:
    - `plan` (str): The plan generated by the planner, the CtrlExMem flow should follow the plan.
    - `memory_files` (dict): mem_name - memfile_path pairs, the memory files that the memory reader will read from.
    - `goal` (str): The goal of the flow
    - `logs` (str): Execution history of the flow, contains all actions taken by each subflow of the flow.

    *Output Interface*:
    - `result` (str): The result of the flow, will be returned to the controller of the caller flow.
    - `summary` (str): The summary of the flow, will be written to the logs of the caller flow.

    """
    @abstractmethod
    def _on_reach_max_round(self):
        """This function will be called when the flow reaches the maximum number of rounds."""
        pass

    @abstractmethod
    @CircularFlow.output_msg_payload_processor
    def detect_finish_or_continue(self, output_payload: Dict[str, Any], src_flow) -> Dict[str, Any]:
        """This function is called after the Controller, it is used to:
        1. Check if the Controller decides to early-exit, if so, implement necessary logics for exiting e.g. drafting result and summary, etc.
        2. For other commands called by the controller, implement necessary logics for the flow to continue e.g. providing necessary information to the branching flow.
        :param output_payload: The output payload of the Controller.
        :type output_payload: Dict[str, Any]
        :param src_flow: The source flow of the Controller.
        :type src_flow: Flow
        :return: The input payload of the Executor.
        """
        pass