Skip to content

experiment_pipeline

ExperimentPipeline

A class for the experiment pipeline process.

Parameters:

Name Type Description Default
settings Settings

Settings for the pipeline which includes the data folder locations, the maximum number of queries to send per minute, the maximum number of attempts when retrying, and whether to run the experiment in parallel

 required
Source code in src/prompto/experiment_pipeline.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
class ExperimentPipeline:
    """
    A class for the experiment pipeline process.

    Parameters
    ----------
    settings : Settings
        Settings for the pipeline which includes the data folder locations,
        the maximum number of queries to send per minute, the maximum number
        of attempts when retrying, and whether to run the experiment in parallel
    """

    def __init__(
        self,
        settings: Settings,
    ):
        self.settings: Settings = settings
        self.average_per_query_processing_times: list[float] = []
        self.overall_avg_proc_times: float = 0.0
        self.experiment_files: list[str] = []

    def run(self) -> None:
        """
        Run the pipeline process of continually by checking for
        new experiment files and running the experiments sequentially
        in the order that the files were created.

        The process will continue to run until the program is stopped.
        """
        while True:
            # obtain experiment files sorted by creation/change time
            self.update_experiment_files()

            if len(self.experiment_files) != 0:
                # obtain the next experiment to process
                next_experiment = Experiment(
                    file_name=self.experiment_files[0], settings=self.settings
                )

                # create the output folder for the experiment
                create_folder(next_experiment.output_folder)

                # log the estimated time of completion of the next experiment
                self.log_estimate(experiment=next_experiment)

                # process the next experiment
                _, avg_query_processing_time = asyncio.run(next_experiment.process())

                # keep track of the average processing time per query for the experiment
                self.average_per_query_processing_times.append(
                    avg_query_processing_time
                )

                # update the overall average processing time per query
                self.overall_avg_proc_times = sum(
                    self.average_per_query_processing_times
                ) / len(self.average_per_query_processing_times)

                # log the progress of the queue of experiments
                self.log_progress(experiment=next_experiment)

    def update_experiment_files(self) -> None:
        """
        Function to update the list of experiment files by sorting
        the files by creation/change time (using `os.path.getctime`).
        """
        self.experiment_files = sort_jsonl_files_by_creation_time(
            input_folder=self.settings.input_folder
        )

    def log_estimate(
        self,
        experiment: Experiment,
    ) -> None:
        """
        Function to log the estimated time of completion of the next experiment.

        Parameters
        ----------
        experiment : Experiment
            The experiment that is being processed
        """
        now = datetime.now()
        if self.overall_avg_proc_times == 0:
            estimated_completion_time = "[unknown]"
            estimated_completion = "[unknown]"
        else:
            estimated_completion_time = round(
                self.overall_avg_proc_times * experiment.number_queries, 3
            )
            estimated_completion = (
                now + timedelta(seconds=estimated_completion_time)
            ).strftime("%d-%m-%Y, %H:%M")

        # log the estimated time of completion of the next experiment
        log_message = (
            f"Next experiment: {experiment}, "
            f"Number of queries: {experiment.number_queries}, "
            f"Estimated completion time: {estimated_completion_time}, "
            f"Estimated completion by: {estimated_completion}"
        )
        write_log_message(log_file=experiment.log_file, log_message=log_message)

    def log_progress(
        self,
        experiment: Experiment,
    ) -> None:
        """
        Function to log the progress of the queue of experiments.

        Parameters
        ----------
        experiment : Experiment
            The experiment that was just processed
        """
        # log completion of experiment
        logging.info(f"Completed experiment: {experiment}!")
        logging.info(
            f"- Overall average time per query: {round(self.overall_avg_proc_times, 3)} seconds"
        )

        # log remaining of experiments
        self.update_experiment_files()
        logging.info(f"- Remaining number of experiments: {len(self.experiment_files)}")
        logging.info(f"- Remaining experiments: {self.experiment_files}")

log_estimate 

log_estimate(experiment: Experiment) -> None

Function to log the estimated time of completion of the next experiment.

Parameters:

Name Type Description Default
experiment Experiment

The experiment that is being processed

 required
Source code in src/prompto/experiment_pipeline.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def log_estimate(
    self,
    experiment: Experiment,
) -> None:
    """
    Function to log the estimated time of completion of the next experiment.

    Parameters
    ----------
    experiment : Experiment
        The experiment that is being processed
    """
    now = datetime.now()
    if self.overall_avg_proc_times == 0:
        estimated_completion_time = "[unknown]"
        estimated_completion = "[unknown]"
    else:
        estimated_completion_time = round(
            self.overall_avg_proc_times * experiment.number_queries, 3
        )
        estimated_completion = (
            now + timedelta(seconds=estimated_completion_time)
        ).strftime("%d-%m-%Y, %H:%M")

    # log the estimated time of completion of the next experiment
    log_message = (
        f"Next experiment: {experiment}, "
        f"Number of queries: {experiment.number_queries}, "
        f"Estimated completion time: {estimated_completion_time}, "
        f"Estimated completion by: {estimated_completion}"
    )
    write_log_message(log_file=experiment.log_file, log_message=log_message)

log_progress 

log_progress(experiment: Experiment) -> None

Function to log the progress of the queue of experiments.

Parameters:

Name Type Description Default
experiment Experiment

The experiment that was just processed

 required
Source code in src/prompto/experiment_pipeline.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def log_progress(
    self,
    experiment: Experiment,
) -> None:
    """
    Function to log the progress of the queue of experiments.

    Parameters
    ----------
    experiment : Experiment
        The experiment that was just processed
    """
    # log completion of experiment
    logging.info(f"Completed experiment: {experiment}!")
    logging.info(
        f"- Overall average time per query: {round(self.overall_avg_proc_times, 3)} seconds"
    )

    # log remaining of experiments
    self.update_experiment_files()
    logging.info(f"- Remaining number of experiments: {len(self.experiment_files)}")
    logging.info(f"- Remaining experiments: {self.experiment_files}")

run 

run() -> None

Run the pipeline process of continually by checking for new experiment files and running the experiments sequentially in the order that the files were created.

The process will continue to run until the program is stopped.

Source code in src/prompto/experiment_pipeline.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def run(self) -> None:
    """
    Run the pipeline process of continually by checking for
    new experiment files and running the experiments sequentially
    in the order that the files were created.

    The process will continue to run until the program is stopped.
    """
    while True:
        # obtain experiment files sorted by creation/change time
        self.update_experiment_files()

        if len(self.experiment_files) != 0:
            # obtain the next experiment to process
            next_experiment = Experiment(
                file_name=self.experiment_files[0], settings=self.settings
            )

            # create the output folder for the experiment
            create_folder(next_experiment.output_folder)

            # log the estimated time of completion of the next experiment
            self.log_estimate(experiment=next_experiment)

            # process the next experiment
            _, avg_query_processing_time = asyncio.run(next_experiment.process())

            # keep track of the average processing time per query for the experiment
            self.average_per_query_processing_times.append(
                avg_query_processing_time
            )

            # update the overall average processing time per query
            self.overall_avg_proc_times = sum(
                self.average_per_query_processing_times
            ) / len(self.average_per_query_processing_times)

            # log the progress of the queue of experiments
            self.log_progress(experiment=next_experiment)

update_experiment_files 

update_experiment_files() -> None

Function to update the list of experiment files by sorting the files by creation/change time (using os.path.getctime).

Source code in src/prompto/experiment_pipeline.py 
75
76
77
78
79
80
81
82
def update_experiment_files(self) -> None:
    """
    Function to update the list of experiment files by sorting
    the files by creation/change time (using `os.path.getctime`).
    """
    self.experiment_files = sort_jsonl_files_by_creation_time(
        input_folder=self.settings.input_folder
    )