Stopping Tasks Early

When working with complex AI workflows, it’s often necessary to repeatedly invoke an agent until a task is completed. This pattern allows for flexible, iterative approaches to task completion, especially useful when the number of steps required is unknown or variable.

Here’s how you can implement this pattern in ControlFlow:

import controlflow as cf

@cf.flow
def iterative_task_flow():
    task = cf.Task("Generate a comprehensive report on AI trends")
    
    while task.is_incomplete():
        task.run_once()
        
        # Optionally, you can add logic here to modify the task,
        # create new tasks, or make decisions based on other results
        
        if some_condition:
            break  # Allows for early termination if needed

    return task.result

result = iterative_task_flow()

This pattern is particularly useful when:

• The task may require multiple attempts or refinements
• You want to implement custom logic between iterations
• You need to dynamically adjust the workflow based on other results

By using a while loop with task.is_incomplete(), you ensure that the agent continues working until the task is either marked as complete or fails. The run_once() method allows for step-by-step execution, giving you control over each iteration of the agentic loop.