Skip to content

ResultRecorder

ResultRecorder

An abstract base class for recording experiment results, including configuration, metrics, and model outputs.

This class defines the interface for different result recording implementations, such as saving to a local directory, uploading to wandb, or integrating with MLflow.

Source code in flexeval/core/result_recorder/base.py 
 7
 8
 9
10
11
12
13
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
class ResultRecorder(ABC):
    """
    An abstract base class for recording experiment results, including configuration,
    metrics, and model outputs.

    This class defines the interface for different result recording implementations,
    such as saving to a local directory, uploading to wandb, or integrating with MLflow.
    """

    @abstractmethod
    def record_config(self, config: dict[str, Any], group: str | None = None) -> None:
        """
        Record the configuration parameters of the experiment.

        Args:
            config: A dictionary containing the configuration
                parameters of the evaluation.
            group: An optional group name to organize the configuration.
        """

    @abstractmethod
    def record_metrics(self, metrics: dict[str, Any], group: str | None = None) -> None:
        """
        Record the evaluation metrics of the experiment.

        Args:
            metrics: A dictionary containing the evaluation metrics,
                where keys are metric names and values are the corresponding results.
            group: An optional group name to organize the metrics.
        """

    @abstractmethod
    def record_model_outputs(self, model_outputs: list[dict[str, Any]], group: str | None = None) -> None:
        """
        Record the outputs generated by the model during evaluation.

        Args:
            model_outputs: A list of dictionaries, where each
                dictionary represents a single model output. The structure of these
                dictionaries may vary depending on the specific model and task.
            group: An optional group name to organize the model outputs.
        """

    def is_metrics_saved(self, group: str | None = None) -> bool:
        """
        Check if the evaluation metrics have already been saved.

        Args:
            group: An optional group name to check for saved metrics.
        """
        return False

    def acquire_lock(self, group: str | None = None) -> bool:
        """
        Try to acquire an exclusive lock for the evaluation run of the given group,
        so that multiple processes can share the same output location without running the same evaluation.

        Returns:
            True if the lock is acquired (or the recorder does not support locking),
            False if another process already holds the lock.
        """
        return True

    def release_lock(self, group: str | None = None) -> None:  # noqa: B027
        """
        Release the lock acquired by `acquire_lock`.
        """

record_config abstractmethod 

record_config(
    config: dict[str, Any], group: str | None = None
) -> None

Record the configuration parameters of the experiment.

Parameters:

  • config (dict[str, Any]) –

    A dictionary containing the configuration parameters of the evaluation.

  • group (str | None, default: None ) –

    An optional group name to organize the configuration.

Source code in flexeval/core/result_recorder/base.py 
16
17
18
19
20
21
22
23
24
25
@abstractmethod
def record_config(self, config: dict[str, Any], group: str | None = None) -> None:
    """
    Record the configuration parameters of the experiment.

    Args:
        config: A dictionary containing the configuration
            parameters of the evaluation.
        group: An optional group name to organize the configuration.
    """

record_metrics abstractmethod 

record_metrics(
    metrics: dict[str, Any], group: str | None = None
) -> None

Record the evaluation metrics of the experiment.

Parameters:

  • metrics (dict[str, Any]) –

    A dictionary containing the evaluation metrics, where keys are metric names and values are the corresponding results.

  • group (str | None, default: None ) –

    An optional group name to organize the metrics.

Source code in flexeval/core/result_recorder/base.py 
27
28
29
30
31
32
33
34
35
36
@abstractmethod
def record_metrics(self, metrics: dict[str, Any], group: str | None = None) -> None:
    """
    Record the evaluation metrics of the experiment.

    Args:
        metrics: A dictionary containing the evaluation metrics,
            where keys are metric names and values are the corresponding results.
        group: An optional group name to organize the metrics.
    """

record_model_outputs abstractmethod 

record_model_outputs(
    model_outputs: list[dict[str, Any]],
    group: str | None = None,
) -> None

Record the outputs generated by the model during evaluation.

Parameters:

  • model_outputs (list[dict[str, Any]]) –

    A list of dictionaries, where each dictionary represents a single model output. The structure of these dictionaries may vary depending on the specific model and task.

  • group (str | None, default: None ) –

    An optional group name to organize the model outputs.

Source code in flexeval/core/result_recorder/base.py 
38
39
40
41
42
43
44
45
46
47
48
@abstractmethod
def record_model_outputs(self, model_outputs: list[dict[str, Any]], group: str | None = None) -> None:
    """
    Record the outputs generated by the model during evaluation.

    Args:
        model_outputs: A list of dictionaries, where each
            dictionary represents a single model output. The structure of these
            dictionaries may vary depending on the specific model and task.
        group: An optional group name to organize the model outputs.
    """

is_metrics_saved 

is_metrics_saved(group: str | None = None) -> bool

Check if the evaluation metrics have already been saved.

Parameters:

  • group (str | None, default: None ) –

    An optional group name to check for saved metrics.

Source code in flexeval/core/result_recorder/base.py 
50
51
52
53
54
55
56
57
def is_metrics_saved(self, group: str | None = None) -> bool:
    """
    Check if the evaluation metrics have already been saved.

    Args:
        group: An optional group name to check for saved metrics.
    """
    return False

acquire_lock 

acquire_lock(group: str | None = None) -> bool

Try to acquire an exclusive lock for the evaluation run of the given group, so that multiple processes can share the same output location without running the same evaluation.

Returns:

  • bool –

    True if the lock is acquired (or the recorder does not support locking),

  • bool –

    False if another process already holds the lock.

Source code in flexeval/core/result_recorder/base.py 
59
60
61
62
63
64
65
66
67
68
def acquire_lock(self, group: str | None = None) -> bool:
    """
    Try to acquire an exclusive lock for the evaluation run of the given group,
    so that multiple processes can share the same output location without running the same evaluation.

    Returns:
        True if the lock is acquired (or the recorder does not support locking),
        False if another process already holds the lock.
    """
    return True

release_lock 

release_lock(group: str | None = None) -> None

Release the lock acquired by acquire_lock.

Source code in flexeval/core/result_recorder/base.py 
70
71
72
73
def release_lock(self, group: str | None = None) -> None:  # noqa: B027
    """
    Release the lock acquired by `acquire_lock`.
    """

LocalRecorder

A class to record the results in JSON format.

Parameters:

  • output_dir (str) –

    The directory to save the results.

  • force (bool, default: False ) –

    Overwrite the existing results.

  • lock_timeout (float, default: 300.0 ) –

    A lock file (.lock) that has not been updated for this many seconds is regarded as stale (left by a dead process) and can be taken over.

  • heartbeat_interval (float | None, default: None ) –

    The interval in seconds to update the lock file while it is held. Defaults to lock_timeout / 20.

Source code in flexeval/core/result_recorder/local_recorder.py 
 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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
class LocalRecorder(ResultRecorder):
    """
    A class to record the results in JSON format.

    Args:
        output_dir: The directory to save the results.
        force: Overwrite the existing results.
        lock_timeout: A lock file (`.lock`) that has not been updated for this many seconds
            is regarded as stale (left by a dead process) and can be taken over.
        heartbeat_interval: The interval in seconds to update the lock file while it is held.
            Defaults to `lock_timeout / 20`.
    """

    def __init__(
        self,
        output_dir: str,
        force: bool = False,
        lock_timeout: float = 300.0,
        heartbeat_interval: float | None = None,
    ) -> None:
        self.output_dir = Path(output_dir)
        self.force = force
        self.lock_timeout = lock_timeout
        self.heartbeat_interval = heartbeat_interval if heartbeat_interval is not None else lock_timeout / 20
        self._heartbeats: dict[Path, tuple[threading.Thread, threading.Event]] = {}

    @staticmethod
    def _check_output_dir_exists(output_dir: str | PathLike[str], checked_files: list[str]) -> None:
        output_dir = Path(output_dir)
        for file_name in checked_files:
            if (output_dir / file_name).exists():
                msg = (
                    f"`{output_dir / file_name}` already exists. If you want to overwrite it, "
                    f"please specify `--force true` from CLI or `force=True` when initializing the recorder."
                )
                raise FileExistsError(msg)

    def record_config(self, config: dict[str, Any], group: str | None = None) -> None:
        output_dir = self.output_dir
        if group is not None:
            output_dir = self.output_dir / group

        if not self.force:
            self._check_output_dir_exists(output_dir, [CONFIG_FILE_NAME])

        save_json(config, output_dir / CONFIG_FILE_NAME)
        logger.info(f"Saved the config to {output_dir / CONFIG_FILE_NAME}")

    def record_metrics(self, metrics: dict[str, Any], group: str | None = None) -> None:
        output_dir = self.output_dir
        if group is not None:
            output_dir = self.output_dir / group

        if not self.force:
            self._check_output_dir_exists(output_dir, [METRIC_FILE_NAME])

        save_json(metrics, output_dir / METRIC_FILE_NAME)
        logger.info(f"Saved the metrics to {output_dir / METRIC_FILE_NAME}")

    def record_model_outputs(self, model_outputs: list[dict[str, Any]], group: str | None = None) -> None:
        output_dir = self.output_dir
        if group is not None:
            output_dir = output_dir / group

        if not self.force:
            self._check_output_dir_exists(output_dir, [OUTPUTS_FILE_NAME])

        save_jsonl(model_outputs, output_dir / OUTPUTS_FILE_NAME)
        logger.info(f"Saved the outputs to {output_dir / OUTPUTS_FILE_NAME}")

    def is_metrics_saved(self, group: str | None = None) -> bool:
        output_dir = self.output_dir
        if group is not None:
            output_dir = self.output_dir / group
        return (output_dir / METRIC_FILE_NAME).exists()

    def _lock_path(self, group: str | None) -> Path:
        output_dir = self.output_dir
        if group is not None:
            output_dir = self.output_dir / group
        return output_dir / LOCK_FILE_NAME

    @staticmethod
    def _create_lock_file(lock_path: Path) -> bool:
        try:
            fd = os.open(lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
        except FileExistsError:
            return False
        os.close(fd)
        return True

    def _is_stale(self, path: Path) -> bool:
        try:
            return time.time() - path.stat().st_mtime > self.lock_timeout
        except FileNotFoundError:
            return True

    def _start_heartbeat(self, lock_path: Path) -> None:
        stop_event = threading.Event()

        def _update_mtime() -> None:
            while not stop_event.wait(self.heartbeat_interval):
                try:
                    os.utime(lock_path)
                except OSError:  # noqa: PERF203
                    logger.warning(f"Failed to update the lock file: {lock_path}")

        thread = threading.Thread(target=_update_mtime, daemon=True)
        thread.start()
        self._heartbeats[lock_path] = (thread, stop_event)

    def _take_over_stale_lock(self, lock_path: Path) -> bool:
        guard_path = lock_path.with_name(LOCK_STEAL_GUARD_FILE_NAME)
        if not self._create_lock_file(guard_path):
            if self._is_stale(guard_path):
                guard_path.unlink(missing_ok=True)
            return False
        try:
            if not self._is_stale(lock_path):
                return False
            logger.warning(f"Taking over the stale lock: {lock_path}")
            lock_path.unlink(missing_ok=True)
            return self._create_lock_file(lock_path)
        finally:
            guard_path.unlink(missing_ok=True)

    def acquire_lock(self, group: str | None = None) -> bool:
        lock_path = self._lock_path(group)
        lock_path.parent.mkdir(parents=True, exist_ok=True)
        if not self._create_lock_file(lock_path) and not (
            self._is_stale(lock_path) and self._take_over_stale_lock(lock_path)
        ):
            return False
        self._start_heartbeat(lock_path)
        return True

    def release_lock(self, group: str | None = None) -> None:
        lock_path = self._lock_path(group)
        heartbeat = self._heartbeats.pop(lock_path, None)
        if heartbeat is not None:
            thread, stop_event = heartbeat
            stop_event.set()
            thread.join()
        lock_path.unlink(missing_ok=True)

output_dir instance-attribute 

output_dir = Path(output_dir)

force instance-attribute 

force = force

lock_timeout instance-attribute 

lock_timeout = lock_timeout

heartbeat_interval instance-attribute 

heartbeat_interval = (
    heartbeat_interval
    if heartbeat_interval is not None
    else lock_timeout / 20
)

__init__ 

__init__(
    output_dir: str,
    force: bool = False,
    lock_timeout: float = 300.0,
    heartbeat_interval: float | None = None,
) -> None
Source code in flexeval/core/result_recorder/local_recorder.py 
59
60
61
62
63
64
65
66
67
68
69
70
def __init__(
    self,
    output_dir: str,
    force: bool = False,
    lock_timeout: float = 300.0,
    heartbeat_interval: float | None = None,
) -> None:
    self.output_dir = Path(output_dir)
    self.force = force
    self.lock_timeout = lock_timeout
    self.heartbeat_interval = heartbeat_interval if heartbeat_interval is not None else lock_timeout / 20
    self._heartbeats: dict[Path, tuple[threading.Thread, threading.Event]] = {}

record_config 

record_config(
    config: dict[str, Any], group: str | None = None
) -> None
Source code in flexeval/core/result_recorder/local_recorder.py 
83
84
85
86
87
88
89
90
91
92
def record_config(self, config: dict[str, Any], group: str | None = None) -> None:
    output_dir = self.output_dir
    if group is not None:
        output_dir = self.output_dir / group

    if not self.force:
        self._check_output_dir_exists(output_dir, [CONFIG_FILE_NAME])

    save_json(config, output_dir / CONFIG_FILE_NAME)
    logger.info(f"Saved the config to {output_dir / CONFIG_FILE_NAME}")

record_metrics 

record_metrics(
    metrics: dict[str, Any], group: str | None = None
) -> None
Source code in flexeval/core/result_recorder/local_recorder.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
def record_metrics(self, metrics: dict[str, Any], group: str | None = None) -> None:
    output_dir = self.output_dir
    if group is not None:
        output_dir = self.output_dir / group

    if not self.force:
        self._check_output_dir_exists(output_dir, [METRIC_FILE_NAME])

    save_json(metrics, output_dir / METRIC_FILE_NAME)
    logger.info(f"Saved the metrics to {output_dir / METRIC_FILE_NAME}")

record_model_outputs 

record_model_outputs(
    model_outputs: list[dict[str, Any]],
    group: str | None = None,
) -> None
Source code in flexeval/core/result_recorder/local_recorder.py 
105
106
107
108
109
110
111
112
113
114
def record_model_outputs(self, model_outputs: list[dict[str, Any]], group: str | None = None) -> None:
    output_dir = self.output_dir
    if group is not None:
        output_dir = output_dir / group

    if not self.force:
        self._check_output_dir_exists(output_dir, [OUTPUTS_FILE_NAME])

    save_jsonl(model_outputs, output_dir / OUTPUTS_FILE_NAME)
    logger.info(f"Saved the outputs to {output_dir / OUTPUTS_FILE_NAME}")

is_metrics_saved 

is_metrics_saved(group: str | None = None) -> bool
Source code in flexeval/core/result_recorder/local_recorder.py 
116
117
118
119
120
def is_metrics_saved(self, group: str | None = None) -> bool:
    output_dir = self.output_dir
    if group is not None:
        output_dir = self.output_dir / group
    return (output_dir / METRIC_FILE_NAME).exists()

acquire_lock 

acquire_lock(group: str | None = None) -> bool
Source code in flexeval/core/result_recorder/local_recorder.py 
172
173
174
175
176
177
178
179
180
def acquire_lock(self, group: str | None = None) -> bool:
    lock_path = self._lock_path(group)
    lock_path.parent.mkdir(parents=True, exist_ok=True)
    if not self._create_lock_file(lock_path) and not (
        self._is_stale(lock_path) and self._take_over_stale_lock(lock_path)
    ):
        return False
    self._start_heartbeat(lock_path)
    return True

release_lock 

release_lock(group: str | None = None) -> None
Source code in flexeval/core/result_recorder/local_recorder.py 
182
183
184
185
186
187
188
189
def release_lock(self, group: str | None = None) -> None:
    lock_path = self._lock_path(group)
    heartbeat = self._heartbeats.pop(lock_path, None)
    if heartbeat is not None:
        thread, stop_event = heartbeat
        stop_event.set()
        thread.join()
    lock_path.unlink(missing_ok=True)

WandBRecorder

A class to record the results to Weights & Biases.

Parameters:

  • init_kwargs (dict[str, Any] | None, default: None ) –

    The arguments for the wandb.init function. Please refer to the official documentation for the details.

Source code in flexeval/core/result_recorder/wandb_recorder.py 
 8
 9
10
11
12
13
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
class WandBRecorder(ResultRecorder):
    """
    A class to record the results to Weights & Biases.

    Args:
        init_kwargs: The arguments for the `wandb.init` function.
            Please refer to [the official documentation](https://docs.wandb.ai/ref/python/init) for the details.
    """

    def __init__(
        self,
        init_kwargs: dict[str, Any] | None = None,
    ) -> None:
        import wandb

        self._wandb = wandb
        init_kwargs = init_kwargs or {}
        self._wandb.init(**init_kwargs)

    def record_config(self, config: dict[str, Any], group: str | None = None) -> None:
        if group:
            self._wandb.config.update({group: config})
        else:
            self._wandb.config.update(config)

    def record_metrics(self, metrics: dict[str, Any], group: str | None = None) -> None:
        if group:
            self._wandb.summary.update({group: metrics})
        else:
            self._wandb.summary.update(metrics)

    def record_model_outputs(self, model_outputs: list[dict[str, Any]], group: str | None = None) -> None:
        table = self._wandb.Table(columns=list(model_outputs[0].keys()))

        for output in model_outputs:
            table.add_data(*output.values())

        table_name = "model_outputs" if group is None else f"{group}/model_outputs"
        self._wandb.log({table_name: table})

    def __del__(self) -> None:
        self._wandb.finish()

__init__ 

__init__(init_kwargs: dict[str, Any] | None = None) -> None
Source code in flexeval/core/result_recorder/wandb_recorder.py 
17
18
19
20
21
22
23
24
25
def __init__(
    self,
    init_kwargs: dict[str, Any] | None = None,
) -> None:
    import wandb

    self._wandb = wandb
    init_kwargs = init_kwargs or {}
    self._wandb.init(**init_kwargs)

record_config 

record_config(
    config: dict[str, Any], group: str | None = None
) -> None
Source code in flexeval/core/result_recorder/wandb_recorder.py 
27
28
29
30
31
def record_config(self, config: dict[str, Any], group: str | None = None) -> None:
    if group:
        self._wandb.config.update({group: config})
    else:
        self._wandb.config.update(config)

record_metrics 

record_metrics(
    metrics: dict[str, Any], group: str | None = None
) -> None
Source code in flexeval/core/result_recorder/wandb_recorder.py 
33
34
35
36
37
def record_metrics(self, metrics: dict[str, Any], group: str | None = None) -> None:
    if group:
        self._wandb.summary.update({group: metrics})
    else:
        self._wandb.summary.update(metrics)

record_model_outputs 

record_model_outputs(
    model_outputs: list[dict[str, Any]],
    group: str | None = None,
) -> None
Source code in flexeval/core/result_recorder/wandb_recorder.py 
39
40
41
42
43
44
45
46
def record_model_outputs(self, model_outputs: list[dict[str, Any]], group: str | None = None) -> None:
    table = self._wandb.Table(columns=list(model_outputs[0].keys()))

    for output in model_outputs:
        table.add_data(*output.values())

    table_name = "model_outputs" if group is None else f"{group}/model_outputs"
    self._wandb.log({table_name: table})

__del__ 

__del__() -> None
Source code in flexeval/core/result_recorder/wandb_recorder.py 
48
49
def __del__(self) -> None:
    self._wandb.finish()