from typing import Dict, Any from aiflows.base_flows import AtomicFlow class FixedReplyFlow(AtomicFlow): """ This class implements a FixedReplyFlow. It's used to reply with a fixed reply. *Configuration Parameters*: - `name` (str): The name of the flow. - `description` (str): A description of the flow. This description is used to generate the help message of the flow. - `fixed_reply` (str): The fixed reply to reply with. - The other configuration parameters are inherited from the default configuration of AtomicFlow (see AtomicFlow) *Input Interface*: - None Output Interface: - `fixed_reply` (str): The fixed reply. :param \**kwargs: The keyword arguments passed to the AtomicFlow constructor. Among these is the flow_config which should also contain the "fixed_reply" key. :type \**kwargs: Dict[str, Any] """ REQUIRED_KEYS_CONFIG = ["fixed_reply"] __default_flow_config = { "input_interface": [], "output_interface": ["fixed_reply"], } def __init__(self, **kwargs): super().__init__(**kwargs) def run(self, input_data: Dict[str, Any]) -> Dict[str, Any]: """ Runs the FixedReplyFlow. It's used to reply with a fixed reply. :param input_data: The input data dictionary :type input_data: Dict[str, Any] :return: The fixed reply :rtype: Dict[str, Any] """ return {"fixed_reply": self.flow_config["fixed_reply"]}