aiflows.base_flows package¶

Submodules¶

aiflows.base_flows.abstract module¶

class aiflows.base_flows.abstract.Flow(flow_config: Dict[str, Any])¶

Bases: ABC

Abstract class inherited by all Flows.

Parameters:: flow_config (Dict[str, Any]) – The configuration of the flow

REQUIRED_KEYS_CONFIG = ['name', 'description']¶

SUPPORTS_CACHING = False¶

flow_config: Dict[str, Any]¶

flow_state: Dict[str, Any]¶

classmethod get_config(**overrides)¶

Returns the default config for the flow, with the overrides applied. The default implementation construct the default config by recursively merging the configs of the base classes.

Parameters:: overrides (Dict[str, Any], optional) – The parameters to override in the default config
Returns:: The default config with the overrides applied
Return type:: Dict[str, Any]

get_interface_description()¶: Returns the input and output interface description of the flow.

history: FlowHistory¶

classmethod instantiate_from_config(config)¶

Instantiates the flow from the given config.

Parameters:: config (Dict[str, Any]) – The config to instantiate the flow from
Returns:: The instantiated flow
Return type:: aiflows.flow.Flow

classmethod instantiate_from_default_config(**overrides: Dict[str, Any] | None)¶

This method is called by the FlowLauncher to build the flow.

Parameters:: overrides (Dict[str, Any], optional) – The parameters to override in the default config
Returns:: The instantiated flow
Return type:: aiflows.flow.Flow

classmethod instantiate_with_overrides(overrides)¶

Instantiates the flow with the given overrides.

Parameters:: overrides (Dict[str, Any], optional) – The parameters to override in the default config
Returns:: The instantiated flow

property name¶

Returns the name of the flow

Returns:: The name of the flow
Return type:: str

reset(full_reset: bool, recursive: bool, src_flow: Flow | str | None = 'Launcher')¶

Reset the flow state and history. If recursive is True, reset all subflows as well.

Parameters:

full_reset – If True, remove all data in flow_state. If False, keep the data in flow_state.
recursive –
src_flow –

Returns:

run(input_data: Dict[str, Any]) → Dict[str, Any]¶

Runs the flow on the given input data. (Not implemented in the base class)

Parameters:: input_data (Dict[str, Any]) – The input data to run the flow on
Returns:: The response of the flow
Return type:: Dict[str, Any]

set_up_flow_state()¶: Sets up the flow state. This method is called when the flow is instantiated, and when the flow is reset.

classmethod type()¶

aiflows.base_flows.atomic module¶

class aiflows.base_flows.atomic.AtomicFlow(**kwargs)¶

Bases: Flow, ABC

AtomicFlow is the minimal execution unit in the Flow framework. It is an encapsulation of a single functionality that takes an input message and returns an output message.

Parameters:: **kwargs – Arguments to be passed to the Flow constructor

flow_config: Dict[str, Any]¶

flow_state: Dict[str, Any]¶

history: FlowHistory¶

classmethod type()¶: Returns the type of the flow.

aiflows.base_flows.branching module¶

class aiflows.base_flows.branching.BranchingFlow(**kwargs)¶

Bases: CompositeFlow

This class implements a branching flow. A branching flow is a composite flow that has multiple subflows. The subflow to be executed is determined by the value of the “branch” key in the input data dictionary passed to the flow.

Parameters:: **kwargs – The keyword arguments passed to the CompositeFlow constructor

run(input_data: Dict[str, Any]) → Dict[str, Any]¶

Runs the branching flow. The subflow to be executed is determined by the value of the “branch” key in the input data dictionary passed to the flow.

Parameters:: input_data (Dict[str, Any]) – The input data dictionary
Returns:: The output data dictionary
Return type:: Dict[str, Any]

subflows: Dict[str, Flow]¶

classmethod type()¶: Returns the type of the flow as a string.

aiflows.base_flows.circular module¶

class aiflows.base_flows.circular.CircularFlow(flow_config: Dict[str, Any], subflows: List[Flow])¶

Bases: CompositeFlow

This class represents a circular flow. It is a composite flow that runs its subflows in a circular fashion.

Parameters:

flow_config (Dict[str, Any]) – The flow configuration dictionary. It must usually should contain the following keys: - ‘max_rounds’ (int): The maximum number of rounds to run the circular flow - ‘early_exit_key’ (str): The key in the flow state that indicates the end of the interaction - ‘topology’ (list[Dict[str, Any]]): The topology of the circular flow (the dictionary describes the topology of one node, see TopologyNode for details) - The keys required by CompositeFlow (subflows_config)
subflows (List[aiflows.base_flows.Flow]) – A list of subflows. This is necessary when instantiating the flow programmatically.
max_rounds (int) – The maximum number of rounds to run the circular flow

Topology:

The topology of the circular flow

REQUIRED_KEYS_CONFIG = ['max_rounds', 'early_exit_key', 'topology']¶

static input_msg_payload_builder(builder_fn)¶

This decorator registers a function as an input message payload builder.

Parameters:: builder_fn (Callable) – The function to register
Returns:: The wrapped function
Return type:: Callable

static output_msg_payload_processor(processor_fn)¶

This decorator registers a function as an output message payload processor.

Parameters:: processor_fn (Callable) – The function to register
Returns:: The wrapped function
Return type:: Callable

run(input_data: Dict[str, Any]) → Dict[str, Any]¶

Runs the circular flow. It runs its subflows in a circular fashion (following the topology).

Parameters:: input_data (Dict[str, Any]) – The input data dictionary
Returns:: The output data dictionary
Return type:: Dict[str, Any]

subflows: Dict[str, Flow]¶

classmethod type()¶: Returns the type of the flow as a string.

class aiflows.base_flows.circular.TopologyNode(goal, input_interface, flow: Flow, output_interface: List[DataTransformation], reset: bool)¶

Bases: object

This class represents a node in the topology of a flows.

Parameters:

goal (str) – The goal of the node
input_interface (aiflows.interfaces.InputInterface) – The input interface of the node’s flow
flow (aiflows.base_flows.Flow) – The flow of the node
output_interface (List[aiflows.data_transformations.DataTransformation]) – The output interface of the node’s flow
reset (bool) – Whether to reset the node’s flow

aiflows.base_flows.composite module¶

class aiflows.base_flows.composite.CompositeFlow(flow_config: Dict[str, Any], subflows: List[Flow])¶

Bases: Flow, ABC

This class implements a composite flow. It is a flow that consists of multiple sub-flows. It is the a parent class for BranchingFlow, SequentialFlow and CircularFlow. Note that the run method of a CompositeFlow is not implemented.

Parameters:

flow_config (Dict[str, Any]) – The configuration of the flow. It must usually contain the following keys: - “subflows_config” (Dict[str,Any]): A dictionary of subflows configurations.The keys are the names of the subflows and the values are the configurations of the subflows. This is necessary when instantiating the flow from a config file. - The parameters required by the constructor of the parent class Flow
subflows (List[Flow]) – A list of subflows. This is necessary when instantiating the flow programmatically.

REQUIRED_KEYS_CONFIG = ['subflows_config']¶

classmethod instantiate_from_config(config)¶

Instantiates the flow from a config file.

Parameters:: config – The configuration of the flow. It must usually contain the following keys: - “subflows_config” (Dict[str,Any]): A dictionary of subflows configurations.The keys are the names of the subflows and the values are the configurations of the subflows. This is necessary when instantiating the flow from a config file. - The parameters required by the constructor of the parent class Flow

subflows: Dict[str, Flow]¶

classmethod type()¶: Returns the type of the flow as a string.

aiflows.base_flows.sequential module¶

class aiflows.base_flows.sequential.SequentialFlow(flow_config: Dict[str, Any], subflows: List[Flow])¶

Bases: CircularFlow

This class implements a sequential flow. It is a flow that consists of multiple sub-flows that are executed sequentially. It is a child class of CircularFlow. The only difference between a SequentialFlow and a CircularFlow is that the SequentialFlow has a max_rounds of 1.

Parameters:

flow_config (Dict[str, Any]) – The configuration of the flow. It must usually contain the following keys: - “subflows_config” (Dict[str,Any]): A dictionary of subflows configurations.The keys are the names of the subflows and the values are the configurations of the subflows. This is necessary when instantiating the flow from a config file. - The parameters required by the constructor of the parent class Flow
subflows (List[Flow]) – A list of subflows. This is necessary when instantiating the flow programmatically.

flow_config: Dict[str, Any]¶

flow_state: Dict[str, Any]¶

history: FlowHistory¶

subflows: Dict[str, Flow]¶

classmethod type()¶: Returns the type of the flow.

Module contents¶

Contains basic flow classes that can be used to build more complex flows.

AtomicFlow is the minimal execution unit in the Flow framework. CompositeFlow is a flow that contains subflows and defines how they are executed.