from flows.base_flows import CircularFlow from flows.utils import logging log = logging.get_logger(f"flows.{__name__}") class ChatHumanFlowModule(CircularFlow): """ This class implements a Chat Human Flow Module. It is a flow that consists of two sub-flows that are executed circularly. It Contains the following subflows: - A User Flow: A flow makes queries to the Assistant Flow. E.g. The user asks the assistant (LLM) a question. - A Assistant Flow: A flow that responds to queries made by the User Flow. E.g. The assistant (LLM) answers the user's question. To end the interaction, the user must type "\<END\>" An illustration of the flow is as follows: |------> User Flow -----------> | ^ | | | | v |<------ Assistant Flow <-------| *Configuration Parameters*: - `name` (str): The name of the flow. Default: "ChatHumanFlowModule" - `description` (str): A description of the flow. This description is used to generate the help message of the flow. Default: "Flow that enables chatting between a ChatAtomicFlow and a user providing the input." - `max_rounds` (int): The maximum number of rounds the flow can run for. Default: None, which means that there is no limit on the number of rounds. - `early_exit_key` (str): The key that is used to exit the flow. Default: "end_of_interaction" - `subflows_config` (Dict[str,Any]): A dictionary of subflows configurations. Default: - `Assistant Flow`: The configuration of the Assistant Flow. By default, it a ChatAtomicFlow. It default parmaters are defined in ChatAtomicFlowModule. - `User Flow`: The configuration of the User Flow. By default, it a HumanStandardInputFlow. It default parmaters are defined in HumanStandardInputFlowModule. - `topology` (str): (List[Dict[str,Any]]): The topology of the flow which is "circular". By default, the topology is the one shown in the illustration above (the topology is also described in ChatHumanFlowModule.yaml). *Input Interface*: - None. By default, the input interface doesn't expect any input. *Output Interface*: - `end_of_interaction` (bool): Whether the interaction is finished or not. :param \**kwargs: Arguments to be passed to the parent class CircularFlow constructor. :type \**kwargs: Dict[str, Any] """ def __init__(self, **kwargs): super().__init__(**kwargs) @classmethod def type(cls): """ This method returns the type of the flow.""" return "OpenAIChatHumanFlowModule"