Table of Contents
run
A simple script to run a Flow that can be used for development and debugging.
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: Truecommands(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_manualfunction 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 onecriticism(str): The criticism of the controller of it's thinking processcommand(str): The command to the executor chooses to callcommand_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_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.
__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: 5doc_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_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
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(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.
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(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).
generate_reply
def generate_reply()
This method generates the reply of the flow. It's called when the flow is finished.
get_next_state
def get_next_state()
set_up_flow_state
def set_up_flow_state()
Sets up the flow state.
call_controller
def call_controller()
Calls the controller: the flow that decides which command to call next.
call_executor
def call_executor()
Calls the flow that executes the command instructed by the ControllerAtomicFlow.
register_data_to_state
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
run
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