|
|
|
|
|
from aiflows.base_flows import CompositeFlow |
|
from aiflows.utils import logging |
|
from aiflows.interfaces import KeyInterface |
|
logging.set_verbosity_debug() |
|
from aiflows.messages import FlowMessage |
|
log = logging.get_logger(__name__) |
|
|
|
|
|
class ChatWithDemonstrationsFlow(CompositeFlow): |
|
""" A Chat with Demonstrations Flow. It is a flow that consists of multiple sub-flows that are executed sequentially. |
|
It's parent class is SequentialFlow. |
|
|
|
It Contains the following subflows: |
|
- A Demonstration Flow: It is a flow that passes demonstrations to the ChatFlow |
|
- A Chat Flow: It is a flow that uses the demonstrations to answer queries asked by the user/human. |
|
|
|
An illustration of the flow is as follows: |
|
|
|
-------> Demonstration Flow -------> Chat Flow -------> |
|
|
|
*Configuration Parameters*: |
|
|
|
- `name` (str): The name of the flow. Default: "ChatAtomic_Flow_with_Demonstrations" |
|
- `description` (str): A description of the flow. This description is used to generate the help message of the flow. |
|
Default: "A sequential flow that answers questions with demonstrations" |
|
- `subflows_config` (Dict[str,Any]): A dictionary of subflows configurations of the sequential Flow. Default: |
|
- `Demonstration Flow`: The configuration of the Demonstration Flow. By default, it a DemonstrationsAtomicFlow. |
|
Its default parmaters are defined in DemonstrationsAtomicFlow). |
|
- `Chat Flow`: The configuration of the Chat Flow. By default, its a ChatAtomicFlow. |
|
Its default parmaters are defined in ChatAtomicFlowModule (see Flowcard, i.e. README.md, of ChatAtomicFlowModule). |
|
- `topology` (str): The topology of the flow which is "sequential". By default, the topology is the one shown in the |
|
illustration above (the topology is also described in ChatWithDemonstrationsFlow.yaml). |
|
|
|
*Input Interface*: |
|
|
|
- `query` (str): A query asked to the flow (e.g. "What is the capital of France?") |
|
|
|
Output Interface: |
|
|
|
- `answer` (str): The answer of the flow to the query |
|
|
|
:param \**kwargs: Arguments to be passed to the parent class SequentialFlow constructor. |
|
""" |
|
def __init__(self,**kwargs): |
|
super().__init__(**kwargs) |
|
self.output_interface = KeyInterface( |
|
keys_to_rename={"api_output": "answer"} |
|
) |
|
|
|
def set_up_flow_state(self): |
|
""" Set up the flow state. It sets the last state of the flow to None. |
|
""" |
|
super().set_up_flow_state() |
|
self.flow_state["last_state"] = None |
|
|
|
def run(self,input_message): |
|
""" Runs the flow. |
|
:param input_message: The input message of the flow. |
|
:type input_message: FlowMessage |
|
""" |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if self.flow_state["last_state"] is None: |
|
|
|
self.flow_state["initial_message"] = input_message |
|
|
|
msg = self.package_input_message( |
|
input_message.data |
|
) |
|
|
|
self.subflows["demonstration_flow"].get_reply( |
|
msg, |
|
) |
|
self.flow_state["last_state"] = "demonstration_flow" |
|
|
|
elif self.flow_state["last_state"] == "demonstration_flow": |
|
|
|
msg = self.package_input_message( |
|
input_message.data |
|
) |
|
|
|
self.subflows["chat_flow"].get_reply( |
|
msg, |
|
) |
|
|
|
self.flow_state["last_state"] = "chat_flow" |
|
|
|
else: |
|
self.flow_state["last_state"] = None |
|
|
|
reply = self.package_output_message( |
|
self.flow_state["initial_message"], |
|
response = self.output_interface(input_message).data |
|
) |
|
|
|
self.send_message( |
|
reply |
|
) |
|
|
|
|
|
|
