New RunLLM response generated: >>>*# Question* We have a workflow that is composed (for now) of 2 parts - generic one and a custom one. I want to have a parameter to the top-level workflow that will determine which workflow is running after the generic part. Trying to put it into code:

@dynamic
def do_next_step(next_step: str, generic_output: ...):
  # ...somehow get the reference launch-plan
  ref_lp = get_ref_lp(next_step)
  ref_lp()

@workflow
def primary(generic_args..., next_step: str):
  generic_output = generic_task(generic_args...)
  do_next_step(next_step, generic_output)

I managed to do something like that:

@dynamic
def do_next(next_step_module: str, next_step_func: str) -> None:
    next_step_mod = importlib.import_module(next_step_module)
    next_step = getattr(next_step_mod, next_step_func)
    next_step()

where the module+func are ref plan, e.g.:

@reference_launch_plan(
    project="flytesnacks",
    domain="development",
    name="data_types_and_io.file.normalize_csv_file",
    version="-ism2z-QiRgPisO-KY_YTA"
)
def normalize_csv_file(
        csv_url: FlyteFile,
        column_names: List[str],
        columns_to_normalize: List[str],
        output_location: str,
) -> FlyteFile:
    ...

That worked but means that I have to maintain launch plans for each sync next step. Essentially, there's no reason I shouldn't be able to get the latest launch plan in runtime using Flyte client which is part of the current context. But I didn't find a way to convert the Flyte API object to something a dynamic workflow can trigger. Another thing I tried is figuring out how to trigger a workflow from within a task. I can then "wait" on the workflow to finish. That's less ideal but also an option. Lastly, I figured we can use a different approach, and generate the workflow dynamically using imperative workflow and register it if needed, then run it. I hope I'm making sense. In more simple terms - I'm trying to determine and execute next steps in a workflow (dynamic or otherwise) according to input arguments. # Answer To determine and execute the next steps in a Flyte workflow according to input arguments, you can use

reference_launch_plan

and Flyte's dynamic workflows. This approach allows you to dynamically select and run a launch plan based on input parameters. Here's a step-by-step breakdown: 1. Define Reference Launch Plans: Predefine the launch plans you want to dynamically select. 2. Dynamic Workflow: Implement a dynamic workflow to select and execute the appropriate launch plan based on input arguments. 3. Primary Workflow: Use the dynamic workflow within your primary workflow to handle the conditional execution logic. Here's an example code implementation: Step 1: Define Reference Launch Plans Define the reference launch plans for the workflows you want to dynamically select.

from flytekit import reference_launch_plan, workflow, dynamic

@reference_launch_plan(