allegroai / clearml

ClearML - Auto-Magical CI/CD to streamline your AI workload. Experiment Management, Data Management, Pipeline, Orchestration, Scheduling & Serving in one MLOps/LLMOps solution
https://clear.ml/docs
Apache License 2.0
5.43k stars 643 forks source link

Add option to create pipeline step as draft task #1226

Closed CharlesFrankum closed 3 months ago

CharlesFrankum commented 4 months ago

Patch Description

Added an option to allow the end user to create a pipeline step as a draft instead of it being automatically enqueued. This enables us to create pipelines where the parameters can be edited by a user before execution i.e. selecting the models to run in a testing step which can only be known once the training step has completed.

Testing Instructions

from clearml import PipelineController

def func(user_input):
    print('This step has been executed!')
    return user_input

if __name__ == '__main__':
    pipe = PipelineController(
        name='development',
        project='testing',
        version='0.1',
    )
    pipe.add_function_step(
        name='run_immediately',
        function=func,
        function_kwargs={'user_input': 'Hello World!'},
        function_return=['user_input'],
    )
    pipe.add_function_step(
        name='draft',
        function=func,
        function_kwargs={'user_input': 'Hello World!'},
        function_return=['user_input'],
        parents=['run_immediately'],
        draft=True,
    )
    pipe.add_function_step(
        name='wait_for_draft',
        function=func,
        function_kwargs={'user_input': 'Hello World!'},
        function_return=['user_input'],
        parents=['draft'],
    )
    pipe.start()
from clearml import PipelineDecorator

@PipelineDecorator.component(
    name='run_immediately',
    return_values=['user_input'],
)
def step_1(user_input='Hello World!'):
    print('This step has been executed!')
    return user_input

@PipelineDecorator.component(
    name='draft',
    return_values=['user_input'],
    draft=True,
    parents=['run_immediately'],
)
def step_2(user_input='Hello World!'):
    print('This step has been executed!')
    return user_input

@PipelineDecorator.component(
    name='wait_for_draft',
    return_values=['user_input'],
    parents=['draft'],
)
def step_3(user_input='Hello World!'):
    print('This step has been executed!')
    return user_input

@PipelineDecorator.pipeline(
    name='development',
    project='testing',
    version='0.1',
)
def main():
    step_1()
    step_2()
    step_3()

if __name__ == '__main__':
    main()