metadata

license: mit

ControllerAtomicFlow
- Command
- ControllerAtomicFlow
  - instantiate_from_config
  - run
__init__
WikiSearchAtomicFlow
- WikiSearchAtomicFlow
  - run
wikipediaAPI
- WikipediaAPIWrapper
ControllerExecutorFlow
- ControllerExecutorFlow
  - detect_finish_or_continue

ControllerAtomicFlow

Command Objects

@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.

ControllerAtomicFlow Objects

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.

instantiate_from_config

@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.

run

def run(input_data: Dict[str, Any]) -> Dict[str, Any]

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_data (Dict[str, Any]): The input data of the flow.

Returns:

Dict[str, Any]: The output data of the flow (thought, reasoning, criticism, command, command_args)

init

WikiSearchAtomicFlow

WikiSearchAtomicFlow Objects

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

run

def run(input_data: Dict[str, Any]) -> Dict[str, Any]

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

Arguments:

input_data (Dict[str, Any]): The input data dictionary

Returns:

Dict[str, Any]: The output data dictionary

wikipediaAPI

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

WikipediaAPIWrapper Objects

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.

validate_environment

@root_validator()
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.

run

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.

search_page_titles

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.

ControllerExecutorFlow

ControllerExecutorFlow Objects

class ControllerExecutorFlow(CircularFlow)

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).

detect_finish_or_continue

@CircularFlow.output_msg_payload_processor
def detect_finish_or_continue(
        output_payload: Dict[str, Any],
        src_flow: ControllerAtomicFlow) -> Dict[str, Any]

This method is called when the ExecutorAtomicFlow receives a message from the ControllerAtomicFlow. It checks if the flow should finish or continue.

Arguments:

output_payload (Dict[str, Any]): The output payload of the ControllerAtomicFlow.
src_flow (ControllerAtomicFlow): The ControllerAtomicFlow.

Returns:

The output payload of the ControllerAtomicFlow.

aiflows
/

ControllerExecutorFlowModule

Table of Contents

ControllerAtomicFlow

Command Objects

ControllerAtomicFlow Objects

instantiate_from_config

run

init

WikiSearchAtomicFlow

WikiSearchAtomicFlow Objects

run

wikipediaAPI

WikipediaAPIWrapper Objects

validate_environment

run

search_page_titles

ControllerExecutorFlow

ControllerExecutorFlow Objects

detect_finish_or_continue

Table of Contents

ControllerAtomicFlow

Command Objects

ControllerAtomicFlow Objects

instantiate_from_config

run

__init__

WikiSearchAtomicFlow

WikiSearchAtomicFlow Objects

run

wikipediaAPI

WikipediaAPIWrapper Objects

validate_environment

run

search_page_titles

ControllerExecutorFlow

ControllerExecutorFlow Objects

detect_finish_or_continue

init