aiflows/ReplanningFlowModule

Table of Contents

• Structure of ReplanningFlow
• run_replanning
• ReplanningAskUserFlow
  • ReplanningAskUserFlow
    • run
• NewPlanGenFlow
  • NewPlanGenFlow
    • __init__
• ReplanningFlow
  • ReplanningFlow
    • detect_finish_or_continue
• __init__

Structure of ReplanningFlow

goal (info on the old plan), plan (old plan), plan_file_location
                |
                v
        +---------------+
        |  Controller   | --------<<<<-----------+
        +---------------+                        |
                |                                |
                | (command, command args)        |
                |                                |
                v                                |
        +------------------+                     |
        |   Executor       |  Each branch is an  |
        | (Tree Structure) |  executor           |
        +------------------+                     |
                |                                ^
                | (summary)                      |
                |                                |
                v                                |             
                |                                |
                +-> goes back to the Controller>-+

Structure of the Executors:

         +-------------------+
         |   Branching       |
         |    Executor       |
         +-------------------+
        /                    \
       /                      \
      /                        \
     /                          \
write_plan                    ask_user

About the branches:

• ask_user: Ask user for info / confirmation, etc.
• write_plan: Generates plan (user edit is allowed) and fetches user feedback.
  • The PlanGenerator of write_plan is replaced with NewPlanGenFlow to re-plan instead of write plan.

How it works: Controller calls write_plan until user is satisfied in the feedback, finish.

run_replanning

ReplanningAskUserFlow

ReplanningAskUserFlow Objects

class ReplanningAskUserFlow(HumanStandardInputFlow)

Refer to: https://huggingface.co/aiflows/ExtendLibraryFlowModule/blob/main/ExtLibAskUserFlow.py

Input Interface:

• question

Output Interface:

• feedback
• new_plan

Configuration Parameters:

• request_multi_line_input_flag: if True, the user can input multiple lines of text. Default: False
• end_of_input_string: the string that the user can input to indicate that the input is finished. Default: EOI
• query_message_prompt_template: the template of the message that is sent to the user to ask for input.

run

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

Run the flow module.

Arguments:

• input_data: the input data

Returns:

the output data

NewPlanGenFlow

NewPlanGenFlow Objects

class NewPlanGenFlow(PlanGeneratorAtomicFlow)

This flow module is responsible for generating a new plan based on the previous plan and the given requirements. This flow is inherited from PlanGeneratorAtomicFlow. (https://huggingface.co/aiflows/PlanGeneratorFlowModule)

Input Interface Non Initialized:

• goal
• old_plan

Input Interface Initialized:

• goal
• old_plan

Output Interface:

• new_plan

Configuration Parameters:

• system_message_prompt_template: The template of the system message.
• init_human_message_prompt_template: The template of the init user message.
• human_message_prompt_template: The template of the user message.

__init__

def __init__(**kwargs)

Initialization of the flow module.

Arguments:

• kwargs: The configuration parameters of the flow module.

ReplanningFlow

ReplanningFlow Objects

class ReplanningFlow(PlanWriterFlow)

This flow inherits from PlanWriterFlow. By changing prompts and injecting proper information to the controller and the PlanGenerator, we are able to achieve the replanning.

Input Interface:

• goal (str): information on the old plan (e.g. what is wrong)
• plan (str): the old plan
• plan_file_location (str): the location of the old plan file

Output Interface:

• plan (str): the new plan
• status: "finished" or "unfinished"
• summary (str): summary of the flow, will be written to the log file of the caller flow.
• result (str): result of the flow, will be passed to the controller of the caller flow.

detect_finish_or_continue

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

This function is called when the replanning flow receives a message from the PlanGenerator.

It checks the command in the message and decides whether to finish the flow or continue the flow.

Arguments:

• output_payload: the message received from the PlanGenerator
• src_flow: the PlanGenerator

Returns:

the message to be sent to the controller of the caller flow

__init__