| 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"]} | |