---
license: mit
---
# Table of Contents

* [run](#run)
* [ControllerAtomicFlow](#ControllerAtomicFlow)
  * [Command](#ControllerAtomicFlow.Command)
  * [ControllerAtomicFlow](#ControllerAtomicFlow.ControllerAtomicFlow)
    * [instantiate\_from\_config](#ControllerAtomicFlow.ControllerAtomicFlow.instantiate_from_config)
    * [run](#ControllerAtomicFlow.ControllerAtomicFlow.run)
* [\_\_init\_\_](#__init__)
* [WikiSearchAtomicFlow](#WikiSearchAtomicFlow)
  * [WikiSearchAtomicFlow](#WikiSearchAtomicFlow.WikiSearchAtomicFlow)
    * [run](#WikiSearchAtomicFlow.WikiSearchAtomicFlow.run)
* [wikipediaAPI](#wikipediaAPI)
  * [WikipediaAPIWrapper](#wikipediaAPI.WikipediaAPIWrapper)
    * [validate\_environment](#wikipediaAPI.WikipediaAPIWrapper.validate_environment)
    * [run](#wikipediaAPI.WikipediaAPIWrapper.run)
    * [search\_page\_titles](#wikipediaAPI.WikipediaAPIWrapper.search_page_titles)
* [ControllerExecutorFlow](#ControllerExecutorFlow)
  * [ControllerExecutorFlow](#ControllerExecutorFlow.ControllerExecutorFlow)
    * [generate\_reply](#ControllerExecutorFlow.ControllerExecutorFlow.generate_reply)
    * [get\_next\_state](#ControllerExecutorFlow.ControllerExecutorFlow.get_next_state)
    * [set\_up\_flow\_state](#ControllerExecutorFlow.ControllerExecutorFlow.set_up_flow_state)
    * [call\_controller](#ControllerExecutorFlow.ControllerExecutorFlow.call_controller)
    * [call\_executor](#ControllerExecutorFlow.ControllerExecutorFlow.call_executor)
    * [register\_data\_to\_state](#ControllerExecutorFlow.ControllerExecutorFlow.register_data_to_state)
    * [run](#ControllerExecutorFlow.ControllerExecutorFlow.run)

<a id="run"></a>

# run

A simple script to run a Flow that can be used for development and debugging.

<a id="ControllerAtomicFlow"></a>

# ControllerAtomicFlow

<a id="ControllerAtomicFlow.Command"></a>

## Command Objects

```python
@dataclass
class Command()
```

The command class is used to store the information about the commands that the user can give to the controller.

**Arguments**:

- `name` (`str`): The name of the command.
- `description` (`str`): The description of the command.
- `input_args` (`List[str]`): The input arguments of the command.

<a id="ControllerAtomicFlow.ControllerAtomicFlow"></a>

## ControllerAtomicFlow Objects

```python
class ControllerAtomicFlow(ChatAtomicFlow)
```

The ControllerAtomicFlow is an atomic flow that, given an observation and a goal, can call a set of commands and arguments which are then usually executed by an ExecutorAtomicFlow (branching flow).

*Configuration Parameters*

- `name` (str): The name of the flow. Default: "ControllerFlow"
- `description` (str): A description of the flow. This description is used to generate the help message of the flow.
Default: "Proposes the next action to take towards achieving the goal, and prepares the input for the executor."
- `enable_cache` (bool): Whether to enable caching or not. Default: True
- `commands` (List[Dict[str,Any]]): A list of commands that the controller can call. Default: []
- `finish` (Dict[str,Any]): The configuration of the finish command. Default parameters: No default parameters.
- `system_message_prompt_template` (Dict[str, Any]): The prompt template used to generate the system message.
By default, it's type is aiflows.prompt_template.JinjaPrompt. It's default parameters are:
    - `template` (str): The template of the prompt. Default: see ControllerAtomicFlow.yaml for the default template.
    - `input_variables` (List[str]): The input variables of the prompt. Default: ["commands"]. Note that the commands are the commands of the executor 
    (subflows of branching flow) and are actually to the system prompt template via the `_build_commands_manual` function of this class.
- `human_message_prompt_template` (Dict[str, Any]): The prompt template of the human/user message (message used everytime the except the first time in).
It's passed as the user message to the LLM. By default its of type aiflows.prompt_template.JinjaPrompt and has the following parameters:
    - `template` (str): The template of the prompt. Default: see ControllerAtomicFlow.yaml for the default template.
    - `input_variables` (List[str]): The input variables of the prompt. Default: ["observation"]
- init_human_message_prompt_template` (Dict[str, Any]):  The prompt template of the human/user message used to initialize the conversation
(first time in). It is used to generate the human message. It's passed as the user message to the LLM.
By default its of type aiflows.prompt_template.JinjaPrompt and has the following parameters:
    - `template` (str): The template of the prompt. Default: see ControllerAtomicFlow.yaml for the default template.
    - `input_variables` (List[str]): The input variables of the prompt. Default: ["goal"]
- All other parameters are inherited from the default configuration of ChatAtomicFlow (see Flowcard, i.e. README.md, of ChatAtomicFlowModule).

*Initial Input Interface (this is the interface used the first time the flow is called)*:
- `goal` (str): The goal of the controller. Usually asked by the user/human (e.g. "I want to know the occupation and birth date of Michael Jordan.")

*Input Interface (this is the interface used after the first time the flow is called)*:
- `observation` (str): The observation of the controller's previous action. Usually the response of the ExecutorAtomicFlow (e.g. "The result of a wikipedia search (if the ExecutorAtomicFlow has a WikipediaExecutorAtomicFlow).")

*Output Interface:*
- `thought` (str): The thought of the controller on what to do next (which command to call)
- `reasoning` (str): The reasoning of the controller on why it thinks the command it wants to call is the right one
- `criticism` (str): The criticism of the controller of it's thinking process
- `command` (str): The command to the executor chooses to call
- `command_args` (Dict[str, Any]): The arguments of the command to call

**Arguments**:

- `commands` (`List[Command]`): The commands that the controller can call (typically the commands of the executor).
- `\**kwargs` (`Dict[str, Any]`): The parameters specific to the ChatAtomicFlow.

<a id="ControllerAtomicFlow.ControllerAtomicFlow.instantiate_from_config"></a>

#### instantiate\_from\_config

```python
@classmethod
def instantiate_from_config(cls, config)
```

This method instantiates the flow from a configuration file.

**Arguments**:

- `config` (`Dict[str, Any]`): The configuration of the flow.

**Returns**:

`ControllerAtomicFlow`: The instantiated flow.

<a id="ControllerAtomicFlow.ControllerAtomicFlow.run"></a>

#### run

```python
def run(input_message: FlowMessage)
```

This method runs the flow. Note that the response of the LLM is in the JSON format, but it's not a hard constraint (it can hallucinate and return an invalid JSON)

**Arguments**:

- `input_message` (`FlowMessage`): The input data of the flow.

<a id="__init__"></a>

# \_\_init\_\_

<a id="WikiSearchAtomicFlow"></a>

# WikiSearchAtomicFlow

<a id="WikiSearchAtomicFlow.WikiSearchAtomicFlow"></a>

## WikiSearchAtomicFlow Objects

```python
class WikiSearchAtomicFlow(AtomicFlow)
```

This class implements a WikiSearch Atomic Flow. It's used to execute a Wikipedia search and get page summaries.

*Configuration Parameters*:

- `name` (str): The name of the flow. Default: "WikiSearchAtomicFlow"
- `description` (str): A description of the flow. This description is used to generate the help message of the flow.
Default: "A Flow that queries the wikipedia API for a page content."
- `lang` (str): The language of the Wikipedia page. Default: "en"
- `top_k_results` (int): The number of top results to return. Default: 5
- `doc_content_chars_max` (int): The maximum number of characters of the content of the Wikipedia page. Default: 3000
- Other parameters are inherited from the default configuration of AtomicFlow (see AtomicFlow)

*input_interface*:

    - `search_term` (str): The search term to search for.

*output_interface*:

    - `wiki_content` (str): The content of the Wikipedia page.

**Arguments**:

- `\**kwargs`: The keyword arguments passed to the AtomicFlow constructor

<a id="WikiSearchAtomicFlow.WikiSearchAtomicFlow.run"></a>

#### run

```python
def run(input_message: FlowMessage)
```

Runs the WikiSearch Atomic Flow. It's used to execute a Wikipedia search and get page summaries.

**Arguments**:

- `input_message` (`FlowMessage`): The input message

<a id="wikipediaAPI"></a>

# wikipediaAPI

Util that calls Wikipedia. references: https://github.com/hwchase17/langchain/blob/9b615022e2b6a3591347ad77a3e21aad6cf24c49/docs/extras/modules/agents/tools/integrations/wikipedia.ipynb#L36

<a id="wikipediaAPI.WikipediaAPIWrapper"></a>

## WikipediaAPIWrapper Objects

```python
class WikipediaAPIWrapper(BaseModel)
```

Wrapper around WikipediaAPI.

To use, you should have the ``wikipedia`` python package installed.
This wrapper will use the Wikipedia API to conduct searches and
fetch page summaries. By default, it will return the page summaries
of the top-k results.
It limits the Document content by doc_content_chars_max.

**Arguments**:

- `top_k_results` (`int`): The number of results to return.
- `lang` (`str`): The language to use for the Wikipedia API.
- `doc_content_chars_max` (`int`): The maximum number of characters in the Document content.

<a id="wikipediaAPI.WikipediaAPIWrapper.validate_environment"></a>

#### validate\_environment

```python
@root_validator(pre=True)
def validate_environment(cls, values: Dict) -> Dict
```

Validate that the python package exists in environment.

**Arguments**:

- `values` (`Dict`): The values to validate.

**Raises**:

- `ImportError`: If the package is not installed.

**Returns**:

`Dict`: The validated values.

<a id="wikipediaAPI.WikipediaAPIWrapper.run"></a>

#### run

```python
def run(query: str) -> str
```

Run Wikipedia search and get page summaries.

**Arguments**:

- `query` (`str`): The query to search for.

**Returns**:

`str`: The page summaries.

<a id="wikipediaAPI.WikipediaAPIWrapper.search_page_titles"></a>

#### search\_page\_titles

```python
def search_page_titles(query: str) -> List[str]
```

Run Wikipedia search and get page summaries.

**Arguments**:

- `query` (`str`): The query to search for.

**Returns**:

`List[str]`: The page titles.

<a id="ControllerExecutorFlow"></a>

# ControllerExecutorFlow

<a id="ControllerExecutorFlow.ControllerExecutorFlow"></a>

## ControllerExecutorFlow Objects

```python
class ControllerExecutorFlow(CompositeFlow)
```

This class implements a ControllerExecutorFlow. It's composed of a ControllerAtomicFlow and an ExecutorFlow.

Where typically the ControllerAtomicFlow is uses a LLM to decide which command to call and the ExecutorFlow (branching flow) is used to execute the command.

It contains the following subflows:

- A Controller Atomic Flow: It is a flow that to decides which command to get closer to completing it's task of accomplishing a given goal.
- An Executor Flow: It is a branching flow that uses the executes the command instructed by the ControllerAtomicFlow.

An illustration of the flow is as follows:

       goal -----|-----> ControllerFlow----->|-----> (anwser,status)
                 ^                           |
                 |                           |
                 |                           v
                 |<----- ExecutorFlow <------|

*Configuration Parameters*:

- `name` (str): The name of the flow. Default: "CtrlEx"
- `description` (str): A description of the flow. This description is used to generate the help message of the flow.
Default: "ControllerExecutor (i.e., MRKL, ReAct) interaction implementation with Flows
that approaches the problem solving in two phases: one Flow chooses the next step and another Flow executes it.
This is repeated until the controller Flow concludes on an answer."
- `max_rounds` (int): The maximum number of rounds the flow can run for. 
Default: 30.
- `subflows_config` (Dict[str,Any]): A dictionary of the subflows configurations. Default:
    - `Controller`: The configuration of the Controller Flow. By default, it a ControllerAtomicFlow. Default parameters:
        - `finish` (Dict[str,Any]): The configuration of the finish command. Default parameters:
            - `description` (str): The description of the command.
            Default: "Signal that the objective has been satisfied, and returns the answer to the user."
            - `input_args` (List[str]): The input arguments of the command. Default: ["answer"]
        -  All other parameters are inherited from the default configuration of ControllerAtomicFlow (see ControllerAtomicFlow)
    - `Executor`: The configuration of the Executor Flow. By default, it's a BranchingFlow. There are no default parameters, the flow
    parameter to to be defined is:
        - `subflows_config` (Dict[str,Any]): A dictionary of the configuration of the subflows of the branching flow.
        These subflows are typically also the possible commands of the Controller Flow. Default: []
- `early_exit_key` (str): The key that is used to exit the flow. Default: "EARLY_EXIT"
- `topology` (str): 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 ControllerExecutorFlow.yaml).


*Input Interface*:

- `goal` (str): The goal of the controller. Usually asked by the user/human (e.g. "I want to know the occupation and birth date of Michael Jordan.")

*Output Interface*:

- `answer` (str): The answer of the flow to the query (e.g. "Michael Jordan is a basketball player and business man. He was born on February 17, 1963.")
- `status` (str): The status of the flow. It can be "finished" or "unfinished". If the status is "unfinished", it's usually because the maximum amount of rounds was reached before the model found an answer.

**Arguments**:

- `flow_config`: The configuration of the flow (see Configuration Parameters).
- `subflows`: A list of subflows. Required when instantiating the subflow programmatically (it replaces subflows_config from flow_config).

<a id="ControllerExecutorFlow.ControllerExecutorFlow.generate_reply"></a>

#### generate\_reply

```python
def generate_reply()
```

This method generates the reply of the flow. It's called when the flow is finished.

<a id="ControllerExecutorFlow.ControllerExecutorFlow.get_next_state"></a>

#### get\_next\_state

```python
def get_next_state()
```



<a id="ControllerExecutorFlow.ControllerExecutorFlow.set_up_flow_state"></a>

#### set\_up\_flow\_state

```python
def set_up_flow_state()
```

Sets up the flow state.

<a id="ControllerExecutorFlow.ControllerExecutorFlow.call_controller"></a>

#### call\_controller

```python
def call_controller()
```

Calls the controller: the flow that decides which command to call next.

<a id="ControllerExecutorFlow.ControllerExecutorFlow.call_executor"></a>

#### call\_executor

```python
def call_executor()
```

Calls the flow that executes the command instructed by the ControllerAtomicFlow.

<a id="ControllerExecutorFlow.ControllerExecutorFlow.register_data_to_state"></a>

#### register\_data\_to\_state

```python
def register_data_to_state(input_message)
```

This method registers the input message data to the flow state. It's everytime a new input message is received.

**Arguments**:

- `input_message` (`FlowMessage`): The input message

<a id="ControllerExecutorFlow.ControllerExecutorFlow.run"></a>

#### run

```python
def run(input_message: FlowMessage)
```

Runs the WikiSearch Atomic Flow. It's used to execute a Wikipedia search and get page summaries.

**Arguments**:

- `input_message` (`FlowMessage`): The input message